From: Max msuraev@sysmocom.de
Factor out 2, add 3 functions. Those functions are simple wrappers around hex strings specific to IPA protocol. Not all of them are utilized at the moment but they were checked with wireshark while working on the tests. It might come in handy if we'd like to further expand related test harness in future. The same goes for optional verbosity argument which is not used right now but will be handy for future debugging. --- openbsc/tests/vty_test_runner.py | 29 +++++++++++++++++++++++++++-- 1 file changed, 27 insertions(+), 2 deletions(-)
diff --git a/openbsc/tests/vty_test_runner.py b/openbsc/tests/vty_test_runner.py index ecf5204..d574129 100644 --- a/openbsc/tests/vty_test_runner.py +++ b/openbsc/tests/vty_test_runner.py @@ -736,13 +736,13 @@ class TestVTYNAT(TestVTYGenericBSC): self.assertEqual(data, "\x00\x01\xfe\x04")
print "Going to send ID_RESP response" - res = ussdSocket.send("\x00\x07\xfe\x05\x00\x04\x01\x6b\x65\x79") + res = ipa_send_resp(ussdSocket, "\x6b\x65\x79") self.assertEqual(res, 10)
# initiating PING/PONG cycle to know, that the ID_RESP message has been processed
print "Going to send PING request" - res = ussdSocket.send("\x00\x01\xfe\x00") + res = ipa_send_ping(ussdSocket) self.assertEqual(res, 4)
print "Expecting PONG response" @@ -1007,6 +1007,31 @@ def add_nat_test(suite, workdir): test = unittest.TestLoader().loadTestsFromTestCase(TestVTYNAT) suite.addTest(test)
+def ipa_send_pong(x, verbose = False): + if (verbose): + print "\tBSC -> NAT: PONG!" + return x.send("\x00\x01\xfe\x01") + +def ipa_send_ping(x, verbose = False): + if (verbose): + print "\tBSC -> NAT: PING?" + return x.send("\x00\x01\xfe\x00") + +def ipa_send_ack(x, verbose = False): + if (verbose): + print "\tBSC -> NAT: IPA ID ACK" + return x.send("\x00\x01\xfe\x06") + +def ipa_send_reset(x, verbose = False): + if (verbose): + print "\tBSC -> NAT: RESET" + return x.send("\x00\x12\xfd\x09\x00\x03\x05\x07\x02\x42\xfe\x02\x42\xfe\x06\x00\x04\x30\x04\x01\x20") + +def ipa_send_resp(x, tk, verbose = False): + if (verbose): + print "\tBSC -> NAT: IPA ID RESP" + return x.send("\x00\x07\xfe\x05\x00\x04\x01" + tk) + 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")
From: Max msuraev@sysmocom.de
Add vty tests for BSC configuration reloading. Load BSCs configuration on bscs-config-file command: * remove all runtime configured BSC not in the config file * close connections to all BSC with updated token value
Fixes: OS#1670 Sponsored-by: On-Waves ehf --- openbsc/include/openbsc/bsc_nat.h | 5 ++ openbsc/src/osmo-bsc_nat/bsc_nat_vty.c | 29 ++++++++ openbsc/tests/vty_test_runner.py | 120 ++++++++++++++++++++++++++++++++- 3 files changed, 153 insertions(+), 1 deletion(-)
diff --git a/openbsc/include/openbsc/bsc_nat.h b/openbsc/include/openbsc/bsc_nat.h index 5ccc02e..94ab0e5 100644 --- a/openbsc/include/openbsc/bsc_nat.h +++ b/openbsc/include/openbsc/bsc_nat.h @@ -35,6 +35,7 @@ #include <osmocom/gsm/protocol/gsm_04_08.h>
#include <regex.h> +#include <stdbool.h>
#define DIR_BSC 1 #define DIR_MSC 2 @@ -164,6 +165,10 @@ struct bsc_config { /* audio handling */ int max_endpoints;
+ /* used internally for reload handling */ + bool remove; + bool token_updated; + /* backpointer */ struct bsc_nat *nat;
diff --git a/openbsc/src/osmo-bsc_nat/bsc_nat_vty.c b/openbsc/src/osmo-bsc_nat/bsc_nat_vty.c index b7b49e6..55f3dbf 100644 --- a/openbsc/src/osmo-bsc_nat/bsc_nat_vty.c +++ b/openbsc/src/osmo-bsc_nat/bsc_nat_vty.c @@ -39,6 +39,7 @@ #include <osmocom/sccp/sccp.h>
#include <stdlib.h> +#include <stdbool.h>
static struct bsc_nat *_nat;
@@ -513,6 +514,8 @@ DEFUN(cfg_nat_include, { char *path; int rc; + struct bsc_config *cf1, *cf2; + struct bsc_connection *con1, *con2;
if ('/' == argv[0][0]) bsc_replace_string(_nat, &_nat->resolved_path, argv[0]); @@ -523,13 +526,32 @@ DEFUN(cfg_nat_include, talloc_free(path); }
+ llist_for_each_entry_safe(cf1, cf2, &_nat->bsc_configs, entry) { + cf1->remove = true; + cf1->token_updated = false; + } + rc = vty_read_config_file(_nat->resolved_path, NULL); if (rc < 0) { vty_out(vty, "Failed to parse the config file %s: %s%s", _nat->resolved_path, strerror(-rc), VTY_NEWLINE); return CMD_WARNING; } + bsc_replace_string(_nat, &_nat->include_file, argv[0]); + + llist_for_each_entry_safe(con1, con2, &_nat->bsc_connections, + list_entry) { + if (con1->cfg) + if (con1->cfg->token_updated || con1->cfg->remove) + bsc_close_connection(con1); + } + + llist_for_each_entry_safe(cf1, cf2, &_nat->bsc_configs, entry) { + if (cf1->remove) + bsc_config_free(cf1); + } + return CMD_SUCCESS; }
@@ -846,6 +868,7 @@ DEFUN(cfg_bsc, cfg_bsc_cmd, "bsc BSC_NR", if (!bsc) return CMD_WARNING;
+ bsc->remove = false; vty->index = bsc; vty->node = NAT_BSC_NODE;
@@ -858,6 +881,12 @@ DEFUN(cfg_bsc_token, cfg_bsc_token_cmd, "token TOKEN", { struct bsc_config *conf = vty->index;
+ if (strncmp(conf->token, argv[0], 128) != 0) { + vty_out(vty, "updated token: %s -> %s%s", conf->token, argv[0], + VTY_NEWLINE); + conf->token_updated = true; + } + bsc_replace_string(conf, &conf->token, argv[0]); return CMD_SUCCESS; } diff --git a/openbsc/tests/vty_test_runner.py b/openbsc/tests/vty_test_runner.py index d574129..59fdd2f 100644 --- a/openbsc/tests/vty_test_runner.py +++ b/openbsc/tests/vty_test_runner.py @@ -620,12 +620,61 @@ class TestVTYBSC(TestVTYGenericBSC): class TestVTYNAT(TestVTYGenericBSC):
def vty_command(self): - return ["./src/osmo-bsc_nat/osmo-bsc_nat", "-c", + return ["./src/osmo-bsc_nat/osmo-bsc_nat", "-l", "127.0.0.1", "-c", "doc/examples/osmo-bsc_nat/osmo-bsc_nat.cfg"]
def vty_app(self): return (4244, "src/osmo-bsc_nat/osmo-bsc_nat", "OsmoBSCNAT", "nat")
+ def testBSCreload(self): + # use separate ip to avoid interference with other tests + ip = "127.0.0.4" + self.vty.enable() + bscs1 = self.vty.command("show bscs-config") + nat_bsc_reload(self) + bscs2 = self.vty.command("show bscs-config") + # check that multiple calls to bscs-config-file give the same result + self.assertEquals(bscs1, bscs2) + + # add new bsc + self.vty.command("configure terminal") + self.vty.command("nat") + self.vty.command("bsc 5") + self.vty.command("token key") + self.vty.command("location_area_code 666") + self.vty.command("end") + + # update bsc token + self.vty.command("configure terminal") + self.vty.command("nat") + self.vty.command("bsc 1") + self.vty.command("token xyu") + self.vty.command("end") + + nat_msc_ip(self, ip) + msc = nat_msc_test(self, ip) + b0 = nat_bsc_sock_test(0, "lol") + b1 = nat_bsc_sock_test(1, "xyu") + b2 = nat_bsc_sock_test(5, "key") + + self.assertEquals("3 BSCs configured", self.vty.command("show nat num-bscs-configured")) + self.assertTrue(3 == nat_bsc_num_con(self)) + self.assertEquals("MSC is connected: 1", self.vty.command("show msc connection")) + + nat_bsc_reload(self) + bscs2 = self.vty.command("show bscs-config") + # check that the reset to initial config succeeded + self.assertEquals(bscs1, bscs2) + + self.assertEquals("2 BSCs configured", self.vty.command("show nat num-bscs-configured")) + self.assertTrue(1 == nat_bsc_num_con(self)) + rem = self.vty.command("show bsc connections").split(' ') + # remaining connection is for BSC0 + self.assertEquals('0', rem[2]) + # remaining connection is authorized + self.assertEquals('1', rem[4]) + self.assertEquals("MSC is connected: 1", self.vty.command("show msc connection")) + def testVtyTree(self): self.vty.enable() self.assertTrue(self.vty.verify('configure terminal', [''])) @@ -1032,6 +1081,75 @@ def ipa_send_resp(x, tk, verbose = False): print "\tBSC -> NAT: IPA ID RESP" return x.send("\x00\x07\xfe\x05\x00\x04\x01" + tk)
+def nat_bsc_reload(x): + x.vty.command("configure terminal") + x.vty.command("nat") + x.vty.command("bscs-config-file bscs.config") + x.vty.command("end") + +def nat_msc_ip(x, ip): + x.vty.command("configure terminal") + x.vty.command("nat") + x.vty.command("msc ip " + ip) + x.vty.command("end") + +def data2str(d): + return "".join("{:02x}".format(ord(c)) for c in d) + +def nat_msc_test(x, ip, verbose = False): + msc = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + msc.settimeout(32) + msc.bind((ip, 5000)) + msc.listen(5) + if (verbose): + print "MSC is ready at " + ip + while "MSC is connected: 0" == x.vty.command("show msc connection"): + conn, addr = msc.accept() + if (verbose): + print "MSC got connection from ", addr + return conn + +def ipa_handle_small(x, verbose = False): + s = data2str(x.recv(4)) + if "0001fe00" == s: + if (verbose): + print "\tBSC <- NAT: PING?" + ipa_send_pong(x, verbose) + elif "0001fe06" == s: + if (verbose): + print "\tBSC <- NAT: IPA ID ACK" + ipa_send_ack(x, verbose) + elif "0001fe00" == s: + if (verbose): + print "\tBSC <- NAT: PONG!" + else: + if (verbose): + print "\tBSC <- NAT: ", s + +def ipa_handle_resp(x, tk, verbose = False): + s = data2str(x.recv(38)) + if "0023fe040108010701020103010401050101010011" in s: + ipa_send_resp(x, tk, verbose) + else: + if (verbose): + print "\tBSC <- NAT: ", s + +def nat_bsc_num_con(x): + return len(x.vty.command("show bsc connections").split('\n')) + +def nat_bsc_sock_test(nr, tk, verbose = False): + bsc = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + bsc.bind(('127.0.0.1' + str(nr), 0)) + bsc.connect(('127.0.0.1', 5000)) + if (verbose): + print "BSC%d " %nr + print "\tconnected to %s:%d" % bsc.getpeername() + ipa_handle_small(bsc, verbose) + ipa_handle_resp(bsc, tk, verbose) + bsc.recv(27) # MGCP msg + ipa_handle_small(bsc, verbose) + return bsc + 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")
On 13 Apr 2016, at 05:36, msuraev@sysmocom.de wrote:
# use separate ip to avoid interference with other testsip = "127.0.0.4"
First question with this comment: What other tests? Do you have an example? If yes, it should be in the comment. :)
====================================================================== ERROR: testBSCreload (__main__.TestVTYNAT) ---------------------------------------------------------------------- Traceback (most recent call last): File "./vty_test_runner.py", line 655, in testBSCreload msc = nat_msc_test(self, ip) File "./vty_test_runner.py", line 1102, in nat_msc_test msc.bind((ip, 5000)) File "<string>", line 1, in bind error: [Errno 99] Cannot assign requested address
---------------------------------------------------------------------- Ran 41 tests in 41.953s
http://jenkins.osmocom.org/jenkins/job/OpenBSC/2146/MGCP=--disable-mgcp-tran...
-
Holger Freyther -
msuraev＠sysmocom.de