From 0760a83910a487e0bdd4e958829616fd491d95ba Mon Sep 17 00:00:00 2001 From: Jacob Erlbeck Date: Mon, 16 Sep 2013 11:20:28 +0200 Subject: [PATCH] ctrl: Add test script for the BSC control interface This script is similar to vty_test_runner.py but tests the control interface instead. It currently tests some error cases, BTS status queries, and setting/clearing rf_locked. --- openbsc/tests/Makefile.am | 3 +- openbsc/tests/ctrl_test_runner.py | 239 ++++++++++++++++++++++++++++++ 2 files changed, 241 insertions(+), 1 deletion(-) create mode 100644 openbsc/tests/ctrl_test_runner.py diff --git a/openbsc/tests/Makefile.am b/openbsc/tests/Makefile.am index 0597c14d0..f2dc057fa 100644 --- a/openbsc/tests/Makefile.am +++ b/openbsc/tests/Makefile.am @@ -27,7 +27,7 @@ $(srcdir)/package.m4: $(top_srcdir)/configure.ac echo ' [$(PACKAGE_URL)])'; \ } >'$(srcdir)/package.m4' -EXTRA_DIST = testsuite.at $(srcdir)/package.m4 $(TESTSUITE) vty_test_runner.py +EXTRA_DIST = testsuite.at $(srcdir)/package.m4 $(TESTSUITE) vty_test_runner.py ctrl_test_runner.py TESTSUITE = $(srcdir)/testsuite DISTCLEANFILES = atconfig @@ -36,6 +36,7 @@ python-tests: $(BUILT_SOURCES) osmotestvty.py -p $(abs_top_srcdir) -w $(abs_top_builddir) -v osmotestconfig.py -p $(abs_top_srcdir) -w $(abs_top_builddir) -v $(PYTHON) $(srcdir)/vty_test_runner.py -w $(abs_top_builddir) -v + $(PYTHON) $(srcdir)/ctrl_test_runner.py -w $(abs_top_builddir) -v rm -f $(top_builddir)/hlr.sqlite3 else python-tests: $(BUILT_SOURCES) diff --git a/openbsc/tests/ctrl_test_runner.py b/openbsc/tests/ctrl_test_runner.py new file mode 100644 index 000000000..24732e031 --- /dev/null +++ b/openbsc/tests/ctrl_test_runner.py @@ -0,0 +1,239 @@ +#!/usr/bin/env python + +# (C) 2013 by Jacob Erlbeck +# based on vty_test_runner.py: +# (C) 2013 by Katerina Barone-Adesi +# (C) 2013 by Holger Hans Peter Freyther +# based on bsc_control.py. + +# 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 . + +import os +import time +import unittest +import socket +import sys +import struct + +import osmopy.obscvty as obscvty +import osmopy.osmoutil as osmoutil + +confpath = '.' +verbose = False + +class TestCtrlBase(unittest.TestCase): + + def ctrl_command(self): + raise Exception("Needs to be implemented by a subclass") + + def ctrl_app(self): + raise Exception("Needs to be implemented by a subclass") + + def setUp(self): + osmo_ctrl_cmd = self.ctrl_command()[:] + config_index = osmo_ctrl_cmd.index('-c') + if config_index: + cfi = config_index + 1 + osmo_ctrl_cmd[cfi] = os.path.join(confpath, osmo_ctrl_cmd[cfi]) + + try: + print "Launch: %s from %s" % (' '.join(osmo_ctrl_cmd), os.getcwd()) + self.proc = osmoutil.popen_devnull(osmo_ctrl_cmd) + except OSError: + print >> sys.stderr, "Current directory: %s" % os.getcwd() + print >> sys.stderr, "Consider setting -b" + time.sleep(2) + + appstring = self.ctrl_app()[2] + appport = self.ctrl_app()[0] + self.connect("127.0.0.1", appport) + self.next_id = 1000 + + def tearDown(self): + self.disconnect() + osmoutil.end_proc(self.proc) + + def prefix_ipa_ctrl_header(self, data): + return struct.pack(">HBB", len(data)+1, 0xee, 0) + data + + def remove_ipa_ctrl_header(self, data): + if (len(data) < 4): + raise BaseException("Answer too short!") + (plen, ipa_proto, osmo_proto) = struct.unpack(">HBB", data[:4]) + if (plen + 3 > len(data)): + print "Warning: Wrong payload length (expected %i, got %i)" % (plen, len(data) - 3) + if (ipa_proto != 0xee or osmo_proto != 0): + raise BaseException("Wrong protocol in answer!") + + return data[4:plen+3], data[plen+3:] + + def disconnect(self): + if not (self.sock is None): + self.sock.close() + + def connect(self, host, port): + if verbose: + print "Connecting to host %s:%i" % (host, port) + + sck = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sck.setblocking(1) + sck.connect((host, port)) + self.sock = sck + return sck + + def send(self, data): + if verbose: + print "Sending \"%s\"" %(data) + data = self.prefix_ipa_ctrl_header(data) + return self.sock.send(data) == len(data) + + def send_set(self, var, value, id): + setmsg = "SET %s %s %s" %(id, var, value) + return self.send(setmsg) + + def send_get(self, var, id): + getmsg = "GET %s %s" %(id, var) + return self.send(getmsg) + + def do_set(self, var, value): + id = self.next_id + self.next_id += 1 + self.send_set(var, value, id) + return self.recv_msgs()[id] + + def do_get(self, var): + id = self.next_id + self.next_id += 1 + self.send_get(var, id) + return self.recv_msgs()[id] + + def recv_msgs(self): + responses = {} + data = self.sock.recv(4096) + while (len(data)>0): + (answer, data) = self.remove_ipa_ctrl_header(data) + if verbose: + print "Got message:", answer + (mtype, id, msg) = answer.split(None, 2) + id = int(id) + rsp = {'mtype': mtype, 'id': id} + if mtype == "ERROR": + rsp['error'] = msg + else: + [rsp['var'], rsp['value']] = msg.split(None, 2) + + responses[id] = rsp + + if verbose: + print "Decoded replies: ", responses + + return responses + + +class TestCtrlBSC(TestCtrlBase): + + def tearDown(self): + TestCtrlBase.tearDown(self) + os.unlink("tmp_dummy_sock") + + def ctrl_command(self): + return ["./src/osmo-bsc/osmo-bsc", "-r", "tmp_dummy_sock", "-c", + "doc/examples/osmo-bsc/osmo-bsc.cfg"] + + def ctrl_app(self): + return (4249, "./src/osmo-bsc/osmo-bsc", "OsmoBSC", "bsc") + + def testCtrlErrs(self): + r = self.do_get('invalid') + self.assertEquals(r['mtype'], 'ERROR') + self.assertEquals(r['error'], 'Command not found') + + r = self.do_get('bts') + self.assertEquals(r['mtype'], 'ERROR') + self.assertEquals(r['error'], 'Error while parsing the index.') + + r = self.do_get('bts.999') + self.assertEquals(r['mtype'], 'ERROR') + self.assertEquals(r['error'], 'Error while resolving object') + + def testRfLock(self): + r = self.do_get('bts.0.rf_state') + self.assertEquals(r['mtype'], 'GET_REPLY') + self.assertEquals(r['var'], 'bts.0.rf_state') + self.assertEquals(r['value'], 'inoperational,unlocked,on') + + r = self.do_set('rf_locked', '1') + self.assertEquals(r['mtype'], 'SET_REPLY') + self.assertEquals(r['var'], 'rf_locked') + self.assertEquals(r['value'], '1') + + time.sleep(1.5) + + r = self.do_get('bts.0.rf_state') + self.assertEquals(r['mtype'], 'GET_REPLY') + self.assertEquals(r['var'], 'bts.0.rf_state') + self.assertEquals(r['value'], 'inoperational,locked,off') + + r = self.do_set('rf_locked', '0') + self.assertEquals(r['mtype'], 'SET_REPLY') + self.assertEquals(r['var'], 'rf_locked') + self.assertEquals(r['value'], '0') + + time.sleep(1.5) + + r = self.do_get('bts.0.rf_state') + self.assertEquals(r['mtype'], 'GET_REPLY') + self.assertEquals(r['var'], 'bts.0.rf_state') + self.assertEquals(r['value'], 'inoperational,unlocked,on') + +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") + return + test = unittest.TestLoader().loadTestsFromTestCase(TestCtrlBSC) + suite.addTest(test) + +if __name__ == '__main__': + import argparse + import sys + + workdir = '.' + + parser = argparse.ArgumentParser() + parser.add_argument("-v", "--verbose", dest="verbose", + action="store_true", help="verbose mode") + parser.add_argument("-p", "--pythonconfpath", dest="p", + help="searchpath for config") + parser.add_argument("-w", "--workdir", dest="w", + help="Working directory") + args = parser.parse_args() + + verbose_level = 1 + if args.verbose: + verbose_level = 2 + verbose = True + + if args.w: + workdir = args.w + + if args.p: + confpath = args.p + + print "confpath %s, workdir %s" % (confpath, workdir) + os.chdir(workdir) + print "Running tests for specific control commands" + suite = unittest.TestSuite() + add_bsc_test(suite, workdir) + res = unittest.TextTestRunner(verbosity=verbose_level).run(suite) + sys.exit(len(res.errors) + len(res.failures))