aboutsummaryrefslogtreecommitdiffstats
path: root/tests/vty/vty_test_runner.py
diff options
context:
space:
mode:
authorPau Espin Pedrol <pespin@sysmocom.de>2019-10-15 17:16:22 +0200
committerlaforge <laforge@osmocom.org>2019-10-22 19:38:53 +0000
commit2c3ff8c81bcbf284da40dc7b3828bf878bf1be89 (patch)
tree6b9ac69b32218148cd5a61fe5c2787ae356c4445 /tests/vty/vty_test_runner.py
parentdec38c141694412f7c114bcbf01cf71beacb73e0 (diff)
tests: Introduce test for multi-homing STP features
Config file sets omo-stp instance to bind on 2 IP addresses, and then the test verfies through linux /proc/net/sctp/* that binding is done correctly and that it can be reached from another remote address to one of the configured addresses. Change-Id: Ifa11b1fc882dff415405f62024e94bed67228866
Diffstat (limited to 'tests/vty/vty_test_runner.py')
-rwxr-xr-xtests/vty/vty_test_runner.py173
1 files changed, 173 insertions, 0 deletions
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