This is a set of test scripts for osmocom projects.
Currently, it's tested on openbsc and osmo-pcu.
Scripts: osmotestvty.py osmodumpdoc.py osmotestconfig.py
The scripts are designed to be run from make check,
but can be run independently as well.
As a general rule, run them in the top dir of a project.
diff --git a/osmopy/__init__.py b/osmopy/__init__.py
new file mode 100644
index 0000000..f6ed99b
--- /dev/null
+++ b/osmopy/__init__.py
@@ -0,0 +1,5 @@
+#!/usr/bin/env python
+__version__ = '0.0.1'
+
+__all__ = ['obscvty', 'osmodumpdoc', 'osmotestconfig', 'osmotestvty',
+ 'osmoutil']
diff --git a/osmopy/obscvty.py b/osmopy/obscvty.py
new file mode 100755
index 0000000..b402cfe
--- /dev/null
+++ b/osmopy/obscvty.py
@@ -0,0 +1,107 @@
+# Copyright (C) 2012 Holger Hans Peter Freyther
+# Copyright (C) 2013 Katerina Barone-Adesi
+# 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 2 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/>.
+
+#
+# VTY helper code for OpenBSC
+#
+import socket
+
+
+class VTYInteract(object):
+ def __init__(self, name, host, port):
+ self.name = name
+ self.host = host
+ self.port = port
+
+ self.socket = None
+ self.norm_end = '\r\n%s> ' % self.name
+ self.priv_end = '\r\n%s# ' % self.name
+
+ def _close_socket(self):
+ self.socket.close()
+ self.socket = None
+
+ def _is_end(self, text, ends):
+ for end in ends:
+ if text.endswith(end):
+ return end
+ return ""
+
+ def _common_command(self, request, close=False, ends=None):
+ if not ends:
+ ends = [self.norm_end, self.priv_end]
+ if not self.socket:
+ self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ self.socket.setblocking(1)
+ self.socket.connect((self.host, self.port))
+ self.socket.recv(4096)
+
+ # Now send the command
+ self.socket.send("%s\r" % request)
+ res = ""
+ end = ""
+
+ # Unfortunately, timeout and recv don't always play nicely
+ while True:
+ data = self.socket.recv(4096)
+ res = "%s%s" % (res, data)
+ if not res: # yes, this is ugly
+ raise IOError("Failed to read data (did the app crash?)")
+ end = self._is_end(res, ends)
+ if end:
+ break
+
+ if close:
+ self._close_socket()
+ return res[len(request) + 2: -len(end)]
+
+ # There's no close parameter, as close=True makes this useless
+ def enable(self):
+ self.command("enable")
+
+ """Run a command on the vty"""
+ def command(self, request, close=False):
+ return self._common_command(request, close)
+
+ """Run enable, followed by another command"""
+ def enabled_command(self, request, close=False):
+ self.enable()
+ return self._common_command(request, close)
+
+ """Verify, ignoring leading/trailing whitespace"""
+ # inspired by diff -w, though not identical
+ def w_verify(self, command, results, close=False, loud=True):
+ return self.verify(command, results, close, loud, lambda x: x.strip())
+
+ """Verify that a command has the expected results
+
+ command = the command to verify
+ results = the expected results [line1, line2, ...]
+ close = True to close the socket after running the verify
+ loud = True to show what was expected and what actually happend, stdout
+ f = A function to run over the expected and actual results, before compare
+
+ Returns True iff the expected and actual results match"""
+ def verify(self, command, results, close=False, loud=True, f=None):
+ res = self.command(command, close).split('\r\n')
+ if f:
+ res = map(f, res)
+ results = map(f, results)
+
+ if loud:
+ if res != results:
+ print "Rec: %s\nExp: %s" % (res, results)
+
+ return res == results
diff --git a/osmopy/osmodumpdoc.py b/osmopy/osmodumpdoc.py
new file mode 100755
index 0000000..31ddafd
--- /dev/null
+++ b/osmopy/osmodumpdoc.py
@@ -0,0 +1,97 @@
+#!/usr/bin/env python
+
+# Make sure this code is in sync with the BTS directory.
+# Fixes may need to be applied to both.
+
+"""Start the process and dump the documentation to the doc dir."""
+
+import subprocess
+import time
+import os
+import sys
+
+import osmopy.obscvty as obscvty
+import osmopy.osmoutil as osmoutil
+
+
+def dump_doc(name, port, filename):
+ vty = obscvty.VTYInteract(name, "127.0.0.1", port)
+ xml = vty.command("show online-help")
+ # Now write everything until the end to the file
+ out = open(filename, 'w')
+ out.write(xml)
+ out.close()
+
+
+"""Dump the config of all the apps.
+
+Returns the number of apps configs could not be dumped for."""
+
+
+def dump_configs(apps, configs):
+ failures = 0
+ successes = 0
+
+ try: # make sure the doc directory exists
+ os.mkdir('doc')
+ except OSError: # it probably does
+ pass
+
+ for app in apps:
+ appname = app[3]
+ print "Starting app for %s" % appname
+ proc = None
+ cmd = [app[1], "-c", configs[appname][0]]
+ try:
+ proc = subprocess.Popen(cmd, stdin=None, stdout=None)
+ except OSError: # Probably a missing binary
+ print >> sys.stderr, "Skipping app %s" % appname
+ failures += 1
+ else:
+ time.sleep(1)
+ try:
+ dump_doc(app[2], app[0], 'doc/%s_vty_reference.xml' % appname)
+ successes += 1
+ except IOError: # Generally a socket issue
+ print >> sys.stderr, "%s: couldn't connect, skipping" % appname
+ failures += 1
+ finally:
+ osmoutil.end_proc(proc)
+
+ return (failures, successes)
+
+
+if __name__ == '__main__':
+ import argparse
+
+ confpath = "."
+ workdir = "."
+
+ parser = argparse.ArgumentParser()
+ parser.add_argument("-p", "--pythonconfpath", dest="p",
+ help="searchpath for config (osmoappdesc)")
+ parser.add_argument("-w", "--workdir", dest="w",
+ help="Working directory to run in")
+ args = parser.parse_args()
+
+ if args.p:
+ confpath = args.p
+
+ if args.w:
+ workdir = args.w
+
+ osmoappdesc = None
+ try:
+ osmoappdesc = osmoutil.importappconf(confpath, "osmoappdesc")
+ except ImportError as e:
+ print >> sys.stderr, "osmoappdesc not found, set searchpath with -p"
+ sys.exit(1)
+
+ os.chdir(workdir)
+ num_fails, num_sucs = dump_configs(
+ osmoappdesc.apps, osmoappdesc.app_configs)
+ if num_fails > 0:
+ print >> sys.stderr, "Warning: Skipped %s apps" % num_fails
+ if 0 == num_sucs:
+ print >> sys.stderr, "Nothing run, wrong working dir? Set with -w"
+ sys.exit(num_fails)
diff --git a/osmopy/osmotestconfig.py b/osmopy/osmotestconfig.py
new file mode 100755
index 0000000..f04f534
--- /dev/null
+++ b/osmopy/osmotestconfig.py
@@ -0,0 +1,197 @@
+#!/usr/bin/env python
+
+# (C) 2013 by Katerina Barone-Adesi <kat.obsc@gmail.com>
+# 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 os.path
+import time
+import sys
+import tempfile
+
+import osmopy.obscvty as obscvty
+import osmopy.osmoutil as osmoutil
+
+
+# Return true iff all the tests for the given config pass
+def test_config(app_desc, config, tmpdir, verbose=True):
+ try:
+ test_config_atest(app_desc, config, verify_doc, verbose)
+
+ newconfig = copy_config(tmpdir, config)
+ test_config_atest(app_desc, newconfig, write_config, verbose)
+ test_config_atest(app_desc, newconfig, token_vty_command, verbose)
+ return 0
+
+ # If there's a socket error, skip the rest of the tests for this config
+ except IOError:
+ return 1
+
+
+def test_config_atest(app_desc, config, run_test, verbose=True):
+ proc = None
+ ret = None
+ try:
+ cmd = [app_desc[1], "-c", config]
+ if verbose:
+ print "Verifying %s, test %s" % (' '.join(cmd), run_test.__name__)
+
+ proc = osmoutil.popen_devnull(cmd)
+ time.sleep(1)
+ end = app_desc[2]
+ port = app_desc[0]
+ vty = obscvty.VTYInteract(end, "127.0.0.1", port)
+ ret = run_test(vty)
+
+ except IOError as se:
+ print >> sys.stderr, "Failed to verify %s" % ' '.join(cmd)
+ print >> sys.stderr, "Error was %s" % se
+ raise se
+
+ finally:
+ if proc:
+ osmoutil.end_proc(proc)
+
+ return ret
+
+
+def copy_config(dirname, config):
+ try:
+ os.stat(dirname)
+ except OSError:
+ os.mkdir(dirname)
+ else:
+ remove_tmpdir(dirname)
+ os.mkdir(dirname)
+
+ prefix = os.path.basename(config)
+ tmpfile = tempfile.NamedTemporaryFile(
+ dir=dirname, prefix=prefix, delete=False)
+ tmpfile.write(open(config).read())
+ tmpfile.close()
+ # This works around the precautions NamedTemporaryFile is made for...
+ return tmpfile.name
+
+
+def write_config(vty):
+ new_config = vty.enabled_command("write")
+ return new_config.split(' ')[-1]
+
+
+# The only purpose of this function is to verify a working vty
+def token_vty_command(vty):
+ vty.command("help")
+ return True
+
+
+# This may warn about the same doc missing multiple times, by design
+def verify_doc(vty):
+ xml = vty.command("show online-help")
+ split_at = "<command"
+ all_errs = []
+ for command in xml.split(split_at):
+ if "(null)" in command:
+ lines = command.split("\n")
+ cmd_line = split_at + lines[0]
+ err_lines = []
+ for line in lines:
+ if '(null)' in line:
+ err_lines.append(line)
+
+ all_errs.append(err_lines)
+
+ print >> sys.stderr, \
+ "Documentation error (missing docs): \n%s\n%s\n" % (
+ cmd_line, '\n'.join(err_lines))
+
+ return (len(all_errs), all_errs)
+
+
+# Skip testing the configurations of anything that hasn't been compiled
+def app_exists(app_desc):
+ cmd = app_desc[1]
+ return os.path.exists(cmd)
+
+
+def remove_tmpdir(tmpdir):
+ files = os.listdir(tmpdir)
+ for f in files:
+ os.unlink(os.path.join(tmpdir, f))
+ os.rmdir(tmpdir)
+
+
+def check_configs_tested(basedir, app_configs):
+ configs = []
+ for root, dirs, files in os.walk(basedir):
+ for f in files:
+ if f.endswith(".cfg"):
+ configs.append(os.path.join(root, f))
+ for config in configs:
+ found = False
+ for app in app_configs:
+ if config in app_configs[app]:
+ found = True
+ if not found:
+ print >> sys.stderr, "Warning: %s is not being tested" % config
+
+
+def test_all_apps(apps, app_configs, tmpdir="writtenconfig", verbose=True,
+ rmtmp=False):
+ check_configs_tested("doc/examples/", app_configs)
+ errors = 0
+ for app in apps:
+ if not app_exists(app):
+ print >> sys.stderr, "Skipping app %s (not found)" % app[1]
+ continue
+
+ configs = app_configs[app[3]]
+ for config in configs:
+ errors |= test_config(app, config, tmpdir, verbose)
+
+ if rmtmp:
+ remove_tmpdir(tmpdir)
+
+ return errors
+
+
+if __name__ == '__main__':
+ import argparse
+
+ confpath = "."
+
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--e1nitb", action="store_true", dest="e1nitb")
+ parser.add_argument("-v", "--verbose", dest="verbose",
+ action="store_true", help="verbose mode")
+ parser.add_argument("-p", "--pythonconfpath", dest="p",
+ help="searchpath for config")
+ args = parser.parse_args()
+
+ if args.p:
+ confpath = args.p
+
+ osmoappdesc = None
+ try:
+ osmoappdesc = osmoutil.importappconf(confpath, "osmoappdesc")
+ except ImportError as e:
+ print >> sys.stderr, "osmoappdesc not found, set searchpath with -p"
+ sys.exit(1)
+
+ apps = osmoappdesc.apps
+ configs = osmoappdesc.app_configs
+
+ if args.e1nitb:
+ configs['nitb'].extend(osmoappdesc.nitb_e1_configs)
+
+ sys.exit(test_all_apps(apps, configs, verbose=args.verbose))
diff --git a/osmopy/osmotestvty.py b/osmopy/osmotestvty.py
new file mode 100755
index 0000000..2ee877a
--- /dev/null
+++ b/osmopy/osmotestvty.py
@@ -0,0 +1,89 @@
+#!/usr/bin/env python
+
+# (C) 2013 by Katerina Barone-Adesi <kat.obsc@gmail.com>
+# 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 time
+import unittest
+
+import osmopy.obscvty as obscvty
+import osmopy.osmoutil as osmoutil
+
+"""Test a VTY. Warning: osmoappdesc must be imported first."""
+
+
+class TestVTY(unittest.TestCase):
+ def setUp(self):
+ osmo_vty_cmd = osmoappdesc.vty_command
+ try:
+ self.proc = osmoutil.popen_devnull(osmo_vty_cmd)
+ except OSError:
+ print >> sys.stderr, "Current directory: %s" % os.getcwd()
+ print >> sys.stderr, "Consider setting -w"
+ time.sleep(1)
+
+ appstring = osmoappdesc.vty_app[2]
+ appport = osmoappdesc.vty_app[0]
+ self.vty = obscvty.VTYInteract(appstring, "127.0.0.1", appport)
+
+ def tearDown(self):
+ self.vty = None
+ osmoutil.end_proc(self.proc)
+
+ def test_history(self):
+ t1 = "show version"
+ self.vty.command(t1)
+ test_str = "show history"
+ assert(self.vty.w_verify(test_str, [t1]))
+
+ def test_unknown_command(self):
+ test_str = "help show"
+ assert(self.vty.verify(test_str, ['% Unknown command.']))
+
+ def test_terminal_length(self):
+ test_str = "terminal length 20"
+ assert(self.vty.verify(test_str, ['']))
+
+
+if __name__ == '__main__':
+ import argparse
+ import os
+ import sys
+
+ workdir = "."
+ confpath = "."
+
+ parser = argparse.ArgumentParser()
+ parser.add_argument("-p", "--pythonconfpath", dest="p",
+ help="searchpath for config")
+ parser.add_argument("-w", "--workdir", dest="w",
+ help="Working directory to run in")
+ args = parser.parse_args()
+
+ if args.w:
+ workdir = args.w
+
+ if args.p:
+ confpath = args.p
+ osmoappdesc = None
+ try:
+ osmoappdesc = osmoutil.importappconf(confpath, "osmoappdesc")
+ except ImportError as e:
+ print >> sys.stderr, "osmoappdesc not found, set searchpath with -p"
+ sys.exit(1)
+
+ os.chdir(workdir)
+ suite = unittest.TestLoader().loadTestsFromTestCase(TestVTY)
+ res = unittest.TextTestRunner(verbosity=1).run(suite)
+ sys.exit(len(res.errors) + len(res.failures))
diff --git a/osmopy/osmoutil.py b/osmopy/osmoutil.py
new file mode 100755
index 0000000..791506d
--- /dev/null
+++ b/osmopy/osmoutil.py
@@ -0,0 +1,50 @@
+#!/usr/bin/env python
+
+# (C) 2013 by Katerina Barone-Adesi <kat.obsc@gmail.com>
+# 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 subprocess
+import os
+import sys
+import importlib
+
+
+"""Run a command, with stdout and stderr directed to devnull"""
+
+
+def popen_devnull(cmd):
+ devnull = open(os.devnull, 'w')
+ return subprocess.Popen(cmd, stdout=devnull, stderr=devnull)
+
+
+"""End a process.
+
+If the process doesn't appear to exist (for instance, is None), do nothing"""
+
+
+def end_proc(proc):
+ if proc:
+ proc.kill()
+ proc.wait()
+
+
+"""Add a directory to sys.path, try to import a config file.
+
+This may throw ImportError if the config file is not found."""
+
+
+def importappconf(dirname, confname):
+ if dirname not in sys.path:
+ sys.path.append(dirname)
+ return importlib.import_module(confname)