cco/cybertools
3
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

remove obsolete agent and twisted packages

Browse source
This commit is contained in:
Helmut Merz 2011-10-10 12:27:00 +02:00
parent 78a0a83c2f
commit 78441e0444
82 changed files with 1 additions and 5056 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

226
agent/README.txt
View file

@ -1,226 +0,0 @@
================================================
Agents for Job Execution and Communication Tasks
================================================
Agents do some work specified by jobs, the main task being to collect
information objects from the local machine or some external source and
transfer them e.g. to a loops server on the same machine or another.
($Id$)
This package does not depend on Zope but represents a standalone application.
Sub-Packages
============
Top-level
Generic interfaces, ``commponents``: adapter registry,
``tests`` module, this ``README.txt`` file.
base
Base and sample classes.
core
Agent and scheduling implementations.
control
Communication with an external agent control and job database application.
crawl
Scanning/crawling some system, e.g. the database of an email application,
the local file system, or an external document source.
transport
Transfer of information objects to agents on another machine or
to an information management application (e.g. loops).
util
Various utility modules, e.g. a backport of the
``twisted.internet.task.coiterate()`` function from Twisted 2.5 so that
we can use the Twisted version coming with Zope 3.3.1 for
cybertools.agent.
All sub-packages except ``base`` depend on Twisted.
Object Structure and Control Flow
=================================
::
---------- ------------
-------- | |<---1| |
| config |--> | master |<---2| controller |
-------- /| |4--->| |
/ ---------- ------------
/ 1 ^ 2
/ | | \ -----------
/ | | `---->| scheduler |
v v 4 -----------
----- --------- 3
| log |<--| agent |<----------´
----- ---------
(1) Agent specifications control creation and setup of agents
(2) Job specifications control scheduling of jobs
(3) Scheduler triggers job execution by agent
(4) Results are recorded, possibly triggering further actions
Basic Stuff
===========
While the real application is based on the asynchronous communication
framework Twisted there is some basic stuff (mainly interfaces and
base classes with basic, sample, or dummy implementations) that is
independent of Twisted.
The code for this resides in in the ``base`` sub-package.
Master Agent and Configuration
------------------------------
All activity is controlled by the master agent.
The master agent is set up according to the specifications in a
configuration file.
The configuration typically provides only basic informations, e.g. about
the controller(s) and logger(s) to be used; details on jobs and agent
configuration are provided by the controller.
>>> from cybertools.agent.tests import baseDir
>>> import os
>>> configFile = open(os.path.join(baseDir, 'base', 'sample.cfg'))
So we are now ready to create a master agent and configure it by supplying
the path to the configuration file.
>>> from cybertools.agent.main import setup
>>> master = setup(configFile)
Starting agent application...
Using controllers base.sample.
>>> configFile.close()
>>> master.config
controller.names = ['base.sample']
logger.name = 'default'
logger.standard = 30
scheduler.name = 'sample'
Controllers
-----------
Creation of agents and scheduling of jobs is controlled by controller
objects. These are typically associated with a sort of control storage that
provides agent and job specifications and receives the results of job
execution.
>>> master.controllers
[<cybertools.agent.base.control.SampleController object ...>]
We make the contollers provide the specifications via the master agent's
``setup()`` method.
>>> master.setup()
Starting agent application...
Using controllers base.sample.
Other Agents
------------
The above ``setup()`` call has triggered the creation of one child agent -
that is all the sample controller provides.
>>> master.children
{'sample01': <cybertools.agent.base.agent.SampleAgent object ...>}
Let's check a few attributes of the newly created agent.
>>> agent01 = master.children['sample01']
>>> agent01.master is master
True
>>> agent01.config is master.config
True
Job Scheduling and Execution
----------------------------
A scheduler is responsible for triggering the execution of a job at the
appropriate time. The master agent schedules the jobs based upon the
information (job specifications) it gets from the controller. There
is just one scheduler associated with the master agent.
>>> master.scheduler
<cybertools.agent.base.schedule.Scheduler object ...>
We schedule a sample job by calling an internal method of the agent's
controller. In addition to the output of the job execution itself we
also get a note from the controller about the feedback it received
about the outcome of the job execution.
>>> master.controllers[0].enterJob('sample', 'sample01')
Job 00001 on agent sample01 has been executed.
Job 00001 completed; result: None;
Logging
-------
>>> master.logger
<cybertools.agent.base.log.Logger object ...>
>>> agent01.logger is master.logger
True
>>> master.config.logger.standard = 20
>>> master.logger.setup()
>>> master.controllers[0].enterJob('sample', 'sample01')
Job 00002 on agent sample01 has been executed.
2... agent:sample01 job:00002 message:job execution result:OK
Job 00002 completed; result: None;
>>> for r in master.logger.records:
... print r
2... agent:sample01 job:00001 message:job execution result:OK
2... agent:sample01 job:00002 message:job execution result:OK
Using the Twisted-based Scheduler
=================================
By specifying the core scheduler in the agent's configuration this will be
used automatically for scheduling.
In addition, we use another sample controller, now also the twisted-based
from the core package. This will in turn set up a queueable agent from
the core package so that now everything is running under the control of
the twisted reactor.
>>> config = '''
... controller(names=['core.sample'])
... scheduler(name='core')
... logger(name='default', standard=30)
... '''
>>> master = setup(config)
Starting agent application...
Using controllers core.sample.
>>> master.scheduler
<cybertools.agent.core.schedule.Scheduler object ...>
We enter the same job specification as above.
>>> master.controllers[0].enterJob('sample', 'sample01')
Now the job is not executed immediately - we have to hand over control to
the twisted reactor first. The running of the reactor is simulated by
the ``iterate()`` method provided for testing.
>>> from cybertools.agent.tests import tester
>>> tester.iterate()
Job 00001 on agent sample01 has been executed.
Job 00001 completed; result: Done;

3
agent/__init__.py
View file

@ -1,3 +0,0 @@
"""
$Id$
"""

10
agent/agent.cfg
View file

@ -1,10 +0,0 @@
#
# Standard configuration for agent application
#
# $Id$
#
controller(names=['cmdline', 'telnet'])
controller.telnet.port = 5001
scheduler(name='core')
logger(name='default', standard=30)

31
agent/app.py
View file

@ -1,31 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Agent application.
$Id$
"""
from twisted.application import service
from cybertools.agent import main
application = main.application = service.Application('Agent Application')
main.setup()

4
agent/base/__init__.py
View file

@ -1,4 +0,0 @@
"""
$Id$
"""

121
agent/base/agent.py
View file

@ -1,121 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Agent base and sample classes.
$Id$
"""
from zope.interface import implements
from cybertools.agent.common import states
from cybertools.agent.components import agents, controllers, jobs
from cybertools.agent.components import loggers, schedulers
from cybertools.agent.components import servers, clients
from cybertools.agent.interfaces import IAgent
from cybertools.util.config import Configurator
class Agent(object):
implements(IAgent)
name = '???'
master = None
config = None
logger = None
def __init__(self, master):
self.master = master
self.config = master.config
self.logger = master.logger
def send(self, job):
self.execute(job)
def execute(self, job):
pass
def log(self, job, result='OK'):
self.logger.log(dict(message='job execution', job=job.identifier,
agent=self.name, result=result))
class Master(Agent):
name = 'master'
scheduler = None
logger = None
def __init__(self, configuration):
if isinstance(configuration, Configurator):
self.config = configuration
else: # configuration is path to config file
self.config = Configurator()
self.config.load(configuration)
self.master = self
self.controllers = []
self.children = {}
self.servers = []
def setup(self):
config = self.config
self.logger = loggers(self, name=config.logger.name)
print 'Starting agent application...'
for n in config.controller.names:
self.controllers.append(controllers(self, n))
self.scheduler = schedulers(self, name=config.scheduler.name)
for cont in self.controllers:
cont.setup()
print 'Using controllers %s.' % ', '.join(config.controller.names)
for n in config.talk.server.names:
server = servers(self, n)
self.servers.append(server)
server.setup()
def setupAgents(self, controller, agentSpecs):
for spec in agentSpecs:
agent = agents(self, spec.type)
agent.name = spec.name
self.children[spec.name] = agent
def setupJobs(self, controller, jobSpecs):
for spec in jobSpecs:
job = jobs(self.scheduler, spec.type)
job.agent = self.children[spec.agent]
job.identifier = spec.identifier
job.controller = controller
job.params = spec.params
self.scheduler.schedule(job, spec.startTime)
def notify(self, job, result=None, message=''):
if job.state.hasFinished():
job.controller.notify(job.identifier, job.state, result, message)
class SampleAgent(Agent):
def execute(self, job):
job.state = states.running
print 'Job %s on agent %s has been executed.' % (job.identifier, self.name)
self.log(job)
job.state = states.completed
self.master.notify(job)
agents.register(SampleAgent, Master, name='base.sample')

100
agent/base/control.py
View file

@ -1,100 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Base/sample controller implementation.
$Id$
"""
from zope.interface import implements
from cybertools.agent.base.agent import Master
from cybertools.agent.components import controllers
from cybertools.agent.interfaces import IController
class Controller(object):
implements(IController)
def __init__(self, agent):
self.agent = agent
def setup(self):
self.agent.setupAgents(self, self._getAgents())
self.agent.setupJobs(self, self._getCurrentJobs())
def _getAgents(self):
return []
def _getCurrentJobs(self):
return []
def notify(self, identifier, state, result=None, message=''):
pass
class SampleController(Controller):
jobNumber = 0
result = None
agents = (('sample01', 'base.sample'),)
def notify(self, identifier, state, result=None, message=''):
self.result = result
msg = ('Job %s %s; result: %s; %s' %
(identifier, state, result, message))
print msg
def _getAgents(self):
return [AgentSpecification(name, type) for name, type in self.agents]
def createAgent(self, agentType, name):
spec = AgentSpecification(name, agentType)
self.agent.setupAgents(self, [spec])
def enterJob(self, jobType, agent, **kw):
self.jobNumber += 1
spec = JobSpecification(jobType, '%05i' % self.jobNumber, agent=agent, **kw)
self.agent.setupJobs(self, [spec])
controllers.register(SampleController, Master, name='base.sample')
class AgentSpecification(object):
def __init__(self, name, type, **kw):
self.name = name
self.type = type
for k, v in kw.items():
setattr(self, k, v)
class JobSpecification(object):
startTime = None
def __init__(self, type, identifier, **kw):
self.type = type
self.identifier = identifier
self.params = kw.pop('params', {})
for k, v in kw.items():
setattr(self, k, v)

63
agent/base/job.py
View file

@ -1,63 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
The real agent stuff.
$Id$
"""
from zope.interface import implements
from cybertools.agent.base.schedule import Scheduler
from cybertools.agent.common import states
from cybertools.agent.components import jobs
from cybertools.agent.interfaces import IScheduledJob
class Job(object):
implements(IScheduledJob)
identifier = '???'
agent = None
startTime = None
repeat = 0
state = states.initialized
def __init__(self, scheduler):
self.scheduler = scheduler
self.params = {}
self.successors = []
def execute(self):
if self.agent is not None:
self.agent.send(self)
self.state = states.submitted
def reschedule(self, startTime=None):
self.scheduler.schedule(self.copy(), startTime)
def copy(self):
newJob = self.__class__(self.scheduler)
newJob.agent = self.agent
newJob.params = self.params
newJob.repeat = self.repeat
newJob.successors = [s.copy() for s in self.successors]
jobs.register(Job, Scheduler, name='sample')

81
agent/base/log.py
View file

@ -1,81 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Log information management.
$Id$
"""
import logging
import sys
import time
from zope.interface import implements
from cybertools.agent.base.agent import Agent
from cybertools.agent.components import loggers
from cybertools.agent.interfaces import ILogger, ILogRecord
class LogRecord(object):
implements(ILogRecord)
datefmt = '%Y-%m-%dT%H:%S'
def __init__(self, logger, data):
self.logger = logger
self.data = data
self.timeStamp = time.time()
def __str__(self):
msg = [str(time.strftime(self.datefmt, time.localtime(self.timeStamp)))]
for k in sorted(self.data):
msg.append('%s:%s' % (str(k), str(self.data[k])))
return ' '.join(msg)
class Logger(object):
implements(ILogger)
recordFactory = LogRecord
def __init__(self, agent):
self.agent = agent
self.records = []
self.setup()
self.externalLoggers = []
def setup(self):
conf = self.agent.config.logger
self.externalLoggers = []
if conf.standard:
logger = logging.getLogger()
logger.level = conf.standard
logger.addHandler(logging.StreamHandler(sys.stdout))
self.externalLoggers.append(logger)
def log(self, data):
record = self.recordFactory(self, data)
self.records.append(record)
for logger in self.externalLoggers:
logger.info(str(record))
loggers.register(Logger, Agent, name='default')

9
agent/base/sample.cfg
View file

@ -1,9 +0,0 @@
#
# sample.cfg - agent configuration for demonstration and testing purposes
#
# $Id$
#
controller(names=['base.sample'])
scheduler(name='sample')
logger(name='default', standard=30)

46
agent/base/schedule.py
View file

@ -1,46 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Basic (sample) job scheduler.
$Id$
"""
from time import time
from zope.interface import implements
from cybertools.agent.base.agent import Master
from cybertools.agent.common import states
from cybertools.agent.components import schedulers
from cybertools.agent.interfaces import IScheduler
class Scheduler(object):
implements(IScheduler)
def __init__(self, agent):
self.agent = agent
def schedule(self, job, startTime=None):
job.startTime = startTime or int(time())
job.state = states.scheduled
job.execute() # the sample scheduler does not care about startTime
schedulers.register(Scheduler, Master, name='sample')

50
agent/common.py
View file

@ -1,50 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Miscellaneous common stuff.
$Id$
"""
from cybertools.util.jeep import Jeep
class JobState(object):
def __init__(self, name, value):
self.name = name
self.value = value
def hasError(self):
return self.value < 0
def hasFinished(self):
return (self.value <= states.aborted.value or
self.value >= states.completed.value)
def __str__(self):
return self.name
def __repr__(self):
return '<JobState %s>' % self.name
states = Jeep([JobState(n, v) for n, v in
(('initialized', 0), ('scheduled', 1), ('submitted', 2),
('running', 3), ('completed', 4), ('aborted', -1))])

34
agent/components.py
View file

@ -1,34 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Component registries.
$Id$
"""
from cybertools.util.adapter import AdapterFactory
agents = AdapterFactory()
controllers = AdapterFactory()
jobs = AdapterFactory()
loggers = AdapterFactory()
schedulers = AdapterFactory()
servers = AdapterFactory()
clients = AdapterFactory()

4
agent/control/__init__.py
View file

@ -1,4 +0,0 @@
"""
$Id$
"""

145
agent/control/cmdline.py
View file

@ -1,145 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Base/sample controller implementation.
$Id$
"""
from twisted.application import service, internet
from twisted.internet import protocol, reactor, stdio
from twisted.protocols import basic
from zope.interface import implements
from cybertools.agent.base.agent import Master
from cybertools.agent.core.control import SampleController
from cybertools.agent.components import controllers
class CmdlineController(SampleController):
def setup(self):
super(CmdlineController, self).setup()
prot = CmdlineProtocol()
prot.controller = self
stdio.StandardIO(prot)
self.results = {}
controllers.register(CmdlineController, Master, name='cmdline')
class TelnetController(CmdlineController):
def setup(self):
super(CmdlineController, self).setup()
port = self.agent.config.controller.telnet.port
from cybertools.agent.main import application
if application is None:
reactor.listenTCP(port, TelnetServerFactory(self))
else:
service = internet.TCPServer(port, TelnetServerFactory(self))
service.setServiceParent(application)
controllers.register(TelnetController, Master, name='telnet')
class CmdlineProtocol(basic.LineReceiver):
delimiter = '\n'
controller = None
def connectionMade(self):
self.sendLine("Agent console. Type 'help' for help.")
self.transport.write('> ')
def lineReceived(self, line):
if not line:
return
commandParts = line.split()
command = commandParts[0].lower()
args = commandParts[1:]
posArgs = []
kwArgs = {}
for arg in args:
if '=' in arg: # directory='...'
key, value = arg.split('=', 1)
if value in ('True', 'False'):
value = eval(value)
#elif value.isdigit():
# value = int(value)
kwArgs[key] = value
else:
posArgs.append(arg)
try:
method = getattr(self, 'do_' + command)
except AttributeError, e:
self.sendLine('Error: no such command.')
else:
try:
method(*posArgs, **kwArgs)
except Exception, e:
self.sendLine('Error: ' + str(e))
self.transport.write('> ')
def do_help(self, command=None):
"Help"
if command:
self.sendLine(getattr(self, 'do_' + command).__doc__)
else:
commands = [cmd[3:] for cmd in dir(self) if cmd.startswith('do_')]
self.sendLine("Valid commands: " +" ".join(commands))
def do_shutdown(self):
"Shut down"
self.sendLine('Shutting down.')
reactor.stop()
def do_agent(self, agentType='crawl.filesystem', name='a1'):
"Create agent"
self.controller.createAgent(agentType, name)
def do_job(self, jobType='sample', agent='a1', **kw):
"Enter job"
self.controller.enterJob(jobType, agent, params=kw)
def do_showresult(self):
"Show last result"
print self.controller.result
class TelnetProtocol(CmdlineProtocol):
delimiter = '\r\n'
def do_quit(self):
self.sendLine('Goodbye.')
self.transport.loseConnection()
class TelnetServerFactory(protocol.ServerFactory):
def __init__(self, controller):
self.controller = controller
def protocol(self, *args, **kw):
prot = TelnetProtocol(*args, **kw)
prot.controller = self.controller
return prot

38
agent/control/remote.py
View file

@ -1,38 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Controller that accepts and processes requests via XML-RPC.
$Id$
"""
from zope.interface import implements
from cybertools.agent.base.agent import Master
from cybertools.agent.core.control import SampleController
from cybertools.agent.components import controllers
class RemoteController(SampleController):
def setup(self):
super(RemoteController, self).setup()
controllers.register(RemoteController, Master, name='remote')

4
agent/core/__init__.py
View file

@ -1,4 +0,0 @@
"""
$Id$
"""

87
agent/core/agent.py
View file

@ -1,87 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Queueable agent base/sample classes.
$Id$
"""
from twisted.internet.defer import succeed
from zope.interface import implements
from cybertools.agent.base.agent import Agent, Master
from cybertools.agent.common import states
from cybertools.agent.components import agents
from cybertools.agent.interfaces import IQueueableAgent
class QueueableAgent(Agent):
implements(IQueueableAgent)
currentJob = None
def __init__(self, master):
super(QueueableAgent, self).__init__(master)
self.queue = []
def send(self, job):
if self.currentJob is None:
if self.queue: # this should not happen...
self.queue.insert(0, job)
job = self.queue.pop()
self.execute(job)
else:
self.queue.insert(0, job)
def execute(self, job):
job.state = states.running
self.currentJob = job
self.params = job.params
d = self.process()
d.addCallbacks(self.completed, self.error)
def process(self):
# do something with the current job, return a deferred
print ('Job %s on agent %s has been executed.'
% (self.currentJob.identifier, self.name))
return succeed('Done')
def completed(self, result):
job = self.currentJob
job.state = states.completed
self.log(job)
self.master.notify(job, result)
self.finishJob()
def error(self, result):
print '*** error', result
job = self.currentJob
job.state = states.aborted
self.log(self.currentJob, result='Error')
self.master.notify(job, result)
self.finishJob()
def finishJob(self):
self.currentJob = None
if self.queue:
job = self.queue.pop()
self.execute(job, job.params)
agents.register(QueueableAgent, Master, name='core.sample')

36
agent/core/control.py
View file

@ -1,36 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Base/sample controller implementation.
$Id$
"""
from zope.interface import implements
from cybertools.agent.base.agent import Master
from cybertools.agent.base.control import SampleController, AgentSpecification
from cybertools.agent.components import controllers
class SampleController(SampleController):
agents = (('sample01', 'core.sample'),)
controllers.register(SampleController, Master, name='core.sample')

57
agent/core/schedule.py
View file

@ -1,57 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Basic job scheduler using twisted.
$Id$
"""
from time import time
from zope.interface import implements
from cybertools.agent.base.agent import Master
from cybertools.agent.base.schedule import Scheduler as BaseScheduler
from cybertools.agent.common import states
from cybertools.agent.components import schedulers
from cybertools.agent.interfaces import IScheduler
from twisted.internet import reactor
from twisted.internet.defer import Deferred
class Scheduler(BaseScheduler):
implements(IScheduler)
def __init__(self, agent):
self.agent = agent
def schedule(self, job, startTime=None):
job.startTime = startTime or int(time())
if startTime is None:
startTime = int(time())
job.startTime = startTime
job.state = states.scheduled
reactor.callLater(startTime-int(time()), job.execute)
return startTime
def getJobsToExecute(startTime=0):
return [j for j in self.queue.values() if startTime <= j.startTime]
schedulers.register(Scheduler, Master, name='core')

46
agent/crawl/README.txt
View file

@ -1,46 +0,0 @@
================================================
Agents for Job Execution and Communication Tasks
================================================
($Id$)
>>> config = '''
... controller(names=['core.sample'])
... scheduler(name='core')
... logger(name='default', standard=30)
... '''
>>> from cybertools.agent.main import setup
>>> master = setup(config)
Starting agent application...
Using controllers core.sample.
Crawler
=======
The agent uses Twisted's cooperative multitasking model.
Crawler is the base class for all derived crawlers like the filesystem crawler
and the mailcrawler. The SampleCrawler returns a deferred that already had a
callback being called, so it will return at once.
Returns a deferred that must be supplied with a callback method (and in
most cases also an errback method).
We create the sample crawler via the master's controller. The sample
controller provides a simple method for this purpose.
>>> controller = master.controllers[0]
>>> controller.createAgent('crawl.sample', 'crawler01')
In the next step we request the start of a job, again via the controller.
>>> controller.enterJob('sample', 'crawler01')
The job is not executed immediately - we have to hand over control to
the twisted reactor first.
>>> from cybertools.agent.tests import tester
>>> tester.iterate()
SampleCrawler is collecting.
Job 00001 completed; result: [];

4
agent/crawl/__init__.py
View file

@ -1,4 +0,0 @@
"""
$Id$
"""

92
agent/crawl/base.py
View file

@ -1,92 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Crawl base and sample classes.
$Id$
"""
from zope.interface import implements
from cybertools.agent.base.agent import Master
from cybertools.agent.core.agent import QueueableAgent
from cybertools.agent.interfaces import ICrawler
from cybertools.agent.interfaces import IResource, IMetadataSet
from cybertools.agent.components import agents
from twisted.internet.defer import succeed
class Crawler(QueueableAgent):
implements(ICrawler)
def __init__(self, master, params={}):
super(Crawler, self).__init__(master)
def process(self):
return self.collect()
def collect(self, filter=None):
d = defer.succeed([])
return d
class SampleCrawler(Crawler):
def collect(self, filter=None):
print 'SampleCrawler is collecting.'
d = succeed([])
return d
agents.register(SampleCrawler, Master, name='crawl.sample')
class Resource(object):
implements(IResource)
data = file = path = None
type = 'sample'
contentType = 'text/plain'
encoding = ''
application = 'sample'
metadata = None
def __init__(self, data=None, **kw):
if data is not None:
self.data = data
for k, v in kw.items():
setattr(self, k, v)
self.subResources = []
class Metadata(dict):
implements(IMetadataSet)
def __init__(self, data=dict()):
self.update(data)
def asXML(self):
# TODO...
return ''
def set(self, key, value):
self['key'] = value

85
agent/crawl/filesystem.py
View file

@ -1,85 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Filesystem crawler.
$Id$
"""
import os
from fnmatch import filter
from datetime import datetime
from twisted.internet.defer import Deferred
from zope.interface import implements
from cybertools.agent.base.agent import Master
from cybertools.agent.components import agents
from cybertools.agent.crawl.base import Resource, Metadata
from cybertools.agent.crawl.base import Crawler
from cybertools.agent.util.task import coiterate
class FilesystemCrawler(Crawler):
def collect(self):
self.collected = []
coiterate(self.crawlFilesystem()).addCallback(self.finished)
# TODO: addErrback()
self.deferred = Deferred()
return self.deferred
def finished(self, result):
self.deferred.callback(self.collected)
def crawlFilesystem(self):
directory = self.params.get('directory')
pattern = self.params.get('pattern') or '*'
lastRun = self.params.get('lastrun') or datetime(1980, 1, 1)
for path, dirs, files in os.walk(directory):
if '.svn' in dirs:
del dirs[dirs.index('.svn')]
for x in self.loadFiles(path, files, pattern, lastRun):
yield None
def loadFiles(self, path, files, pattern, lastRun):
for f in filter(files, pattern):
filename = os.path.join(path, f)
mtime = datetime.fromtimestamp(os.path.getmtime(filename))
if mtime <= lastRun: # file not changed
continue
meta = dict(
path=filename,
)
self.collected.append(FileResource(path=filename, metadata=Metadata(meta)))
yield None
agents.register(FilesystemCrawler, Master, name='crawl.filesystem')
class FileResource(Resource):
type = 'file'
application = 'filesystem'
@property
def data(self):
f = open(self.path, 'r')
text = f.read()
f.close()
return text

42
agent/crawl/filesystem.txt
View file

@ -1,42 +0,0 @@
================================================
Agents for Job Execution and Communication Tasks
================================================
($Id$)
>>> import os
>>> from time import time
>>> from cybertools.agent.tests import tester, baseDir
>>> config = '''
... controller(names=['core.sample'])
... scheduler(name='core')
... logger(name='default', standard=30)
... '''
>>> from cybertools.agent.main import setup
>>> master = setup(config)
Starting agent application...
Using controllers core.sample.
Filesystem Crawler
==================
>>> controller = master.controllers[0]
>>> controller.createAgent('crawl.filesystem', 'sample03')
In the next step we request the start of a job, again via the controller.
>>> path = os.path.join(baseDir, 'testing', 'data')
>>> controller.enterJob('sample', 'sample03', params=dict(directory=path))
The job is not executed immediately - we have to hand over control to
the twisted reactor first.
>>> from cybertools.agent.tests import tester
>>> tester.iterate()
Job 00001 completed; result: [..., ...];
>>> r0 = controller.result[0]
>>> r0.metadata, r0.data
({'path': '...file1.txt'}, 'Data from file1.txt')

76
agent/crawl/mail.py
View file

@ -1,76 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Crawl base and sample classes.
$Id$
"""
from zope.interface import implements
from cybertools.agent.base.agent import Agent, Master
from cybertools.agent.crawl.base import Resource
from cybertools.agent.crawl.base import Metadata
from cybertools.agent.crawl.base import Crawler
from cybertools.agent.components import agents
from twisted.internet.defer import succeed
class MailCrawler(Crawler):
def __init__(self, master):
super(MailCrawler, self).__init__(master)
self.result = []
def collect(self, filter=None):
print 'MailCrawler is collecting.'
d = self.crawlFolders()
return d
def fetchCriteria(self):
pass
def crawlFolders(self):
return succeed([])
def loadMailsFromFolder(self, folder):
pass
def createResource(self, data, path=None, application=None, metadata=None):
resource = MailResource(data=data, path=path, application=application,
metadata=metadata)
self.result.append(resource)
def createMetadata(self, metadata):
metadata = Metadata(metadata)
## for k, v in kw.items():
## metadata[k] = v
return metadata
def login(self):
pass
agents.register(MailCrawler, Master, name='crawl.mail')
class MailResource(Resource):
type = 'email'
application = 'mailclient'

231
agent/crawl/outlook.py
View file

@ -1,231 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Outlook Crawler Class.
$Id$
"""
import re
from email import MIMEMultipart
import tempfile
import os
from zope.interface import implements
from twisted.internet import defer
#from pywintypes import com_error
#The watsup import is needed as soon as we start handling the Outlook Pop-Up
#again
#This should also be integrated within the wrapper-api for doctests
#from watsup.winGuiAuto import findTopWindow, findControl, findControls, clickButton, \
# getComboboxItems, selectComboboxItem, setCheckBox
from cybertools.agent.base.agent import Agent, Master
from cybertools.agent.crawl.mail import MailCrawler
from cybertools.agent.crawl.mail import MailResource
from cybertools.agent.crawl.filesystem import FileResource
from cybertools.agent.components import agents
from cybertools.agent.system.windows import api
from cybertools.agent.util.task import coiterate
from cybertools.agent.system.windows.codepages import codepages
# some constants
COMMASPACE = ', '
class OutlookCrawler(MailCrawler):
keys = ""
inbox = ""
subfolders = ""
pattern = ""
def collect(self, filter=None):
self.result = []
self.d = defer.Deferred()
self.oOutlookApp = None
if self.findOutlook():
self.fetchCriteria()
coiterate(self.crawlFolders()).addCallback(self.finished).addErrback(self.error)
else:
pass
#self.d.addErrback([])
return self.d
def error(self, reason):
print '***** error',
print reason
def finished(self, result):
self.d.callback(self.result)
def fetchCriteria(self):
criteria = self.params
self.keys = criteria.get('keys')
self.inbox = criteria.get('inbox') #boolean
self.subfolders = criteria.get('subfolders') #boolean
self.pattern = criteria.get('pattern')
if self.pattern != '' and self.pattern != None:
self.pattern = re.compile(criteria.get('pattern') or '.*')
def crawlFolders(self):
onMAPI = self.oOutlookApp.GetNamespace("MAPI")
ofInbox = \
onMAPI.GetDefaultFolder(api.client.constants.olFolderInbox)
# fetch mails from inbox
if self.inbox:
for m in self.loadMailsFromFolder(ofInbox):
yield None
# fetch mails of inbox subfolders
if self.subfolders and self.pattern is None:
lInboxSubfolders = getattr(ofInbox, 'Folders')
for of in range(lInboxSubfolders.__len__()):
# get a MAPI-subfolder object and load its emails
for m in self.loadMailsFromFolder(lInboxSubfolders.Item(of + 1)):
yield None
elif self.subfolders and self.pattern:
lInboxSubfolders = getattr(ofInbox, 'Folders')
for of in range(lInboxSubfolders.__len__()):
# get specified MAPI-subfolder object and load its emails
if self.pattern.match(getattr(lInboxSubfolders.Item(of + 1), 'Name')):
for m in self.loadMailsFromFolder(lInboxSubfolders.Item(of + 1)):
yield None
def loadMailsFromFolder(self, folder):
# get items of the folder
folderItems = getattr(folder, 'Items')
for item in range(len(folderItems)):
mail = folderItems.Item(item+1)
if mail.Class == api.client.constants.olMail:
if self.keys is None:
self.keys = []
for key in mail._prop_map_get_.items():
try:
if isinstance(key[0], (int, str, unicode, bool)):
self.keys.append(key[0])
except api.com_error:
pass
record = {}
for key in self.keys:
try:
if (hasattr(mail, key)):
value = getattr(mail, key)
if isinstance(value, (int, str, unicode, bool)):
record[key] = value
else:
record[key] = None
except:
pass
metadata = self.assembleMetadata(folder, record)
# Create a resource and append it to the result list
self.createResource(mail, folder, metadata)
yield None
def findOutlook(self):
outlookFound = False
try:
self.oOutlookApp = \
api.client.gencache.EnsureDispatch("Outlook.Application")
outlookFound = True
except com_error:
pass
return outlookFound
def assembleMetadata(self, folder, mailAttr):
meta = {}
for key in mailAttr.keys():
if isinstance(mailAttr[key], (str, unicode))\
and mailAttr[key] != 'Body' and mailAttr[key] != 'HTMLBody':
meta[key] = mailAttr[key].encode('utf-8')
elif isinstance(mailAttr[key], (list, tuple, dict)):
lst = []
for rec in mailAttr[key]:
lst.append(rec)
meta[key] = COMMASPACE.join(lst)
else:
meta[key] = mailAttr[key]
meta["path"] = folder
metadata = self.createMetadata(meta)
return metadata
def createResource(self, mail, folder, metadata):
enc = None
textType = "application/octet-stream"
attachments = []
mailContent = ""
ident = None
if (hasattr(mail, 'BodyFormat')):
value = getattr(mail, 'BodyFormat')
if value == 1:
#1: it is a plain text mail, that is maybe decorated with
#some html Tags by Outlook for formatting
#so save it as plain text mail
if hasattr(mail, 'Body'):
mailContent = getattr(mail, 'Body')
textType = "text/plain"
else:
mailContent = ""
textType = "text/plain"
elif value == 2:
#2: it is a HTML mail
if hasattr(mail, 'HTMLBody'):
mailContent = getattr(mail, 'HTMLBody')
textType = "text/html"
else:
mailContent = ""
textType = "text/html"
else:
#Could not determine BodyFormat. Try to retrieve plain text
if hasattr(mail, 'Body'):
mailContent = getattr(mail, 'Body')
else:
mailContent = ""
if hasattr(mail, 'InternetCodepage'):
Codepage = getattr(mail, 'InternetCodepage')
if codepages.has_key(Codepage):
enc = codepages[Codepage]
if hasattr(mail, 'EntryID'):
ident = getattr(mail, 'EntryID')
if hasattr(mail, 'Attachments'):
attachedElems = getattr(mail, 'Attachments')
for item in range(1, len(attachedElems)+1):
fileHandle, filePath = tempfile.mkstemp(prefix="outlook")
attachedItem = attachedElems.Item(item)
attachedItem.SaveAsFile(filePath)
os.close(fileHandle)
metadat = self.createMetadata(dict(filename=filePath))
fileRes = FileResource(data=None,
path=filePath,
metadata=metadat)
attachments.append(fileRes)
fileHandle, filePath = tempfile.mkstemp(prefix="olmail")
filePointer = os.fdopen(fileHandle, "w")
mailContent = mailContent.encode('utf-8')
filePointer.write(mailContent)
filePointer.close()
resource = MailResource(data=mailContent,
contentType=textType,
encoding=enc,
path=filePath,
application='outlook',
identifier=ident,
metadata=metadata,
subResources=attachments)
self.result.append(resource)
agents.register(OutlookCrawler, Master, name='crawl.outlook')

53
agent/crawl/outlook.txt
View file

@ -1,53 +0,0 @@
================================================
Agents for Job Execution and Communication Tasks
================================================
($Id$)
>>> config = '''
... controller(names=['core.sample'])
... scheduler(name='core')
... logger(name='default', standard=30)
... system.winapi = 'testing'
... '''
>>> from cybertools.agent.main import setup
>>> master = setup(config)
Starting agent application...
Using controllers core.sample.
OutlookCrawler
==============
The agent uses Twisted's cooperative multitasking model.
OutlookCrawler is derived from MailCrawler. The OutlookCrawler returns a deferred
which itself holds a list of MailResource Objects.
Returns a deferred that must be supplied with a callback method (and in
most cases also an errback method).
The TestCase here is using subsidiary methods which replace calls to the "real Outlook
dlls".
>>> controller = master.controllers[0]
>>> controller.createAgent('crawl.outlook', 'sample02')
In the next step we request the start of a job, again via the controller.
>>> controller.enterJob('sample', 'sample02', params=dict(inbox=True))
The job is not executed immediately - we have to hand over control to
the twisted reactor first.
>>> from cybertools.agent.tests import tester
>>> tester.iterate()
Outlook.Application retrieved
Namespace MAPI retrieved
retrieving Outlook default folder
collecting Mails from folder
Attachment: Invitation.pdf
Attachment: 21.pdf
Attachment saved
Attachment saved
Job 00001 completed; result: [<...MailResource...>, <...MailResource...>, <...MailResource...>];

278
agent/interfaces.py
View file

@ -1,278 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
cybertools agent interfaces.
$Id$
"""
from zope.interface import Interface, Attribute
from cybertools.util.jeep import Jeep
# agents
class IAgent(Interface):
""" An agent waits for jobs to execute.
"""
name = Attribute('A name identifying the agent.')
master = Attribute('IMaster instance.')
config = Attribute('Configuration settings.')
logger = Attribute('Logger instance to be used for recording '
'job execution and execution results.')
def send(job):
""" If the agent supports queueing and the agent is busy put
job in queue, otherwise just execute the job.
"""
def execute(job):
""" Execute a job using the parameters given.
"""
class IQueueableAgent(Interface):
""" An agent that keeps a queue of jobs. A queueable agent
executes not more than one job at a time; when a running
job is finished the next one will be taken from the queue.
"""
queue = Attribute('A sequence of jobs to execute.')
def process():
""" Do the real work asynchronously, returning a deferred.
This method will be called by execute().
"""
class IMaster(IAgent):
""" The top-level controller agent.
"""
config = Attribute('Central configuration settings.')
controllers = Attribute('Collection of IController instances.')
scheduler = Attribute('IScheduler instance.')
children = Attribute('A collection of agents that are managed by this '
'master.')
def setup():
""" Set up the master agent by triggering all assigned controllers.
Each controller will then call the master agent's callback
methods ``setupAgents()`` and ``setupJobs()``.
"""
def setupAgents(controller, agentSpecs):
""" Callback for loading agent specifications from the controller
and setting up the corresponding agents.
Will be called upon agent setup and later when the controller
wants to provide new agent information.
"""
def setupJobs(controller, jobSpecs):
""" Callback for loading the specifications of active jobs from
the controller and scheduling the corresponding jobs.
Will be called upon agent setup and later when the controller
wants to provide new job information.
"""
def notify(job, result=None, message=''):
""" Callback for informing the master about the state of a job.
"""
class ICrawler(IAgent):
""" Collects resources.
"""
def collect(filter=None):
""" Return a deferred that upon callback will provide a
collection of resource objects that should be transferred
to the server.
Use the selection criteria given to filter the resources that
should be collected.
"""
class ITransporter(IAgent):
""" Transfers one or more collected resources and corresponding
metadata to another entity - a remote agent or another application.
"""
serverURL = Attribute('URL of the server the resources will be '
'transferred to. The URL also determines the '
'transfer protocol, e.g. HTTP or SFTP.')
method = Attribute('Transport method, e.g. PUT.')
machineName = Attribute('Name under which the local machine is '
'known to the server.')
userName = Attribute('User name for logging in to the server.')
password = Attribute('Password for logging in to the server.')
def transfer(resource):
""" Transfer the resource (an object providing IResource)
to the server and return a Deferred.
"""
# job control
class IController(Interface):
""" Fetches agent and job specifications from a control
storage and updates the storage with the status and result
information.
"""
def setup():
""" Set up the controller; e.g. create agents and jobs by calling
the agent's callback methods.
"""
class IScheduler(Interface):
""" Manages jobs and cares that they are started at the appropriate
time by the agents responsible for it.
"""
def schedule(job, startTime=None):
""" Register the job given for execution at the intended start
date/time (an integer timestamp) and return the job.
If the start time is not given schedule the job for immediate
start. Return the start time with which the job has been
scheduled - this may be different from the start time
supplied.
"""
# jobs
class IScheduledJob(Interface):
""" A job that will be executed on some external triggering at
a predefined date and time - this is the basic job interface.
"""
identifier = Attribute('A name/ID unique within the realm of the '
'controller.')
scheduler = Attribute('Scheduler that controls this job.')
agent = Attribute('Agent responsible for executing the job.')
controller = Attribute('Controller that issued the job.')
startTime = Attribute('Date/time at which the job should be executed.')
params = Attribute('Mapping with key/value pairs to be used by '
'the ``execute()`` method.')
state = Attribute('An object representing the current state of the job.')
repeat = Attribute('Number of seconds after which the job should be '
'rescheduled. Do not repeat if 0.')
successors = Attribute('Jobs to execute immediately after this '
'one has been finished.')
def execute():
""" Execute the job, typically by calling the ``execute()`` method
of the agent responsible for it.
"""
def reschedule(startTime):
""" Re-schedule the job, setting the date/time the job should be
executed again.
"""
# information objects
class IResource(Interface):
""" Represents a data object that is collected by a crawler and
will be transferred to the server.
"""
data = Attribute('A string representation of the '
'resource\'s content; may be None if the receiver of '
'the information can retrieve the data from the file or path '
'attribute.')
file = Attribute('A file-like object providing the data via its read() '
'method; may be None if the data or path attribute '
'is given.')
path = Attribute('A filesystem path for accessing the resource; may be '
'None if the data or file attribute is given.')
identifier = Attribute('A string (usually derived from the path) that '
'uniquely identifies the resource.')
type = Attribute('A string denoting the type of the resource, e.g. '
'"file" or "email".')
contentType = Attribute('A string denoting the MIME type of the data, '
'e.g. "text/plain" or "application/octet-stream"')
encoding = Attribute('Optional: a string denoting the encoding of the '
'file data, e.g. "UTF-8".')
application = Attribute('The name of the application that provided '
'the resource, e.g. "filesystem" or "mail".')
metadata = Attribute('Information describing this resource; '
'should be an IMetadataSet object.')
subResources = Attribute('A collection of resources that are inherently '
'connected to or parts of this resource, e.g. attachments '
'of an email. Will be None or empty in most cases.')
class IMetadataSet(Interface):
""" Metadata associated with a resource; a mapping.
"""
def asXML():
""" Return an XML string representing the metadata set.
If this metadata set contains other metadata sets
(nested metadata) these will be converted to XML as well.
"""
def set(key, value):
""" Set a metadata element.
The value may be a string or another metadata set
(nested metadata).
"""
# logging
class ILogger(Interface):
""" Ordered collection (list) of log records, probably stored on some
external device.
"""
externalLoggers = Attribute('A collection of logger objects '
'to which the logging records should be written.')
def setup():
""" Initialize the logger with the current configuration settings.
"""
def log(data):
""" Record the information given by the ``data`` argument
(a mapping).
"""
class ILogRecord(Interface):
"""
"""
def __str__():
""" Return a string representation suitable for writing to a
log file.
"""

77
agent/main.py
View file

@ -1,77 +0,0 @@
#! /usr/bin/env python2.4
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Agent application.
$Id$
"""
import os
from twisted.internet import reactor
from cybertools.agent.base.agent import Master
application = None # contains application object if started via twistd
def getConfig():
agentHome = os.path.abspath(os.path.dirname(__file__))
configName = 'agent.cfg'
configFile = open(os.path.join(agentHome, configName))
config = configFile.read()
configFile.close()
return config
def setup(configInfo=None):
if configInfo is None:
configInfo = getConfig()
master = Master(configInfo)
setupEnvironment(master.config)
master.setup()
return master
def setupEnvironment(config):
# API registration:
from cybertools.agent.system.windows import api
api.setup(config)
from cybertools.agent.system import http, xmlrpc, sftp
http.setup(config)
xmlrpc.setup(config)
sftp.setup(config)
# self registration of components:
from cybertools.agent.base import agent, control, job, log, schedule
from cybertools.agent.core import agent, control, schedule
from cybertools.agent.control import cmdline, remote
from cybertools.agent.crawl import base, filesystem, outlook
from cybertools.agent.talk import http
from cybertools.agent.transport import remote, loops
def startReactor():
reactor.run()
print 'Agent application has been stopped.'
if __name__ == '__main__':
setup()
startReactor()

4
agent/system/__init__.py
View file

@ -1,4 +0,0 @@
"""
$Id$
"""

31
agent/system/http.py
View file

@ -1,31 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Configuration-controlled import of HTTP communication functionality.
$Id$
"""
def setup(config):
global listener, getPage
if config.talk.http.handler == 'testing':
from cybertools.agent.testing.http import listener, getPage
else:
from twisted.internet import reactor as listener
from twisted.web.client import getPage

30
agent/system/sftp.py
View file

@ -1,30 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Configuration-controlled import of sftp functionality.
$Id: rpcapi.py
"""
def setup(config):
global FileTransfer
if config.transport.remote.sftp == 'testing':
from cybertools.agent.testing.sftp import FileTransfer
else:
from cybertools.agent.transport.file.sftp import FileTransfer

4
agent/system/windows/__init__.py
View file

@ -1,4 +0,0 @@
"""
$Id$
"""

39
agent/system/windows/api.py
View file

@ -1,39 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Conficuration-controlled import of Windows API functions.
$Id$
"""
def setup(config):
global client, ctypes, win32api, win32process, win32con, com_error
if config.system.winapi == 'testing':
from cybertools.agent.testing.winapi import \
client, ctypes, win32api, win32process, win32con, com_error
else:
try:
from win32com import client
import ctypes
import win32api, win32process, win32con
from pywintypes import com_error
except ImportError:
from cybertools.agent.testing.winapi import \
client, ctypes, win32api, win32process, win32con, com_error

59
agent/system/windows/codepages.py
View file

@ -1,59 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Codepages Module
$Id$
"""
codepages = {28596: 'iso-8859-6',\
1256: 'windows-1256',\
28594: 'iso-8859-4',\
1257: 'windows-1257',\
28592: 'iso-8859-2',\
1250: 'windows-1250',\
936: 'gb2312',\
52936: 'hz-gb-2312',\
950: 'big5',\
28595: 'iso-8859-5',\
20866: 'koi8-r',\
21866: 'koi8-u',\
1251: 'windows-1251',\
28597: 'iso-8859-7',\
1253: 'windows-1253',\
38598: 'iso-8859-8-i',\
1255: 'windows-1255',\
51932: 'euc-jp',\
50220: 'iso-2022-jp',\
50221: 'csISO2022JP',\
932: 'iso-2022-jp',\
949: 'ks_c_5601-1987',\
51949: 'euc-kr',\
28593: 'iso-8859-3',\
28605: 'iso-8859-15',\
874: 'windows-874',\
28599: 'iso-8859-9',\
1254: 'windows-1254',\
65000: 'utf-7',\
65001: 'utf-8',\
20127: 'us-ascii',\
1258: 'windows-1258',\
28591: 'iso-8859-1',\
1252: 'Windows-1252'
}

56
agent/system/windows/outlookdialog.py
View file

@ -1,56 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Module for handling Outlook dialoges
$Id$
"""
def login(self):
pass
def handleOutlookDialog(self):
"""
This function handles the outlook dialog, which appears if someone
tries to access to MS Outlook.
"""
hwnd = None
while True:
hwnd = api.ctypes.windll.user32.FindWindowExA(None, hwnd, None, None)
if hwnd == None:
break
else:
val = u"\0" * 1024
api.ctypes.windll.user32.GetWindowTextW(hwnd, val, len(val))
val = val.replace(u"\000", u"")
if val and repr(val) == "u'Microsoft Office Outlook'":
print repr(val)
# get the Main Control
form = api.findTopWindow(wantedText='Microsoft Office Outlook')
controls = findControls(form)
# get the check box
checkBox = findControl(form, wantedText='Zugriff')
setCheckBox(checkBox, 1)
# get the combo box
comboBox = findControl(form, wantedClass='ComboBox')
items = getComboboxItems(comboBox)
selectComboboxItem(comboBox, items[3])#'10 Minuten'
# finally get the button and click it
button = findControl(form, wantedText = 'Erteilen')
clickButton(button)
break

30
agent/system/xmlrpc.py
View file

@ -1,30 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Configuration controlled import of twisted xmlrpc functionality
$Id: rpcapi.py
"""
def setup(config):
global xmlrpc
if config.transport.remote.server == 'testing':
from cybertools.agent.testing.rpcserver import xmlrpc
else:
from twisted.web import xmlrpc

73
agent/talk/README.txt
View file

@ -1,73 +0,0 @@
================================================
Agents for Job Execution and Communication Tasks
================================================
($Id$)
>>> from cybertools.agent.tests import tester
Communication Handling
======================
Communication services are provided by handlers specified in the ``talk``
package.
Set up and start an agent with a server
---------------------------------------
>>> config = '''
... controller(names=['core.sample'])
... scheduler(name='core')
... logger(name='default', standard=30)
... talk.server(names=['http'])
... talk.server.http(port=8081)
... talk.http(handler='testing')
... '''
>>> from cybertools.agent.main import setup
>>> master = setup(config)
Starting agent application...
Using controllers core.sample.
Setting up HTTP handler for port 8081.
>>> master.servers
[<cybertools.agent.talk.http.HttpServer object...>]
We also provide a class to be used for creating subscribers, i.e. objects
that receive messages.
>>> class Subscriber(object):
... def __init__(self, name):
... self.name = name
... def onMessage(self, interaction, data):
... print ('%s receiving: interaction=%s, data=%s' %
... (self.name, interaction, data))
... tester.stop()
>>> serverSub = Subscriber('server')
>>> master.servers[0].subscribe(serverSub, 'testing')
Set up a client
---------------
In order to simplify the testing we do not set up a separate agent to
work with the client but handle the client directly.
>>> from cybertools.agent.talk.http import HttpClient
>>> client = HttpClient(master)
>>> clientSub = Subscriber('client')
>>> session = client.connect(clientSub, 'http://localhost:8081/')
Run the communication dialog
----------------------------
>>> tester.run()
client receiving: interaction=None, data={u'status': u'OK'}
Fin de Partie
=============
>>> tester.stopThreads()

4
agent/talk/__init__.py
View file

@ -1,4 +0,0 @@
"""
$Id$
"""

93
agent/talk/base.py
View file

@ -1,93 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Handling asynchronous communication tasks - common and base classes.
$Id$
"""
from twisted.web.client import getPage
from zope.interface import implements
from cybertools.agent.talk.interfaces import ISession, IInteraction
from cybertools.util import json
class Session(object):
implements(ISession)
def __init__(self, id, manager, subscriber, url):
self.id = id
self.manager = manager
self.subscriber = subscriber
self.url = url
self.state = 'logon'
self.sending = False
self.queue = []
self.interactions = {}
self.interactionCount = 0
def received(self, data):
data = json.loads(data)
# TODO: check data; notify sender?
self.sending = False
self._processQueue()
def send(self, data, interaction):
data['interaction'] = interaction.id
if self.sending or self.queue:
self.queue.append(data)
else:
self._sendData(data)
def processQueue(self):
if not self.queue:
return
self._sendData(self.queue.pop(0))
def sendData(self, data, command='send'):
self.sending = True
content = dict(id=self.id, command=command, data=data)
d = getPage(self.url, postdata=json.dumps(content))
d.addCallback(s.received)
def connected(self, data):
data = json.loads(data)
self.state = 'open'
self.subscriber.onMessage(None, data)
self.sending = False
self.processQueue()
def generateInteractionId(self):
self.interactionCount += 1
return '%07i' % self.interactionCount
class Interaction(object):
implements(IInteraction)
finished = False
def __init__(self, session):
self.session = session
self.id = self.session.generateInteractionId()
self.session.interactions[self.id] = self

181
agent/talk/http.py
View file

@ -1,181 +0,0 @@
#
# Copyright (c) 2009 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Handling asynchronous and possibly asymmetric communication tasks via HTTP.
$Id$
"""
from time import time
from twisted.web.client import getPage
from twisted.web.resource import Resource
from twisted.web.server import Site, NOT_DONE_YET
from zope.interface import implements
from cybertools.agent.base.agent import Master
from cybertools.agent.components import servers, clients
from cybertools.agent.system.http import listener
from cybertools.agent.talk.base import Session, Interaction
from cybertools.agent.talk.interfaces import IServer, IClient
from cybertools.util import json
# server implementation
#@server
class HttpServer(object):
implements(IServer)
def __init__(self, agent):
self.agent = agent
self.port = agent.config.talk.server.http.port
self.subscribers = {}
self.sessions = {}
self.site = Site(RootResource(self))
def setup(self):
print 'Setting up HTTP handler for port %i.' % self.port
listener.listenTCP(self.port, self.site)
def subscribe(self, subscriber, aspect):
subs = self.subscribers.setdefault(aspect, [])
if subscriber not in subs:
subs.append(subscriber)
def unsubscribe(self, subscriber, aspect):
pass
def send(self, session, data, interaction=None):
if interaction is None:
interaction = Interaction(session)
# check session's queue
# check open poll - write response
return interaction
def process(self, client, data):
action = data.get('action')
if not action:
return self._error('missing action')
amethod = self.actions.get(action)
if amethod is None:
return self._error('illegal action %r' % action)
sid = data.get('session')
if not sid:
return self._error('missing session id')
sessionId = ':'.join((client, sid))
message = amethod(self, sessionId, client, data)
if message:
return self._error(message)
return '{"status": "OK"}'
def _connect(self, sessionId, client, data):
if sessionId in self.sessions:
return 'duplicate session id %r' % sessionId
self.sessions[sessionId] = HttpServerSession(sessionId, self, None, client)
# TODO: notify subscribers
def _poll(self, sessionId, client, data):
# record deferred with session
return NOT_DONE_YET
def _send(self, sessionId, client, data):
for sub in self.subscribers.values():
sub.onMessage(data)
def _error(self, message):
return json.dumps(dict(status='error', message=message))
actions = dict(connect=_connect, poll=_poll, send=_send)
servers.register(HttpServer, Master, name='http')
class RootResource(Resource):
isLeaf = True
def __init__(self, server):
self.server = server
def render(self, request):
client = request.getClient()
data = json.loads(request.content.read())
return self.server.process(client, data)
class HttpServerSession(Session):
pass
# client implementation
#@client
class HttpClient(object):
implements(IClient)
def __init__(self, agent):
self.agent = agent
self.sessions = {}
def connect(self, subscriber, url, credentials=None):
id = self.generateSessionId()
s = HttpClientSession(self, id, subscriber, url)
self.sessions[id] = s
data = dict(action='connect', session=id)
if credentials is not None:
data.update(credentials)
# s.send(data, None)
d = getPage(url, postdata=json.dumps(data))
d.addCallback(s.connected)
return s
def disconnect(self, session):
pass
def send(self, session, data, interaction=None):
if interaction is None:
interaction = Interaction(session)
session.send(data, interaction)
return interaction
def generateSessionId(self):
return '%.7f' % time()
clients.register(HttpClient, Master, name='http')
class HttpClientSession(Session):
def connected(self, data):
super(HttpClientSession, self).connected(data)
# self.poll()
def pollReceived(self, data):
data = json.loads(data)
if data.get('action') != 'idle':
self.subscriber.onMessage(interaction, data)
# self.poll()
def poll(self):
content = dict(id=self.id, command='poll')
d = getPage(self.url, postdata=json.dumps(content))
d.addCallback(s.pollReceived)

121
agent/talk/interfaces.py
View file

@ -1,121 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Interfaces for handling asynchronous communication tasks.
$Id$
"""
from zope.interface import Interface, Attribute
class IServer(Interface):
""" A server waits for connection requests from a client. A connected
client may then send data to or receive messages from the server.
"""
def subscribe(subscriber, aspect):
""" The subscriber will receive messages via its ``onMesssage`` method.
The aspect is a dotted string used to select the kind of
sessions/remote clients the subscriber wants to receive messages
from.
"""
def unsubscribe(subscriber, aspect):
""" Stop receiving messages.
"""
def send(session, data, interaction=None):
""" Send data to the remote client specified via the session given.
The session has to be created previously by a connect attempt
from the client.
If interaction is None, create a new one.
Return the interaction.
"""
class IClient(Interface):
""" A client initiates a connection (session) to a server and may then
sent data to or receive data from the server.
"""
def connect(subscriber, url, credentials=None):
""" Connect to a server using the URL given, optionally logging in
with the credentials given.
The subscriber will receive messages via its ``onMesssage`` callback.
Return a an ISession implementation that may be used for sending
data to the server.
"""
def disconnect(session):
""" Close the connection for the session given.
"""
def send(session, data, interaction=None):
""" Send data to the server specified via the session given.
If interaction is None, create a new one.
Return the interaction.
Sending an interaction with ``finished`` set to True signifies
the last message of an interaction.
"""
# auxiliary interfaces
class ISubscriber(Interface):
""" May receive message notifications.
"""
def onMessage(interaction, data):
""" Callback method for message notifications.
"""
def onError(interaction, data):
""" Callback method for error notifications.
"""
class ISession(Interface):
""" Represents the connection to a server within a client or
a remote client connection within a server.
"""
manager = Attribute("""The server or client object, respectively, that
created the session.""")
subscriber = Attribute("The subscriber that initiated the session.")
url = Attribute("The URL of the server (or client) the session connects to.")
state = Attribute("""A string specifying the current state of the session:
'logon': The remote client is trying to connect/log in,
data may contain credential information;
'logoff': The remote client is closing the connection;
'open': The connection is open.""")
class IInteraction(Interface):
""" Represents a set of message exchanges belonging together.
"""
session = Attribute("The session the interaction belongs to.")
finished = Attribute("The interaction is finished, interaction data may be cleared.")

4
agent/testing/__init__.py
View file

@ -1,4 +0,0 @@
"""
$Id$
"""

9
agent/testing/agent.cfg
View file

@ -1,9 +0,0 @@
#
# Standard configuration for agent application
#
# $Id$
#
controller(names=['test'])
scheduler(name='core')
logger(name='default', standard=30)

43
agent/testing/control.py
View file

@ -1,43 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Controller for testing purposes.
$Id$
"""
from zope.interface import implements
from cybertools.agent.base.agent import Master
from cybertools.agent.base.control import SampleController, JobSpecification
from cybertools.agent.components import controllers
from cybertools.agent.crawl.mail import MailResource
class Controller(SampleController):
agents = (('tr1', 'transport.remote'),)
def _getCurrentJobs(self):
print '_getCurrentJobs'
return [JobSpecification('sample', '00001', agent='tr1',
params=dict(resource=MailResource()))]
controllers.register(Controller, Master, name='test')

1
agent/testing/data/file1.txt
View file

@ -1 +0,0 @@
Data from file1.txt

1
agent/testing/data/subdir/file2.txt
View file

@ -1 +0,0 @@
Data from file2.txt

44
agent/testing/http.py
View file

@ -1,44 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Fake testing objects/functions for HTTP communication.
$Id$
"""
from twisted.internet.defer import Deferred, succeed
class Listener(object):
site = port = None
def listenTCP(self, port, site):
self.port = port
self.site = site
self.resource = site.resource
deferred = self.deferred = Deferred()
return deferred
listener = Listener()
def getPage(url, contextFactory=None, method='GET', postdata=None, **kwargs):
return succeed('{"message": "OK"}')

73
agent/testing/main.py
View file

@ -1,73 +0,0 @@
#! /usr/bin/env python2.4
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Agent application.
$Id$
"""
import os
from twisted.internet import reactor
from cybertools.agent.base.agent import Master
application = None # contains application object if started via twistd
def getConfig():
home = os.path.abspath(os.path.dirname(__file__))
configName = 'agent.cfg'
configFile = open(os.path.join(home, configName))
config = configFile.read()
configFile.close()
return config
def setup(configInfo=None):
if configInfo is None:
configInfo = getConfig()
master = Master(configInfo)
setupEnvironment(master.config)
master.setup()
print 'Starting agent application...'
print 'Using controllers %s.' % ', '.join(master.config.controller.names)
return master
def setupEnvironment(config):
from cybertools.agent.base import agent, control, job, log, schedule
from cybertools.agent.core import agent, control, schedule
from cybertools.agent.control import cmdline, remote
from cybertools.agent.transport import remote, loops
from cybertools.agent.testing import control
from cybertools.agent.system.windows import api
api.setup(config)
from cybertools.agent.crawl import base, filesystem, outlook
def startReactor():
reactor.run()
print 'Agent application has been stopped.'
if __name__ == '__main__':
setup()
startReactor()

74
agent/testing/main_outlook.py
View file

@ -1,74 +0,0 @@
#! /usr/bin/env python2.4
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Agent application.
$Id$
"""
import os
from twisted.internet import reactor
from cybertools.agent.base.agent import Master
application = None # contains application object if started via twistd
def getConfig():
agentHome = os.path.abspath(os.path.dirname(__file__))
configName = 'outlook.cfg'
configFile = open(os.path.join(agentHome, configName))
config = configFile.read()
configFile.close()
return config
def setup(configInfo=None):
if configInfo is None:
configInfo = getConfig()
master = Master(configInfo)
setupEnvironment(master.config)
master.setup()
print 'Starting agent application...'
print 'Using controllers %s.' % ', '.join(master.config.controller.names)
return master
def setupEnvironment(config):
from cybertools.agent.base import agent, control, job, log, schedule
from cybertools.agent.core import agent, control, schedule
from cybertools.agent.control import cmdline
from cybertools.agent.system.windows import api
api.setup(config)
from cybertools.agent.crawl import base, outlook
def startReactor():
reactor.run()
print 'Agent application has been stopped.'
if __name__ == '__main__':
master = setup()
controller = master.controllers[0]
controller.createAgent('crawl.outlook', 'sample02')
controller.enterJob('sample', 'sample02', params=dict(inbox=True))
startReactor()

82
agent/testing/main_transport.py
View file

@ -1,82 +0,0 @@
#! /usr/bin/env python2.4
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Agent application.
$Id$
"""
import sys
import os
from twisted.internet import reactor
from cybertools.agent.base.agent import Master
from cybertools.agent.crawl.base import Metadata, Resource
application = None # contains application object if started via twistd
def getConfig():
agentHome = os.path.abspath(os.path.dirname(__file__))
configName = 'transporter.cfg'
configFile = open(os.path.join(agentHome, configName))
config = configFile.read()
configFile.close()
return config
def setup(configInfo=None):
if configInfo is None:
configInfo = getConfig()
master = Master(configInfo)
setupEnvironment(master.config)
master.setup()
print 'Starting agent application...'
print 'Using controllers %s.' % ', '.join(master.config.controller.names)
return master
def setupEnvironment(config):
from cybertools.agent.base import agent, control, job, log, schedule
from cybertools.agent.core import agent, control, schedule
from cybertools.agent.control import cmdline
from cybertools.agent.system import rpcapi
rpcapi.setup(config)
from cybertools.agent.system import sftpapi
sftpapi.setup(config)
from cybertools.agent.transport import remote
def startReactor():
reactor.run()
print 'Agent application has been stopped.'
if __name__ == '__main__':
master = setup()
controller = master.controllers[0]
controller.createAgent('transport.remote', 'sample03')
metadata01 = Metadata(dict(filename='dummy.txt'))
res01 = Resource()
res01.metadata = metadata01
res01.path = 'data/file1.txt'
controller.enterJob('sample', 'sample03', params=dict(resource=res01))
startReactor()

10
agent/testing/outlook.cfg
View file

@ -1,10 +0,0 @@
#
# Standard configuration for agent application
#
# $Id: outlook.cfg 2496 2008-04-04 08:07:22Z helmutm $
#
controller(names=['core.sample'])
scheduler(name='core')
logger(name='default', standard=30)
system.winapi = 'use_outlook'

135
agent/testing/rpcserver.py
View file

@ -1,135 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Fake rpcserver for testing purposes
$Id$
"""
from twisted.internet.defer import succeed
class RPCServer(object):
serverURL = ''
method = ''
machineName = ''
userName = ''
password = ''
controller = ''
def __init__(self, serverURL = '', method = '', machineName = '',
userName = '', password = '', controlObj= None):
self.serverURL = serverURL
self.method = method
self.machineName = machineName
self.userName = userName
self.password = password
self.controller = controlObj
def getMetadata(self, metadata):
if self.controller is not None:
# pass metadata to controller
# this is done AFTER the resource (like e.g. file or mail)
# is handed over
pass
deferred = succeed('Metadata accepted by server')
return deferred
def xmlrpc_shutdownRPCServer():
return "xmlrRPC server shutdown completed!"
class XmlRpc(object):
Proxy = None
XMLRPC = None
Handler = None
XMLRPCIntrospection = None
QueryProtocol = None
_QueryFactory = None
def __init__(self):
self.Proxy = Proxy
#self.XMLRPC = XMLRPC()
#self.Handler = Handler()
#self.XMLRPCIntrospection = XMLRPCIntrospection()
#self.QueryProtocol = QueryProtocol()
#self._QueryFactory = _QueryFactory()
def addIntrospection(self, xmlrpc):
pass
class Proxy(object):
url = ''
user = None
password = None
allowNone = False
queryFactory = None
def __init__(self, url, user=None, password=None, allowNone=False):
self.url = url
self.user = user
self.password = password
self.allowNone = allowNone
self.RPCServer = RPCServer()
def callRemote(self, methodName, *params):
"""
intended to simulate the callRemote command of a real xmlrpcserver
that takes a method name and calls the method, returning the results
as xml formatted strings
"""
method = getattr(self.RPCServer, methodName)
return method(*params)
xmlrpc = XmlRpc()
class XMLRPC(object):
def __init__(self):
pass
class Handler(object):
def __init__(self):
pass
class XMLRPCIntrospection(object):
def __init__(self):
pass
class QueryProtocol(object):
def __init__(self):
pass
class _QueryFactory(object):
def __init__(self):
pass

37
agent/testing/sftp.py
View file

@ -1,37 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Fake sftp class for testing purposes
$Id$
"""
from twisted.internet.defer import succeed
class FileTransfer(object):
def __init__(self, host, port, username, password):
pass
def upload(self, localPath, remotePath):
deferred = succeed('Upload completed')
return deferred

80
agent/testing/test_rpcserver.py
View file

@ -1,80 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Providing access for remote agent instances by listening for requests
from remote transport agents.
$Id$
"""
from twisted.web import xmlrpc, server, resource
from twisted.internet import defer, reactor
from cybertools.agent.base.agent import Agent
application = None
class RPCServer(xmlrpc.XMLRPC):
serverURL = ''
method = ''
machineName = ''
userName = ''
password = ''
controller = ''
close = reactor.stop
def __init__(self, serverURL = '', method = '', machineName = '',
userName = '', password = '', controlObj= None):
self.serverURL = serverURL
self.method = method
self.machineName = machineName
self.userName = userName
self.password = password
self.controller = controlObj
xmlrpc.XMLRPC.__init__(self)
def xmlrpc_transfer(self, resource):
if self.controller is not None:
# pass resource object to controller
# this is done BEFORE the metadata is handed over
# call notify method of controller
pass
print resource
return "Resource received: ", resource
def xmlrpc_getMetadata(self, metadata):
if self.controller is not None:
# pass metadata to controller
# this is done AFTER the resource (like e.g. file or mail)
# is handed over
pass
print '*** metadata', metadata
metadata = "Echo: ", metadata
return metadata
def xmlrpc_shutdownRPCServer():
self.close()
if __name__ == '__main__':
from twisted.internet import reactor
site = RPCServer()
reactor.listenTCP(8082, server.Site(site))
print '*** listening...'
reactor.run()

18
agent/testing/test_sftp.py
View file

@ -1,18 +0,0 @@
from twisted.internet import reactor
from cybertools.agent.transport.file.sftp import FileTransfer
def output(x):
print x
ft = FileTransfer('cy05.de', 22, 'scrat', '...')
d = ft.upload('d:\\text2.rtf', 'text.txt')
d.addCallback(output)
reactor.callLater(21, ft.close)
reactor.callLater(32, reactor.stop)
reactor.run()

17
agent/testing/transporter.cfg
View file

@ -1,17 +0,0 @@
#
# sample.cfg - agent configuration for demonstration and testing purposes
#
# $Id$
#
# transportserver.xmlrpc='testing'
controller(names=['core.sample'])
scheduler(name='core')
logger(name='default', standard=30)
#transport.remote.server = 'testing'
transport.remote.url = 'http://localhost:8082'
transport.remote.ftp.url = 'cy05.de'
transport.remote.ftp.user = 'scrat'
transport.remote.ftp.password = '...'
transport.remote.sftp = 'http://cy05.de'
transport.remote.chunksize = 4096

240
agent/testing/winapi.py
View file

@ -1,240 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Fake Windows API functions for testing purposes.
$Id$
"""
win32api = win32process = win32con = None
class com_error(Exception):
pass
class Attachments(list):
elemCount = 0
data = []
def __init__(self, params=[]):
for elem in params:
fileitem = Attachment(filename=elem[0], ParentMail=elem[1])
self.data.append(fileitem)
print "Attachment: ", fileitem.FileName
@property
def Application(self):
print "Outlook application instance"
return "Outlook application instance"
def Item(self, index):
return self.data[index-1]
@property
def count(self):
return len(data)
def __len__(self):
return len(self.data)
def __iter__(self):
yield self.data
def __getitem__(self, idx):
return self.data[idx]
class Attachment(object):
File = ""
parentMailObject = None
def __init__(self, ParentMail, filename=""):
self.File = filename
self.parentMailObject = ParentMail
def SaveAsFile(self, path=""):
print "Attachment saved"
@property
def Parent(self):
" return value of Attribute Parent is of type _MailItem"
return self.parentMailObject
@property
def Type(self):
pass
@property
def Size(self):
# the size property is not available in Outlook 2000
pass
@property
def Application(self):
" Actual instance of Outlook application"
return None
@property
def FileName(self):
return self.File
class Mail(object):
#this is just a guess what a Outlook Mail Object Probably returns
#Class = client.constants.olMail
def __init__(self, subj="", sendName="", to="", body="", **kw):
self.Class = client.constants.olMail
self.Subject = subj
self.SenderName = sendName
self.To = to
self.Body = body
for k, v in kw.items():
setattr(self, k, v)
def addAttachment(self, **kw):
"""
this is a method which probably does not exist in a real mail
Currently this is a work around to add attachments to a mail
"""
for k, v in kw.items():
setattr(self, k, v)
@property
def _prop_map_get_(self):
#here it is necessary of what attributes (called keys in outlok.py)
#an Outlook Mail typically has
return self.__dict__
class Items(object):
temp = {}
data = []
def __init__(self):
self.data.append(Mail(subj="Python Training",
sendName="Mark Pilgrim",
to="allPythonics@python.org",
body="The training will take place on Wed, 21st Dec.\
Kindly check the enclosed invitation.",
BodyFormat=1
))
self.data[0].addAttachment(Attachments=Attachments([("Invitation.pdf", self.data[0]), ("21.pdf", self.data[0])]))
self.data.append(Mail(subj="Information Technolgies Inc. Test it!",
sendName="IT.org",
to="allUser@internet.com",
BodyFormat=2,
HTMLBody="<html>\
<head>\
<title>Test-HTML-Mail</title>\
</head>\
<body>\
<h1>Das ist eine HTML-Mail</h1>\
<div align='center'>Hier steht \
<b>Beispiel</b>-Text</div>\
</body>\
</html>",
SentOn="21.04.07"
))
self.data.append(Mail(subj="@ Product Details @",
sendName="",
senderEmailAddress="custominfo@enterprise.com",
to="recipient1@mail.com, customer@web.de",
BodyFormat=1,
body="Dear customer,\
Hereby we submit you the information you ordered.\
Please feel free to ask anytime you want.\
Sincerely, Customer Support",
SentOn="30.07.07"
))
def Item(self, idx):
return self.data[idx-1]
def __len__(self):
return len(self.data)
class OutlookFolder(object):
# Folders defines in Outlook the sub folders under the "Main" Folder
Folders = None
def __init__(self):
print "collecting Mails from folder"
self.Items = Items()
class OutlookNamespace(object):
def __init__(self):
pass
def GetDefaultFolder(self, message=""):
print "retrieving Outlook default folder"
folder = OutlookFolder()
return folder
class OutlookApp(object):
def __init__(self):
pass
def GetNamespace(self, message=""):
print "Namespace " + message + " retrieved"
oNamespace = OutlookNamespace()
return oNamespace
class Message(object):
olFolderInbox = None
# esp. for olMail, for further dummy implementations it is necessary
# to find out, what class is expected. Meaning what type of object has
# to be faked and what attributes it has. see outlook.py
# loadMailsfromFolder
olMail = Mail
def __init__(self):
pass
def EnsureDispatch(self, message=""):
print message + " retrieved"
oApp = OutlookApp()
return oApp
class client(object):
gencache = Message()
constants = Message()
def __init__(self):
pass
class ctypes(object):
def __init__(self):
pass

74
agent/tests.py
View file

@ -1,74 +0,0 @@
#! /usr/bin/env python
# $Id$
import os, time
import unittest, doctest
from zope.testing.doctestunit import DocFileSuite
from twisted.internet import reactor
#from twisted.internet.defer import Deferred
#from twisted.trial import unittest as trial_unittest
baseDir = os.path.dirname(__file__)
class Tester(object):
""" Used for controlled execution of reactor iteration cycles.
"""
stopped = False
def iterate(self, n=10, delay=0):
self.stopped = False
for i in range(n):
if self.stopped:
return
reactor.iterate(delay)
def run(self, maxduration=1.0, delay=0):
self.stopped = False
end = time.time() + maxduration
while not self.stopped:
reactor.iterate(delay)
if time.time() >= end:
return
def stop(self):
self.stopped = True
def stopThreads(self):
reactor.threadpool.stop()
reactor.threadpool = None
tester = Tester()
class Test(unittest.TestCase):
"Basic tests for the cybertools.agent package."
def setUp(self):
pass
def tearDown(self):
pass
def testBasicStuff(self):
pass
def test_suite():
flags = doctest.NORMALIZE_WHITESPACE | doctest.ELLIPSIS
testSuite = unittest.TestSuite((
unittest.makeSuite(Test),
DocFileSuite('README.txt', optionflags=flags),
DocFileSuite('crawl/README.txt', optionflags=flags),
DocFileSuite('crawl/filesystem.txt', optionflags=flags),
DocFileSuite('crawl/outlook.txt', optionflags=flags),
DocFileSuite('transport/transporter.txt', optionflags=flags),
DocFileSuite('talk/README.txt', optionflags=flags),
))
return testSuite
if __name__ == '__main__':
standard_unittest.main(defaultTest='test_suite')

4
agent/transport/__init__.py
View file

@ -1,4 +0,0 @@
"""
$Id$
"""

4
agent/transport/file/__init__.py
View file

@ -1,4 +0,0 @@
"""
$Id$
"""

160
agent/transport/file/sftp.py
View file

@ -1,160 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Transferring files to a remote site via SFTP.
$Id$
"""
from twisted.conch.ssh import channel, common, connection
from twisted.conch.ssh import filetransfer, transport, userauth
from twisted.internet import defer, protocol, reactor
CHUNKSIZE = 4096
class FileTransfer(protocol.ClientFactory):
""" Transfers files to a remote SCP/SFTP server.
"""
channel = None
def __init__(self, host, port, username, password):
self.username = username
self.password = password
self.queue = []
reactor.connectTCP(host, port, self)
def buildProtocol(self, addr):
protocol = self.protocol = ClientTransport(self)
return protocol
def upload(self, localPath, remotePath):
""" Copies a file, returning a deferred.
"""
d = self.deferred = defer.Deferred()
# we put everything in a queue so that more than one file may
# be transferred in one connection.
self.queue.append(dict(deferred=d,
command='upload',
localPath=localPath,
remotePath=remotePath))
if len(self.queue) == 1 and self.channel is not None:
# the channel has emptied the queue
self.channel.execute()
return d
def close(self):
# TODO: put in queue...
self.protocol.transport.loseConnection()
print 'connection closed'
class SFTPChannel(channel.SSHChannel):
""" An SSH channel using the SFTP subsystem for transferring files
and issuing other filesystem requests.
"""
name = 'session'
remFile = ''
remOffset = 0
def channelOpen(self, data):
d = self.conn.sendRequest(self, 'subsystem', common.NS('sftp'), wantReply=1)
d.addCallback(self.channelOpened)
def channelOpened(self, data):
self.client = filetransfer.FileTransferClient()
self.client.makeConnection(self)
self.dataReceived = self.client.dataReceived
self.execute()
self.conn.factory.channel = self
def execute(self):
queue = self.conn.factory.queue
if queue:
command = queue.pop()
commandName = command.pop('command')
method = getattr(self, 'command_' + commandName, None)
if method is not None:
self.params = command
method()
def command_upload(self):
params = self.params
remotePath = params['remotePath']
localPath = params['localPath']
self.localFile = open(localPath, 'rb')
d = self.client.openFile(remotePath,
filetransfer.FXF_WRITE | filetransfer.FXF_CREAT, {})
d.addCallbacks(self.writeChunk, self.logError)
def writeChunk(self, remoteFile):
if isinstance(remoteFile, tuple) == False:
self.remFile = remoteFile
data = self.localFile.read(CHUNKSIZE)
if len(data) < CHUNKSIZE:
self.d = self.remFile.writeChunk(self.remOffset, data)
self.d.addCallbacks(self.finished, self.logError)
else:
self.d = self.remFile.writeChunk(self.remOffset, data)
self.remOffset = self.remOffset + CHUNKSIZE
self.d.addCallbacks(self.writeChunk, self.logError)
def logError(self, reason):
print 'error', reason
def finished(self, result):
self.localFile.close()
self.remFile.close()
#self.d.callback('finished')
self.conn.factory.deferred.callback('finished')
# classes for managing the SSH protocol and connection
class ClientTransport(transport.SSHClientTransport):
def __init__(self, factory):
self.factory = factory
def verifyHostKey(self, pubKey, fingerprint):
# this is insecure!!!
return defer.succeed(True)
def connectionSecure(self):
self.requestService(UserAuth(self.factory, ClientConnection(self.factory)))
class ClientConnection(connection.SSHConnection):
def __init__(self, factory):
connection.SSHConnection.__init__(self)
self.factory = factory
def serviceStarted(self):
self.openChannel(SFTPChannel(conn=self))
class UserAuth(userauth.SSHUserAuthClient):
def __init__(self, factory, connection):
userauth.SSHUserAuthClient.__init__(self, factory.username, connection)
self.password = factory.password
def getPassword(self, prompt=None):
return defer.succeed(self.password)

27
agent/transport/loops.py
View file

@ -1,27 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Transferring information to a loops site on a local Zope instance.
$Id$
"""
from zope.interface import implements

92
agent/transport/remote.py
View file

@ -1,92 +0,0 @@
#
# Copyright (c) 2008 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Transferring information to or requesting information from a remote
cybertools.agent instance by transferring files to the remote system
and sending requests to a corresponding remote controller.
$Id$
"""
from twisted.internet import defer
from zope.interface import implements
import os
from cybertools.agent.system import xmlrpc
from cybertools.agent.system import sftp
from cybertools.agent.base.agent import Master
from cybertools.agent.core.agent import QueueableAgent
from cybertools.agent.interfaces import ITransporter
from cybertools.agent.crawl.base import Metadata
from cybertools.agent.crawl.mail import MailResource
from cybertools.agent.crawl.filesystem import FileResource
from cybertools.agent.components import agents
from cybertools.util.config import Configurator
class Transporter(QueueableAgent):
implements(ITransporter)
port = 22
machineName = ''
def __init__(self, master):
super(Transporter, self).__init__(master)
config = master.config
serverURL = config.transport.remote.url
self.server = xmlrpc.xmlrpc.Proxy(serverURL)
userName = config.transport.remote.ftp.user
password = config.transport.remote.ftp.password
host = config.transport.remote.ftp.url
self.ftpServer = sftp.FileTransfer(host, self.port, userName, password)
def process(self):
return self.transfer(self.params['resource'])
def transfer(self, resource):
""" Transfer the resource (an object providing IResource)
to the server and return a Deferred.
"""
self.deferred = defer.Deferred()
remoteFile = os.path.basename(resource.path)
d = self.ftpServer.upload(resource.path, remoteFile)
d.addErrback(self.errorHandler)
d.addCallback(lambda result:
self.server.callRemote('getMetadata', dict(resource.metadata)))
d.addCallback(self.transferDone)
return self.deferred
def errorHandler(self, errorInfo):
"""
Invoked as a callback from self.transfer
Error handler.
"""
print errorInfo
#self.server.close()
def transferDone(self, result):
"""
Invoked as a callback from self.transfer
This callback method is called when resource and metadata
have been transferred successfully.
"""
self.deferred.callback(result)
agents.register(Transporter, Master, name='transport.remote')

52
agent/transport/transporter.txt
View file

@ -1,52 +0,0 @@
================================================
Agents for Job Execution and Communication Tasks
================================================
($Id$)
>>> config = '''
... controller(names=['core.sample'])
... scheduler(name='core')
... logger(name='default', standard=30)
... transport.remote.server = 'testing'
... transport.remote.sftp = 'testing'
... transport.remote.url = 'http://localhost:8123'
... '''
>>> from cybertools.agent.main import setup
>>> master = setup(config)
Starting agent application...
Using controllers core.sample.
Transporter
===========
The agent uses Twisted's cooperative multitasking model.
The Transporter is used to contact an xmlrpc Server and transmit the metadata
to the other loops system. The Transporter is derived from Queueable agent
to ensure that only one item at a time is transmitted.
Returns a deferred that must be supplied with a callback method (and in
most cases also an errback method).
This Testcase is using subsidiary methods to simulate a real xmlrpc server.
>>> controller = master.controllers[0]
>>> controller.createAgent('transport.remote', 'sample03')
In the next step we request the start of a job, again via the controller.
>>> from cybertools.agent.crawl.base import Metadata, Resource
>>> md01 = Metadata(dict(filename='dummy.txt'))
>>> r01 = Resource()
>>> r01.metadata = md01
>>> r01.path = 'resource.txt'
>>> controller.enterJob('sample', 'sample03', params=dict(resource=r01))
The job is not executed immediately - we have to hand over control to
the twisted reactor first.
>>> from cybertools.agent.tests import tester
>>> tester.iterate()
Job 00001 completed; result: Metadata accepted by server;

4
agent/util/__init__.py
View file

@ -1,4 +0,0 @@
"""
$Id$
"""

368
agent/util/task.py
View file

@ -1,368 +0,0 @@
# -*- test-case-name: twisted.test.test_task -*-
# Copyright (c) 2001-2004 Twisted Matrix Laboratories.
# See LICENSE for details.
"""Scheduling utility methods and classes.
API Stability: Unstable
@author: U{Jp Calderone<mailto:exarkun@twistedmatrix.com>}
"""
__metaclass__ = type
import time
from twisted.python.runtime import seconds
from twisted.python import reflect
from twisted.internet import base, defer
class LoopingCall:
"""Call a function repeatedly.
@ivar f: The function to call.
@ivar a: A tuple of arguments to pass the function.
@ivar kw: A dictionary of keyword arguments to pass to the function.
If C{f} returns a deferred, rescheduling will not take place until the
deferred has fired. The result value is ignored.
"""
call = None
running = False
deferred = None
interval = None
count = None
starttime = None
def _callLater(self, delay):
from twisted.internet import reactor
return reactor.callLater(delay, self)
_seconds = staticmethod(seconds)
def __init__(self, f, *a, **kw):
self.f = f
self.a = a
self.kw = kw
def start(self, interval, now=True):
"""Start running function every interval seconds.
@param interval: The number of seconds between calls. May be
less than one. Precision will depend on the underlying
platform, the available hardware, and the load on the system.
@param now: If True, run this call right now. Otherwise, wait
until the interval has elapsed before beginning.
@return: A Deferred whose callback will be invoked with
C{self} when C{self.stop} is called, or whose errback will be
invoked when the function raises an exception or returned a
deferred that has its errback invoked.
"""
assert not self.running, ("Tried to start an already running "
"LoopingCall.")
if interval < 0:
raise ValueError, "interval must be >= 0"
self.running = True
d = self.deferred = defer.Deferred()
self.starttime = self._seconds()
self.count = 0
self.interval = interval
if now:
self()
else:
self._reschedule()
return d
def stop(self):
"""Stop running function.
"""
assert self.running, ("Tried to stop a LoopingCall that was "
"not running.")
self.running = False
if self.call is not None:
self.call.cancel()
self.call = None
d, self.deferred = self.deferred, None
d.callback(self)
def __call__(self):
def cb(result):
if self.running:
self._reschedule()
else:
d, self.deferred = self.deferred, None
d.callback(self)
def eb(failure):
self.running = False
d, self.deferred = self.deferred, None
d.errback(failure)
self.call = None
d = defer.maybeDeferred(self.f, *self.a, **self.kw)
d.addCallback(cb)
d.addErrback(eb)
def _reschedule(self):
if self.interval == 0:
self.call = self._callLater(0)
return
fromNow = self.starttime - self._seconds()
while self.running:
self.count += 1
fromStart = self.count * self.interval
delay = fromNow + fromStart
if delay > 0:
self.call = self._callLater(delay)
return
def __repr__(self):
if hasattr(self.f, 'func_name'):
func = self.f.func_name
if hasattr(self.f, 'im_class'):
func = self.f.im_class.__name__ + '.' + func
else:
func = reflect.safe_repr(self.f)
return 'LoopingCall<%r>(%s, *%s, **%s)' % (
self.interval, func, reflect.safe_repr(self.a),
reflect.safe_repr(self.kw))
class SchedulerStopped(Exception):
"""
The operation could not complete because the scheduler was stopped in
progress or was already stopped.
"""
class _Timer(object):
MAX_SLICE = 0.01
def __init__(self):
self.end = time.time() + self.MAX_SLICE
def __call__(self):
return time.time() >= self.end
_EPSILON = 0.00000001
def _defaultScheduler(x):
from twisted.internet import reactor
return reactor.callLater(_EPSILON, x)
class Cooperator(object):
"""
Cooperative task scheduler.
"""
def __init__(self,
terminationPredicateFactory=_Timer,
scheduler=_defaultScheduler,
started=True):
"""
Create a scheduler-like object to which iterators may be added.
@param terminationPredicateFactory: A no-argument callable which will
be invoked at the beginning of each step and should return a
no-argument callable which will return False when the step should be
terminated. The default factory is time-based and allows iterators to
run for 1/100th of a second at a time.
@param scheduler: A one-argument callable which takes a no-argument
callable and should invoke it at some future point. This will be used
to schedule each step of this Cooperator.
@param started: A boolean which indicates whether iterators should be
stepped as soon as they are added, or if they will be queued up until
L{Cooperator.start} is called.
"""
self.iterators = []
self._metarator = iter(())
self._terminationPredicateFactory = terminationPredicateFactory
self._scheduler = scheduler
self._delayedCall = None
self._stopped = False
self._started = started
def coiterate(self, iterator, doneDeferred=None):
"""
Add an iterator to the list of iterators I am currently running.
@return: a Deferred that will fire when the iterator finishes.
"""
if doneDeferred is None:
doneDeferred = defer.Deferred()
if self._stopped:
doneDeferred.errback(SchedulerStopped())
return doneDeferred
self.iterators.append((iterator, doneDeferred))
self._reschedule()
return doneDeferred
def _tasks(self):
terminator = self._terminationPredicateFactory()
while self.iterators:
for i in self._metarator:
yield i
if terminator():
return
self._metarator = iter(self.iterators)
def _tick(self):
"""
Run one scheduler tick.
"""
self._delayedCall = None
for taskObj in self._tasks():
iterator, doneDeferred = taskObj
try:
result = iterator.next()
except StopIteration:
self.iterators.remove(taskObj)
doneDeferred.callback(iterator)
except:
self.iterators.remove(taskObj)
doneDeferred.errback()
else:
if isinstance(result, defer.Deferred):
self.iterators.remove(taskObj)
def cbContinue(result, taskObj=taskObj):
self.coiterate(*taskObj)
result.addCallbacks(cbContinue, doneDeferred.errback)
self._reschedule()
_mustScheduleOnStart = False
def _reschedule(self):
if not self._started:
self._mustScheduleOnStart = True
return
if self._delayedCall is None and self.iterators:
self._delayedCall = self._scheduler(self._tick)
def start(self):
"""
Begin scheduling steps.
"""
self._stopped = False
self._started = True
if self._mustScheduleOnStart:
del self._mustScheduleOnStart
self._reschedule()
def stop(self):
"""
Stop scheduling steps. Errback the completion Deferreds of all
iterators which have been added and forget about them.
"""
self._stopped = True
for iterator, doneDeferred in self.iterators:
doneDeferred.errback(SchedulerStopped())
self.iterators = []
if self._delayedCall is not None:
self._delayedCall.cancel()
self._delayedCall = None
_theCooperator = Cooperator()
def coiterate(iterator):
"""
Cooperatively iterate over the given iterator, dividing runtime between it
and all other iterators which have been passed to this function and not yet
exhausted.
"""
return _theCooperator.coiterate(iterator)
class Clock:
"""
Provide a deterministic, easily-controlled implementation of
L{IReactorTime.callLater}. This is commonly useful for writing
deterministic unit tests for code which schedules events using this API.
"""
rightNow = 0.0
def __init__(self):
self.calls = []
def seconds(self):
"""
Pretend to be time.time(). This is used internally when an operation
such as L{IDelayedCall.reset} needs to determine a a time value
relative to the current time.
@rtype: C{float}
@return: The time which should be considered the current time.
"""
return self.rightNow
def callLater(self, when, what, *a, **kw):
"""
See L{twisted.internet.interfaces.IReactorTime.callLater}.
"""
self.calls.append(
base.DelayedCall(self.seconds() + when,
what, a, kw,
self.calls.remove,
lambda c: None,
self.seconds))
self.calls.sort(lambda a, b: cmp(a.getTime(), b.getTime()))
return self.calls[-1]
def advance(self, amount):
"""
Move time on this clock forward by the given amount and run whatever
pending calls should be run.
@type amount: C{float}
@param amount: The number of seconds which to advance this clock's
time.
"""
self.rightNow += amount
while self.calls and self.calls[0].getTime() <= self.seconds():
call = self.calls.pop(0)
call.called = 1
call.func(*call.args, **call.kw)
def pump(self, timings):
"""
Advance incrementally by the given set of times.
@type timings: iterable of C{float}
"""
for amount in timings:
self.advance(amount)
__all__ = [
'LoopingCall',
'Clock',
'SchedulerStopped', 'Cooperator', 'coiterate',
]

2
relation/tests.py
View file

@ -1,6 +1,6 @@
# $Id$ # $Id$
import unittest import unittest, doctest
from zope.app.testing import ztapi from zope.app.testing import ztapi
from zope.interface.verify import verifyClass from zope.interface.verify import verifyClass
from zope.interface import implements from zope.interface import implements

62
twisted/README.txt
View file

@ -1,62 +0,0 @@
===============================
Zope 3 extensions using Twisted
===============================
$Id$
manhole
=======
A simple twisted manhole that allows you to access a running Zope 3
instance via a python command line without having to run ZEO.
You may start it for testing purposes via `python manhole.py` (note that
the twisted library must be reachable via your PYTHONPATH) and log in
from another console window using `ssh -p 5001 admin@localhost`. The
password is defined in the "reactor.listenTCP()" statement of the
manhole.py script.
Note that this will open up a serious security hole on your computer
as now anybody knowing this password may login from remote to the Python
console and get full access to the system with the permissions of the user
running the manhole script.
The script may be stopped with Ctrl-C.
In order to use it with Zope copy the cybertools.twisted-configure.zcml
to the etc/package-includes directory of your Zope instance and restart
Zope. You can then log in with ssh like shown above, using the username
and password of the zope.manager principal defined in your principals.zcml.
After logging in use the `help` command to get more information.
Dependencies
------------
Zope 3.2+ with Twisted as server component
PyOpenSSL: http://pyopenssl.sourceforge.net
PyCrypto: http://www.amk.ca/python/code/crypto.html
Installation
------------
Create a directory `cybertools` somewhere in your Python path, typically
in lib/python of your Zope instance, and put an empty __init__.py there.
In this directory, check out the the cybertools.twisted package:
svn co svn://svn.cy55.de/Zope3/src/cybertools/trunk/twisted
In order to use it with Zope copy the cybertools.twisted-configure.zcml
to the etc/package-includes directory of your Zope instance and restart
Zope.
Acknowledgements
================
Thanks to Abe Fettig who provides a good introduction to Twisted and some
of the code used for this package with his book
"Twisted Network Programming Essentials".

4
twisted/__init__.py
View file

@ -1,4 +0,0 @@
"""
$Id$
"""

18
twisted/configure.zcml
View file

@ -1,18 +0,0 @@
<!-- $Id$ -->
<configure
xmlns="http://namespaces.zope.org/zope"
xmlns:browser="http://namespaces.zope.org/browser"
i18n_domain="zope"
>
<browser:page
for="zope.app.applicationcontrol.interfaces.IApplicationControl"
name="manhole_control.html"
template="manhole_control.pt"
class=".manhole_control.ManholeControlView"
permission="zope.ManageApplication"
menu="zmi_views" title="Manhole Control"
/>
</configure>

1
twisted/cybertools.twisted-configure.zcml
View file

@ -1 +0,0 @@
<include package="cybertools.twisted" />

72
twisted/deferring.py
View file

@ -1,72 +0,0 @@
#
# Copyright (c) 2007 Helmut Merz helmutm@cy55.de
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
#
"""
Objects that defer attribute access.
$Id$
"""
from twisted.internet import reactor
from twisted.internet.defer import Deferred
class Deferring(Deferred):
def __init__(self, context):
self.context = context
Deferred.__init__(self)
def __getattr__(self, attr):
d = Deferring(self)
reactor.callLater(0, d.getattr, self, attr)
return d
def __call__(self, *args, **kw):
d = Deferring(None)
reactor.callLater(0, d.call, self, *args, **kw)
return d
def __repr__(self):
return '<Deferring on %s>' % repr(self.context)
__str__ = __repr__
def makeCallback(self, method, *args, **kw):
#print '*** makeCallback', self, method, args, kw
def cb(result, method=method, args=args, kw=kw):
value = method(result, *args, **kw)
#print '*** cb', self, result, method, args, kw, value
self.callback(value)
return cb
def getattr(self, deferring, attr):
ctx = deferring.context
if isinstance(ctx, Deferring):
ctx.addCallback(self.makeCallback(self.getattr, attr))
else:
value = getattr(deferring.context, attr)
self.context = value
self.callback(value)
def call(self, deferring, *args, **kw):
ctx = deferring.context
value = ctx(*args, **kw)
self.context = value
self.callback(value)

55
twisted/deferring.txt
View file

@ -1,55 +0,0 @@
======================
Working with Deferreds
======================
$Id$
>>> from cybertools.twisted.tests import tester
>>> from cybertools.twisted.deferring import Deferring
>>> class Demo(object):
... color = 'green'
>>> class Demo2(object):
... color = 'red'
... other = Demo()
>>> demo = Demo()
>>> demoDef = Deferring(demo)
>>> col = demoDef.color
>>> col
<Deferring on ...>
>>> print col
<Deferring on ...>
>>> def printIt(result):
... print 'result:', str(result)
>>> col.addCallback(printIt)
<Deferring on ...>
>>> tester.iterate()
result: green
>>> col
<Deferring on 'green'>
>>> demo2 = Demo2()
>>> demo2Def = Deferring(demo2)
>>> other = demo2Def.other
>>> other
<Deferring on <Deferring on ...>>
>>> ocol = other.color
>>> ocol.addCallback(printIt)
<Deferring on ...>
>>> tester.iterate()
result: green
>>> ocol
<Deferring on 'green'>
>>> other
<Deferring on <Demo object ...>>

157
twisted/manhole.py
View file

@ -1,157 +0,0 @@
"""
A simple twisted manhole that allows you to access a running Zope 3
instance via a python command line without having to run ZEO.
You may run it for testing purposes via `python manhole.py` (note that
the twisted library must be reachable via your PYTHONPATH) and log in
from another console window using `ssh -p 5001 admin@localhost`. The
password is defined below in the "reactor.listenTCP()" statement. The
manhole script may be stopped with Ctrl-C.
Note that this will open up a serious security hole on your computer
as now anybody knowing this password may login to the Python console
and get full access to the system with the permissions of the user
running the manhole script.
In order to use it with Zope copy the cybertools.twisted-configure.zcml
to the etc/package-includes directory of your Zope instance and restart
Zope. Open the manhole via the "Manhole Control" Tab of the "Manage process"
menu.
You can then log in with ssh like shown above, using the username
and password of the zope.manager principal defined in your principals.zcml.
After logging in use the `help` command to get more information.
$Id$
"""
from twisted.internet import reactor, protocol, defer
from twisted.protocols import basic
from twisted.cred import portal, checkers, credentials, error as credError
from twisted.conch import manhole as manhole, manhole_ssh
from zope.interface import implements
try:
from zope.app.publication.zopepublication import ZopePublication
from zope.app.component.hooks import setSite
from zope.app.security.principalregistry import principalRegistry
from zope.app.security import settings
from zope.app.security.interfaces import IAuthentication
from zope.app.securitypolicy.principalrole import principalRoleManager
from zope.app import zapi
import transaction
hasZope = True
except:
hasZope = False
import time
import sys
from cStringIO import StringIO
listener = None
factory = None
printLog = None
port = 5001
def getManholeFactory(namespace, **passwords):
realm = manhole_ssh.TerminalRealm()
def getManhole(_):
#return manhole.ColoredManhole(namespace)
return manhole.Manhole(namespace)
realm.chainedProtocolFactory.protocolFactory = getManhole
p = portal.Portal(realm)
checker = (hasZope and ZopeManagerChecker() or
checkers.InMemoryUsernamePasswordDatabaseDontUse(**passwords))
p.registerChecker(checker)
return manhole_ssh.ConchFactory(p)
class ZopeManagerChecker(object):
implements(checkers.ICredentialsChecker)
credentialInterfaces = (credentials.IUsernamePassword,)
def requestAvatarId(self, credentials):
login = credentials.username
password = credentials.password
# TODO: This should be based on the official Zope API stuff, e.g. via:
#principalRegistry = zapi.getUtility(IAuthentication)
principal = principalRegistry.getPrincipalByLogin(login)
if principal.validate(password):
roles = principalRoleManager.getRolesForPrincipal(principal.id)
for role, setting in roles:
if role == 'zope.Manager' and setting == settings.Allow:
return defer.succeed(login)
return defer.fail(credError.UnauthorizedLogin(
'Insufficient permissions'))
return defer.fail(credError.UnauthorizedLogin(
'User/password not correct'))
def printTime():
global printLog
print '***', time.strftime('%H:%M:%S'), '- twisted.manhole open ***'
printLog = reactor.callLater(600, printTime)
class Help(object):
def __repr__(self):
info = """
Use `dir()` to see what variables and functions are available.
"""
zopeInfo = """
You may use `x = zapi.traverse(root, 'path/to/object')` to get an
object in your folder hierarchy. Then you may call any method or
access any attribute of this object.
In order to get access to local utilities and adapters you may
issue a `setSite(root)`. Don't forget to call `setSite()` before
finishing your session in order to reset this setting.
If you change an object stored in the ZODB you should issue a
`transaction.commit()` to make your changes permanent or a
`transaction.abort()` to undo them.
"""
return info + (hasZope and zopeInfo or '')
def __call__(self):
print self
help = Help()
def open(port=5001, request=None):
global hasZope, factory, listener
printTime()
d = globals()
if hasZope and request is not None:
database = request.publication.db
connection = database.open()
root = connection.root()[ZopePublication.root_name]
else:
hasZope = False
d.update(locals())
namespace = {}
for key in ('__builtins__', 'connection', 'event', 'setSite', 'hasZope',
'zapi', 'transaction', 'root', '__doc__', 'help',
'manholeFactory', 'context'):
if key in d:
namespace[key] = d[key]
# TODO: get admin password from somewhere else or use a real checker.
factory = getManholeFactory(namespace, admin='aaa')
listener = reactor.listenTCP(port, factory)
def close():
global listener
listener.stopListening()
listener = None
if __name__ == '__main__':
port = 5001
if len(sys.argv) > 1:
port = int(sys.argv[-1])
startup(port=port)
reactor.run()

43
twisted/manhole_control.pt
View file

@ -1,43 +0,0 @@
<html metal:use-macro="context/@@standard_macros/view"
i18n:domain="zope">
<head>
<title i18n:translate="">ZODB Controller</title>
</head>
<body>
<div metal:fill-slot="body">
<tal:update condition="view/update" />
<div class="row" tal:condition="view/isOpen">
<div class="field" i18n:translate="">Manhole is open.</div>
</div>
<div class="row" tal:condition="not:view/isOpen">
<div class="field" i18n:translate="">Manhole is closed.</div>
</div>
<form action="." method="post" tal:attributes="action request/URL">
<input type="hidden" name="form_submitted" value="true" />
<div class="row">
<span class="label"
i18n:translate="">Port</span>:
<input type="text" size="5" name="manhole.port"
tal:attributes="value view/port" />
</div>
<div class="row">
<input type="checkbox" size="5" name="manhole.setting" value="true" />
<span class="label" i18n:translate=""
tal:condition="not:view/isOpen">Open manhole</span>
<span class="label" i18n:translate=""
tal:condition="view/isOpen">Close manhole</span>
</div>
<div class="row">
<div class="controls">
<input type="submit" name="APPLY" value="Apply"
i18n:attributes="value apply-button"/>
</div>
</div>
</form>
</div>
</body>
</html>

43
twisted/manhole_control.py
View file

@ -1,43 +0,0 @@
##############################################################################
#
# Copyright (c) 2001, 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
""" Manhole Control View
$Id$
"""
__docformat__ = 'restructuredtext'
from cybertools.twisted import manhole
class ManholeControlView(object):
def isOpen(self):
return manhole.listener is not None
def toggleState(self):
if self.isOpen():
manhole.close()
else:
manhole.open(self.port(), self.request)
def port(self):
return manhole.port
def update(self):
if not self.request.get('form_submitted'):
return
port = self.request.get('manhole.port')
if port:
manhole.port = int(port)
if self.request.get('manhole.setting', False):
self.toggleState()

36
twisted/tests.py
View file

@ -1,36 +0,0 @@
# $Id$
import unittest
import doctest
from twisted.internet import reactor
import cybertools.twisted.deferring
class Tester(object):
def iterate(self, n=10, delays={}):
for i in range(n):
delay = delays.get(i, 0)
reactor.iterate(delay)
tester = Tester()
class Test(unittest.TestCase):
"Basic tests for modules in the util package."
def testBasicStuff(self):
pass
def test_suite():
flags = doctest.NORMALIZE_WHITESPACE | doctest.ELLIPSIS
return unittest.TestSuite((
#unittest.makeSuite(Test), # we don't need this
#doctest.DocTestSuite(cybertools.util.property, optionflags=flags),
doctest.DocFileSuite('deferring.txt', optionflags=flags),
))
if __name__ == '__main__':
unittest.main(defaultTest='test_suite')

4
twisted/z2/__init__.py
View file

@ -1,4 +0,0 @@
"""
$Id$
"""

11
twisted/z2/configure.zcml
View file

@ -1,11 +0,0 @@
<!-- $Id$ -->
<configure
xmlns:zope="http://namespaces.zope.org/zope"
xmlns:browser="http://namespaces.zope.org/browser">
<zope:subscriber
for="zope.app.appsetup.interfaces.IDatabaseOpenedWithRootEvent"
handler="cybertools.twisted.z2.startup.startup" />
</configure>

38
twisted/z2/startup.py
View file

@ -1,38 +0,0 @@
"""
Code to be called at Zope startup.
$Id$
"""
from twisted.internet import reactor
from OFS.Application import AppInitializer
from zope.event import notify
from zope.app.appsetup.interfaces import DatabaseOpenedWithRoot
def startup(event):
print '*** Database opened: %s' % event.database
#twisted_target(event.database)
base_install_standards = AppInitializer.install_standards
def install_standards(self):
result = base_install_standards(self)
notify(DatabaseOpenedWithRoot(self.getApp()._p_jar.db()))
return result
#AppInitializer.install_standards = install_standards
#print '*** AppInitializer monkey patch installed'
def twisted_target(db):
reactor.callLater(5, show_application, db)
def show_application(db):
c = db.open()
print '*** Application: %s' % c.root()['Application']
c.close()
reactor.callLater(5, show_application, db)