158 lines
5.9 KiB
Python
158 lines
5.9 KiB
Python
# 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
|