Move Test class to its own test.py module
Make the code more reachable for newcomers, as well as more organized
for people who work a lot with it.
SuiteRun in suite.py is already quite big, and having the Test class in
there make it unnecessarily more big, and makes it difficult to find
stuff.
At the same time, having a test.py which does actually not contain the
Test class but other stuff, makes it even more confusing.
Change-Id: I9c8d67f598466ba52a4827ff77027b9eae85929a
diff --git a/src/osmo_gsm_tester/suite.py b/src/osmo_gsm_tester/suite.py
index 5b9df76..25aef35 100644
--- a/src/osmo_gsm_tester/suite.py
+++ b/src/osmo_gsm_tester/suite.py
@@ -20,11 +20,9 @@
import os
import sys
import time
-import traceback
import pprint
-from . import config, log, template, util, resource, schema, event_loop
-from . import osmo_nitb, osmo_hlr, osmo_mgcpgw, osmo_mgw, osmo_msc, osmo_bsc, osmo_stp, modem, esme, sms
-from . import testenv
+from . import config, log, template, util, resource, schema, event_loop, test
+from . import osmo_nitb, osmo_hlr, osmo_mgcpgw, osmo_mgw, osmo_msc, osmo_bsc, osmo_stp, modem, esme
class Timeout(Exception):
pass
@@ -59,94 +57,6 @@
continue
self.test_basenames.append(basename)
-
-class Test(log.Origin):
- UNKNOWN = 'UNKNOWN'
- SKIP = 'skip'
- PASS = 'pass'
- FAIL = 'FAIL'
-
- _run_dir = None
-
- def __init__(self, suite_run, test_basename):
- self.basename = test_basename
- super().__init__(log.C_TST, self.basename)
- self.suite_run = suite_run
- self.path = os.path.join(self.suite_run.definition.suite_dir, self.basename)
- self.status = Test.UNKNOWN
- self.start_timestamp = 0
- self.duration = 0
- self.fail_type = None
- self.fail_message = None
-
- def get_run_dir(self):
- if self._run_dir is None:
- self._run_dir = util.Dir(self.suite_run.get_run_dir().new_dir(self._name))
- return self._run_dir
-
- def run(self):
- try:
- log.large_separator(self.suite_run.trial.name(), self.suite_run.name(), self.name(), sublevel=3)
- self.status = Test.UNKNOWN
- self.start_timestamp = time.time()
- testenv.setup(self.suite_run, self, sys.modules[__name__], event_loop, sms)
- with self.redirect_stdout():
- util.run_python_file('%s.%s' % (self.suite_run.definition.name(), self.basename),
- self.path)
- if self.status == Test.UNKNOWN:
- self.set_pass()
- except Exception as e:
- if hasattr(e, 'msg'):
- msg = e.msg
- else:
- msg = str(e)
- if isinstance(e, AssertionError):
- # AssertionError lacks further information on what was
- # asserted. Find the line where the code asserted:
- msg += log.get_src_from_exc_info(sys.exc_info())
- # add source file information to failure report
- if hasattr(e, 'origins'):
- msg += ' [%s]' % e.origins
- tb_str = traceback.format_exc()
- if isinstance(e, resource.NoResourceExn):
- tb_str += self.suite_run.resource_status_str()
- self.set_fail(type(e).__name__, msg, tb_str, log.get_src_from_exc_info())
- except BaseException as e:
- # when the program is aborted by a signal (like Ctrl-C), escalate to abort all.
- self.err('TEST RUN ABORTED: %s' % type(e).__name__)
- raise
-
- def name(self):
- l = log.get_line_for_src(self.path)
- if l is not None:
- return '%s:%s' % (self._name, l)
- return super().name()
-
- def set_fail(self, fail_type, fail_message, tb_str=None, src=4):
- self.status = Test.FAIL
- self.duration = time.time() - self.start_timestamp
- self.fail_type = fail_type
- self.fail_message = fail_message
-
- if tb_str is None:
- # populate an exception-less call to set_fail() with traceback info
- tb_str = ''.join(traceback.format_stack()[:-1])
-
- self.fail_tb = tb_str
- self.err('%s: %s' % (self.fail_type, self.fail_message), _src=src)
- if self.fail_tb:
- self.log(self.fail_tb, _level=log.L_TRACEBACK)
- self.log('Test FAILED (%.1f sec)' % self.duration)
-
- def set_pass(self):
- self.status = Test.PASS
- self.duration = time.time() - self.start_timestamp
- self.log('Test passed (%.1f sec)' % self.duration)
-
- def set_skip(self):
- self.status = Test.SKIP
- self.duration = 0
-
class SuiteRun(log.Origin):
UNKNOWN = 'UNKNOWN'
PASS = 'PASS'
@@ -176,7 +86,7 @@
def load_tests(self):
self.tests = []
for test_basename in self.definition.test_basenames:
- self.tests.append(Test(self, test_basename))
+ self.tests.append(test.Test(self, test_basename))
def register_for_cleanup(self, *obj):
assert all([hasattr(o, 'cleanup') for o in obj])
@@ -243,12 +153,12 @@
event_loop.register_poll_func(self.poll)
if not self.reserved_resources:
self.reserve_resources()
- for test in self.tests:
- if names and not test.name() in names:
- test.set_skip()
+ for t in self.tests:
+ if names and not t.name() in names:
+ t.set_skip()
continue
- self.current_test = test
- test.run()
+ self.current_test = t
+ t.run()
self.stop_processes()
self.objects_cleanup()
self.reserved_resources.put_all()
@@ -284,10 +194,10 @@
passed = 0
skipped = 0
failed = 0
- for test in self.tests:
- if test.status == Test.PASS:
+ for t in self.tests:
+ if t.status == test.Test.PASS:
passed += 1
- elif test.status == Test.FAIL:
+ elif t.status == test.Test.FAIL:
failed += 1
else:
skipped += 1