118 lines
4.0 KiB
Python
118 lines
4.0 KiB
Python
# osmo_ms_driver: Main test runner
|
|
#
|
|
# 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/>.
|
|
|
|
# Local modules
|
|
from .event_server import EventServer
|
|
from .simple_loop import SimpleLoop
|
|
from .location_update_test import MassUpdateLocationTest
|
|
from .cdf import cdfs
|
|
from .starter import BinaryOptions, MobileTestStarter
|
|
from .test_support import imsi_ki_gen
|
|
from osmo_gsm_tester.core import log, util
|
|
from osmo_gsm_tester.obj import ms_osmo_mobile
|
|
|
|
# System modules
|
|
from datetime import timedelta
|
|
|
|
import argparse
|
|
import atexit
|
|
import datetime
|
|
import signal
|
|
import tempfile
|
|
import os.path
|
|
import os
|
|
|
|
def parser():
|
|
parser = argparse.ArgumentParser(epilog=__doc__, formatter_class=argparse.RawTextHelpFormatter)
|
|
parser.add_argument('-s', '--launch-duration', dest='launch_duration',
|
|
default=60, type=int,
|
|
help="Time launching applications should take in seconds")
|
|
parser.add_argument('-i', '--launch-interval', dest='launch_interval',
|
|
default=100, type=int,
|
|
help="Time between launching in milliseconds")
|
|
parser.add_argument('-t', '--test-duration', dest="test_duration",
|
|
default=120, type=int,
|
|
help="Time of the test duration in seconds")
|
|
parser.add_argument('-d', '--distribution', dest="cdf_name",
|
|
choices=cdfs.keys(), default="ease_in_out",
|
|
help="Curve to use for starting within launch duration")
|
|
parser.add_argument('-m', '--number-ms', dest="num_ms",
|
|
default=10, type=int,
|
|
help="Number of MobileStations to simulate")
|
|
return parser
|
|
|
|
def main():
|
|
# Create a default log to stdout
|
|
log.LogTarget().style(src=False)
|
|
|
|
args = parser().parse_args()
|
|
|
|
# We don't care what is happening to child processes we spawn!
|
|
signal.signal(signal.SIGCHLD, signal.SIG_IGN)
|
|
|
|
loop = SimpleLoop()
|
|
|
|
tmp_dir = util.Dir(tempfile.mkdtemp(suffix="osmo-ms-driver"))
|
|
log.log("Going to store files in ", tmp_dir=tmp_dir)
|
|
|
|
# How long should starting all apps take
|
|
time_start=datetime.timedelta(seconds=args.launch_duration)
|
|
# In which steps to start processes
|
|
time_step=datetime.timedelta(milliseconds=args.launch_interval)
|
|
|
|
# Event server path
|
|
event_server_path = os.path.join(str(tmp_dir), "osmo_ms_driver.unix")
|
|
|
|
# The function that decides when to start something
|
|
cdf = cdfs[args.cdf_name](time_start, time_step)
|
|
|
|
# Event server to handle MS->test events
|
|
ev_server = EventServer("ev_server", event_server_path)
|
|
ev_server.listen(loop)
|
|
|
|
# Just a single test for now.
|
|
options = BinaryOptions("virtphy", "mobile", os.environ)
|
|
result = {}
|
|
starter = MobileTestStarter("lu_test", options, cdf, ev_server, tmp_dir, result)
|
|
test = MassUpdateLocationTest("lu_test", ev_server, result)
|
|
|
|
# Add subscribers to the test.
|
|
imsi_gen = imsi_ki_gen()
|
|
for i in range(0, args.num_ms):
|
|
imsi, ki = next(imsi_gen)
|
|
conf = {
|
|
'imsi': imsi,
|
|
'ki': ki,
|
|
'auth_algo': 'comp128v1',
|
|
'run_lu_test': False,
|
|
}
|
|
starter.subscriber_add(ms_osmo_mobile.MSOsmoMobile("ms_%d" % i, conf))
|
|
starter.configure_tasks()
|
|
test.configure(args.num_ms)
|
|
|
|
atexit.register(starter.stop_all)
|
|
|
|
# Run until everything has been launched
|
|
deadline = starter.start_all(loop, timedelta(seconds=args.test_duration))
|
|
test.wait_for_test(loop, deadline)
|
|
|
|
# Print stats
|
|
test.print_stats()
|
|
|
|
if __name__ == '__main__':
|
|
main()
|