Commit ae96e2a6 authored by Jérome Perrin's avatar Jérome Perrin

runUnitTest: introduce --processing_node_loop=timerserver

This allows running tests using a timer server loop, that will invoke
process timer on all subscribers.
The need for this is to test a custom tool using timer server.
parent 2015426c
......@@ -390,3 +390,26 @@ class ProcessingNodeTestCase(ZopeTestCase.TestCase):
LOG('Invoking Activity Tool', ERROR, '', error=True)
except KeyboardInterrupt:
pass
def timerserver(self):
"""Main loop using timer server.
"""
import Products.TimerService
timerserver_thread = None
try:
while not Lifetime._shutdown_phase:
time.sleep(.3)
transaction.begin()
try:
self.portal = self.app[self.app.test_portal_name]
except (AttributeError, KeyError):
continue
self._setUpDummyMailHost()
if not timerserver_thread:
timerserver_thread = Products.TimerService.timerserver.TimerServer.TimerServer(
module='Zope2',
interval=0.1,
)
except KeyboardInterrupt:
pass
......@@ -141,6 +141,16 @@ Options:
activities.
--zeo_server=[[HOST:]PORT] Bind the ZEO server to the given host/port.
--zeo_client=[HOST:]PORT Use specified ZEO server as storage.
--processing_node_loop=LOOP
Make ZEO clients execute the given loop, one of:
- processing_node: process activities only.
this is the default.
- timerserver: start a timer server thread,
which will typically execute activities,
alarms and everything else registered on
timer service.
This option only makes sense with --activity_node=
or when not specifying a test to run.
--zserver=ADDRESS[,...] Make ZServer listen on given IPv4 address.
Adresses can be given in the following syntaxs:
- HOST:PORT
......@@ -638,7 +648,8 @@ def runUnitTestList(test_list, verbosity=1, debug=0, run_only=None):
if zeo_server_pid == 0:
suite = ZEOServerTestCase('asyncore_loop')
elif node_pid_list is None or not test_list:
suite = ProcessingNodeTestCase('processing_node')
processing_node_loop = os.environ.get('processing_node_loop', 'processing_node')
suite = ProcessingNodeTestCase(processing_node_loop)
if not (dummy or load):
_print('WARNING: either --save or --load should be used because static'
' files are only reloaded by the node installing business'
......@@ -763,6 +774,7 @@ def main(argument_list=None):
"live_instance=",
"zeo_client=",
"zeo_server=",
"processing_node_loop=",
"zserver=",
"zserver_frontend_url=",
"neo_storage",
......@@ -878,6 +890,8 @@ def main(argument_list=None):
os.environ["zeo_client"] = arg
elif opt == "--zeo_server":
os.environ["zeo_server"] = arg
elif opt == "--processing_node_loop":
os.environ["processing_node_loop"] = arg
elif opt == "--zserver":
os.environ["zserver"] = arg
elif opt == "--zserver_frontend_url":
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment