src/osmo_gsm_tester/obj/ms_driver.py

# ms_driver: Launch OsmocomBB mobile's virtually connected to a BTS
#
# 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 datetime import timedelta
from ..core import log, util
from osmo_ms_driver.cdf import cdfs
from osmo_ms_driver.event_server import EventServer
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
import tempfile

class MsDriver(log.Origin):

    def __init__(self, testenv):
        super().__init__(log.C_RUN, 'ms-driver')
        self._testenv = testenv

        # TODO: take config out of the test scenario
        self._time_start = timedelta(seconds=60)
        self._time_step = timedelta(milliseconds=100)
        self._test_duration = timedelta(seconds=120)
        self._cdf = cdfs["ease_in_out"](self._time_start, self._time_step)
        self._loop = SimpleLoop()
        self._executor = TestExecutor()
        self.event_server_sk_tmp_dir = None
        self._subscribers = []
        self._configured = False
        self._results = {}

        # 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')
        return os.path.join(self.event_server_sk_tmp_dir, 'osmo_ms_driver.unix')

    def build_binary_options(self):
        """Builds an instance of BinaryOptions.

        Populates the BinaryOptions by searching the virtphy and mobile
        application within the trial directory.
        """

        # Get the base directory for the virtphy/mobile application
        inst = util.Dir(os.path.abspath(self._testenv.suite().trial().get_inst('osmocom-bb')))

        # Assume these are dynamically linked and verify there is a lib dir.
        lib = inst.child('lib')
        if not os.path.isdir(lib):
            raise RuntimeError('No lib/ in %r' % inst)
        env = { 'LD_LIBRARY_PATH': util.prepend_library_path(lib) }

        def check_and_return_binary(name):
            """Checks the binary exists and returns the path."""
            binary = inst.child('bin', name)
            if not os.path.isfile(binary):
                raise RuntimeError('Binary missing: %r' % binary)
            return binary

        virtphy = check_and_return_binary('virtphy')
        mobile = check_and_return_binary('mobile')
        return BinaryOptions(virtphy, mobile, env)

    def subscriber_add(self, subscriber):
        """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 == 'ul_test':
            test = MassUpdateLocationTest("mass",
                                          self._ev_server, self._results)

        # Verify that a test was instantiated.
        if test_name is None:
            raise log.Error("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.
        """
        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, testenv=self._testenv)

        for sub in self._subscribers:
            self._starter.subscriber_add(sub)

        self._starter.configure_tasks()
        self._executor.configure(self._subscribers, self._starter.mobiles())
        self._configured = True

    def run_test(self):
        """
        Runs the configured tests by starting the configured amount of mobile
        devices according to their schedule. Returns once all tests succeeded
        or the configured timeout has passed.
        """
        if not self._configured:
            self.configure()
        self._executor.before_start()
        deadline = self._starter.start_all(self._loop, self._test_duration)
        self._executor.after_start()
        self._executor.wait_for_test(self._loop, deadline)

    def print_stats(self):
        """
        Prints statistics about the test run.
        """
        self._executor.print_stats()

    def cleanup(self):
        """
        Cleans up the driver (e.g. AF_UNIX files).
        """

        # Clean-up the temporary directory.
        if self.event_server_sk_tmp_dir:
            shutil.rmtree(path=self.event_server_sk_tmp_dir)

# vim: expandtab tabstop=4 shiftwidth=4