Commit 1c5710e4 authored by Thomas Gambier's avatar Thomas Gambier 🚴🏼

tests: Teach test driver to pass testWendelinCore when run with wendelin.core 2

See merge request !1445
parents b2c47ad2 171a66f2
......@@ -29,6 +29,7 @@ except KeyError:
data_fs_path = os.environ.get('erp5_tests_data_fs_path',
os.path.join(instance_home, 'var', 'Data.fs'))
with_wendelin_core = int(os.environ.get('with_wendelin_core', 0))
load = int(os.environ.get('erp5_load_data_fs', 0))
save = int(os.environ.get('erp5_save_data_fs', 0))
save_mysql = int(os.environ.get('erp5_dump_sql') or not zeo_client) or None
......@@ -83,12 +84,16 @@ else:
zeo_server_pid = None
node_pid_list = []
in_forked_process = False
def fork():
global in_forked_process
pid = os.fork()
if pid:
# make sure parent and child have 2 different RNG
instance_random.seed(instance_random.random())
else:
in_forked_process = True
return pid
def forkNodes():
......@@ -173,5 +178,18 @@ else:
except TypeError: # BBB: ZEO<5
Storage = ClientStorage(zeo_client)
# launch WCFS server if wendelin.core usage is requested and we are running
# with wendelin.core 2.
wcfs_server = None
if with_wendelin_core and not in_forked_process:
try:
from wendelin import wcfs
except ImportError:
pass # wendelin.core 1
else:
from wendelin.lib.zodb import zstor_2zurl
zurl = zstor_2zurl(Storage)
wcfs_server = wcfs.start(zurl)
if node_pid_list is not None:
_print("Instance at %r loaded ... " % instance_home)
......@@ -159,6 +159,8 @@ Options:
extend sys.path
--instance_home=PATH Create/use test instance in given path
--log_directory=PATH Create log files in given path
--with_wendelin_core Pass for tests that use wendelin.core.
This option is likely to become a noop in the future.
When no unit test is specified, only activities are processed.
"""
......@@ -602,13 +604,14 @@ def runUnitTestList(test_list, verbosity=1, debug=0, run_only=None):
from Products.ERP5Type.tests.utils import DbFactory
root_db_name, = cfg.dbtab.databases.keys()
DbFactory(root_db_name).addMountPoint('/')
db_factory = DbFactory(root_db_name)
db_factory.addMountPoint('/')
TestRunner = unittest.TextTestRunner
import Lifetime
from Zope2.custom_zodb import Storage, save_mysql, \
node_pid_list, neo_cluster, zeo_server_pid
node_pid_list, neo_cluster, zeo_server_pid, wcfs_server
def shutdown(signum, frame, signum_set=set()):
Lifetime.shutdown(0)
signum_set.add(signum)
......@@ -681,6 +684,7 @@ def runUnitTestList(test_list, verbosity=1, debug=0, run_only=None):
raise
finally:
ProcessingNodeTestCase.unregisterNode()
db_factory.close()
Storage.close()
if node_pid_list is not None:
# Wait that child processes exit. Stop ZEO storage (if any) after all
......@@ -694,6 +698,10 @@ def runUnitTestList(test_list, verbosity=1, debug=0, run_only=None):
os.waitpid(zeo_server_pid, 0)
if neo_cluster:
neo_cluster.stop()
if wcfs_server is not None:
# Stop WCFS server (if any) after all Zope nodes are stopped and
# disconnected from it.
wcfs_server.stop()
if coverage_config:
coverage_process.stop()
......@@ -761,7 +769,8 @@ def main(argument_list=None):
"products_path=",
"sys_path=",
"instance_home=",
"log_directory="
"log_directory=",
"with_wendelin_core"
])
except getopt.GetoptError, msg:
usage(sys.stderr, msg)
......@@ -883,6 +892,8 @@ def main(argument_list=None):
instance_home = os.path.abspath(arg)
elif opt == "--log_directory":
_log_directory = os.path.abspath(arg)
elif opt == "--with_wendelin_core":
os.environ["with_wendelin_core"] = "1"
bt5_path_list += filter(None,
os.environ.get("erp5_tests_bt5_path", "").split(','))
......
# -*- coding: utf-8 -*-
from glob import glob
import os, subprocess, re
# test_suite is provided by 'run_test_suite'
......@@ -89,6 +90,19 @@ class ERP5(_ERP5):
if not status_dict['status_code']:
status_dict = self.runUnitTest('--load', '--activity_node=2', full_test)
return status_dict
elif test.startswith('testWendelinCore'):
# Combining Zope and WCFS working together requires data to be on a real
# storage, not on in-RAM MappingStorage inside Zope's Python process.
# Force this via --load --save for now.
#
# Also manually indicate via --with_wendelin_core, that this test needs
# WCFS server - corresponding to ZODB test storage - to be launched.
#
# In the future we might want to rework custom_zodb.py to always use
# FileStorage on tmpfs instead of δ=MappingStorage in DemoStorage(..., δ),
# and to always spawn WCFS for all tests, so that this hack becomes
# unnecessary.
return self.runUnitTest('--load', '--save', '--with_wendelin_core', full_test)
elif test.startswith('testFunctional'):
return self._updateFunctionalTestResponse(self.runUnitTest(full_test))
elif test == 'testUpgradeInstanceWithOldDataFs':
......
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