Commit f0023bef authored by Łukasz Nowak's avatar Łukasz Nowak

Implement directory based JSON task distribution tool.

JSON allows to exchange python dictionaries in safe and simple way.

So thanks to using directory where test results are posted, it is possible that call
external tool will call erunTestSuite, which would later attach to such directory
"x" and do "something" more with stored test results.
parent cac06a88
import json
import threading
import os
class DummyTaskDistributionTool(object):
......@@ -25,4 +27,46 @@ class DummyTaskDistributionTool(object):
self.lock.release()
def stopUnitTest(self, test_path, status_dict):
pass
\ No newline at end of file
pass
def jsondump(obj, fp):
json.dump(obj, fp, indent=4, sort_keys=True)
class JSONFSTaskDistributionTool(object):
"""Uses directory to store information"""
def __init__(self, directory):
if not os.path.isdir(directory):
raise ValueError('Directory %r does not exists.' % directory)
self.directory = os.path.abspath(directory)
self.lock = threading.Lock()
self.file_index = 1
def createTestResult(self, name, revision, test_name_list, allow_restart,
*args):
jsondump(dict(name=name, revision=revision, test_name_list=test_name_list,
allow_restart=allow_restart, args=args),
open(os.path.join(self.directory, 'createTestResult.json'), 'w'))
self.test_name_list = list(test_name_list)
return self.directory, revision
def updateTestResult(self, name, revision, test_name_list):
raise NotImplementedError
def startUnitTest(self, test_result_path, exclude_list=()):
self.lock.acquire()
try:
for i, test in enumerate(self.test_name_list):
if test not in exclude_list:
jsondump(dict(test_result_path=test_result_path,
exclude_list=exclude_list,),
open(os.path.join(self.directory, 'startUnitTest.%s.json' % test),
'w'))
del self.test_name_list[i]
return test, test
finally:
self.lock.release()
def stopUnitTest(self, test_path, status_dict):
jsondump(dict(test_path=test_path, status_dict=status_dict),
open(os.path.join(self.directory, 'stopUnitTest.%s.json' % test_path),
'w'))
#!/usr/bin/python2.6
import argparse, pprint, socket, sys, time, xmlrpclib
from DummyTaskDistributionTool import DummyTaskDistributionTool
import DummyTaskDistributionTool
from ERP5TypeTestSuite import ERP5TypeTestSuite
def makeSuite(node_quantity=None, test_suite=None, revision=None,
......@@ -55,6 +55,9 @@ def main():
parser.add_argument('--master_url',
help='The Url of Master controling many suites',
default=None)
parser.add_argument('--master_directory',
help='Directory to simulate master',
default=None)
parser.add_argument('--db_list', help='A list of sql connection strings')
# parameters that needs to be passed to runUnitTest
parser.add_argument('--conversion_server_hostname', default=None)
......@@ -73,8 +76,11 @@ def main():
(master_url, 'portal_task_distribution'),
allow_none=1)
assert master.getProtocolRevision() == 1
elif args.master_directory is not None:
master = DummyTaskDistributionTool.JSONFSTaskDistributionTool(
args.master_directory)
else:
master = DummyTaskDistributionTool()
master = DummyTaskDistributionTool.DummyTaskDistributionTool()
test_suite_title = args.test_suite_title or args.test_suite
revision = args.revision
suite = makeSuite(test_suite=args.test_suite,
......@@ -94,4 +100,4 @@ def main():
suite.start(test[1], lambda status_dict, __test_path=test[0]:
safeRpcCall(master.stopUnitTest, __test_path, status_dict))
elif not suite.running:
break
\ No newline at end of file
break
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