initial import
The original osmo-gsm-tester was an internal development at sysmocom, mostly by
D. Laszlo Sitzer <dlsitzer@sysmocom.de>, of which this public osmo-gsm-tester
is a refactoring / rewrite.
This imports an early state of the refactoring and is not functional yet. Bits
from the earlier osmo-gsm-tester will be added as needed. The earlier commit
history is not imported.
diff --git a/src/osmo_gsm_tester/suite.py b/src/osmo_gsm_tester/suite.py
new file mode 100644
index 0000000..fb7c34d
--- /dev/null
+++ b/src/osmo_gsm_tester/suite.py
@@ -0,0 +1,150 @@
+# osmo_gsm_tester: test suite
+#
+# Copyright (C) 2016-2017 by sysmocom - s.f.m.c. GmbH
+#
+# Author: Neels Hofmeyr <neels@hofmeyr.de>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+import os
+from . import config, log, template, utils
+
+class Suite(log.Origin):
+ '''A test suite reserves resources for a number of tests.
+ Each test requires a specific number of modems, BTSs etc., which are
+ reserved beforehand by a test suite. This way several test suites can be
+ scheduled dynamically without resource conflicts arising halfway through
+ the tests.'''
+
+ CONF_FILENAME = 'suite.conf'
+
+ CONF_SCHEMA = {
+ 'resources.nitb_iface': config.INT,
+ 'resources.nitb': config.INT,
+ 'resources.bts': config.INT,
+ 'resources.msisdn': config.INT,
+ 'resources.modem': config.INT,
+ 'defaults.timeout': config.STR,
+ }
+
+ class Results:
+ def __init__(self):
+ self.passed = []
+ self.failed = []
+ self.all_passed = None
+
+ def add_pass(self, test):
+ self.passed.append(test)
+
+ def add_fail(self, test):
+ self.failed.append(test)
+
+ def conclude(self):
+ self.all_passed = bool(self.passed) and not bool(self.failed)
+ return self
+
+ def __init__(self, suite_dir):
+ self.set_log_category(log.C_CNF)
+ self.suite_dir = suite_dir
+ self.set_name(os.path.basename(self.suite_dir))
+ self.read_conf()
+
+ def read_conf(self):
+ with self:
+ if not os.path.isdir(self.suite_dir):
+ raise RuntimeError('No such directory: %r' % self.suite_dir)
+ self.conf = config.read(os.path.join(self.suite_dir,
+ Suite.CONF_FILENAME),
+ Suite.CONF_SCHEMA)
+ self.load_tests()
+
+ def load_tests(self):
+ with self:
+ self.tests = []
+ for basename in os.listdir(self.suite_dir):
+ if not basename.endswith('.py'):
+ continue
+ self.tests.append(Test(self, basename))
+
+ def add_test(self, test):
+ with self:
+ if not isinstance(test, Test):
+ raise ValueError('add_test(): pass a Test() instance, not %s' % type(test))
+ if test.suite is None:
+ test.suite = self
+ if test.suite is not self:
+ raise ValueError('add_test(): test already belongs to another suite')
+ self.tests.append(test)
+
+ def run_tests(self):
+ results = Suite.Results()
+ for test in self.tests:
+ self._run_test(test, results)
+ return results.conclude()
+
+ def run_tests_by_name(self, *names):
+ results = Suite.Results()
+ for name in names:
+ basename = name
+ if not basename.endswith('.py'):
+ basename = name + '.py'
+ for test in self.tests:
+ if basename == test.basename:
+ self._run_test(test, results)
+ break
+ return results.conclude()
+
+ def _run_test(self, test, results):
+ try:
+ with self:
+ test.run()
+ results.add_pass(test)
+ except:
+ results.add_fail(test)
+ self.log_exn()
+
+class Test(log.Origin):
+
+ def __init__(self, suite, test_basename):
+ self.suite = suite
+ self.basename = test_basename
+ self.set_name(self.basename)
+ self.set_log_category(log.C_TST)
+ self.path = os.path.join(self.suite.suite_dir, self.basename)
+ with self:
+ with open(self.path, 'r') as f:
+ self.script = f.read()
+
+ def run(self):
+ with self:
+ self.code = compile(self.script, self.path, 'exec')
+ with self.redirect_stdout():
+ exec(self.code, self.test_globals())
+ self._success = True
+
+ def test_globals(self):
+ test_globals = {
+ 'this': utils.dict2obj({
+ 'suite': self.suite.suite_dir,
+ 'test': self.basename,
+ }),
+ 'resources': utils.dict2obj({
+ }),
+ }
+ return test_globals
+
+def load(suite_dir):
+ return Suite(suite_dir)
+
+# vim: expandtab tabstop=4 shiftwidth=4