sysmocom: Introduce suite to test emergency calls

Change-Id: I2e851c94311ded0abd4ff072b8cc72316d972750
This commit is contained in:
Pau Espin 2020-10-14 14:49:05 +02:00 committed by pespin
parent 83a2fdca9b
commit 680ba03038
8 changed files with 109 additions and 2 deletions

View File

@ -19,7 +19,10 @@
import copy
from abc import ABCMeta, abstractmethod
from ..core import log, config, schema
from ..core import log
from ..core import config
from ..core import schema
from ..core import util
def on_register_schemas():
resource_schema = {
@ -31,6 +34,7 @@ def on_register_schemas():
'ciphers[]': schema.CIPHER,
'channel_allocator': schema.CHAN_ALLOCATOR,
'gprs_mode': schema.GPRS_MODE,
'emergency_calls_allowed': schema.BOOL_STR,
'num_trx': schema.UINT,
'max_trx': schema.UINT,
'trx_list[].addr': schema.IPV4,
@ -132,6 +136,8 @@ class Bts(log.Origin, metaclass=ABCMeta):
if self.bvci is not None:
config.overlay(values, { 'bvci': self.bvci })
config.overlay(values, { 'emergency_calls_allowed': util.str2bool(values.get('emergency_calls_allowed', 'false')) } )
conf = copy.deepcopy(self.conf)
trx_list = conf.get('trx_list')
if trx_list and len(trx_list) != self.num_trx():

View File

@ -120,5 +120,8 @@ class MS(log.Origin, metaclass=ABCMeta):
self.set_msisdn(self.testenv.msisdn())
return self._msisdn
def emergency_numbers(self):
return ['112']
def get_counter(self, counter_name):
raise log.Error('get_counter() not implemented!')

View File

@ -834,4 +834,10 @@ class Modem(MS):
service_type, response = ss.Initiate(command)
return response
def emergency_numbers(self):
cmgr = self.dbus.interface(I_CALLMGR)
props = cmgr.GetProperties()
self.dbg('got Call properties', props)
return props.get('EmergencyNumbers', [])
# vim: expandtab tabstop=4 shiftwidth=4

View File

@ -27,6 +27,7 @@ from . import osmo_ctrl, pcap_recorder, smsc
def on_register_schemas():
resource_schema = {
'path': schema.STR,
'emergency_call_msisdn': schema.MSISDN,
}
schema.register_resource_schema('modem', resource_schema)
@ -43,6 +44,7 @@ class OsmoMsc(log.Origin):
self.use_osmux = "off"
self.testenv = testenv
self.ip_address = ip_address
self._emergency_call_msisdn = None
self.hlr = hlr
self.mgw = mgw
self.stp = stp
@ -94,7 +96,8 @@ class OsmoMsc(log.Origin):
if self.authentication is not None:
config.overlay(values, dict(msc=dict(net=dict(authentication=self.authentication))))
config.overlay(values, dict(msc=dict(use_osmux=self.use_osmux)))
if self._emergency_call_msisdn is not None:
config.overlay(values, dict(msc=dict(emergency_call_msisdn=self._emergency_call_msisdn)))
self.config = values
@ -146,6 +149,10 @@ class OsmoMsc(log.Origin):
def imsi_list_attached(self):
return OsmoMscCtrl(self).subscriber_list_active()
def set_emergency_call_msisdn(self, msisdn):
self.dbg('Setting Emergency Call MSISDN', msisdn=msisdn)
self._emergency_call_msisdn = msisdn
def running(self):
return not self.process.terminated()

View File

@ -58,6 +58,11 @@ network
% endif
ip.access unit_id ${bts.ipa_unit_id} 0
oml ip.access stream_id ${bts.stream_id} line 0
% if bts.get('emergency_calls_allowed', False):
rach emergency call allowed 1
% else:
rach emergency call allowed 0
%endif
% if bts.get('sgsn', False) and bts['gprs_mode'] != 'none':
gprs mode ${bts.gprs_mode}
gprs routing area ${bts.routing_area_code}
@ -105,6 +110,7 @@ network
msc
core-mobile-country-code ${bsc.net.mcc}
core-mobile-network-code ${bsc.net.mnc}
allow-emergency allow
ip.access rtp-base 25000
mgw remote-ip ${mgw.ip_address.addr}
mgw remote-port 2427

View File

@ -31,6 +31,9 @@ msc
mgw bts-base 8000
osmux ${msc.use_osmux}
assign-tmsi
%if msc.get('emergency_call_msisdn', None) is not None:
emergency-call route-to-msisdn ${msc.emergency_call_msisdn}
%endif
cs7-instance-iu 0
cs7-instance-a 0
ctrl

View File

@ -0,0 +1,64 @@
#!/usr/bin/env python3
from osmo_gsm_tester.testenv import *
hlr = tenv.hlr()
bts = tenv.bts()
mgw_msc = tenv.mgw()
mgw_bsc = tenv.mgw()
stp = tenv.stp()
msc = tenv.msc(hlr, mgw_msc, stp)
bsc = tenv.bsc(msc, mgw_bsc, stp)
ms_mo = tenv.modem()
ms_mt = tenv.modem()
hlr.start()
stp.start()
# Set MSC to route emergency call to ms_mt:
msc.set_emergency_call_msisdn(ms_mt.msisdn())
msc.start()
mgw_msc.start()
mgw_bsc.start()
bsc.bts_add(bts)
bsc.start()
bts.start()
wait(bsc.bts_is_connected, bts)
hlr.subscriber_add(ms_mo)
hlr.subscriber_add(ms_mt)
ms_mo.connect(msc.mcc_mnc())
ms_mt.connect(msc.mcc_mnc())
ms_mo.log_info()
ms_mt.log_info()
print('waiting for modems to attach...')
wait(ms_mo.is_registered, msc.mcc_mnc())
wait(ms_mt.is_registered, msc.mcc_mnc())
wait(msc.subscriber_attached, ms_mo, ms_mt)
assert len(ms_mo.call_id_list()) == 0 and len(ms_mt.call_id_list()) == 0
# Calling emergency number should be redirected to ms_mt as configured further above:
emerg_numbers = ms_mo.emergency_numbers()
assert len(emerg_numbers) > 0
print('dialing Emergency Number %s' % (emerg_numbers[0]))
mo_cid = ms_mo.call_dial(emerg_numbers[0])
mt_cid = ms_mt.call_wait_incoming(ms_mo)
print('dial success')
assert not ms_mo.call_is_active(mo_cid) and not ms_mt.call_is_active(mt_cid)
ms_mt.call_answer(mt_cid)
wait(ms_mo.call_is_active, mo_cid)
wait(ms_mt.call_is_active, mt_cid)
print('answer success, call established and ongoing')
sleep(5) # maintain the call active for 5 seconds
assert ms_mo.call_is_active(mo_cid) and ms_mt.call_is_active(mt_cid)
ms_mt.call_hangup(mt_cid)
wait(lambda: len(ms_mo.call_id_list()) == 0 and len(ms_mt.call_id_list()) == 0)
print('hangup success')

View File

@ -0,0 +1,12 @@
resources:
ip_address:
- times: 6 # msc, bsc, hlr, stp, mgw*2
bts:
- times: 1
modem:
- times: 2
features:
- voice
modifiers:
bts:
- emergency_calls_allowed: true