This is merely a historical archive of years 2008-2021, before the migration to mailman3.
A maintained and still updated list archive can be found at https://lists.osmocom.org/hyperkitty/list/gerrit-log@lists.osmocom.org/.
Holger Freyther gerrit-no-reply at lists.osmocom.orgHolger Freyther has uploaded this change for review. ( https://gerrit.osmocom.org/13826
Change subject: virtual: Make it possible to add tests to the ms driver
......................................................................
virtual: Make it possible to add tests to the ms driver
Introduce an Executor that forwards all testcase related methods to
a list of testcases. Allow to instantiate them by name and use the
result to access the statistics.
Change-Id: Ia65ee53987e92b24e6b8c40e1376bc74dc260180
---
M src/osmo_gsm_tester/ms_driver.py
M src/osmo_ms_driver/test_support.py
M suites/nitb_netreg_mass/register_default_mass.py
3 files changed, 82 insertions(+), 33 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/osmo-gsm-tester refs/changes/26/13826/1
diff --git a/src/osmo_gsm_tester/ms_driver.py b/src/osmo_gsm_tester/ms_driver.py
index dddedf7..355a03e 100644
--- a/src/osmo_gsm_tester/ms_driver.py
+++ b/src/osmo_gsm_tester/ms_driver.py
@@ -22,6 +22,7 @@
from osmo_ms_driver.simple_loop import SimpleLoop
from osmo_ms_driver.location_update_test import MassUpdateLocationTest
from osmo_ms_driver.starter import BinaryOptions, MobileTestStarter
+from osmo_ms_driver.test_support import TestExecutor
import os.path
import shutil
@@ -39,14 +40,20 @@
self._test_duration = timedelta(seconds=120)
self._cdf = cdfs["ease_in_out"](self._time_start, self._time_step)
self._loop = SimpleLoop()
- self._test_case = None
+ self._executor = TestExecutor()
self.event_server_sk_tmp_dir = None
self._subscribers = []
self._configured = False
+ self._results = {}
- if len(self.event_server_path().encode()) > 107:
+ # Set-up and start the event server
+ event_server_path = self.event_server_path()
+ if len(event_server_path.encode()) > 107:
raise log.Error('Path for event_server socket is longer than max allowed len for unix socket path (107):', self.event_server_path())
+ self._ev_server = EventServer("ev_server", event_server_path)
+ self._ev_server.listen(self._loop)
+
def event_server_path(self):
if self.event_server_sk_tmp_dir is None:
self.event_server_sk_tmp_dir = tempfile.mkdtemp('', 'ogteventserversk')
@@ -83,27 +90,39 @@
"""Adds a subscriber to the list of subscribers."""
self._subscribers.append(subscriber)
+ def add_test(self, test_name, **kwargs):
+ """
+ Instantiates and returns a test for the given name.
+
+ The instance created and added will be returned.
+ """
+ if test_name == 'update_location':
+ test = MassUpdateLocationTest("mass",
+ self._ev_server, self._results)
+
+ # Verify that a test was instantiated.
+ if test_name is None:
+ raise Exception("Unknown test_name: " + test_name)
+
+ # Add it to the executor and return it.
+ self._executor.add_test(test)
+ return test
+
def configure(self):
"""
Configures the subscribers, tests and registration server. Needs to be
called after the complete configuration of this driver.
"""
- event_server_path = self.event_server_path()
-
- self._ev_server = EventServer("ev_server", event_server_path)
- self._ev_server.listen(self._loop)
- self._results = {}
options = self.build_binary_options()
self._starter = MobileTestStarter("mass", options, self._cdf,
self._ev_server,
util.Dir(self.event_server_sk_tmp_dir),
self._results, suite_run=self._suite_run)
- self._test_case = MassUpdateLocationTest("mass", self._ev_server, self._results)
for sub in self._subscribers:
self._starter.subscriber_add(sub)
- self._test_case.configure(len(self._subscribers))
+ self._executor.configure(len(self._subscribers))
self._configured = True
def run_test(self):
@@ -114,34 +133,16 @@
"""
if not self._configured:
self.configure()
- self._test_case.before_start()
+ self._executor.before_start()
deadline = self._starter.start_all(self._loop, self._test_duration)
- self._test_case.after_start()
- self._test_case.wait_for_test(self._loop, deadline)
+ self._executor.after_start()
+ self._executor.wait_for_test(self._loop, deadline)
def print_stats(self):
"""
Prints statistics about the test run.
"""
- self._test_case.print_stats()
-
- def get_stats(self):
- """
- Returns a statistical summary of the test.
- """
- return self._test_case.get_stats()
-
- def get_result_values(self):
- """
- Returns the raw result values of the test run in any order.
- """
- return self._test_case.get_result_values()
-
- def lus_less_than(self, acceptable_delay):
- """
- Returns the results that completed their LU within the acceptable delay.
- """
- return self._test_case.lus_less_than(acceptable_delay)
+ self._executor.print_stats()
def cleanup(self):
"""
diff --git a/src/osmo_ms_driver/test_support.py b/src/osmo_ms_driver/test_support.py
index cbfd444..f7910dd 100644
--- a/src/osmo_ms_driver/test_support.py
+++ b/src/osmo_ms_driver/test_support.py
@@ -18,6 +18,8 @@
from abc import ABCMeta
from osmo_gsm_tester import log
+import time
+
def imsi_ki_gen():
"""
Generate IMSIs and KIs to be used by test.
@@ -93,3 +95,48 @@
def print_stats(self):
"""Prints statistics/results of the test."""
pass
+
+
+class TestExecutor(log.Origin):
+ """Execute/Wait for a list of tests to complete."""
+
+ def __init__(self):
+ super().__init__(log.C_RUN, "executor")
+ self._tests = []
+
+ def add_test(self, test):
+ self._tests.append(test)
+
+ def configure(self, num_subscriber):
+ for test in self._tests:
+ test.configure(num_subscriber)
+
+ def before_start(self):
+ for test in self._tests:
+ test.before_start()
+
+ def after_start(self):
+ for test in self._tests:
+ test.after_start()
+
+ def print_stats(self):
+ """Prints statistics/results of the test."""
+ for test in self._tests:
+ test.print_stats()
+
+ def all_tests_completed(self):
+ """Returns true if all tests completed."""
+ for test in self._tests:
+ if not test.has_completed():
+ return False
+ return True
+
+ def wait_for_test(self, loop, deadline):
+ """Waits up to the absolute deadline for all tests to complete."""
+ while not self.all_tests_completed():
+ now_time = time.clock_gettime(time.CLOCK_MONOTONIC)
+ sleep_time = deadline - now_time
+ if sleep_time < 0:
+ break
+ loop.schedule_timeout(sleep_time)
+ loop.select()
diff --git a/suites/nitb_netreg_mass/register_default_mass.py b/suites/nitb_netreg_mass/register_default_mass.py
index f4e5e80..d6782e7 100644
--- a/suites/nitb_netreg_mass/register_default_mass.py
+++ b/suites/nitb_netreg_mass/register_default_mass.py
@@ -10,6 +10,7 @@
nitb = suite.nitb()
bts = suite.bts()
ms_driver = suite.ms_driver()
+ul = ms_driver.add_test('update_location')
modems = suite.all_resources(suite.modem)
print('Launching a simple network')
@@ -35,7 +36,7 @@
#
# 99% of LUs should complete
# 99% of successful LUs should complete within 10s.
-stats = ms_driver.get_stats()
+stats = ul.get_stats()
if len(modems) > 0 and stats.num_completed < 1:
raise Exception("No run completed.")
completion_ratio = stats.num_completed / stats.num_attempted
@@ -46,7 +47,7 @@
# Check how many results are below our threshold.
acceptable_delay = timedelta(seconds=30)
-quick_enough = len(ms_driver.lus_less_than(acceptable_delay))
+quick_enough = len(ul.lus_less_than(acceptable_delay))
latency_ratio = quick_enough / stats.num_attempted
if latency_ratio < 0.99:
raise Exception("Latency ratio of %f%% lower than threshold." % (latency_ratio * 100.0))
--
To view, visit https://gerrit.osmocom.org/13826
To unsubscribe, or for help writing mail filters, visit https://gerrit.osmocom.org/settings
Gerrit-Project: osmo-gsm-tester
Gerrit-Branch: master
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ia65ee53987e92b24e6b8c40e1376bc74dc260180
Gerrit-Change-Number: 13826
Gerrit-PatchSet: 1
Gerrit-Owner: Holger Freyther <holger at freyther.de>
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://lists.osmocom.org/pipermail/gerrit-log/attachments/20190430/4e711a45/attachment.htm>