ms_srs: refactor UE state getter functions

rename UE helper functions to query RRC connected/idle state
and adapt tests accordingly

also add helper to retrieve assinged IP address

Change-Id: I6cd057e34b4df6a1a73695355dd6406d7e039546

src/osmo_gsm_tester/obj/ms.py

@@ -77,6 +77,20 @@ class MS(log.Origin, metaclass=ABCMeta):
             raise log.Error('MS type not supported:', ms_type)
         return ms_class(testenv, conf)
 
+    @abstractmethod
+    def is_registered(self, mcc_mnc=None):
+        '''Check whether MS is considered registered with the target network. In
+        2G networks, and MS is registered if it had a successful Location Update
+        in CS. In 4G networks, an UE is considered registered with the core
+        network if it has obtained and IP address. If MCC/MNC are given it tries
+        to manually register against that specific network.'''
+        pass
+
+    @abstractmethod
+    def get_assigned_addr(self, ipv6=False):
+        ''' Returns last assigned IP address '''
+        pass
+
 ###################
 # PUBLIC (test API included)
 ###################

src/osmo_gsm_tester/obj/ms_amarisoft.py

@@ -296,13 +296,19 @@ class AmarisoftUE(MS):
             self.rem_host.scp('scp-cfg-rf-to-remote', self.config_rf_file, self.remote_config_rf_file)
             self.rem_host.scp('scp-ifup-to-remote', self.ifup_file, self.remote_ifup_file)
 
-    def is_connected(self, mcc_mnc=None):
+    def is_registered(self, mcc_mnc=None):
         # lteue doesn't call the ifup script until after it becomes attached, so
         # simply look for our ifup script output at the end of it:
         return 'netns %s configured' % (self.netns()) in (self.process.get_stdout() or '')
 
+    def is_rrc_connected(self):
+        return self.is_registered()
+
     def is_attached(self):
-        return self.is_connected()
+        return self.is_registered()
+
+    def get_assigned_addr(self, ipv6=False):
+        raise log.Error('API not implemented!')
 
     def running(self):
         return not self.process.terminated()

src/osmo_gsm_tester/obj/ms_ofono.py

@@ -487,6 +487,9 @@ class Modem(MS):
             return self._apn_ipaddr
         return 'dynamic'
 
+    def get_assigned_addr(self, ipv6=False):
+        raise log.Error('API not implemented!')
+
     def features(self):
         return self._conf.get('features', [])
 
@@ -501,7 +504,7 @@ class Modem(MS):
     def _on_netreg_property_changed(self, name, value):
         self.dbg('%r.PropertyChanged() -> %s=%s' % (I_NETREG, name, value))
 
-    def is_connected(self, mcc_mnc=None):
+    def is_registered(self, mcc_mnc=None):
         netreg = self.dbus.interface(I_NETREG)
         prop = netreg.GetProperties()
         status = prop.get('Status')
@@ -516,6 +519,10 @@ class Modem(MS):
             return True
         return False
 
+    def is_connected(self, mcc_mnc=None):
+        '''Convenience helper to keep old test API'''
+        return self.is_registered(mcc_mnc)
+
     def schedule_scan_register(self, mcc_mnc):
         if self.register_attempts > NETREG_MAX_REGISTER_ATTEMPTS:
             raise log.Error('Failed to find Network Operator', mcc_mnc=mcc_mnc, attempts=self.register_attempts)

src/osmo_gsm_tester/obj/ms_srs.py

@@ -19,6 +19,7 @@
 
 import os
 import pprint
+import re
 
 from ..core import log, util, config, template, process, remote
 from ..core import schema
@@ -300,11 +301,26 @@ class srsUE(MS):
             self.rem_host.recreate_remote_dir(self.remote_run_dir)
             self.rem_host.scp('scp-cfg-to-remote', self.config_file, self.remote_config_file)
 
-    def is_connected(self, mcc_mnc=None):
+    def is_rrc_connected(self):
+        ''' Check whether UE is RRC connected using console message '''
+        pos_connected = (self.process.get_stdout() or '').rfind('RRC Connected')
+        pos_released = (self.process.get_stdout() or '').rfind('RRC IDLE')
+        return pos_connected > pos_released
+
+    def is_registered(self, mcc_mnc=None):
+        ''' Checks if UE is EMM registered '''
         return 'Network attach successful.' in (self.process.get_stdout() or '')
 
-    def is_attached(self):
-        return self.is_connected()
+    def get_assigned_addr(self, ipv6=False):
+        if ipv6:
+            raise log.Error('IPv6 not implemented!')
+        else:
+            stdout_lines = (self.process.get_stdout() or '').splitlines()
+            for line in reversed(stdout_lines):
+                if line.find('Network attach successful. IP: ') != -1:
+                    ipv4_addr = re.findall( r'[0-9]+(?:\.[0-9]+){3}', line)
+                    return ipv4_addr[0]
+            return None
 
     def running(self):
         return not self.process.terminated()

sysmocom/suites/4g/handover.py

@@ -46,7 +46,7 @@ iperf3srv.start()
 proc = iperf3cli.prepare_test_proc(iperf3cli.DIR_UL, ue.netns(), duration + 30)
 
 print('waiting for UE to attach...')
-wait(ue.is_connected, None)
+wait(ue.is_rrc_connected)
 print('UE is attached')
 
 rfemu_cell1 = enb.get_rfemu(0)

sysmocom/suites/4g/iperf3_dl.py

@@ -26,7 +26,7 @@ iperf3srv.start()
 proc = iperf3cli.prepare_test_proc(iperf3cli.DIR_DL, ue.netns(), bitrate=max_rate)
 
 print('waiting for UE to attach...')
-wait(ue.is_connected, None)
+wait(ue.is_rrc_connected)
 print('UE is attached')
 
 print("Running iperf3 client to %s through %s" % (str(iperf3cli), ue.netns()))

sysmocom/suites/4g/iperf3_ul.py

@@ -26,7 +26,7 @@ iperf3srv.start()
 proc = iperf3cli.prepare_test_proc(iperf3cli.DIR_UL, ue.netns(), bitrate=max_rate)
 
 print('waiting for UE to attach...')
-wait(ue.is_connected, None)
+wait(ue.is_rrc_connected)
 print('UE is attached')
 
 print("Running iperf3 client to %s through %s" % (str(iperf3cli), ue.netns()))

sysmocom/suites/4g/ping.py

@@ -16,7 +16,7 @@ print('ENB is connected to EPC')
 
 ue.connect(enb)
 print('waiting for UE to attach...')
-wait(ue.is_connected, None)
+wait(ue.is_rrc_connected)
 print('UE is attached')
 
 proc = ue.run_netns_wait('ping', ('ping', '-c', '10', epc.tun_addr()))