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
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
|
# 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 Subscriber(log.Origin):
def __init__(self, imsi, ki):
super().__init__(log.C_RUN, 'subscriber')
self._imsi = imsi
self._ki = ki
self._auth_algo = "comp128v1"
self._msisdn = None
def msisdn(self):
return self._msisdn
def set_msisdn(self, msisdn):
self._msisdn = msisdn
def imsi(self):
return self._imsi
def ki(self):
return self._ki
def auth_algo(self):
return self._auth_algo
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._num_ms = 10
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
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 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._num_ms, self._cdf,
self._ev_server,
util.Dir(self.event_server_sk_tmp_dir),
suite_run=self._suite_run)
# TODO: We should pass subscribers down to the test and not get it from
# there.
self._subs = [Subscriber(imsi=mob.imsi(), ki=mob.ki()) for mob in self._test_case.mobiles()]
def ms_subscribers(self):
"""
Returns a list of 'subscribers' that were configured in the
current scenario.
"""
if not hasattr(self, '_subs'):
self.configure()
return self._subs
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 hasattr(self, '_subs'):
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 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
|
