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.
This commit is contained in:
commit
a7185c6c72
|
@ -0,0 +1,5 @@
|
|||
#!/usr/bin/env python
|
||||
__version__ = '0.0.1'
|
||||
|
||||
__all__ = ['obscvty', 'osmodumpdoc', 'osmotestconfig', 'osmotestvty',
|
||||
'osmoutil']
|
|
@ -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
|
|
@ -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)
|
|
@ -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))
|
|
@ -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))
|
|
@ -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)
|
|
@ -0,0 +1,31 @@
|
|||
#!/usr/bin/env python
|
||||
# Osmopython: test utilities for osmocom programs
|
||||
# Copyright (C) 2013 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 Affero 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 Affero General Public License for more details.
|
||||
|
||||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
from distutils.core import setup
|
||||
from osmopy import __version__
|
||||
|
||||
setup(
|
||||
name = 'osmopython',
|
||||
version = __version__,
|
||||
packages = ["osmopy"],
|
||||
scripts = ["osmopy/osmodumpdoc.py", "osmopy/osmotestconfig.py",
|
||||
"osmopy/osmotestvty.py"],
|
||||
license = "AGPLv3",
|
||||
description = "Osmopython: osmocom testing scripts",
|
||||
author = "Katerina Barone-Adesi",
|
||||
author_email = "kat.obsc@gmail.com"
|
||||
)
|
Loading…
Reference in New Issue