src/osmo_ms_driver/test_support.py

# osmo_ms_driver: Test helpers and base classes
#
# Copyright (C) 2018 by Holger Hans Peter Freyther
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

from abc import ABCMeta
from osmo_gsm_tester.core import log

import time

def imsi_ki_gen():
    """
    Generate IMSIs and KIs to be used by test.
    """
    n = 1010000000000
    while True:
        yield ("%.15d" % n, "00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00")
        n += 1

class ResultStore(log.Origin):
    """
    The class for results. There should be one result class per test subject.
    Specific tests can use add_result to add their outcome to this object.
    """

    def __init__(self, name):
        super().__init__(log.C_RUN, name)
        self._time_of_registration = None
        self._time_of_launch = None
        self._results = {}

    def set_start_time(self, time):
        assert self._time_of_registration is None
        self._time_of_registration = time

    def set_launch_time(self, time):
        assert self._time_of_launch is None
        self._time_of_launch = time

    def start_time(self):
        return self._time_of_registration or 0

    def launch_time(self):
        return self._time_of_launch or 0

    def set_result(self, key, value):
        """Sets a result with the given key and value."""
        self._results[key] = value

    def get_result(self, key, default=None):
        """Returns the result for the given key or default."""
        return self._results.get(key, default)

    def has_result(self, key):
        """Returns true if there is a value for the key."""
        return self._results.get(key) is not None


class TestBase(log.Origin, metaclass=ABCMeta):
    """Base class for all mass test cases."""

    def __init__(self, name, event_server, results):
        super().__init__(log.C_RUN, name)
        self._event_server = event_server
        self._results = results

    def configure(self, subscribers, mobiles):
        """
        Configures the test given the subscribers.

        The subscriber at index _i_ belongs to the mobile at the
        same index. subscribers[i] == mobiles[i].subscriber().
        """
        pass

    def before_start(self):
        """Prepares the test for starting."""
        pass

    def after_start(self):
        """Finishes the test after starting."""
        pass

    def has_completed(self):
        """Returns true if the test has completed."""
        pass

    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, subscribers, mobiles):
        for test in self._tests:
            test.configure(subscribers, mobiles)

    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()