#!${:python_for_test} import argparse import configparser import importlib import json import logging import time import unittest import erp5.util.taskdistribution import slapos.client def main(): parser = argparse.ArgumentParser() parser.add_argument('--test_suite') parser.add_argument('--test_suite_title') parser.add_argument('--test_node_title') parser.add_argument('--project_title') parser.add_argument('--revision') parser.add_argument('--master_url') args, _ = parser.parse_known_args() # ignore other arguments if not args.master_url: unittest.main() # exits module = importlib.import_module(__name__) suite = unittest.defaultTestLoader.loadTestsFromModule(module) all_tests = [t for s in suite for t in s] task_distributor = erp5.util.taskdistribution.TaskDistributor( portal_url=args.master_url ) test_result = task_distributor.createTestResult( revision = args.revision, test_name_list = [t.id() for t in all_tests], node_title = args.test_node_title, test_title = args.test_suite_title, project_title = args.project_title, ) runner = unittest.TextTestRunner() result = runner.run(suite) errors = dict(result.errors) failures = dict(result.failures) skipped = dict(result.skipped) print(errors) print(failures) print(skipped) for t in all_tests: kind = [errors, failures, skipped] count = [0, 0, 0] output = 'OK' for i in range(len(kind)): try: output = kind[i][t] count[i] = 1 break except KeyError: pass print(t.id()) print(count) print(output) test_result_line = test_result.start() test_result_line.stop( test_count = 1, error_count = count[0], failure_count = count[1], skip_count = count[2], duration = 0, command = '', stdout = output, stderr = '', html_test_result = '', ) class EndToEndTestCase(unittest.TestCase): @classmethod def setUpClass(cls): configp = configparser.ConfigParser() configp.read('${slapos-client.cfg:output}') args = type("empty_args", (), {})() conf = slapos.client.ClientConfig(args, configp) local = slapos.client.init(conf, logging.getLogger(__name__)) cls.slap = local['slap'] cls.supply = staticmethod(local['supply']) cls._request = staticmethod(local['request']) cls.product = staticmethod(local['product']) cls._requested = {} @classmethod def tearDownClass(cls): for args, kw in cls._requested.values(): kw['state'] = 'stopped' cls._request(*args, **kw) @classmethod def request(cls, *args, **kw): instance_name = args[1] cls._requested[instance_name] = (args, kw) partition = cls._request(*args, **kw) return cls.unwrapConnectionDict(partition.getConnectionParameterDict()) @staticmethod def unwrapConnectionDict(connection_dict): try: connection_json = connection_dict['_'] except KeyError: return connection_dict return json.loads(connection_json) @classmethod def getInstanceInfos(cls, instance_name): # adapated from cli/info infos = cls.slap.registerOpenOrder().getInformation(instance_name) class Infos: def __init__(self, **kw): self.__dict__.update(kw) connection_dict = { e['connection_key'] : e['connection_value'] for e in infos._connection_dict } return Infos( software_url = infos._software_release_url, software_type = infos._source_reference, shared = bool(infos._root_slave), requested_state = infos._requested_state, parameter_dict = infos._parameter_dict, connection_dict = cls.unwrapConnectionDict(connection_dict), news = infos._news, ) @classmethod def getInstanceNews(cls, instance_name): try: news = cls.slap.registerOpenOrder().getInformation(instance_name)._news except Exception: return () return news['instance'] @classmethod def getInstanceStatus(cls, instance_name): # adapated from cli/info status = 0b00 for e in cls.getInstanceNews(instance_name): text = e.get('text', '') if text.startswith('#access'): status |= 0b01 elif text.startswith('#error'): status |= 0b10 if status == 0b11: break return ('none', 'green', 'red', 'orange')[status] @classmethod def waitUntilGreen(cls, instance_name, timeout=80): t0 = time.time() while (status := cls.getInstanceStatus(instance_name)) != 'green': msg = 'Instance %s status is still %s' % (instance_name, status) print(msg) if (time.time() - t0) > 60 * timeout: raise TimeoutError(msg) time.sleep(60) class KvmTest(EndToEndTestCase): def test(self): # instance_name = time.strftime('e2e-test-kvm-%Y-%B-%d-%H:%M:%S') instance_name = 'e2e-kvm-test' # avoid timestamp to reuse instance self.request(self.product.kvm, instance_name) self.waitUntilGreen(instance_name) connection_dict = self.request(self.product.kvm, instance_name) self.assertIn('url', connection_dict) class Fail(EndToEndTestCase): def test(self): self.assertEqual(0, 1) if __name__ == '__main__': main()