aboutsummaryrefslogtreecommitdiffstats
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rw-r--r--tests/vty/Makefile.am15
-rwxr-xr-xtests/vty/vty_test_runner.py173
2 files changed, 186 insertions, 2 deletions
diff --git a/tests/vty/Makefile.am b/tests/vty/Makefile.am
index 28e9dc2..be67293 100644
--- a/tests/vty/Makefile.am
+++ b/tests/vty/Makefile.am
@@ -5,7 +5,10 @@ AM_LDFLAGS = -static
LDADD = $(top_builddir)/src/libosmo-sigtran.la \
$(LIBOSMOCORE_LIBS) $(LIBOSMOVTY_LIBS) $(LIBOSMONETIF_LIBS) $(LIBSCTP_LIBS)
-EXTRA_DIST = ss7_asp_test.vty
+EXTRA_DIST = \
+ ss7_asp_test.vty \
+ vty_test_runner.py \
+ $(NULL)
noinst_PROGRAMS = ss7_asp_vty_test
@@ -19,14 +22,22 @@ ext-tests:
echo "Not running python-based external tests (determined at configure-time)"
endif
+vty-python-test: $(BUILT_SOURCES)
+ $(srcdir)/vty_test_runner.py -w $(abs_top_builddir) -v
+
# To update the VTY script from current application behavior,
# pass -u to osmo_verify_transcript_vty.py by doing:
# make vty-test U=-u
-vty-test: ss7_asp_vty_test
+vty-transcript-test: ss7_asp_vty_test
osmo_verify_transcript_vty.py -v \
-p 42043 \
-r "$(builddir)/ss7_asp_vty_test" \
$(U) $(srcdir)/ss7_asp_*.vty
+# don't run multiple tests concurrently so that the ports don't conflict
+vty-test:
+ $(MAKE) vty-python-test
+ $(MAKE) vty-transcript-test
+
check-local:
$(MAKE) $(AM_MAKEFLAGS) ext-tests
diff --git a/tests/vty/vty_test_runner.py b/tests/vty/vty_test_runner.py
new file mode 100755
index 0000000..47ce23c
--- /dev/null
+++ b/tests/vty/vty_test_runner.py
@@ -0,0 +1,173 @@
+#!/usr/bin/env python2
+
+# (C) 2013 by Katerina Barone-Adesi <kat.obsc@gmail.com>
+# (C) 2013 by Holger Hans Peter Freyther
+# (C) 2019 by sysmocom s.f.m.c. GmbH
+# 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, sys
+import time
+import unittest
+import socket
+import subprocess
+import time
+
+import osmopy.obscvty as obscvty
+import osmopy.osmoutil as osmoutil
+from osmopy.osmo_ipa import IPA
+
+# to be able to find $top_srcdir/doc/...
+confpath = os.path.join(sys.path[0], '..')
+
+class TestVTYBase(unittest.TestCase):
+
+ def checkForEndAndExit(self):
+ res = self.vty.command("list")
+ #print ('looking for "exit"\n')
+ self.assert_(res.find(' exit\r') > 0)
+ #print 'found "exit"\nlooking for "end"\n'
+ self.assert_(res.find(' end\r') > 0)
+ #print 'found "end"\n'
+
+ def vty_command(self):
+ raise Exception("Needs to be implemented by a subclass")
+
+ def vty_app(self):
+ raise Exception("Needs to be implemented by a subclass")
+
+ def setUp(self):
+ osmo_vty_cmd = self.vty_command()[:]
+ config_index = osmo_vty_cmd.index('-c')
+ if config_index:
+ cfi = config_index + 1
+ osmo_vty_cmd[cfi] = os.path.join(confpath, osmo_vty_cmd[cfi])
+
+ try:
+ self.proc = osmoutil.popen_devnull(osmo_vty_cmd)
+ except OSError:
+ print >> sys.stderr, "Current directory: %s" % os.getcwd()
+ print >> sys.stderr, "Consider setting -b"
+
+ appstring = self.vty_app()[2]
+ appport = self.vty_app()[0]
+ self.vty = obscvty.VTYInteract(appstring, "127.0.0.1", appport)
+
+ def tearDown(self):
+ if self.vty:
+ self.vty._close_socket()
+ self.vty = None
+ osmoutil.end_proc(self.proc)
+
+class TestVTYSTP(TestVTYBase):
+
+ def vty_command(self):
+ return ["./stp/osmo-stp", "-c",
+ "../doc/examples/osmo-stp-multihome.cfg"]
+
+ def vty_app(self):
+ return (4239, "./stp/osmo-stp", "OsmoSTP", "stp")
+
+ def check_sctp_sock_local(self, laddr_list, lport):
+ path = "/proc/net/sctp/eps"
+ try:
+ with open(path, "r") as fp:
+ #drop first line, contains column names:
+ fp.readline()
+ while True:
+ # Read next line
+ line = fp.readline().strip()
+ if not line:
+ return False
+ print "%s: parsing line: %s" %(path, line)
+ it = line.split()
+ if lport == int(it[5]):
+ print "%s: local port %d found" %(path, lport)
+ itaddr_list = it[8:]
+ if len(itaddr_list) != len(laddr_list):
+ print "%s: addr list mismatch: %r vs %r" % (path, repr(itaddr_list), repr(laddr_list))
+ continue
+ for addr in laddr_list:
+ if addr not in itaddr_list:
+ print "%s: addr not found in list: %s vs %r" % (path, addr, repr(itaddr_list))
+ return False
+ return True
+ return False
+ except IOError as e:
+ print "I/O error({0}): {1}".format(e.errno, e.strerror)
+ return False
+
+ def testMultiHome(self):
+ # first check if STP is listening in required addresses:
+ found = False
+ for i in range(5):
+ if self.check_sctp_sock_local(['127.0.0.1', '127.0.0.2'], 2905):
+ found = True
+ break
+ else:
+ print "[%d] osmo-stp not yet available, retrying in a second" % i
+ time.sleep(1)
+ self.assert_(found)
+ try:
+ proto = socket.IPPROTO_SCTP
+ except AttributeError: # it seems to be not defined under python2?
+ proto = 132
+ s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, proto)
+ s.bind(('127.0.0.3', 0))
+ try:
+ s.connect(('127.0.0.2',2905))
+ except socket.error as msg:
+ s.close()
+ self.assert_(False)
+ print "Connected to STP through SCTP"
+ s.close()
+
+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")
+ parser.add_argument("test_name", nargs="*", help="(parts of) test names to run, case-insensitive")
+ args = parser.parse_args()
+
+ verbose_level = 1
+ if args.verbose:
+ verbose_level = 2
+
+ 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 VTY commands"
+ suite = unittest.TestSuite()
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(TestVTYSTP))
+
+ if args.test_name:
+ osmoutil.pick_tests(suite, *args.test_name)
+
+ res = unittest.TextTestRunner(verbosity=verbose_level, stream=sys.stdout).run(suite)
+ sys.exit(len(res.errors) + len(res.failures))
+
+# vim: shiftwidth=4 expandtab nocin ai