diff options
| -rw-r--r-- | tests/Makefile.am | 21 | ||||
| -rw-r--r-- | tests/ctrl_test_runner.py | 241 | ||||
| -rw-r--r-- | tests/test_subscriber.ctrl | 27 |
3 files changed, 46 insertions, 243 deletions
diff --git a/tests/Makefile.am b/tests/Makefile.am index cddea0b..e71a250 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -28,7 +28,8 @@ EXTRA_DIST = \ $(TESTSUITE) \ test_nodes.vty \ test_subscriber.vty \ - ctrl_test_runner.py \ + test_subscriber.sql \ + test_subscriber.ctrl \ $(NULL) TESTSUITE = $(srcdir)/testsuite @@ -41,7 +42,7 @@ if ENABLE_EXT_TESTS python-tests: # don't run vty and ctrl tests concurrently so that the ports don't conflict $(MAKE) vty-test - $(PYTHON) $(srcdir)/ctrl_test_runner.py -w $(abs_top_builddir) -v + $(MAKE) ctrl-test VTY_TEST_DB = hlr_vty_test.db @@ -56,6 +57,22 @@ vty-test: -r "$(top_builddir)/src/osmo-hlr -c $(top_srcdir)/doc/examples/osmo-hlr.cfg -l hlr_vty_test.db" \ $(U) $(srcdir)/*.vty -rm -f $(VTY_TEST_DB) + +CTRL_TEST_DB = hlr_ctrl_test.db + +# To update the CTRL script from current application behavior, +# pass -u to ctrl_script_runner.py by doing: +# make ctrl-test U=-u +ctrl-test: + -rm -f $(CTRL_TEST_DB) + sqlite3 $(CTRL_TEST_DB) < $(top_srcdir)/sql/hlr.sql + sqlite3 $(CTRL_TEST_DB) < $(srcdir)/test_subscriber.sql + osmo_verify_transcript_ctrl.py -v \ + -p 4259 \ + -r "$(top_builddir)/src/osmo-hlr -c $(top_srcdir)/doc/examples/osmo-hlr.cfg -l hlr_ctrl_test.db" \ + $(U) $(srcdir)/*.ctrl + -rm -f $(CTRL_TEST_DB) + else python-tests: echo "Not running python-based tests (determined at configure-time)" diff --git a/tests/ctrl_test_runner.py b/tests/ctrl_test_runner.py deleted file mode 100644 index 1c5ec2a..0000000 --- a/tests/ctrl_test_runner.py +++ /dev/null @@ -1,241 +0,0 @@ -#!/usr/bin/env python - -# (C) 2013 by Jacob Erlbeck <jerlbeck@sysmocom.de> -# (C) 2014 by Holger Hans Peter Freyther -# based on vty_test_runner.py: -# (C) 2013 by Katerina Barone-Adesi <kat.obsc@gmail.com> -# (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 <http://www.gnu.org/licenses/>. - -import os -import time -import unittest -import socket -import sys -import struct -import subprocess - -import osmopy.osmoutil as osmoutil - -# add $top_srcdir/contrib to find ipa.py -sys.path.append(os.path.join(sys.path[0], '..', 'contrib')) - -from ipa import Ctrl, IPA - -# to be able to find $top_srcdir/doc/... -confpath = os.path.join(sys.path[0], '..') -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: - 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 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) - - retries = 30 - while True: - try: - sck = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sck.setblocking(1) - sck.connect((host, port)) - except IOError: - retries -= 1 - if retries <= 0: - raise - time.sleep(.1) - continue - break - self.sock = sck - return sck - - def send(self, data): - if verbose: - print "Sending \"%s\"" %(data) - data = Ctrl().add_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 assert_reply(self, r, mtype, var, val): - expect = dict(mtype=mtype, var=var, value=val) - result_matches = all([r.get(k) == expect.get(k) for k in expect.keys()]) - if not result_matches: - print('\nError details:\nGot reply: %r\nExpected reply: %r\n' % (r, expect)) - self.assertTrue(result_matches) - - def assert_set(self, var, val, result_val): - r = self.do_set(var, val) - self.assert_reply(r, 'SET_REPLY', var, result_val) - - def assert_get(self, var, result_val): - r = self.do_get(var) - self.assert_reply(r, 'GET_REPLY', var, result_val) - - def recv_msgs(self): - responses = {} - data = self.sock.recv(4096) - while (len(data)>0): - (head, data) = IPA().split_combined(data) - answer = Ctrl().rem_header(head) - 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: - split = msg.split(None, 1) - rsp['var'] = split[0] - if len(split) > 1: - rsp['value'] = split[1] - else: - rsp['value'] = None - responses[id] = rsp - - if verbose: - print "Decoded replies: ", responses - - return responses - - -class TestCtrlHLR(TestCtrlBase): - - HLR_DB = 'hlr_ctrl_test.db' - HLR_SQL = '%s/sql/hlr.sql' % confpath - HLR_TEST_SQL = '%s/tests/test_subscriber.sql' % confpath - - def setUp(self): - print('\n') - print(os.getcwd()) - assert subprocess.call('sqlite3 %s < %s' % (self.HLR_DB, self.HLR_SQL), shell=True) == 0 - assert subprocess.call('sqlite3 %s < %s' % (self.HLR_DB, self.HLR_TEST_SQL), shell=True) == 0 - super(TestCtrlHLR, self).setUp() - - def tearDown(self): - super(TestCtrlHLR, self).tearDown() - os.unlink(self.HLR_DB) - - def ctrl_command(self): - return ["./src/osmo-hlr", "-c", "doc/examples/osmo-hlr.cfg", '-l', 'hlr_ctrl_test.db'] - - def ctrl_app(self): - return (4259, "./src/osmo-hlr", "OsmoHLR", "hlr") - - def testCtrlErrs(self): - r = self.do_get('invalid') - self.assertEquals(r['mtype'], 'ERROR') - self.assertEquals(r['error'], 'Command not found') - - def testEnableDisablePs(self): - self.assert_set('enable-ps', '901990000000001', 'OK') - self.assert_set('status-ps', '901990000000001', '1') - self.assert_set('enable-ps', '901990000000001', 'OK') - self.assert_set('status-ps', '901990000000001', '1') - self.assert_set('disable-ps', '901990000000001', 'OK') - self.assert_set('status-ps', '901990000000001', '0') - self.assert_set('disable-ps', '901990000000001', 'OK') - self.assert_set('status-ps', '901990000000001', '0') - self.assert_set('enable-ps', '901990000000001', 'OK') - self.assert_set('status-ps', '901990000000001', '1') - -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() - test = unittest.TestLoader().loadTestsFromTestCase(TestCtrlHLR) - suite.addTest(test) - res = unittest.TextTestRunner(verbosity=verbose_level).run(suite) - sys.exit(len(res.errors) + len(res.failures)) - -# vim: tabstop=4 shiftwidth=4 expandtab diff --git a/tests/test_subscriber.ctrl b/tests/test_subscriber.ctrl new file mode 100644 index 0000000..3284ae5 --- /dev/null +++ b/tests/test_subscriber.ctrl @@ -0,0 +1,27 @@ +GET 1 invalid +ERROR 1 Command not found +SET 2 invalid nonsense +ERROR 2 Command not found + +SET 3 enable-ps 901990000000001 +SET_REPLY 3 enable-ps OK +SET 4 status-ps 901990000000001 +SET_REPLY 4 status-ps 1 +SET 5 enable-ps 901990000000001 +SET_REPLY 5 enable-ps OK +SET 6 status-ps 901990000000001 +SET_REPLY 6 status-ps 1 + +SET 7 disable-ps 901990000000001 +SET_REPLY 7 disable-ps OK +SET 8 status-ps 901990000000001 +SET_REPLY 8 status-ps 0 +SET 9 disable-ps 901990000000001 +SET_REPLY 9 disable-ps OK +SET 10 status-ps 901990000000001 +SET_REPLY 10 status-ps 0 + +SET 11 enable-ps 901990000000001 +SET_REPLY 11 enable-ps OK +SET 12 status-ps 901990000000001 +SET_REPLY 12 status-ps 1 |