Commit ba22def5 authored by Julien Muchembled's avatar Julien Muchembled

Unit tests: start webdav server

git-svn-id: https://svn.erp5.org/repos/public/erp5/trunk@40394 20353a03-c40f-0410-a6d1-a30d3c3de9de
parent 8e7e6d90
...@@ -917,7 +917,7 @@ class ERP5TypeTestCase(ProcessingNodeTestCase, PortalTestCase): ...@@ -917,7 +917,7 @@ class ERP5TypeTestCase(ProcessingNodeTestCase, PortalTestCase):
from Products import DeadlockDebugger from Products import DeadlockDebugger
except ImportError: except ImportError:
pass pass
self.serverhost, self.serverport = self.startZServer() self.serverhost, self.serverport = self.startZServer(verbose=True)
self._registerNode(distributing=1, processing=1) self._registerNode(distributing=1, processing=1)
self._updateConnectionStrings() self._updateConnectionStrings()
......
...@@ -85,16 +85,24 @@ class ProcessingNodeTestCase(backportUnittest.TestCase, ZopeTestCase.TestCase): ...@@ -85,16 +85,24 @@ class ProcessingNodeTestCase(backportUnittest.TestCase, ZopeTestCase.TestCase):
pass pass
Lifetime.graceful_shutdown_loop() Lifetime.graceful_shutdown_loop()
def startZServer(self): def startZServer(self, verbose=False):
"""Start HTTP ZServer in background""" """Start HTTP ZServer in background"""
utils = ZopeTestCase.utils utils = ZopeTestCase.utils
if utils._Z2HOST is None: if utils._Z2HOST is None:
_print = lambda hs: verbose and ZopeTestCase._print(
"Running %s server at %s:%s\n" % (
hs.server_protocol, hs.server_name, hs.server_port))
try: try:
hs = createZServer() hs = createZServer()
except RuntimeError, e: except RuntimeError, e:
ZopeTestCase._print(str(e)) ZopeTestCase._print(str(e))
else: else:
utils._Z2HOST, utils._Z2PORT = hs.server_name, hs.server_port utils._Z2HOST, utils._Z2PORT = hs.server_name, hs.server_port
_print(hs)
try:
_print(createZServer(zserver_type='webdav'))
except RuntimeError, e:
ZopeTestCase._print(str(e))
t = Thread(target=Lifetime.loop) t = Thread(target=Lifetime.loop)
t.setDaemon(1) t.setDaemon(1)
t.start() t.start()
......
...@@ -297,19 +297,25 @@ def parseListeningAddress(host_port=None, default_host='127.0.0.1'): ...@@ -297,19 +297,25 @@ def parseListeningAddress(host_port=None, default_host='127.0.0.1'):
raise RuntimeError("Can't find free port (tried ports %s)\n" raise RuntimeError("Can't find free port (tried ports %s)\n"
% ', '.join(map(str, port_list))) % ', '.join(map(str, port_list)))
def createZServer(log=os.devnull): def createZServer(log=os.devnull, zserver_type='http'):
from ZServer import logger, zhttp_server, zhttp_handler import ZServer
lg = logger.file_logger(log) if zserver_type == 'http':
class new_zhttp_server: zserver_class, zhandler_class = ZServer.zhttp_server, ZServer.zhttp_handler
elif zserver_type == 'webdav':
from ZServer.HTTPServer import zwebdav_server as zserver_class
from ZServer.WebDAVSrcHandler import WebDAVSrcHandler as zhandler_class
else:
raise NotImplementedError
lg = ZServer.logger.file_logger(log)
class new_zserver(zserver_class):
# I can't use __new__ because zhttp_handler is an old-style class :( # I can't use __new__ because zhttp_handler is an old-style class :(
def __init__(self): def __init__(self):
self.__class__ = zhttp_server self.__class__, = self.__class__.__bases__
for ip, port in parseListeningAddress(os.environ.get('zserver')): for ip, port in parseListeningAddress(os.environ.get('zserver')):
hs = new_zhttp_server() hs = new_zserver()
try: try:
hs.__init__(ip, port, resolver=None, logger_object=lg) hs.__init__(ip, port, resolver=None, logger_object=lg)
hs.install_handler(zhttp_handler(module='Zope2', uri_base='')) hs.install_handler(zhandler_class(module='Zope2', uri_base=''))
sys.stderr.write("Running ZServer at %s:%s\n" % (ip, port))
return hs return hs
except socket.error, e: except socket.error, e:
if e[0] != errno.EADDRINUSE: if e[0] != errno.EADDRINUSE:
......
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