Implement per-test timeout guard
Timeout value can be specified by test in suite.conf:
config:
suite:
<suite_name>:
<test_name>:
timeout: 2 # 2 seconds timeout
Change-Id: I522f51f77f8be64ebfdb5d5e07ba92baf82d7706
diff --git a/src/osmo_gsm_tester/core/suite.py b/src/osmo_gsm_tester/core/suite.py
index 9b9062d..938471c 100644
--- a/src/osmo_gsm_tester/core/suite.py
+++ b/src/osmo_gsm_tester/core/suite.py
@@ -44,6 +44,8 @@
self.suite_dir = suite_dir
self.conf = None
self._schema = None
+ self.test_basenames = []
+ self.load_test_basenames()
self.read_conf()
def read_conf(self):
@@ -54,13 +56,16 @@
SuiteDefinition.CONF_FILENAME))
# Drop schema part since it's dynamically defining content, makes no sense to validate it.
self._schema = self.conf.pop('schema', {})
+ # Add per-test 'timeout' attribute:
+ d = {t.rstrip('.py'):{'timeout': schema.DURATION} for t in self.test_basenames}
+ schema.combine(self._schema, d)
+ # Convert config file format to proper schema format and register it:
sdef = schema.config_to_schema_def(self._schema, "%s." % self._suite_name)
schema.register_config_schema('suite', sdef)
+ # Finally validate the file:
schema.validate(self.conf, schema.get_all_schema())
- self.load_test_basenames()
def load_test_basenames(self):
- self.test_basenames = []
for basename in sorted(os.listdir(self.suite_dir)):
if not basename.endswith('.py'):
continue
diff --git a/src/osmo_gsm_tester/core/test.py b/src/osmo_gsm_tester/core/test.py
index 45dfd41..c6d88e6 100644
--- a/src/osmo_gsm_tester/core/test.py
+++ b/src/osmo_gsm_tester/core/test.py
@@ -35,12 +35,12 @@
PASS = 'pass'
FAIL = 'FAIL'
- def __init__(self, suite_run, test_basename, test_specific_config):
+ def __init__(self, suite_run, test_basename, config_test_specific):
self.basename = test_basename
super().__init__(log.C_TST, self.basename)
self._run_dir = None
self.suite_run = suite_run
- self._config_test_specific = test_specific_config
+ self._config_test_specific = config_test_specific
self.path = os.path.join(self.suite_run.definition.suite_dir, self.basename)
self.status = Test.UNKNOWN
self.start_timestamp = 0
@@ -49,6 +49,7 @@
self.fail_message = None
self.log_targets = []
self._report_stdout = None
+ self.timeout = int(config_test_specific['timeout']) if 'timeout' in config_test_specific else None
def module_name(self):
'Return test name without trailing .py'
diff --git a/src/osmo_gsm_tester/testenv.py b/src/osmo_gsm_tester/testenv.py
index 11199c2..77d844a 100644
--- a/src/osmo_gsm_tester/testenv.py
+++ b/src/osmo_gsm_tester/testenv.py
@@ -55,6 +55,8 @@
self.test_import_modules_to_clean_up = []
self.objects_to_clean_up = None
MainLoop.register_poll_func(self.poll)
+ if self._test.timeout is not None: # aimed at firing once
+ MainLoop.register_poll_func(self.timeout_expired, timestep=self._test.timeout)
def test(self):
return self._test
@@ -120,6 +122,11 @@
except Exception:
log_module.log_exn()
+ def timeout_expired(self):
+ # Avoid timeout being called several times:
+ MainLoop.unregister_poll_func(self.timeout_expired)
+ raise log_module.Error('Test Timeout triggered: %d seconds elapsed' % self._test.elapsed_time())
+
def poll(self):
for proc, respawn in self._processes:
if proc.terminated():
@@ -139,6 +146,7 @@
self.objects_cleanup()
self.suite_run.reserved_resources.put_all()
MainLoop.unregister_poll_func(self.poll)
+ MainLoop.unregister_poll_func(self.timeout_expired)
self.test_import_modules_cleanup()
self.set_overlay_template_dir(None)