1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
|
# 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 . 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
import os.path
import shutil
import tempfile
class MsDriver(log.Origin):
def __init__(self, suite_run):
super().__init__(log.C_RUN, 'ms-driver')
self._suite_run = suite_run
# 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._test_case = None
self.event_server_sk_tmp_dir = None
self._subscribers = []
self._configured = False
if len(self.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())
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._suite_run.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 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)
options = self.build_binary_options()
self._test_case = MassUpdateLocationTest("mass", options, self._cdf,
self._ev_server,
util.Dir(self.event_server_sk_tmp_dir),
suite_run=self._suite_run)
for sub in self._subscribers:
self._test_case.subscriber_add(sub)
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._test_case.run_test(self._loop, self._test_duration)
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 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
|
