sgsn/ctrl: Add ctrl interface, implement listing subscribers

Add the control interface with no hierachy right now and implement
the first command to list IMSI + Context Address of active sessions.
sgsn_cmd_handle could share more code with bsc variant.

Fixes: SYS#264, SYS#265

1 files changed, 23 insertions, 0 deletions

diff --git a/openbsc/tests/ctrl_test_runner.py b/openbsc/tests/ctrl_test_runner.py
index 60d90a7c1..70ae0908a 100644
--- a/openbsc/tests/ctrl_test_runner.py
+++ b/openbsc/tests/ctrl_test_runner.py
@@ -383,6 +383,21 @@ class TestCtrlNAT(TestCtrlBase):
         self.assertEquals(r['var'], 'net')
         self.assertEquals(r['value'], None)
 
+class TestCtrlSGSN(TestCtrlBase):
+    def ctrl_command(self):
+        return ["./src/gprs/osmo-sgsn", "-c",
+                "doc/examples/osmo-sgsn/osmo-sgsn.cfg"]
+
+    def ctrl_app(self):
+        return (4251, "./src/gprs/osmo-sgsn", "OsmoSGSN", "sgsn")
+
+    def testListSubscribers(self):
+        # TODO. Add command to mark a subscriber as active
+        r = self.do_get('subscriber-list-active-v1')
+        self.assertEquals(r['mtype'], 'GET_REPLY')
+        self.assertEquals(r['var'], 'subscriber-list-active-v1')
+        self.assertEquals(r['value'], None)
+
 def add_bsc_test(suite, workdir):
     if not os.path.isfile(os.path.join(workdir, "src/osmo-bsc/osmo-bsc")):
         print("Skipping the BSC test")
@@ -401,6 +416,13 @@ def add_nat_test(suite, workdir):
     test = unittest.TestLoader().loadTestsFromTestCase(TestCtrlNAT)
     suite.addTest(test)
 
+def add_sgsn_test(suite, workdir):
+    if not os.path.isfile(os.path.join(workdir, "src/gprs/osmo-sgsn")):
+        print("Skipping the SGSN test")
+        return
+    test = unittest.TestLoader().loadTestsFromTestCase(TestCtrlSGSN)
+    suite.addTest(test)
+
 if __name__ == '__main__':
     import argparse
     import sys
@@ -434,5 +456,6 @@ if __name__ == '__main__':
     add_bsc_test(suite, workdir)
     add_nitb_test(suite, workdir)
     add_nat_test(suite, workdir)
+    add_sgsn_test(suite, workdir)
     res = unittest.TextTestRunner(verbosity=verbose_level).run(suite)
     sys.exit(len(res.errors) + len(res.failures))