=== modified file '.bzrignore'
@@ -5,3 +5,4 @@
*~
*.tmp
*.py[co]
+build
=== modified file 'MANIFEST.in'
@@ -1,4 +1,3 @@
include COPYING
include README
include .testr.conf
-include bin/lava-test
\ No newline at end of file
=== modified file 'README'
@@ -3,20 +3,11 @@
automatically installed, executed, and the results can be parsed and
uploaded to an external server.
-External dependency
--------------------
-The following debian packages are needed:
-* python-setuptools
-* python-apt
-* usbutils
-* python-testrepository - for running unit tests
-
How to install from the source code
===================================
1. Run: ./setup.py install
-
How to setup from the source code for development
=================================================
@@ -32,7 +23,7 @@
3. Add <home>/.local.bin in your PATH
-To install build-in tests
+To install built-in tests
=========================
1. Run: lava-test list-tests
2. Run: lava-test install <test>
@@ -47,4 +38,5 @@
To install test define with a json file
=======================================
1. Run: lava-test register-test file://localhost/<..>/examples/stream.json
-2. Run: lava-test list-tests
\ No newline at end of file
+2. Run: lava-test list-tests
+
=== removed file 'abrek/api.py'
@@ -1,76 +0,0 @@
-"""
-Public API for extending Abrek
-"""
-from abc import abstractmethod, abstractproperty
-
-class ITestProvider(object):
- """
- Abrek test provider.
-
- Abstract source of abrek tests.
- """
-
- @abstractmethod
- def __init__(self, config):
- """
- Initialize test provider with the specified configuration object. The
- configuration object is obtained from the abrek providers registry.
- """
-
- @abstractmethod
- def __iter__(self):
- """
- Iterates over instances of ITest exposed by this provider
- """
-
- @abstractmethod
- def __getitem__(self, test_name):
- """
- Return an instance of ITest with the specified name
- """
-
- @abstractproperty
- def description(self):
- """
- The description string used by abrek list-tests
- """
-
-
-class ITest(object):
- """
- Abrek test.
-
- Something that can be installed and invoked by abre.
- """
-
- @abstractmethod
- def install(self):
- """
- Install the test suite.
-
- This creates an install directory under the user's XDG_DATA_HOME
- directory to mark that the test is installed. The installer's
- install() method is then called from this directory to complete any
- test specific install that may be needed.
- """
-
- @abstractmethod
- def uninstall(self):
- """
- Uninstall the test suite.
-
- Uninstalling just recursively removes the test specific directory under
- the user's XDG_DATA_HOME directory. This will both mark the test as
- removed, and clean up any files that were downloaded or installed under
- that directory. Dependencies are intentionally not removed by this.
- """
-
- @abstractmethod
- def run(self, quiet=False):
- # TODO: Document me
- pass
-
- @abstractmethod
- def parse(self, resultname):
- # TODO: Document me
- pass
=== removed file 'abrek/builtins.py'
@@ -1,175 +0,0 @@
-# Copyright (c) 2010 Linaro
-#
-# 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 sys
-from optparse import make_option
-
-import abrek.command
-import abrek.testdef
-from abrek.config import get_config
-
-
-class cmd_version(abrek.command.AbrekCmd):
- """
- Show the version of abrek
- """
- def run(self):
- import abrek
- print abrek.__version__
-
-
-class cmd_help(abrek.command.AbrekCmd):
- """ Get help on abrek commands
-
- If the command name is ommited, calling the help command will return a
- list of valid commands.
- """
- arglist = ['command', 'subcommand']
- def run(self):
- if len(self.args) < 1:
- print "Available commands:"
- for cmd in abrek.command.get_all_cmds():
- print " %s" % cmd
- print
- print "To access extended help on a command use 'abrek help " \
- "[command]'"
- return
- command_name = self.args.pop(0)
- cmd = abrek.command.get_command(command_name)
- if not cmd:
- print "No command found for '%s'" % command_name
- return
- while self.args:
- subcommand_name = self.args.pop(0)
- cmd = cmd.get_subcommand(subcommand_name)
- if not cmd:
- print "No sub-command of '%s' found for '%s'" % (
- command_name, subcommand_name)
- return
- command_name += ' ' + subcommand_name
- print cmd.help()
-
-
-class cmd_install(abrek.command.AbrekCmd):
- """
- Install a test
- """
- arglist = ['*testname']
-
- def run(self):
- self.checkroot()
- if len(self.args) != 1:
- print "please specify the name of the test to install"
- sys.exit(1)
- test = abrek.testdef.testloader(self.args[0])
- try:
- test.install()
- except RuntimeError as strerror:
- print "Test installation error: %s" % strerror
- sys.exit(1)
-
-
-class cmd_run(abrek.command.AbrekCmd):
- """
- Run tests
- """
- arglist = ['*testname']
- options = [make_option('-q', '--quiet', action='store_true',
- default=False, dest='quiet'),
- make_option('-o', '--output', action='store',
- default=None, metavar="FILE",
- help="Store processed test output to FILE")]
-
- def run(self):
- self.checkroot()
- if len(self.args) != 1:
- print "please specify the name of the test to run"
- sys.exit(1)
- test = abrek.testdef.testloader(self.args[0])
- try:
- result_id = test.run(quiet=self.opts.quiet)
- if self.opts.output:
- from abrek.dashboard import generate_bundle
- import json
- bundle = generate_bundle(result_id)
- with open(self.opts.output, "wt") as stream:
- json.dump(bundle, stream)
- except Exception as strerror:
- print "Test execution error: %s" % strerror
- sys.exit(1)
-
-
-class cmd_uninstall(abrek.command.AbrekCmd):
- """
- Uninstall a test
- """
- arglist = ['*testname']
-
- def run(self):
- if len(self.args) != 1:
- print "please specify the name of the test to uninstall"
- sys.exit(1)
- test = abrek.testdef.testloader(self.args[0])
- try:
- test.uninstall()
- except Exception as strerror:
- print "Test uninstall error: %s" % strerror
- sys.exit(1)
-
-
-class cmd_list_installed(abrek.command.AbrekCmd):
- """
- List tests that are currently installed
- """
- def run(self):
- config = get_config()
- print "Installed tests:"
- try:
- for dir in os.listdir(config.installdir):
- print dir
- except OSError:
- print "No tests installed"
-
-
-class cmd_list_tests(abrek.command.AbrekCmd):
- """
- List all known tests
- """
- def run(self):
- from abrek.testdef import TestLoader
- for provider in TestLoader().get_providers():
- print provider.description
- for test in provider:
- print " - %s" % test
-
-
-class cmd_register_test(abrek.command.AbrekCmd):
- """
- Register declarative tests
- """
-
- arglist = ['test_url']
-
- def run(self):
- if len(self.args) != 1:
- self.parser.error("You need to provide an URL to a test definition file")
- test_url = self.args[0]
- from abrek.providers import RegistryProvider
- try:
- RegistryProvider.register_remote_test(test_url)
- except ValueError as exc:
- print "Unable to register test: %s" % exc
- sys.exit(1)
=== removed file 'abrek/bundle.py'
@@ -1,44 +0,0 @@
-# Copyright (c) 2011 Linaro
-#
-# 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/>.
-
-"""
-This module attempts to use the linaro-dashboard-bundle package, if possible.
-
-Using that package adds proper support for loading and saving bundle
-documents. In particular it supports loosles decimals, better, more stable
-load-modify-write cycles, data validation, transparent migration and many
-other features.
-
-It is not a hard dependency to make it possible to run abrek from a checkout
-without having to install (too many) dependencies.
-"""
-
-try:
- from linaro_dashboard_bundle import DocumentIO
-except ImportError:
- import json
-
- class DocumentIO(object):
- """ Bare replacement DocumentIO without any fancy features """
-
- @classmethod
- def dumps(cls, doc):
- return json.dumps(doc, indent=2)
-
- @classmethod
- def loads(cls, text):
- doc = json.loads(text)
- fmt = doc.get("format")
- return fmt, doc
\ No newline at end of file
=== removed file 'abrek/cache.py'
@@ -1,73 +0,0 @@
-"""
-Cache module for Abrek
-"""
-import contextlib
-import hashlib
-import os
-import urllib2
-
-
-class AbrekCache(object):
- """
- Cache class for Abrek
- """
-
- _instance = None
-
- def __init__(self):
- home = os.environ.get('HOME', '/')
- basecache = os.environ.get('XDG_CACHE_HOME',
- os.path.join(home, '.cache'))
- self.cache_dir = os.path.join(basecache, 'abrek')
-
- @classmethod
- def get_instance(cls):
- if cls._instance is None:
- cls._instance = cls()
- return cls._instance
-
- def open_cached(self, key, mode="r"):
- """
- Acts like open() but the pathname is relative to the
- abrek-specific cache directory.
- """
- if "w" in mode and not os.path.exists(self.cache_dir):
- os.makedirs(self.cache_dir)
- if os.path.isabs(key):
- raise ValueError("key cannot be an absolute path")
- try:
- stream = open(os.path.join(self.cache_dir, key), mode)
- yield stream
- finally:
- stream.close()
-
- def _key_for_url(self, url):
- return hashlib.sha1(url).hexdigest()
-
- def _refresh_url_cache(self, key, url):
- with contextlib.nested(
- contextlib.closing(urllib2.urlopen(url)),
- self.open_cached(key, "wb")) as (in_stream, out_stream):
- out_stream.write(in_stream.read())
-
- @contextlib.contextmanager
- def open_cached_url(self, url):
- """
- Like urlopen.open() but the content may be cached.
- """
- # Do not cache local files, this is not what users would expect
-
- # workaround - not using cache at all.
- # TODO: fix this and use the cache
- # if url.startswith("file://"):
- if True:
- stream = urllib2.urlopen(url)
- else:
- key = self._key_for_url(url)
- try:
- stream = self.open_cached(key, "rb")
- except IOError as exc:
- self._refresh_url_cache(key, url)
- stream = self.open_cached(key, "rb")
- yield stream
- stream.close()
=== removed file 'abrek/command.py'
@@ -1,171 +0,0 @@
-# Copyright (c) 2010 Linaro
-#
-# 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/>.
-
-from optparse import OptionParser
-import os
-import sys
-
-
-class _AbrekOptionParser(OptionParser):
- """
- This is just to override the epilog formatter to allow newlines
- """
- def format_epilog(self, formatter):
- return self.epilog
-
-
-class AbrekCmd(object):
- """ Base class for commands that can be passed to Abrek.
-
- Commands added to abrek should inherit from AbrekCmd. To allow for
- autodiscovery, the name of the class should begin with cmd_.
-
- Arguments allowed by the command can be specified in the 'arglist'.
- These arguments will automatically be listed in the help for that
- command. Required arguments should begin with a '*'. For example:
- arglist = ['*requiredarg', 'optionalarg']
-
- Options may also be specified by using the 'options' list. To add
- arguments, you must use the make_option() function from optparse.
- For example:
- options = [make_option("-b", "--bar", dest="bar")]
-
- Commands also support subcommands. A subcommand is similar to a
- command in abrek, and it should also inherit from AbrekCmd. However,
- a subcommand class should not begin with cmd_. Instead, it should
- be tied to the command that uses it, using the 'subcmds' dict.
- For example:
- class subcmd_bar(AbrekCmd):
- pass
- class cmd_foo(AbrekCmd):
- subcmds = {'bar':subcmd_bar()}
- pass
- """
- options = []
- arglist = []
-
- def __init__(self, name_prefix=''):
- self._name_prefix = name_prefix
- self.parser = _AbrekOptionParser(usage=self._usage(),
- epilog=self._desc())
- for opt in self.options:
- self.parser.add_option(opt)
-
- def main(self, argv):
- (self.opts, self.args) = self.parser.parse_args(argv)
- return self.run()
-
- def name(self):
- return self._name_prefix + _convert_command_name(self.__class__.__name__)
-
- def run(self):
- raise NotImplementedError("%s: command defined but not implemented!" %
- self.name())
-
- def _usage(self):
- usagestr = "Usage: lava-test %s" % self.name()
- for arg in self.arglist:
- if arg[0] == '*':
- usagestr += " %s" % arg[1:].upper()
- else:
- usagestr += " [%s]" % arg.upper()
- return usagestr
-
- def _desc(self):
- from inspect import getdoc
- docstr = getdoc(self)
- if not docstr:
- return ""
- description = "\nDescription:\n"
- description += docstr + "\n"
- return description
-
- def help(self):
- #For some reason, format_help includes an extra \n
- return self.parser.format_help()[:-1]
-
- def get_subcommand(self, name):
- return None
-
- def checkroot(self):
- if os.getuid() != 0:
- print >> sys.stderr, ("**** WARNING: ROOT PERMISSIONS ARE OFTEN"
- "REQUIRED FOR THIS OPERATION ****")
-
-
-class AbrekCmdWithSubcommands(AbrekCmd):
-
- arglist = ['subcommand']
-
- def main(self, argv):
- if not argv:
- print "Missing sub-command." + self._list_subcmds()
- else:
- subcmd = self.get_subcommand(argv[0])
- if subcmd is None:
- # This line might print the help and raise SystemExit if
- # --help is passed or if an invalid option was passed.
- opts, args = self.parser.parse_args(argv)
- # If it didn't, complain.
- print "'%s' not found as a sub-command of '%s'" % (
- args[0], self.name()) + self._list_subcmds()
- else:
- return subcmd.main(argv[1:])
-
- def get_subcommand(self, name):
- subcmd_cls = getattr(self, 'cmd_' + name.replace('_', '-'), None)
- if subcmd_cls is None:
- return None
- return subcmd_cls(self.name() + ' ')
-
- def _usage(self):
- usagestr = AbrekCmd._usage(self)
- usagestr += self._list_subcmds()
- return usagestr
-
- def _list_subcmds(self):
- subcmds = []
- for attrname in self.__class__.__dict__.keys():
- if attrname.startswith('cmd_'):
- subcmds.append(_convert_command_name(attrname))
- if not subcmds:
- return ''
- return "\n\nAvailable sub-commands:\n " + "\n ".join(subcmds)
-
-
-def _convert_command_name(cmd):
- return cmd[4:].replace('_','-')
-
-
-def _find_commands(module):
- cmds = {}
- for name, func in module.__dict__.iteritems():
- if name.startswith("cmd_"):
- real_name = _convert_command_name(name)
- cmds[real_name] = func()
- return cmds
-
-
-def get_all_cmds():
- from abrek import builtins, dashboard, results
- cmds = _find_commands(builtins)
- cmds.update(_find_commands(dashboard))
- cmds.update(_find_commands(results))
- return cmds
-
-
-def get_command(cmd_name):
- cmds = get_all_cmds()
- return cmds.get(cmd_name)
=== removed file 'abrek/config.py'
@@ -1,93 +0,0 @@
-# Copyright (c) 2010 Linaro
-#
-# 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 json
-
-
-class AbrekConfig(object):
-
- def __init__(self):
- home = os.environ.get('HOME', '/')
- baseconfig = os.environ.get('XDG_CONFIG_HOME',
- os.path.join(home, '.config'))
- basedata = os.environ.get('XDG_DATA_HOME',
- os.path.join(home, '.local', 'share'))
- self.configdir = os.path.join(baseconfig, 'abrek')
- self.installdir = os.path.join(basedata, 'abrek', 'installed-tests')
- self.resultsdir = os.path.join(basedata, 'abrek', 'results')
- self.registry = self._load_registry()
-
- @property
- def _registry_pathname(self):
- return os.path.join(self.configdir, "registry.json")
-
- def _load_registry(self):
- try:
- with open(self._registry_pathname) as stream:
- return json.load(stream)
- except (IOError, ValueError) as exc:
- return self._get_default_registry()
-
- def _save_registry(self):
- if not os.path.exists(self.configdir):
- os.makedirs(self.configdir)
- with open(self._registry_pathname, "wt") as stream:
- json.dump(self.registry, stream, indent=2)
-
- def _get_default_registry(self):
- return {
- "format": "Abrek Test Registry 1.0 Experimental",
- "providers": [
- {
- "entry_point": "abrek.providers:BuiltInProvider",
- },
- {
- "entry_point": "abrek.providers:PkgResourcesProvider",
- },
- {
- "entry_point": "abrek.providers:RegistryProvider",
- "config": {
- "entries": []
- }
- }
- ]
- }
-
- def get_provider_config(self, entry_point_name):
- if "providers" not in self.registry:
- self.registry["providers"] = []
- for provider_info in self.registry["providers"]:
- if provider_info.get("entry_point") == entry_point_name:
- break
- else:
- provider_info = {"entry_point": entry_point_name}
- self.registry["providers"].append(provider_info)
- if "config" not in provider_info:
- provider_info["config"] = {}
- return provider_info["config"]
-
-
-_config = None
-
-def get_config():
- global _config
- if _config is not None:
- return _config
- return AbrekConfig()
-
-def set_config(config):
- global _config
- _config = config
=== removed file 'abrek/dashboard.py'
@@ -1,206 +0,0 @@
-# Copyright (c) 2010 Linaro
-#
-# 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 base64
-import os
-import socket
-import sys
-import urllib
-import xmlrpclib
-from ConfigParser import ConfigParser, NoOptionError
-from getpass import getpass
-from optparse import make_option
-
-from abrek.bundle import DocumentIO
-from abrek.command import AbrekCmd, AbrekCmdWithSubcommands
-from abrek.config import get_config
-from abrek.testdef import testloader
-
-
-class DashboardConfig(object):
- """
- Read and write dashboard configuration data
- """
-
- def __init__(self, section="Default Server"):
- self.dashboardconf = ConfigParser()
- self.section = section
- self.config = get_config()
- self.path = os.path.join(self.config.configdir, "dashboard.conf")
- if os.path.exists(self.path):
- self.dashboardconf.read(self.path)
- if not (self.section in self.dashboardconf.sections()):
- self.dashboardconf.add_section(self.section)
-
- def set_host(self, host):
- self.dashboardconf.set(self.section, 'host', host)
-
- def get_host(self):
- try:
- host = self.dashboardconf.get(self.section, 'host')
- return host
- except NoOptionError:
- return ""
-
- host = property(get_host, set_host)
-
- def set_user(self, user):
- self.dashboardconf.set(self.section, 'user', user)
-
- def get_user(self):
- try:
- user = self.dashboardconf.get(self.section, 'user')
- return user
- except NoOptionError:
- return ""
-
- user = property(get_user, set_user)
-
- def set_password(self, password):
- #Not exactly secure, but better than storing in plaintext
- password = base64.encodestring(password).strip()
- self.dashboardconf.set(self.section,'password', password)
-
- def get_password(self):
- try:
- password = self.dashboardconf.get(self.section, 'password')
- return base64.decodestring(password)
- except NoOptionError:
- return ""
-
- password = property(get_password, set_password)
-
- def write(self):
- """
- write the dashboard configuration out to the config file
- """
- if not os.path.exists(self.config.configdir):
- os.makedirs(self.config.configdir)
- with open(self.path, "w") as fd:
- self.dashboardconf.write(fd)
-
-
-class cmd_dashboard(AbrekCmdWithSubcommands):
- """
- Connect to the Launch-control dashboard
- """
-
- class cmd_setup(AbrekCmd):
- """
- Configure information needed to push results to the dashboard
- """
- options = [make_option("-u", "--user", dest="user"),
- make_option("-p", "--password", dest="password")]
- arglist = ["*server"]
-
- def run(self):
- if len(self.args) != 1:
- print "You must specify a server"
- sys.exit(1)
- config = DashboardConfig()
- if self.opts.user:
- user = self.opts.user
- else:
- user = raw_input("Username: ")
- if self.opts.password:
- password = self.opts.password
- else:
- password = getpass()
- config.host = self.args[0]
- config.user = user
- config.password = password
- config.write()
-
-
- class cmd_put(AbrekCmd):
- """
- Push the results from a test to the server
- The stream name must include slashes (e.g. /anonymous/foo/)
- """
- arglist = ["*stream", "*result"]
-
- def run(self):
- if len(self.args) != 2:
- print "You must specify a stream and a result"
- sys.exit(1)
- stream_name = self.args[0]
- result_name = self.args[1]
- bundle = generate_bundle(result_name)
- db_config = DashboardConfig()
- hosturl = urllib.basejoin(db_config.host, "xml-rpc/")
- try:
- server = xmlrpclib.Server(hosturl)
- except IOError, e:
- print ("Error connecting to server at '%s'. Error was: %s, "
- "please run 'lava-test dashboard setup [host]'" % (
- hosturl, e))
- sys.exit(1)
- try:
- result = server.put(DocumentIO.dumps(bundle), result_name,
- stream_name)
- print "Bundle successfully uploaded to id: %s" % result
- except xmlrpclib.Fault as strerror:
- print "Error uploading bundle: %s" % strerror.faultString
- sys.exit(1)
- except socket.error as strerror:
- print "Unable to connect to host: %s" % strerror
- sys.exit(1)
- except xmlrpclib.ProtocolError as strerror:
- print "Connection error: %s" % strerror
- sys.exit(1)
-
-
- class cmd_bundle(AbrekCmd):
- """
- Print JSON output that can be imported into the dashboard
- """
- arglist = ["*result"]
-
- def run(self):
- if len(self.args) != 1:
- print "You must specify a result"
- sys.exit(1)
- bundle = generate_bundle(self.args[0])
- try:
- print DocumentIO.dumps(bundle)
- except IOError:
- pass
-
-
-def generate_bundle(result):
- config = get_config()
- resultdir = os.path.join(config.resultsdir, result)
- if not os.path.exists(resultdir):
- # FIXME: UI and sys.exit mixed with internal implementation, yuck
- print "Result directory not found"
- sys.exit(1)
- with open(os.path.join(resultdir, "testdata.json")) as stream:
- bundle_text = stream.read()
- with open(os.path.join(resultdir, "testoutput.log")) as stream:
- output_text = stream.read()
- fmt, bundle = DocumentIO.loads(bundle_text)
- test = testloader(bundle['test_runs'][0]['test_id'])
- test.parse(result)
- bundle['test_runs'][0].update(test.parser.results)
- bundle['test_runs'][0]["attachments"] = [
- {
- "pathname": "testoutput.log",
- "mime_type": "text/plain",
- "content": base64.standard_b64encode(output_text)
- }
- ]
- return bundle
-
-
=== removed file 'abrek/hwprofile.py'
@@ -1,216 +0,0 @@
-# Copyright (c) 2010 Linaro
-#
-# 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 re
-import sys
-from subprocess import Popen, PIPE
-from utils import read_file, get_machine_type
-
-INTEL_KEYMAP = {
- 'vendor_id': 'cpu_vendor_name',
- 'cpu family': 'cpu_family',
- 'model': 'cpu_model',
- 'model name': 'cpu_model_name',
- 'stepping': 'cpu_stepping',
- 'cpu MHz': 'cpu_mhz',
- 'flags': 'cpu_features',
-}
-
-INTEL_VALMAP = {
- 'cpu family': int,
- 'model': int,
- 'stepping': int,
- 'cpu MHz': float,
-}
-
-ARM_KEYMAP = {
- 'Processor': 'cpu_model_name',
- 'Features': 'cpu_features',
- 'CPU implementer': 'cpu_implementer',
- 'CPU architecture': 'cpu_architecture',
- 'CPU variant': 'cpu_variant',
- 'CPU part': 'cpu_part',
- 'CPU revision': 'cpu_revision',
-}
-
-ARM_VALMAP = {
- 'CPU implementer': lambda value: int(value, 16),
- 'CPU architecture': int,
- 'CPU variant': lambda value: int(value, 16),
- 'CPU part': lambda value: int(value, 16),
- 'CPU revision': int,
-}
-
-
-def _translate_cpuinfo(keymap, valmap, key, value):
- """
- Translate a key and value using keymap and valmap passed in
- """
- newkey = keymap.get(key, key)
- newval = valmap.get(key, lambda x: x)(value)
- return newkey, newval
-
-def get_cpu_devs():
- """
- Return a list of CPU devices
- """
- pattern = re.compile('^(?P<key>.+?)\s*:\s*(?P<value>.*)$')
- cpunum = 0
- devices = []
- cpudevs = []
- cpudevs.append({})
- machine = get_machine_type()
- if machine in ('i686', 'x86_64'):
- keymap, valmap = INTEL_KEYMAP, INTEL_VALMAP
- elif machine.startswith('arm'):
- keymap, valmap = ARM_KEYMAP, ARM_VALMAP
-
- try:
- cpuinfo = read_file("/proc/cpuinfo")
- for line in cpuinfo.splitlines():
- match = pattern.match(line)
- if match:
- key, value = match.groups()
- key, value = _translate_cpuinfo(keymap, valmap,
- key, value)
- if cpudevs[cpunum].get(key):
- cpunum += 1
- cpudevs.append({})
- cpudevs[cpunum][key] = value
- for c in range(len(cpudevs)):
- device = {}
- device['device_type'] = 'device.cpu'
- device['description'] = 'Processor #{0}'.format(c)
- device['attributes'] = cpudevs[c]
- devices.append(device)
- except IOError:
- print >> sys.stderr, "WARNING: Could not read cpu information"
- return devices
-
-
-def get_board_devs():
- """
- Return a list of board devices
- """
- devices = []
- attributes = {}
- device = {}
- machine = get_machine_type()
- if machine in ('i686', 'x86_64'):
- try:
- description = read_file('/sys/class/dmi/id/board_name') or None
- vendor = read_file('/sys/class/dmi/id/board_vendor') or None
- version = read_file('/sys/class/dmi/id/board_version') or None
- if description:
- device['description'] = description.strip()
- if vendor:
- attributes['vendor'] = vendor.strip()
- if version:
- attributes['version'] = version.strip()
- except IOError:
- print >> sys.stderr, "WARNING: Could not read board information"
- return devices
- elif machine.startswith('arm'):
- try:
- cpuinfo = read_file("/proc/cpuinfo")
- if cpuinfo is None:
- return devices
- pattern = re.compile("^Hardware\s*:\s*(?P<description>.+)$", re.M)
- match = pattern.search(cpuinfo)
- if match is None:
- return devices
- device['description'] = match.group('description')
- except IOError:
- print >> sys.stderr, "WARNING: Could not read board information"
- return devices
- else:
- return devices
- if attributes:
- device['attributes'] = attributes
- device['device_type'] = 'device.board'
- devices.append(device)
- return devices
-
-def get_mem_devs():
- """ Return a list of memory devices
-
- This returns up to two items, one for physical RAM and another for swap
- """
- pattern = re.compile('^(?P<key>.+?)\s*:\s*(?P<value>.+) kB$', re.M)
-
- devices = []
- try:
- meminfo = read_file("/proc/meminfo")
- for match in pattern.finditer(meminfo):
- key, value = match.groups()
- if key not in ('MemTotal', 'SwapTotal'):
- continue
- capacity = int(value) << 10 #Kernel reports in 2^10 units
- if capacity == 0:
- continue
- if key == 'MemTotal':
- kind = 'RAM'
- else:
- kind = 'swap'
- description = "{capacity}MiB of {kind}".format(
- capacity=capacity >> 20, kind=kind)
- device = {}
- device['description'] = description
- device['attributes'] = {'capacity': str(capacity), 'kind': kind}
- device['device_type'] = "device.mem"
- devices.append(device)
- except IOError:
- print >> sys.stderr, "WARNING: Could not read memory information"
- return devices
-
-def get_usb_devs():
- """
- Return a list of usb devices
- """
- pattern = re.compile(
- "^Bus \d{3} Device \d{3}: ID (?P<vendor_id>[0-9a-f]{4}):"
- "(?P<product_id>[0-9a-f]{4}) (?P<description>.*)$")
- devices = []
- try:
- for line in Popen('lsusb', stdout=PIPE).communicate()[0].splitlines():
- match = pattern.match(line)
- if match:
- vendor_id, product_id, description = match.groups()
- attributes = {}
- device = {}
- attributes['vendor_id'] = int(vendor_id, 16)
- attributes['product_id'] = int(product_id, 16)
- device['attributes'] = attributes
- device['description'] = description
- device['device_type'] = 'device.usb'
- devices.append(device)
- except OSError:
- print >> sys.stderr, "WARNING: Could not read usb device information, \
-unable to run lsusb, please install usbutils package"
- return devices
-
-def get_hardware_context():
- """
- Return a dict with all of the hardware profile information gathered
- """
- hardware_context = {}
- devices = []
- devices.extend(get_cpu_devs())
- devices.extend(get_board_devs())
- devices.extend(get_mem_devs())
- devices.extend(get_usb_devs())
- hardware_context['devices'] = devices
- return hardware_context
-
=== removed file 'abrek/providers.py'
@@ -1,145 +0,0 @@
-"""
-Test providers.
-
-Allow listing and loading of tests in a generic way.
-"""
-
-from pkg_resources import working_set
-
-from abrek.api import ITestProvider
-from abrek.cache import AbrekCache
-from abrek.config import get_config
-from abrek.testdef import AbrekDeclarativeTest
-
-
-class BuiltInProvider(ITestProvider):
- """
- Test provider that provides tests shipped in the Abrek source tree
- """
-
- _builtin_tests = [
- 'bootchart',
- 'firefox',
- 'glmemperf',
- 'gmpbench',
- 'gtkperf',
- 'ltp',
- 'peacekeeper'
- 'posixtestsuite',
- 'pwrmgmt',
- 'pybench',
- 'smem',
- 'stream',
- 'tiobench',
- 'x11perf',
- 'xrestop',
- ]
-
- def __init__(self, config):
- pass
-
- @property
- def description(self):
- return "Tests built directly into Abrek:"
-
- def __iter__(self):
- return iter(self._builtin_tests)
-
- def __getitem__(self, test_name):
- try:
- module = __import__("abrek.test_definitions.%s" % test_name, fromlist=[''])
- except ImportError:
- raise KeyError(test_name)
- else:
- return module.testobj
-
-
-class PkgResourcesProvider(ITestProvider):
- """
- Test provider that provides tests declared in pkg_resources working_set
-
- By default it looks at the 'abrek.test_definitions' name space but it can
- be changed with custom 'namespace' configuration entry.
- """
-
- def __init__(self, config):
- self._config = config
-
- @property
- def namespace(self):
- return self._config.get("namespace", "abrek.test_definitions")
-
- @property
- def description(self):
- return "Tests provided by installed python packages:"
-
- def __iter__(self):
- for entry_point in working_set.iter_entry_points(self.namespace):
- yield entry_point.name
-
- def __getitem__(self, test_name):
- for entry_point in working_set.iter_entry_points(self.namespace, test_name):
- return entry_point.load().testobj
- raise KeyError(test_name)
-
-
-class RegistryProvider(ITestProvider):
- """
- Test provider that provides declarative tests listed in the test registry.
- """
- def __init__(self, config):
- self._config = config
- self._cache = None
-
- @property
- def entries(self):
- """
- List of URLs to AbrekDeclarativeTest description files
- """
- return self._config.get("entries", [])
-
- @property
- def description(self):
- return "Tests provided by Abrek registry:"
-
- @classmethod
- def register_remote_test(self, test_url):
- config = get_config() # This is a different config object from self._config
- provider_config = config.get_provider_config("abrek.providers:RegistryProvider")
- if "entries" not in provider_config:
- provider_config["entries"] = []
- if test_url not in provider_config["entries"]:
- provider_config["entries"].append(test_url)
- config._save_registry()
- else:
- raise ValueError("This test is already registered")
-
- def _load_remote_test(self, test_url):
- """
- Load AbrekDeclarativeTest from a (possibly cached copy of) test_url
- """
- cache = AbrekCache.get_instance()
- with cache.open_cached_url(test_url) as stream:
- return AbrekDeclarativeTest.load_from_stream(stream)
-
- def _fill_cache(self):
- """
- Fill the cache of all remote tests
- """
- if self._cache is not None:
- return
- self._cache = {}
- for test_url in self.entries:
- test = self._load_remote_test(test_url)
- if test.testname in self._cache:
- raise ValueError("Duplicate test %s declared" % test.testname)
- self._cache[test.testname] = test
-
- def __iter__(self):
- self._fill_cache()
- for test_name in self._cache.iterkeys():
- yield test_name
-
- def __getitem__(self, test_name):
- self._fill_cache()
- return self._cache[test_name]
=== removed file 'abrek/results.py'
@@ -1,111 +0,0 @@
-# Copyright (c) 2010 Linaro
-#
-# 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 shutil
-import sys
-from optparse import make_option
-
-from abrek.command import AbrekCmd, AbrekCmdWithSubcommands
-from abrek.config import get_config
-from abrek.utils import read_file
-
-
-class cmd_results(AbrekCmdWithSubcommands):
- """
- Operate on results of previous test runs stored locally
- """
-
- class cmd_list(AbrekCmd):
- """
- List results of previous runs
- """
- def run(self):
- config = get_config()
- print "Saved results:"
- try:
- for dir in os.listdir(config.resultsdir):
- print dir
- except OSError:
- print "No results found"
-
-
- class cmd_show(AbrekCmd):
- """
- Display the output from a previous test run
- """
- arglist = ['*result']
- def run(self):
- if len(self.args) != 1:
- print "please specify the name of the result dir"
- sys.exit(1)
- config = get_config()
- resultsdir = os.path.join(config.resultsdir, self.args[0])
- testoutput = os.path.join(resultsdir, "testoutput.log")
- if not os.path.exists(testoutput):
- print "No result found for '%s'" % self.args[0]
- sys.exit(1)
- try:
- print(read_file(testoutput))
- except IOError:
- pass
-
-
- class cmd_remove(AbrekCmd):
- """
- Remove the results of a previous test run
- """
- arglist = ['*result']
- options = [make_option('-f', '--force', action='store_true',
- dest='force')]
- def run(self):
- if len(self.args) != 1:
- print "please specify the name of the result dir"
- sys.exit(1)
- config = get_config()
- resultsdir = os.path.join(config.resultsdir, self.args[0])
- if not os.path.exists(resultsdir):
- print "No result found for '%s'" % self.args[0]
- sys.exit(1)
- if not self.opts.force:
- print "Delete result '%s' for good? [Y/N]" % self.args[0],
- response = raw_input()
- if response[0].upper() != 'Y':
- sys.exit(0)
- shutil.rmtree(resultsdir)
-
-
- class cmd_rename(AbrekCmd):
- """
- Rename the results from a previous test run
- """
- arglist = ['*source', '*destination']
-
- def run(self):
- if len(self.args) != 2:
- print "please specify the name of the result, and the new name"
- sys.exit(1)
- config = get_config()
- srcdir = os.path.join(config.resultsdir, self.args[0])
- destdir = os.path.join(config.resultsdir, self.args[1])
- if not os.path.exists(srcdir):
- print "Result directory not found"
- sys.exit(1)
- if os.path.exists(destdir):
- print "Destination result name already exists"
- sys.exit(1)
- shutil.move(srcdir, destdir)
-
-
=== removed file 'abrek/swprofile.py'
@@ -1,65 +0,0 @@
-# Copyright (c) 2010 Linaro
-#
-# 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 apt
-import lsb_release
-from utils import read_file
-
-def get_packages(apt_cache=None):
- """ Get information about the packages installed
-
- apt_cache - if not provided, this will be read from the system
- """
- if apt_cache == None:
- apt_cache = apt.Cache()
- packages = []
- for apt_pkg in apt_cache:
- if hasattr(apt_pkg, 'is_installed'):
- is_installed = apt_pkg.is_installed
- else:
- is_installed = apt_pkg.isInstalled # old style API
- if is_installed:
- pkg = {"name":apt_pkg.name, "version":apt_pkg.installed.version}
- packages.append(pkg)
- return packages
-
-def get_software_context(apt_cache=None, lsb_information=None):
- """ Return dict used for storing software_context information
-
- test_id - Unique identifier for this test
- time_check - whether or not a check was performed to see if
- the time on the system was synced with a time server
- apt_cache - if not provided, this will be read from the system
- lsb_information - if not provided, this will be read from the system
- """
- software_context = {}
- software_context['image'] = get_image(lsb_information)
- software_context['packages'] = get_packages(apt_cache)
- return software_context
-
-def get_image(lsb_information=None):
- """ Get information about the image we are running
-
- If /etc/buildstamp exists, get the image id from that. Otherwise
- just use the lsb-release description for a rough idea.
- """
- try:
- buildstamp = read_file("/etc/buildstamp")
- name = buildstamp.splitlines()[1]
- except IOError:
- if lsb_information == None:
- lsb_information = lsb_release.get_distro_information()
- name = lsb_information['DESCRIPTION']
- return {"name":name}
=== removed file 'abrek/testdef.py'
@@ -1,435 +0,0 @@
-# Copyright (c) 2010 Linaro
-#
-# 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 hashlib
-import json
-import os
-import re
-import shutil
-import decimal
-import sys
-import time
-from commands import getstatusoutput
-from datetime import datetime
-from uuid import uuid4
-
-from abrek import swprofile, hwprofile
-from abrek.api import ITest
-from abrek.bundle import DocumentIO
-from abrek.config import get_config
-from abrek.utils import geturl, run_and_log, write_file
-
-
-class AbrekTest(ITest):
- """Base class for defining tests.
-
- This can be used by test definition files to create an object that
- contains the building blocks for installing tests, running them,
- and parsing the results.
-
- testname - name of the test or test suite
- version - version of the test or test suite
- installer - AbrekInstaller instance to use
- runner - AbrekRunner instance to use
- parser - AbrekParser instance to use
- """
- def __init__(self, testname, version="", installer=None, runner=None,
- parser=None):
- self.testname = testname
- self.version = version
- self.installer = installer
- self.runner = runner
- self.parser = parser
- self.origdir = os.path.abspath(os.curdir)
-
- def install(self):
- """Install the test suite.
-
- This creates an install directory under the user's XDG_DATA_HOME
- directory to mark that the test is installed. The installer's
- install() method is then called from this directory to complete any
- test specific install that may be needed.
- """
- if not self.installer:
- raise RuntimeError("no installer defined for '%s'" %
- self.testname)
- config = get_config()
- installdir = os.path.join(config.installdir, self.testname)
- if os.path.exists(installdir):
- raise RuntimeError("%s is already installed" % self.testname)
- os.makedirs(installdir)
- os.chdir(installdir)
- try:
- self.installer.install()
- except Exception as strerror:
- self.uninstall()
- raise
- finally:
- os.chdir(self.origdir)
-
- def uninstall(self):
- """Uninstall the test suite.
-
- Uninstalling just recursively removes the test specific directory
- under the user's XDG_DATA_HOME directory. This will both mark
- the test as removed, and clean up any files that were downloaded
- or installed under that directory. Dependencies are intentionally
- not removed by this.
- """
- os.chdir(self.origdir)
- config = get_config()
- path = os.path.join(config.installdir, self.testname)
- if os.path.exists(path):
- shutil.rmtree(path)
-
- def _savetestdata(self, analyzer_assigned_uuid):
- TIMEFORMAT = '%Y-%m-%dT%H:%M:%SZ'
- bundle = {
- 'format': 'Dashboard Bundle Format 1.2',
- 'test_runs': [
- {
- 'test_id': self.testname,
- 'analyzer_assigned_date': self.runner.starttime.strftime(TIMEFORMAT),
- 'analyzer_assigned_uuid': analyzer_assigned_uuid,
- 'time_check_performed': False,
- 'hardware_context': hwprofile.get_hardware_context(),
- 'software_context': swprofile.get_software_context(),
- 'test_results': []
- }
- ]
- }
- filename = os.path.join(self.resultsdir, 'testdata.json')
- write_file(DocumentIO.dumps(bundle), filename)
-
- def run(self, quiet=False):
- if not self.runner:
- raise RuntimeError("no test runner defined for '%s'" %
- self.testname)
- config = get_config()
- uuid = str(uuid4())
- installdir = os.path.join(config.installdir, self.testname)
- resultname = (self.testname +
- str(time.mktime(datetime.utcnow().timetuple())))
- self.resultsdir = os.path.join(config.resultsdir, resultname)
- os.makedirs(self.resultsdir)
- try:
- os.chdir(installdir)
- self.runner.run(self.resultsdir, quiet=quiet)
- self._savetestdata(uuid)
- finally:
- os.chdir(self.origdir)
- result_id = os.path.basename(self.resultsdir)
- print("ABREK TEST RUN COMPLETE: Result id is '%s'" % result_id)
- return result_id
-
- def parse(self, resultname):
- if not self.parser:
- raise RuntimeError("no test parser defined for '%s'" %
- self.testname)
- config = get_config()
- resultsdir = os.path.join(config.resultsdir, resultname)
- os.chdir(resultsdir)
- self.parser.parse()
- os.chdir(self.origdir)
-
-
-class AbrekDeclarativeTest(AbrekTest):
- """
- Declarative test is like AbrekTest but cannot contain any python code and
- is completely encapsulated in .json file.
- """
-
- @classmethod
- def load_from_stream(cls, stream):
- return cls(json.load(stream))
-
- def save_to_stream(self, stream):
- json.dumps(self.about, stream, indent="2")
-
- def __init__(self, about):
- self.about = about
- super(AbrekDeclarativeTest, self).__init__(self.about.get('test_id'))
- self.installer = AbrekTestInstaller(**self.about.get('install', {}))
- self.runner = AbrekTestRunner(**self.about.get('run', {}))
- if self.about.get('parse', {}).get('native', False) is True:
- self.parser = AbrekNativeTestParser(self)
- else:
- self.parser = AbrekForeignTestParser(**self.about.get('parse', {}))
-
-
-class AbrekTestInstaller(object):
- """Base class for defining an installer object.
-
- This class can be used as-is for simple installers, or extended for more
- advanced funcionality.
-
- steps - list of steps to be executed in a shell
- deps - list of dependencies to apt-get install before running the steps
- url - location from which the test suite should be downloaded
- md5 - md5sum to check the integrety of the download
- """
- def __init__(self, steps=[], deps=[], url="", md5="", **kwargs):
- self.steps = steps
- self.deps = deps
- self.url = url
- self.md5 = md5
-
- def _installdeps(self):
- if not self.deps:
- return 0
- cmd = "sudo apt-get install -y %s" % " ".join(self.deps)
- rc, output = getstatusoutput(cmd)
- if rc:
- raise RuntimeError("Dependency installation failed. %d : %s" %(rc,output))
-
- def _download(self):
- """Download the file specified by the url and check the md5.
-
- Returns the path and filename if successful, otherwise return None
- """
- if not self.url:
- return 0
- filename = geturl(self.url)
- #If the file does not exist, then the download was not successful
- if not os.path.exists(filename):
- return None
- if self.md5:
- checkmd5 = hashlib.md5()
- with open(filename, 'rb') as fd:
- data = fd.read(0x10000)
- while data:
- checkmd5.update(data)
- data = fd.read(0x10000)
- if checkmd5.hexdigest() != self.md5:
- raise RuntimeError("Unexpected md5sum downloading %s" %
- filename)
- return None
- return filename
-
- def _runsteps(self):
- for cmd in self.steps:
- rc, output = getstatusoutput(cmd)
- if rc:
- raise RuntimeError("Run step '%s' failed. %d : %s" %(cmd,rc,output))
-
-
- def install(self):
- self._installdeps()
- self._download()
- self._runsteps()
-
-
-class AbrekTestRunner(object):
- """Base class for defining an test runner object.
-
- This class can be used as-is for simple execution with the expectation
- that the run() method will be called from the directory where the test
- was installed. Steps, if used, should handle changing directories from
- there to the directory where the test was extracted if necessary.
- This class can also be extended for more advanced funcionality.
-
- steps - list of steps to be executed in a shell
- """
- def __init__(self, steps=[]):
- self.steps = steps
- self.testoutput = []
-
- def _runsteps(self, resultsdir, quiet=False):
- outputlog = os.path.join(resultsdir, 'testoutput.log')
- with open(outputlog, 'a') as fd:
- for cmd in self.steps:
- run_and_log(cmd, fd, quiet)
-
- def run(self, resultsdir, quiet=False):
- self.starttime = datetime.utcnow()
- self._runsteps(resultsdir, quiet=quiet)
- self.endtime = datetime.utcnow()
-
-
-class AbrekForeignTestParser(object):
- """Base class for defining a test parser
-
- This class can be used as-is for simple results parsers, but will
- likely need to be extended slightly for many. If used as it is,
- the parse() method should be called while already in the results
- directory and assumes that a file for test output will exist called
- testoutput.log.
-
- pattern - regexp pattern to identify important elements of test output
- For example: If your testoutput had lines that look like:
- "test01: PASS", then you could use a pattern like this:
- "^(?P<testid>\w+):\W+(?P<result>\w+)"
- This would result in identifying "test01" as testid and
- "PASS" as result. Once parse() has been called,
- self.results.test_results[] contains a list of dicts of all the
- key,value pairs found for each test result
- fixupdict - dict of strings to convert test results to standard strings
- For example: if you want to standardize on having pass/fail results
- in lower case, but your test outputs them in upper case, you could
- use a fixupdict of something like: {'PASS':'pass','FAIL':'fail'}
- appendall - Append a dict to the test_results entry for each result.
- For example: if you would like to add units="MB/s" to each result:
- appendall={'units':'MB/s'}
- """
- def __init__(self, pattern=None, fixupdict=None, appendall={}):
- self.pattern = pattern
- self.fixupdict = fixupdict
- self.results = {'test_results':[]}
- self.appendall = appendall
-
- def _find_testid(self, id):
- for x in self.results['test_results']:
- if x['testid'] == id:
- return self.results['test_results'].index(x)
-
- def parse(self):
- """Parse test output to gather results
-
- Use the pattern specified when the class was instantiated to look
- through the results line-by-line and find lines that match it.
- Results are then stored in self.results. If a fixupdict was supplied
- it is used to convert test result strings to a standard format.
- """
- filename = "testoutput.log"
- try:
- pat = re.compile(self.pattern)
- except Exception as strerror:
- raise RuntimeError(
- "AbrekTestParser - Invalid regular expression '%s' - %s" % (
- self.pattern,strerror))
-
- with open(filename, 'r') as stream:
- for lineno, line in enumerate(stream, 1):
- match = pat.search(line)
- if not match:
- continue
- data = match.groupdict()
- if 'measurement' in data:
- data['measurement'] = decimal.Decimal(data['measurement'])
- data["log_filename"] = filename
- data["log_lineno"] = lineno
- self.results['test_results'].append(data)
- if self.fixupdict:
- self.fixresults(self.fixupdict)
- if self.appendall:
- self.appendtoall(self.appendall)
- self.fixmeasurements()
- self.fixids()
-
- def append(self, testid, entry):
- """Appends a dict to the test_results entry for a specified testid
-
- This lets you add a dict to the entry for a specific testid
- entry should be a dict, updates it in place
- """
- index = self._find_testid(testid)
- self.results['test_results'][index].update(entry)
-
- def appendtoall(self, entry):
- """Append entry to each item in the test_results.
-
- entry - dict of key,value pairs to add to each item in the
- test_results
- """
- for t in self.results['test_results']:
- t.update(entry)
-
- def fixresults(self, fixupdict):
- """Convert results to a known, standard format
-
- pass it a dict of keys/values to replace
- For instance:
- {"TPASS":"pass", "TFAIL":"fail"}
- This is really only used for qualitative tests
- """
- for t in self.results['test_results']:
- if t.has_key("result"):
- t['result'] = fixupdict[t['result']]
-
- def fixmeasurements(self):
- """Measurements are often read as strings, but need to be float
- """
- for id in self.results['test_results']:
- if id.has_key('measurement'):
- id['measurement'] = float(id['measurement'])
-
- def fixids(self):
- """
- Convert spaces to _ in test_case_id and remove illegal characters
- """
- badchars = "[^a-zA-Z0-9\._-]"
- for id in self.results['test_results']:
- if id.has_key('test_case_id'):
- id['test_case_id'] = id['test_case_id'].replace(" ", "_")
- id['test_case_id'] = re.sub(badchars, "", id['test_case_id'])
-
-
-AbrekTestParser = AbrekForeignTestParser
-
-
-class AbrekNativeTestParser(object):
-
- def __init__(self, test_def):
- self.test_def = test_def
-
- def parse(self):
- raise NotImplementedError(self.parse)
-
-
-class TestLoader(object):
- """
- Test loader.
-
- Encapsulates Abrek's knowledge of available tests.
-
- Test can be loaded by name with TetsLoader.get_test_by_name. Test can also
- be listed by TestLoader.get_providers() and then iterating over tests
- returned by each provider.
- """
-
- def __init__(self):
- self._config = get_config()
-
- def get_providers(self):
- """
- Return a generator of available providers
- """
- import pkg_resources
- for provider_info in self._config.registry.get("providers", []):
- entry_point_name = provider_info.get("entry_point")
- module_name, attrs = entry_point_name.split(':', 1)
- attrs = attrs.split('.')
- try:
- entry_point = pkg_resources.EntryPoint(entry_point_name, module_name, attrs,
- dist=pkg_resources.get_distribution("lava-test"))
- provider_cls = entry_point.load()
- provider = provider_cls(provider_info.get("config", {}))
- yield provider
- except pkg_resources.DistributionNotFound as ex:
- raise RuntimeError("lava-test is not properly set up. Please read the REAMME file")
-
- def get_test_by_name(self, test_name):
- """
- Lookup a test with the specified name
- """
- for provider in self.get_providers():
- try:
- return provider[test_name]
- except KeyError:
- pass
- raise ValueError("No such test %r" % test_name)
-
-
-testloader = TestLoader().get_test_by_name
=== removed directory 'bin'
=== removed file 'bin/lava-test'
@@ -1,31 +0,0 @@
-#!/usr/bin/python
-
-# Copyright (c) 2010 Linaro
-#
-# 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 sys
-
-ABREK_BINDIR=os.path.abspath(os.path.dirname(os.path.realpath(sys.argv[0])))
-ABREK_ROOT=os.path.dirname(ABREK_BINDIR)
-ABREK_DIR=os.path.join(ABREK_ROOT,'abrek')
-if os.path.exists(ABREK_DIR) and ABREK_ROOT not in sys.path:
- sys.path.insert(0, ABREK_ROOT)
-
-import abrek.main
-
-if __name__ == '__main__':
- exit_code = abrek.main.main(sys.argv)
- sys.exit(exit_code)
=== added directory 'doc'
=== added file 'doc/changes.rst'
@@ -0,0 +1,25 @@
+Version History
+***************
+
+.. _version_0_2:
+
+Version 0.2
+===========
+
+* Rewrite most of the code and deprecate old Abrek interfaces. This allowed us
+ to clean up the API, rethink some of the design and integrate the code better
+ with other parts of LAVA.
+
+* Improved documentation and code reference. LAVA Test should now have
+ sufficient documentation to help new users and contributors alike.
+
+* Support for installing and running out-of-tree tests.
+
+* Ability to define parsers that add new attachments.
+
+* Unified command line interface with other lava tools thanks to lava-tool.
+
+Version 0.1
+===========
+
+* Initial release (as Abrek)
=== added file 'doc/conf.py'
@@ -0,0 +1,211 @@
+# -*- coding: utf-8 -*-
+#
+# Linaro JSON documentation build configuration file, created by
+# sphinx-quickstart on Mon Dec 27 16:39:47 2010.
+#
+# This file is execfile()d with the current directory set to its containing dir.
+#
+# Note that not all possible configuration values are present in this
+# autogenerated file.
+#
+# All configuration values have a default; values that are commented out
+# serve to show the default.
+
+import sys
+import os
+
+# If extensions (or modules to document with autodoc) are in another directory,
+# add these directories to sys.path here. If the directory is relative to the
+# documentation root, use os.path.abspath to make it absolute, like shown here.
+sys.path.append(os.path.abspath('..'))
+
+# -- General configuration -----------------------------------------------------
+
+# Add any Sphinx extension module names here, as strings. They can be extensions
+# coming with Sphinx (named 'sphinx.ext.*') or your custom ones.
+extensions = [
+ 'sphinx.ext.autodoc',
+ 'sphinx.ext.doctest',
+ 'sphinx.ext.intersphinx',
+ 'sphinx.ext.todo',
+ 'sphinx.ext.coverage',
+ 'sphinx.ext.viewcode']
+
+# Configuration for sphinx.ext.todo
+
+todo_include_todos = True
+
+# Add any paths that contain templates here, relative to this directory.
+templates_path = []
+
+# The suffix of source filenames.
+source_suffix = '.rst'
+
+# The encoding of source files.
+#source_encoding = 'utf-8'
+
+# The master toctree document.
+master_doc = 'index'
+
+# General information about the project.
+project = u'LAVA Test'
+copyright = u'2010-2011, Linaro Limited'
+
+# The version info for the project you're documenting, acts as replacement for
+# |version| and |release|, also used in various other places throughout the
+# built documents.
+#
+# The short X.Y version.
+import versiontools
+import lava_test
+version = "%d.%d" % lava_test.__version__[0:2]
+# The full version, including alpha/beta/rc tags.
+release = versiontools.format_version(lava_test.__version__)
+
+# The language for content autogenerated by Sphinx. Refer to documentation
+# for a list of supported languages.
+#language = None
+
+# There are two options for replacing |today|: either, you set today to some
+# non-false value, then it is used:
+#today = ''
+# Else, today_fmt is used as the format for a strftime call.
+#today_fmt = '%B %d, %Y'
+
+# List of documents that shouldn't be included in the build.
+#unused_docs = []
+
+# List of directories, relative to source directory, that shouldn't be searched
+# for source files.
+exclude_trees = []
+
+# The reST default role (used for this markup: `text`) to use for all documents.
+#default_role = None
+
+# If true, '()' will be appended to :func: etc. cross-reference text.
+#add_function_parentheses = True
+
+# If true, the current module name will be prepended to all description
+# unit titles (such as .. function::).
+#add_module_names = True
+
+# If true, sectionauthor and moduleauthor directives will be shown in the
+# output. They are ignored by default.
+#show_authors = False
+
+# The name of the Pygments (syntax highlighting) style to use.
+pygments_style = 'sphinx'
+
+# A list of ignored prefixes for module index sorting.
+#modindex_common_prefix = []
+
+
+# -- Options for HTML output ---------------------------------------------------
+
+# The theme to use for HTML and HTML Help pages. Major themes that come with
+# Sphinx are currently 'default' and 'sphinxdoc'.
+html_theme = 'default'
+
+# Theme options are theme-specific and customize the look and feel of a theme
+# further. For a list of options available for each theme, see the
+# documentation.
+#html_theme_options = {}
+
+# Add any paths that contain custom themes here, relative to this directory.
+#html_theme_path = []
+
+# The name for this set of Sphinx documents. If None, it defaults to
+# "<project> v<release> documentation".
+#html_title = None
+
+# A shorter title for the navigation bar. Default is the same as html_title.
+#html_short_title = None
+
+# The name of an image file (relative to this directory) to place at the top
+# of the sidebar.
+#html_logo = None
+
+# The name of an image file (within the static path) to use as favicon of the
+# docs. This file should be a Windows icon file (.ico) being 16x16 or 32x32
+# pixels large.
+#html_favicon = None
+
+# Add any paths that contain custom static files (such as style sheets) here,
+# relative to this directory. They are copied after the builtin static files,
+# so a file named "default.css" will overwrite the builtin "default.css".
+html_static_path = []
+
+# If not '', a 'Last updated on:' timestamp is inserted at every page bottom,
+# using the given strftime format.
+#html_last_updated_fmt = '%b %d, %Y'
+
+# If true, SmartyPants will be used to convert quotes and dashes to
+# typographically correct entities.
+#html_use_smartypants = True
+
+# Custom sidebar templates, maps document names to template names.
+#html_sidebars = {}
+
+# Additional templates that should be rendered to pages, maps page names to
+# template names.
+#html_additional_pages = {}
+
+# If false, no module index is generated.
+#html_use_modindex = True
+
+# If false, no index is generated.
+#html_use_index = True
+
+# If true, the index is split into individual pages for each letter.
+#html_split_index = False
+
+# If true, links to the reST sources are added to the pages.
+#html_show_sourcelink = True
+
+# If true, an OpenSearch description file will be output, and all pages will
+# contain a <link> tag referring to it. The value of this option must be the
+# base URL from which the finished HTML is served.
+#html_use_opensearch = ''
+
+# If nonempty, this is the file name suffix for HTML files (e.g. ".xhtml").
+#html_file_suffix = ''
+
+# Output file base name for HTML help builder.
+htmlhelp_basename = 'LAVATestDocumentation'
+
+
+# -- Options for LaTeX output --------------------------------------------------
+
+# The paper size ('letter' or 'a4').
+#latex_paper_size = 'letter'
+
+# The font size ('10pt', '11pt' or '12pt').
+#latex_font_size = '10pt'
+
+# Grouping the document tree into LaTeX files. List of tuples
+# (source start file, target name, title, author, documentclass [howto/manual]).
+latex_documents = [
+ ('index', 'LAVA Test.tex', u'LAVA Test Documentation',
+ u'Zygmunt Krynicki', 'manual'),
+]
+
+# The name of an image file (relative to this directory) to place at the top of
+# the title page.
+#latex_logo = None
+
+# For "manual" documents, if this is true, then toplevel headings are parts,
+# not chapters.
+#latex_use_parts = False
+
+# Additional stuff for the LaTeX preamble.
+#latex_preamble = ''
+
+# Documents to append as an appendix to all manuals.
+#latex_appendices = []
+
+# If false, no module index is generated.
+#latex_use_modindex = True
+
+
+# Example configuration for intersphinx: refer to the Python standard library.
+intersphinx_mapping = {'http://docs.python.org/': None}
=== added file 'doc/index.rst'
@@ -0,0 +1,81 @@
+=======================
+LAVA Test Documentation
+=======================
+
+LAVA Test is a wrapper framework exposing unified API and command line
+interface for running arbitrary tests and storing the results in a structured
+manner.
+
+LAVA Test is a part of the LAVA stack and can be used with other LAVA
+components, most notably the dispatcher (for setting up the test environment
+and controlling execution of multiple tests) and the dashboard (for storing
+
+.. seealso:: To learn more about LAVA see https://launchpad.net/lava
+
+60 second example
+=================
+
+This example will run on Ubuntu Lucid and beyond::
+
+ $ sudo add-apt-repository ppa:linaro-validation/ppa
+ $ sudo apt-get update
+ $ sudo apt-get install lava-test
+ $ lava-test install stream
+ $ lava-test run stream
+
+.. seealso:: For a more thorough description see :ref:`usage`
+.. seealso:: For detailed installation istructions see :ref:`installation`
+
+Features
+========
+
+* Ability to enumerate, install, run and remove tests on a Linux-based system.
+* Support for benchmarks as well as pass/fail tests.
+* Support for capturing environment information such as installed software and
+ hardware information and recording that in a machine-readable manner.
+* Store results in raw form (log files) as well as Linaro Dashboard Bundle
+ format that can be uploaded to the LAVA Dashboard for archiving and analysis.
+* Extensible API for adding new tests (:class:`~lava_test.api.core.ITest`) or even
+ collections of tests (:class:`~lava_test.api.core.ITestProvider`).
+* Ever-growing collection of freely available and generic tests and benchmarks
+
+.. seealso:: See what's new in :ref:`version_0_2`
+
+
+Latest documentation
+====================
+
+This documentation my be out of date, we try to make sure that all the latest
+and greatest releases are always documented on http://lava-test.readthedocs.org/
+
+
+Source code, bugs and patches
+=============================
+
+The project is maintained on Launchpad at http://launchpad.net/lava-test/.
+
+You can get the source code with bazaar using ``bzr branch lp:lava-test``.
+Patches can be submitted using Launchpad merge proposals (for introduction to
+this and topic see https://help.launchpad.net/Code/Review).
+
+Please report all bugs at https://bugs.launchpad.net/lava-test/+filebug.
+
+Most of the team is usually available in ``#linaro`` on ``irc.freenode.net``.
+Feel free to drop by to chat and ask questions.
+
+
+Indices and tables
+==================
+
+.. toctree::
+ :maxdepth: 2
+
+ installation.rst
+ changes.rst
+ usage.rst
+ reference.rst
+ todo.rst
+
+* :ref:`genindex`
+* :ref:`modindex`
+* :ref:`search`
=== added file 'doc/installation.rst'
@@ -0,0 +1,66 @@
+
+.. _installation:
+
+Installation
+============
+
+Prerequisites
+^^^^^^^^^^^^^
+
+The following debian packages are needed to use LAVA Test:
+
+* python-setuptools
+* python-apt
+* usbutils
+* python-testrepository - for running unit tests
+* python-sphinx - for building documentation
+
+
+Installation Options
+^^^^^^^^^^^^^^^^^^^^
+
+There are several installation options available:
+
+
+Using Ubuntu PPAs
+-----------------
+
+For Ubuntu 10.04 onward there is a stable PPA (personal package archive):
+
+* ppa:linaro-validation/ppa
+
+To add a ppa to an Ubuntu system use the add-apt-repository command::
+
+ sudo add-apt-repository ppa:linaro-validation/ppa
+
+After you add the PPA you need to update your package cache::
+
+ sudo apt-get update
+
+Finally you can install the package, it is called `lava-test`::
+
+ sudo apt-get install lava-test
+
+
+Using Python Package Index
+--------------------------
+
+This package is being actively maintained and published in the `Python Package
+Index <http://http://pypi.python.org>`_. You can install it if you have `pip
+<http://pip.openplans.org/>`_ tool using just one line::
+
+ pip install lava-test
+
+
+Using source tarball
+--------------------
+
+To install from source you must first obtain a source tarball from either pypi
+or from `Launchpad <http://launchpad.net/>`_. To install the package unpack the
+tarball and run::
+
+ python setup.py install
+
+You can pass ``--user`` if you prefer to do a local (non system-wide)
+installation. Note that executable programs are placed in ``~/.local/bin/`` and
+this directory is not on ``PATH`` by default.
=== added file 'doc/reference.rst'
@@ -0,0 +1,107 @@
+.. _reference:
+
+=========
+Reference
+=========
+
+.. _command_reference:
+
+Command Reference
+=================
+
+.. automodule:: lava_test.commands
+ :members:
+
+.. todo::
+
+ * Describe basic commands
+ * Describe arguments and options to each command in detail
+
+Pathnames and files
+===================
+
+LAVA Test uses the following files:
+
+* ``$XDG_CONFIG_HOME/lava_test/`` -- configuration files
+* ``$XDG_DATA_HOME/lava_test/installed-tests`` -- installed test programs
+* ``$XDG_DATA_HOME/lava_test/results`` -- artifacts of running tests
+* ``$XDG_CACHE_HOME/lava_test/`` -- download cache
+
+.. _code_reference:
+
+Code reference
+==============
+
+.. todo::
+
+ * Describe general code layout
+ * Describe key API integration points (on a separate page if needed for clarity)
+ * Provide an example test and walk the reader through the meaning of each part
+
+Abstract Interfaces
+^^^^^^^^^^^^^^^^^^^
+
+.. automodule:: lava_test.api.core
+ :members:
+
+.. automodule:: lava_test.api.delegates
+ :members:
+
+.. automodule:: lava_test.api.observers
+ :members:
+
+Test definitions and test providers
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+.. automodule:: lava_test.core.providers
+ :members:
+
+.. automodule:: lava_test.core.tests
+ :members:
+
+Test components (installers, runners and parsers)
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+.. automodule:: lava_test.core.installers
+ :members:
+
+.. automodule:: lava_test.core.runners
+ :members:
+
+.. automodule:: lava_test.core.parsers
+ :members:
+
+Core Modules
+^^^^^^^^^^^^
+
+.. automodule:: lava_test.core.artifacts
+ :members:
+
+.. automodule:: lava_test.core.config
+ :members:
+
+Environment Scanners
+^^^^^^^^^^^^^^^^^^^^
+
+.. automodule:: lava_test.core.hwprofile
+ :members:
+
+.. automodule:: lava_test.core.swprofile
+ :members:
+
+Utilities
+^^^^^^^^^
+
+.. automodule:: lava_test.utils
+ :members:
+
+.. automodule:: lava_test.extcmd
+ :members:
+
+
+Abrek compatibility
+===================
+
+.. automodule:: abrek.testdef
+ :members:
+
=== added file 'doc/todo.rst'
@@ -0,0 +1,4 @@
+List of items that need work
+============================
+
+.. todolist::
=== added file 'doc/usage.rst'
@@ -0,0 +1,195 @@
+.. _usage:
+
+=====
+Usage
+=====
+
+Workflow Overview
+=================
+
+LAVA Test can be used in several different ways. Most notably those are
+standalone (without the LAVA dispatcher) and managed (when LAVA Test is
+installed and controlled by the LAVA dispatcher).
+
+Standalone usage
+^^^^^^^^^^^^^^^^
+
+In standalone mode a human operator installs LAVA Test on some device
+(development board, laptop or other computer or a virtual machine), installs
+the tests that are to be executed and then executes them manually (by manually
+running LAVA test, the actual tests are non-interactive).
+
+Using LAVA to develop and run new tests
++++++++++++++++++++++++++++++++++++++++
+
+This mode is useful for test development (adding new tests, developing custom
+tests especially tailored for LAVA, etc.). Here the typical cycle depends on
+how the tests is wrapped for usage by LAVA and what the test developer is
+focusing on.
+
+While developing the actual test the typical set of commands might look like
+this::
+
+ $ lava-test install my-custom-test
+ $ lava-test run my-custom-test
+ $ lava-test uninstall my-custom-test
+
+Here the developer could observe changes to the test program (that is
+presumably compiled and copied somewhere by the install stage).
+
+Using LAVA to analyze test results
+++++++++++++++++++++++++++++++++++
+
+Developing the test is only half of the story. The other half is developing
+LAVA Test integration code, most importantly the artefact parser / analyzer.
+This part has to be implemented in python (unlike the test program that can be
+implemented in any language and technology). Here the developer is focusing on
+refining the parser to see if the outcome is as indented. Assuming that earlier
+the developer ran the test at least once and wrote down the result identifier
+the set of commands one might use is::
+
+ $ lava-test parse my-custom-test my-custom-test.2011-08-19T23:53:21Z | pager
+
+Here the developer has to pass both the identifier of the test
+(``my-custom-test``) as well as the identifier of the actual result. While
+currently the result identifier starts with the test identifier we wanted to
+avoid magic values like that so both are needed. The test defines which
+artefact parser to use. The result id is used to locate leftovers from running
+that specific test at some previous point in time.
+
+By default parse will print the bundle to standard output for inspection. It
+should be redirected to a pager for easier verification.
+
+.. note::
+
+ While the syntax of the bundle created with `lava-test parse` is always
+ correct (or, if the parser does something really, really strange, a
+ detailed error is reported) the actual contents may not be what you
+ intended it to be. Parsers are ultimately fragile as they mostly deal with
+ unstructured or semi-structured free-form text that most test programs seem
+ to produce. The ultimate goal of a developer should be to produce
+ unambiguous, machine readable format. This level of integration would allow
+ to wrap a whole class of tests in one go (such as all xUnit-XML speaking
+ test frameworks).
+
+Usage with the dispatcher
+^^^^^^^^^^^^^^^^^^^^^^^^^
+
+The dispatcher is useful for automating LAVA Test environment setup, describing
+test scenarios (the list of tests to invoke) and finally storing the results in
+the LAVA dashboard.
+
+Typically this mode is based on the following sequence of commands:
+
+#. Install lava-test (from PPA or source) along with the required dependencies
+#. (optional) for out of tree tests install the additional `test definition` package
+#. Install the test or tests that are to be invoked with ``lava-tool install``.
+#. Run, parse and store in one go with ``lava-tool run --output=FILE``.
+
+Here the whole setup is non-interactive and at the end the dispatcher can copy
+the output bundle for additional processing.
+
+Automation considerations
+^^^^^^^^^^^^^^^^^^^^^^^^^
+
+.. _wrapping_existing_test_or_benchmark:
+
+Wrapping existing test or benchmark
+===================================
+
+LAVA Test can be extended in several different ways. There is no best method,
+each has some pros and cons. In general we welcome any freely redistributable,
+generic tests. Those enrich the LAVA ecosystem and by providing useful
+out-of-the-box features to our users.
+
+Technically all tests are hidden behind a set of abstract interfaces that tell
+LAVA Test what to do in response to operator or dispatcher actions. The primary
+interface is :class:`~lava_test.api.core.ITest` and the three principal
+methods: :meth:`~lava_test.api.core.ITest.install`,
+:meth:`~lava_test.api.core.ITest.run`,
+:meth:`~lava_test.api.core.ITest.parse`.
+
+In practice it is usually much easier to instantiate our pluggable delegate
+test (:class:`lava_test.core.tests.Test`) and define the three delegates that
+know how to install, run and parse. Again for each step we have a base class
+that can be easily customized or even used directly as is. Those classes are
+:class:`~lava_test.core.installers.TestInstaller`,
+:class:`~lava_test.core.runners.TestRunner` and
+:class:`~lava_test.core.parsers.TestParser`. They all implement well-defined
+interfaces (specified in :mod:`lava_test.api.delegates`) so if you wish to
+customize them you should become familiar with the API requirements first.
+
+Contributing new tests to LAVA
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+The most direct way to add a new test is to contribute patches to LAVA Test
+itself. This method will simply add a new test definition to the collection of
+available tests.
+
+This method is recommended for generic tests that rarely change and are
+suitable for wide variety of hardware and software (assuming basic Linux-like
+system, Android tests are a special case).
+
+The advantage is that those tests can be invoked out of the box and will be
+maintained by the LAVA team. The disadvantage is that all changes to those
+tests need to follow Linaro development work flow, get reviewed and finally
+merged. Depending on your situation this may be undesired.
+
+.. todo::
+
+ Describe how tests are discovered, loaded and used. It would be
+ nice to have a tutorial that walks the user through wrapping a
+ simple pass/fail test.
+
+Maintaining out-of-tree tests
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+For some kinds of tests (proprietary, non-generic, in rapid development, fused
+with application code) contributing their definition to upstream LAVA Test
+project would be impractical.
+
+In such cases the test maintainer can still leverage LAVA to actually run and
+process the test without being entangled in the review process or going through
+any public channel.
+
+Because LAVA Test supports pluggable test providers it is easy to add a new
+source of test definitions. Fortunately we ship with a very useful generic
+out-of-tree test provider based on the python `pkg_resources` system.
+
+Any python package (that is a module or package and the corresponding setup.py
+and .egg_info) can define LAVA Test extensions using the `pkg_resurces` entry
+points system.
+
+To do this write your test program as you normally would, write the LAVA Test
+integration code and put this into your integration package setup.py::
+
+ setup(
+ ...,
+ entry_points="""[lava_test.test_definitions]
+ my_test_id=my_package.my_module
+ """)
+
+Here we'd define an entry point in the ``lava_test.test_definitions`` namespace
+that LAVA Test searches by default. In that namespace we define one object
+``my_test_id`` which points at the module ``my_package.my_module``. LAVA Test
+will discover this entry point, import the relevant module and make the test
+definition available.
+
+Maintaining simple declarative tests
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+By registering pure declarative tests at runtime.
+
+.. todo::
+
+ Describe how to use declarative tests. It would be a nice
+ extension of the tutorial once the user feels comfortable with
+ the initial python-based version.
+
+Writing new tests from scratch
+==============================
+
+.. todo::
+
+ Describe considerations for test writers. Using native test
+ format with human-readable output adapters.
=== modified file 'examples/power-management-tests.json'
@@ -1,5 +1,5 @@
{
- "format": "Abrek Test Definition 1.0 Experimental",
+ "format": "Lava-Test Test Definition 1.0",
"test_id": "linaro.pmwg",
"install": {
"steps": ["bzr get lp:~zkrynicki/+junk/linaro-pm-qa-tests"],
=== modified file 'examples/stream.json'
@@ -1,5 +1,5 @@
{
- "format": "Abrek Test Definition Format 1.0 Experimental",
+ "format": "LAVA-Test Test Definition Format",
"test_id": "stream-json",
"install": {
"url": "http://www.cs.virginia.edu/stream/FTP/Code/stream.c",
=== renamed directory 'abrek' => 'lava_test'
=== modified file 'lava_test/__init__.py'
@@ -13,4 +13,4 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
-__version__ = "0.2.0"
+__version__ = (0, 2, 0, "dev", 0)
=== added directory 'lava_test/api'
=== added file 'lava_test/api/__init__.py'
@@ -0,0 +1,24 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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/>.
+
+from abc import ABCMeta
+
+
+class _Interface(object):
+ """
+ Interface class for simplifying usage of interface meta-classes
+ """
+
+ __metaclass__ = ABCMeta
=== added file 'lava_test/api/core.py'
@@ -0,0 +1,164 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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/>.
+
+"""
+:mod:`lava_test.api.core` -- Interface classes for core LAVA Test features
+==========================================================================
+
+.. module: lava_test.api.core
+
+ :synopsis: Interface classes for core LAVA Test features
+"""
+
+from abc import abstractmethod, abstractproperty
+
+from lava_test.api import _Interface
+
+class ITest(_Interface):
+ """
+ Abstract test definition.
+
+ Test definitions allow lava-test to install, remove, run and parse log
+ files of automatic tests. While the interface can be implemented directly
+ you should use :class:`lava_test.core.tests.Test` that implements the core
+ logic and allow you to customize the parts that are needed by providing
+ delegates implementing :class:`~lava_test.api.delegates.ITestInstaller`,
+ :class:`~lava_test.api.delegates.ITestRunner` and
+ :class:`~lava_test.api.delegates.ITestParser`.
+
+ .. seealso:: :ref:`wrapping_existing_test_or_benchmark`
+ """
+
+ @abstractproperty
+ def is_installed(self):
+ """
+ True if this test is installed
+
+ .. versionadded:: 0.2
+ """
+
+ @abstractmethod
+ def install(self, observer):
+ """
+ Install the test program.
+
+ This creates an install directory under the user's XDG_DATA_HOME
+ directory to mark that the test is installed. The installer's
+ install() method is then called from this directory to complete any
+ test specific install that may be needed.
+
+ :param observer:
+ Observer object that makes it possible to monitor the actions
+ performed by the test installer.
+ :type observer: :class:`~lava_test.api.observers.ITestInstallerObserver`
+
+ .. versionadded:: 0.2
+ """
+
+ @abstractmethod
+ def uninstall(self):
+ """
+ Remove the test program
+
+ Recursively remove test specific directory under the user's
+ ``XDG_DATA_HOME directory``. This will both mark the test as removed,
+ and clean up any files that were downloaded or installed under that
+ directory. Dependencies are intentionally not removed by this.
+
+ .. versionadded:: 0.1
+ """
+
+ @abstractmethod
+ def run(self, observer):
+ """
+ Run the test program and store artifacts.
+
+ :param observer:
+ Observer object that makes it possible to monitor the actions
+ performed by the test runner.
+ :type observer: :class:`~lava_test.api.observers.ITestRunnerObserver`
+ :return: Test run artifacts
+ :rtype: :class:`~lava_test.core.artifacts.TestArtifacts`.
+
+ .. versionadded:: 0.2
+ """
+
+ @abstractmethod
+ def parse(self, artifacts):
+ """
+ Parse the artifacts of an earlier run.
+
+ :param artifacts: Object that describes which files should be parsed.
+ :type artifacts: :class:`~lava_test.core.artifacts.TestArtifacts`
+ :return:
+ A dictionary with all the parsed data. In particular this is a
+ TestRun part of the dashboard bundle so it should have the
+ test_results list of all the results parsed from the artifacts.
+ :rtype: :class:`dict`
+
+ .. versionadded:: 0.2
+ """
+
+
+class ITestProvider(_Interface):
+ """
+ Source of ITest instances.
+
+ Test providers can be used to make lava-test aware of arbitrary collections
+ of tests that can be installed and invoked. Internally lava-test uses this
+ class to offer built-in tests (via the
+ :class:`~lava_test.providers.BuiltInProvider`), out-of-tree tests (via the
+ :class:`~lava_test.providers.PkgResourcesProvider`) and declarative tests
+ (via the :class:`~lava_test.providers.RegistryProvider`).
+
+ Normally this is not something you would need to implement. If you have a
+ large collection of existing tests that can be somehow adapted in bulk, or
+ you have your own internal registry of tests that could be adapted this way
+ then you might use this interface to simplify test discovery.
+
+ Test providers need to be registered using pkg-resources entry-point
+ feature and then added to the lava-test configuration file. See
+ :class:`lava_test.config.LavaTestConfig` for details.
+
+ .. versionadded:: 0.2
+ """
+
+ @abstractmethod
+ def __init__(self, config):
+ """
+ Initialize test provider with the specified configuration object. The
+ configuration object is obtained from the test tool providers registry.
+ """
+
+ @abstractmethod
+ def __iter__(self):
+ """
+ Iterates over instances of ITest exposed by this provider
+ """
+
+ @abstractmethod
+ def __getitem__(self, test_id):
+ """
+ Return an instance of ITest with the specified id
+ """
+
+ @abstractproperty
+ def description(self):
+ """
+ The description string used by `lava-test list-tests`
+ """
+
+
+__all__ = ['ITest', 'ITestProvider']
=== added file 'lava_test/api/delegates.py'
@@ -0,0 +1,119 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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/>.
+
+
+"""
+:mod:`lava_test.api.delegates` -- Interface classes for test delegates
+======================================================================
+
+.. module: lava_test.api.delegates
+
+ :synopsis: Interface classes for test delegates
+"""
+
+from abc import abstractmethod, abstractproperty
+
+from lava_test.api import _Interface
+
+
+class ITestInstaller(_Interface):
+ """
+ Test installer delegate class.
+
+ Wraps the knowledge on how to install a test. It is most helpful with
+ :class:`~lava_test.core.tests.Test` that delegates actual actions to helper
+ classes.
+
+ .. versionadded:: 0.2
+ """
+
+ @abstractmethod
+ def install(self, observer):
+ """
+ Install the test program.
+
+ :param observer:
+ Observer object that makes it possible to monitor the actions
+ performed by the test installer.
+ :type observer: :class:`~lava_test.api.observers.ITestInstallerObserver`
+
+ .. versionadded:: 0.2
+ """
+
+
+class ITestRunner(_Interface):
+ """
+ Test runner delegate.
+
+ Wraps the knowledge on how to run a test. It is most helpful with
+ :class:`lava_test.core.tests.Test` that delegates actual actions to
+ helper classes.
+
+ .. versionadded:: 0.2
+ """
+
+ @abstractmethod
+ def run(self, artifacts, observer):
+ """
+ Run the test and create artifacts (typically log files).
+
+ Artifacts must be created in the directory specified by various methods
+ and properties of of :class:`lava_test.core.TestArtifacts`.
+
+ :param artifacts:
+ Object that describes where to store test run artifacts
+ :type artifacts: :class:`~lava_test.core.artifacts.TestArtifacts`.
+ :param observer:
+ Observer object that makes it possible to monitor the actions
+ performed by the test runner.
+ :type observer: :class:`~lava_test.api.observers.ITestRunnerObserver`
+
+ .. versionadded:: 0.2
+ """
+
+
+class ITestParser(_Interface):
+ """
+ Test artefact parser delegate.
+
+ Wraps the knowledge on how to parse the artifacts of a previous test run.
+ It is most helpful with :class:`~lava_test.core.tests.Test` that delegates
+ actual actions to helper classes.
+
+ .. versionadded:: 0.2
+ """
+
+ @abstractmethod
+ def parse(self, artifacts):
+ """
+ Parse the artifacts of a previous test run and return a dictionary with
+ a partial TestRun object.
+
+ :param artifacts:
+ Object that describes where to find test run artifacts
+ :type artifacts: :class:`~lava_test.core.artifacts.TestArtifacts`.
+
+ .. versionadded:: 0.2
+ """
+
+ @abstractproperty
+ def results(self):
+ """
+ Results dictionary to be merged with TestRun object inside the bundle.
+
+ .. seealso:: :meth:`~lava_test.core.artifacts.TestArtifacts.incorporate_parse_results`
+
+ .. versionadded:: 0.1
+ """
=== added file 'lava_test/api/observers.py'
@@ -0,0 +1,120 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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/>.
+
+
+"""
+:mod:`lava_test.api.observers` -- Interface classes for observer classes
+========================================================================
+
+.. module: lava_test.api.observers
+ :synopsis: Interface classes for observer classes
+"""
+
+from abc import abstractmethod
+
+from lava_test.api import _Interface
+
+
+class IShellCommandObserver(_Interface):
+ """
+ Shell command runner observer class.
+
+ Allows the caller to observe shell commands that occur during some
+ operation. It is used by the command line UI.
+
+ .. versionadded:: 0.2
+ """
+
+ @abstractmethod
+ def about_to_run_shell_command(self, cmd):
+ """
+ Method called when a shell command is about to be invoked by the
+ observed object.
+
+ .. versionadded:: 0.2
+ """
+
+ @abstractmethod
+ def did_run_shell_command(self, cmd, returncode):
+ """
+ Method called when a shell command has been invoked by the observed
+ object.
+
+ .. versionadded:: 0.2
+ """
+
+ @abstractmethod
+ def display_subprocess_output(self, stream_name, line):
+ """
+ Method called for each line of stdout/stderr as obtained from a
+ subprocess.
+
+ .. versionadded:: 0.2
+ """
+
+
+class ITestInstallerObserver(IShellCommandObserver):
+ """
+ Test installer observer class.
+
+ Allows the caller to observe interesting actions that occur during
+ installation process. It is used by the command line UI.
+
+ .. versionadded:: 0.2
+ """
+
+ @abstractmethod
+ def about_to_install_packages(self, package_list):
+ """
+ Method called when a list of packages is about to be installed by the
+ installer
+
+ .. versionadded:: 0.2
+ """
+
+ @abstractmethod
+ def did_install_packages(self, package_list):
+ """
+ Method called when a package has been installed by the installer
+
+ .. versionadded:: 0.2
+ """
+
+ @abstractmethod
+ def about_to_download_file(self, url):
+ """
+ Method called when a file is about to be downloaded
+
+ .. versionadded:: 0.2
+ """
+
+ @abstractmethod
+ def did_download_file(self, url):
+ """
+ Method called when a file has been downloaded
+
+ .. versionadded:: 0.2
+ """
+
+
+class ITestRunnerObserver(IShellCommandObserver):
+ """
+ Test runner observer class.
+
+ Allows the caller to observe interesting actions that occur during testing
+ process. It is used by the command line UI.
+
+ .. versionadded:: 0.2
+ """
=== added file 'lava_test/commands.py'
@@ -0,0 +1,383 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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 subprocess
+
+from lava_tool.interface import Command as LavaCommand
+from lava_tool.interface import LavaCommandError
+import versiontools
+
+from lava_test.api.observers import (
+ ITestInstallerObserver,
+ ITestRunnerObserver)
+from lava_test.core.artifacts import TestArtifacts
+from lava_test.core.config import get_config
+from lava_test.core.loader import TestLoader
+
+
+class Command(LavaCommand, ITestInstallerObserver, ITestRunnerObserver):
+
+ def __init__(self, parser, args):
+ super(Command, self).__init__(parser, args)
+ self._config = get_config()
+ self._test_loader = TestLoader(self._config)
+
+ @classmethod
+ def register_arguments(cls, parser):
+ parser.add_argument(
+ "-q", "--quiet",
+ action="store_true",
+ default=False,
+ help="Be less verbose about undertaken actions")
+ parser.add_argument(
+ "-Q", "--quiet-subcommands",
+ action="store_true",
+ default=False,
+ help="Hide the output of all sub-commands (including tests)")
+
+ def say(self, text, *args, **kwargs):
+ print "LAVA:", text.format(*args, **kwargs)
+
+ def about_to_install_packages(self, package_list):
+ if self.args.quiet:
+ return
+ self.say("Installing packages: {0}", ", ".join(package_list))
+
+ def about_to_run_shell_command(self, cmd):
+ if self.args.quiet:
+ return
+ self.say("Running shell command: {0!r}", cmd)
+
+ def about_to_download_file(self, url):
+ if self.args.quiet:
+ return
+ self.say("Downloading file from: {0!r}", url)
+
+ def did_install_packages(self, package_list):
+ pass
+
+ def did_run_shell_command(self, cmd, returncode):
+ if returncode is None:
+ self.say("Command {0!r} was terminated prematurely", cmd)
+ elif returncode != 0:
+ self.say("Command {0!r} returned non-zero exit status {1}",
+ cmd, returncode)
+
+ def did_download_file(self, url):
+ pass
+
+ def display_subprocess_output(self, stream_name, line):
+ if self.args.quiet_subcommands:
+ return
+ if stream_name == 'stdout':
+ self.say('(stdout) {0}', line.rstrip())
+ elif stream_name == 'stderr':
+ self.say('(stderr) {0}', line.rstrip())
+
+
+class list_tests(Command):
+ """
+ List available tests
+
+ .. program:: lava-test list-tests
+
+ Lists all available tests, grouping them by provider.
+ """
+
+ def invoke(self):
+ for provider in self._test_loader.get_providers():
+ test_list = [provider[test_id] for test_id in provider]
+ if not test_list:
+ continue
+ self.say("{0}", provider.description)
+ for test in test_list:
+ self.say(" - {test_id}", test_id=test.test_id)
+
+
+class list_installed(Command):
+ """
+ List installed tests
+ """
+
+ def invoke(self):
+ for provider in self._test_loader.get_providers():
+ test_list = [provider[test_id] for test_id in provider]
+ if not test_list:
+ continue
+ self.say("{0}", provider.description)
+ count = 0
+ for test in test_list:
+ if not test.is_installed:
+ continue
+ self.say(" - {test_id}", test_id=test.test_id)
+ count += 1
+ if not count:
+ self.say("No tests installed")
+
+
+
+class TestAffectingCommand(Command):
+
+ INSTALL_REQUIRED = False
+
+ @classmethod
+ def register_arguments(cls, parser):
+ super(TestAffectingCommand, cls).register_arguments(parser)
+ parser.add_argument("test_id",
+ help="Test or test suite identifier")
+
+ def invoke(self):
+ try:
+ test = self._test_loader[self.args.test_id]
+ except KeyError:
+ raise LavaCommandError("There is no test with the specified ID")
+ return self.invoke_with_test(test)
+
+
+class install(TestAffectingCommand):
+ """
+ Install a test program
+ """
+
+ def invoke_with_test(self, test):
+ if test.is_installed:
+ raise LavaCommandError("This test is already installed")
+ try:
+ test.install(self)
+ except (subprocess.CalledProcessError, RuntimeError) as ex:
+ raise LavaCommandError(str(ex))
+
+
+class uninstall(TestAffectingCommand):
+ """
+ Uninstall a test program
+ """
+
+ def invoke_with_test(self, test):
+ if not test.is_installed:
+ raise LavaCommandError("This test is not installed")
+ test.uninstall()
+
+
+class run(TestAffectingCommand):
+ """
+ Run a previously installed test program
+ """
+
+ @classmethod
+ def register_arguments(cls, parser):
+ super(run, cls).register_arguments(parser)
+ group = parser.add_argument_group("initial bundle configuration")
+ group.add_argument("-S", "--skip-software-context",
+ default=False,
+ action="store_true",
+ help=("Do not store the software context in the"
+ " initial bundle. Typically this saves OS"
+ " image name and all the installed software"
+ " packages."))
+ group.add_argument("-H", "--skip-hardware-context",
+ default=False,
+ action="store_true",
+ help=("Do not store the hardware context in the"
+ " initial bundle. Typically this saves CPU,"
+ " memory and USB device information."))
+ group.add_argument("--trusted-time",
+ default=False,
+ action="store_true",
+ help=("Indicate that the real time clock has"
+ " accurate data. This can differentiate"
+ " test results created on embedded devices"
+ " that often have inaccurate real time"
+ " clock settings."))
+ group = parser.add_argument_group("complete bundle configuration")
+ group.add_argument("-o", "--output",
+ default=None,
+ metavar="FILE",
+ help=("After running the test parse the result"
+ " artifacts, fuse them with the initial"
+ " bundle and finally save the complete bundle"
+ " to the specified FILE."))
+ group.add_argument("-A", "--skip-attachments",
+ default=False,
+ action="store_true",
+ help=("Do not store standard output and standard"
+ " error log files as attachments. This"
+ " option is only affecting the bundle"
+ " created with --output, the initial bundle"
+ " is not affected as it never stores any"
+ " attachments."))
+
+ def invoke_with_test(self, test):
+ if not test.is_installed:
+ raise LavaCommandError("The specified test is not installed")
+ try:
+ artifacts = test.run(self)
+ except subprocess.CalledProcessError as ex:
+ if ex.returncode is None:
+ raise LavaCommandError("Command %r was aborted" % ex.cmd)
+ else:
+ raise LavaCommandError(str(ex))
+ except RuntimeError as ex:
+ raise LavaCommandError(str(ex))
+ self.say("run complete, result_id is {0!r}", artifacts.result_id)
+ artifacts.create_initial_bundle(
+ self.args.skip_software_context,
+ self.args.skip_hardware_context,
+ self.args.trusted_time)
+ artifacts.save_bundle()
+ if self.args.output:
+ parse_results = test.parse(artifacts)
+ artifacts.incorporate_parse_results(parse_results)
+ if not self.args.skip_attachments:
+ artifacts.attach_standard_files_to_bundle()
+ artifacts.save_bundle_as(self.args.output)
+
+
+class parse(TestAffectingCommand):
+ """
+ Parse the results of previous test run
+ """
+
+ @classmethod
+ def register_arguments(cls, parser):
+ super(parse, cls).register_arguments(parser)
+ parser.add_argument("result_id",
+ help="Test run result identifier")
+ group = parser.add_argument_group("complete bundle configuration")
+ group.add_argument("-o", "--output",
+ default=None,
+ metavar="FILE",
+ help=("After running the test parse the result"
+ " artifacts, fuse them with the initial"
+ " bundle and finally save the complete bundle"
+ " to the specified FILE."))
+ group.add_argument("-A", "--skip-attachments",
+ default=False,
+ action="store_true",
+ help=("Do not store standard output and standard"
+ " error log files as attachments. This"
+ " option is only affecting the bundle"
+ " created with --output, the initial bundle"
+ " is not affected as it never stores any"
+ " attachments."))
+
+ def invoke_with_test(self, test):
+ artifacts = TestArtifacts(
+ self.args.test_id, self.args.result_id, self._config)
+ if not os.path.exists(artifacts.bundle_pathname):
+ raise LavaCommandError("Specified result does not exist")
+ artifacts.load_bundle()
+ parse_results = test.parse(artifacts)
+ artifacts.incorporate_parse_results(parse_results)
+ self.say("Parsed {0} test results",
+ len(artifacts.bundle["test_runs"][0]["test_results"]))
+ print artifacts.dumps_bundle()
+ if self.args.output:
+ if not self.args.skip_attachments:
+ artifacts.attach_standard_files_to_bundle()
+ artifacts.save_bundle_as(self.args.output)
+
+
+class show(Command):
+ """
+ Display the output from a previous test run
+ """
+
+ @classmethod
+ def register_arguments(cls, parser):
+ super(show, cls).register_arguments(parser)
+ parser.add_argument("result_id",
+ help="Test run result identifier")
+
+ def invoke(self):
+ artifacts = TestArtifacts(None, self.args.result_id, self._config)
+ if not os.path.exists(artifacts.results_dir):
+ raise LavaCommandError("Specified result does not exist")
+ if os.path.exists(artifacts.stdout_pathname):
+ with open(artifacts.stdout_pathname, "rt") as stream:
+ for line in iter(stream.readline, ''):
+ self.display_subprocess_output("stdout", line)
+ if os.path.exists(artifacts.stderr_pathname):
+ with open(artifacts.stderr_pathname, "rt") as stream:
+ for line in iter(stream.readline, ''):
+ self.display_subprocess_output("stderr", line)
+
+
+class version(Command):
+ """
+ Show LAVA Test version
+ """
+
+ def invoke(self):
+ self.say("version details:")
+ for framework in self._get_frameworks():
+ self.say(" - {framework}: {version}",
+ framework=framework.__name__,
+ version=versiontools.format_version(
+ framework.__version__, framework))
+
+ def _get_frameworks(self):
+ import lava_tool
+ import lava_test
+ import linaro_dashboard_bundle
+ import linaro_json
+ return [
+ lava_test,
+ lava_tool,
+ linaro_dashboard_bundle,
+ linaro_json]
+
+
+class register_test(Command):
+ """
+ Register remote test
+ """
+
+ @classmethod
+ def register_arguments(cls, parser):
+ super(register_test, cls).register_arguments(parser)
+ parser.add_argument("test_url",
+ help="Url for test definition file")
+
+ def invoke(self):
+ try:
+ from lava_test.core.providers import RegistryProvider
+ RegistryProvider.register_remote_test(self.args.test_url)
+ except ValueError as exc:
+ raise LavaCommandError("Unable to register test: %s" % exc)
+ except KeyError:
+ raise LavaCommandError("There is no test_url")
+
+class unregister_test(Command):
+ """
+ Unregister remote test
+ """
+
+ @classmethod
+ def register_arguments(cls, parser):
+ super(unregister_test, cls).register_arguments(parser)
+ parser.add_argument("test_url",
+ help="Url for test definition file")
+
+ def invoke(self):
+ try:
+ from lava_test.core.providers import RegistryProvider
+ RegistryProvider.unregister_remote_test(self.args.test_url)
+ except ValueError as exc:
+ raise LavaCommandError("Unable to unregister test: %s" % exc)
+ except KeyError:
+ raise LavaCommandError("There is no test_url")
=== added directory 'lava_test/core'
=== added file 'lava_test/core/__init__.py'
=== added file 'lava_test/core/artifacts.py'
@@ -0,0 +1,277 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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/>.
+
+from __future__ import absolute_import
+
+import base64
+import datetime
+import logging
+import os
+import uuid
+
+from linaro_dashboard_bundle.io import DocumentIO
+
+from lava_test.core import hwprofile, swprofile
+from lava_test.utils import merge_dict, mkdir_p
+
+
+class TestArtifacts(object):
+ """
+ Class representing test run artifacts, that is, static leftovers
+ independent of the wrapper class that encapsulates test handling.
+
+ .. versionadded:: 0.2
+ """
+
+ def __init__(self, test_id, result_id, config):
+ self._test_id = test_id
+ self._result_id = result_id
+ self._config = config
+ self._bundle = None
+
+ @classmethod
+ def allocate(cls, test_id, config):
+ """
+ Allocate new test artifacts object that corresponds to the specified
+ test_id. This constructs a new result_id and creates the corresponding
+ filesystem directory that holds those artifacts.
+
+ .. versionadded:: 0.2
+ """
+ result_id = (
+ "{test_id}.{time.tm_year:04}-{time.tm_mon:02}-{time.tm_mday:02}T"
+ "{time.tm_hour:02}:{time.tm_min:02}:{time.tm_sec:02}Z").format(
+ test_id=test_id,
+ time=datetime.datetime.utcnow().timetuple())
+ self = cls(test_id, result_id, config)
+ logging.debug("Creating result directory: %r", self.results_dir)
+ mkdir_p(self.results_dir)
+ return self
+
+ @property
+ def test_id(self):
+ """
+ The ID of the test this run is associated with
+
+ .. versionadded:: 0.2
+ """
+ return self._test_id
+
+ @property
+ def result_id(self):
+ """
+ The ID of the test run.
+
+ This field is different from analyzer_assigned_uuid at this time but
+ may change in the future. The purpose of this field is to identify the
+ test run and be able to locate attachments/log files/bundle on the file
+ system.
+
+ .. versionadded:: 0.2
+ """
+ return self._result_id
+
+ @property
+ def results_dir(self):
+ """
+ Pathname of a directory with test run artifacts (log files, crash
+ dumps, etc).
+
+ .. versionadded:: 0.2
+ """
+ return os.path.join(self._config.resultsdir, self.result_id)
+
+ def load_bundle(self):
+ """
+ Load the results bundle from disk.
+
+ The bundle is also validated if linaro-dashboard-bundle library is
+ installed.
+ """
+ with open(self.bundle_pathname, 'rt') as stream:
+ self._bundle = DocumentIO.load(stream)[1]
+
+ def dumps_bundle(self):
+ return DocumentIO.dumps(self._bundle)
+
+ def save_bundle(self):
+ """
+ Save the results bundle to the disk
+
+ The bundle is also validated if linaro-dashboard-bundle library is
+ installed.
+ """
+ self.save_bundle_as(self.bundle_pathname)
+
+ def save_bundle_as(self, pathname):
+ """
+ Save the results bundle to the specified file on disk.
+
+ The bundle should have been created or loaded earlier
+ """
+ with open(pathname, 'wt') as stream:
+ DocumentIO.dump(stream, self._bundle)
+
+ @property
+ def bundle(self):
+ """
+ The deserialized bundle object.
+
+ This can be either created with create_bundle() or loaded
+ from disk with load_bundle()
+ """
+ return self._bundle
+
+ def create_initial_bundle(self,
+ skip_software_context=False,
+ skip_hardware_context=False,
+ time_check_performed=False):
+ """
+ Create the bundle object.
+
+ This creates a typical bundle structure. Optionally it can also add
+ software and hardware context information.
+
+ For a complete bundle you may want to add attachments and incorporate
+ parse results by calling appropriate methods after loading or creating
+ the initial bundle.
+ """
+ TIMEFORMAT = '%Y-%m-%dT%H:%M:%SZ'
+ # Generate UUID and analyzer_assigned_date for the test run
+ analyzer_assigned_uuid = str(uuid.uuid1())
+ analyzer_assigned_date = datetime.datetime.utcnow()
+ # Create basic test run structure
+ test_run = {
+ 'test_id': self.test_id,
+ 'analyzer_assigned_date': analyzer_assigned_date.strftime(
+ TIMEFORMAT),
+ 'analyzer_assigned_uuid': analyzer_assigned_uuid,
+ 'time_check_performed': time_check_performed,
+ "test_results": [],
+ "attachments": [],
+ }
+ # Store hardware and software context if requested
+ if not skip_software_context:
+ test_run['software_context'] = swprofile.get_software_context()
+ if not skip_hardware_context:
+ test_run['hardware_context'] = hwprofile.get_hardware_context()
+ # Create the bundle object
+ self._bundle = {
+ 'format': 'Dashboard Bundle Format 1.2',
+ 'test_runs': [test_run]}
+
+ @property
+ def test_run(self):
+ try:
+ return self._bundle["test_runs"][0]
+ except KeyError:
+ raise AttributeError("test_run can be accessed only after you load"
+ " or create an initial bundle")
+
+ def attach_file(self, real_pathname, stored_pathname, mime_type):
+ """
+ Append an attachment to the test run.
+
+ The file is only attached if real_pathname designates an existing,
+ nonempty file. If the mime_type starts with 'text/' the file is opened
+ in text mode, otherwise binary mode is used.
+ """
+ if not os.path.exists(real_pathname):
+ return
+ if mime_type.startswith('text/'):
+ mode = 'rt'
+ else:
+ mode = 'rb'
+ with open(real_pathname, mode) as stream:
+ data = stream.read()
+ if not data:
+ return
+ self.test_run['attachments'].append({
+ "pathname": stored_pathname,
+ "mime_type": mime_type,
+ "content": base64.standard_b64encode(data)})
+
+ def incorporate_parse_results(self, parse_results):
+ """
+ Merge the data returned by the test parser into the current test run.
+
+ Non-overlapping data is simply added. Overlapping data is either merged
+ (lists are extended, dictionaries are recursively merged) or
+ overwritten (all other types).
+ """
+ assert isinstance(parse_results, dict)
+ # Use whatever the parser gave us to improve the results
+ logging.debug("Using parser data to enrich test run details")
+ merge_dict(self.test_run, parse_results)
+
+ def attach_standard_files_to_bundle(self):
+ """
+ Attach standard output and standard error log files to the bundle.
+
+ Both files are only attached if exist and non-empty. The attachments
+ are actually associated with a test run, not a bundle, but the
+ description is good enough for simplicity.
+ """
+ self.attach_file(self.stdout_pathname, "testoutput.log", "text/plain")
+ self.attach_file(self.stderr_pathname, "testoutput.err", "text/plain")
+
+ @property
+ def bundle_pathname(self):
+ """
+ Pathname of the result bundle.
+
+ The bundle contains the snapshot of environment information as well as
+ test identity and is created when you invoke ITest.run().
+
+ The bundle file name is always "testdata.json"
+
+ .. versionadded:: 0.2
+ """
+ return self.get_artefact_pathname("testdata.json")
+
+ @property
+ def stdout_pathname(self):
+ """
+ Pathname of the log file of the standard output as returned by the test
+ program.
+
+ The log file name is always "testoutput.log"
+
+ .. versionadded:: 0.2
+ """
+ return self.get_artefact_pathname("testoutput.log")
+
+ @property
+ def stderr_pathname(self):
+ """
+ Pathname of the log file of the standard output as returned by the test
+ program.
+
+ The log file name is always "testoutput.err"
+
+ .. versionadded:: 0.2
+ """
+ return self.get_artefact_pathname("testoutput.err")
+
+ def get_artefact_pathname(self, artefact_name):
+ """
+ Return a pathname of a test run artefact file.
+
+ This is more useful than hard-coding the path as it allows the test
+ runner not to worry about the location of the results directory.
+
+ .. versionadded:: 0.2
+ """
+ return os.path.join(self.results_dir, artefact_name)
=== added file 'lava_test/core/config.py'
@@ -0,0 +1,97 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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 json
+
+
+class LavaTestConfig(object):
+
+ def __init__(self):
+ home = os.environ.get('HOME', '/')
+ baseconfig = os.environ.get('XDG_CONFIG_HOME',
+ os.path.join(home, '.config'))
+ basedata = os.environ.get('XDG_DATA_HOME',
+ os.path.join(home, '.local', 'share'))
+ self.configdir = os.path.join(baseconfig, 'lava_test')
+ self.installdir = os.path.join(basedata, 'lava_test', 'installed-tests')
+ self.resultsdir = os.path.join(basedata, 'lava_test', 'results')
+ self.registry = self._load_registry()
+
+ @property
+ def _registry_pathname(self):
+ return os.path.join(self.configdir, "registry.json")
+
+ def _load_registry(self):
+ try:
+ with open(self._registry_pathname) as stream:
+ return json.load(stream)
+ except (IOError, ValueError):
+ return self._get_default_registry()
+
+ def _save_registry(self):
+ if not os.path.exists(self.configdir):
+ os.makedirs(self.configdir)
+ with open(self._registry_pathname, "wt") as stream:
+ json.dump(self.registry, stream, indent=2)
+
+ def _get_default_registry(self):
+ return {
+ "format": "Lava-Test Test Registry 1.0",
+ "providers": [{
+ "entry_point": "lava_test.core.providers:BuiltInProvider"
+ }, {
+ "entry_point": "lava_test.core.providers:PkgResourcesProvider",
+ "config": {"namespace": "lava_test.test_definitions" }
+ },
+ {
+ "entry_point": "lava_test.core.providers:RegistryProvider",
+ "config": {"entries": [] }
+ }]
+ }
+
+ def get_provider_config(self, entry_point_name):
+ if "providers" not in self.registry:
+ self.registry["providers"] = []
+ for provider_info in self.registry["providers"]:
+ if provider_info.get("entry_point") == entry_point_name:
+ break
+ else:
+ provider_info = {"entry_point": entry_point_name}
+ self.registry["providers"].append(provider_info)
+ if "config" not in provider_info:
+ provider_info["config"] = {}
+ return provider_info["config"]
+
+ def get_logging_config_file(self):
+ logging_file = os.path.join(self.configdir, "logging.conf")
+ if os.path.exists(logging_file):
+ return logging_file
+ else:
+ return None
+
+_config = None
+
+
+def get_config():
+ global _config
+ if _config is not None:
+ return _config
+ return LavaTestConfig()
+
+
+def set_config(config):
+ global _config
+ _config = config
=== added file 'lava_test/core/hwprofile.py'
@@ -0,0 +1,223 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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 re
+import sys
+from subprocess import Popen, PIPE
+from lava_test.utils import read_file, get_machine_type
+
+
+INTEL_KEYMAP = {
+ 'vendor_id': 'cpu_vendor_name',
+ 'cpu family': 'cpu_family',
+ 'model': 'cpu_model',
+ 'model name': 'cpu_model_name',
+ 'stepping': 'cpu_stepping',
+ 'cpu MHz': 'cpu_mhz',
+ 'flags': 'cpu_features',
+}
+
+
+INTEL_VALMAP = {
+ 'cpu family': int,
+ 'model': int,
+ 'stepping': int,
+ 'cpu MHz': float,
+}
+
+
+ARM_KEYMAP = {
+ 'Processor': 'cpu_model_name',
+ 'Features': 'cpu_features',
+ 'CPU implementer': 'cpu_implementer',
+ 'CPU architecture': 'cpu_architecture',
+ 'CPU variant': 'cpu_variant',
+ 'CPU part': 'cpu_part',
+ 'CPU revision': 'cpu_revision',
+}
+
+
+ARM_VALMAP = {
+ 'CPU implementer': lambda value: int(value, 16),
+ 'CPU architecture': int,
+ 'CPU variant': lambda value: int(value, 16),
+ 'CPU part': lambda value: int(value, 16),
+ 'CPU revision': int,
+}
+
+
+def _translate_cpuinfo(keymap, valmap, key, value):
+ """
+ Translate a key and value using keymap and valmap passed in
+ """
+ newkey = keymap.get(key, key)
+ newval = valmap.get(key, lambda x: x)(value)
+ return newkey, newval
+
+
+def get_cpu_devs():
+ """
+ Return a list of CPU devices
+ """
+ pattern = re.compile('^(?P<key>.+?)\s*:\s*(?P<value>.*)$')
+ cpunum = 0
+ devices = []
+ cpudevs = []
+ cpudevs.append({})
+ machine = get_machine_type()
+ if machine in ('i686', 'x86_64'):
+ keymap, valmap = INTEL_KEYMAP, INTEL_VALMAP
+ elif machine.startswith('arm'):
+ keymap, valmap = ARM_KEYMAP, ARM_VALMAP
+
+ try:
+ cpuinfo = read_file("/proc/cpuinfo")
+ for line in cpuinfo.splitlines():
+ match = pattern.match(line)
+ if match:
+ key, value = match.groups()
+ key, value = _translate_cpuinfo(keymap, valmap,
+ key, value)
+ if cpudevs[cpunum].get(key):
+ cpunum += 1
+ cpudevs.append({})
+ cpudevs[cpunum][key] = value
+ for c in range(len(cpudevs)):
+ device = {}
+ device['device_type'] = 'device.cpu'
+ device['description'] = 'Processor #{0}'.format(c)
+ device['attributes'] = cpudevs[c]
+ devices.append(device)
+ except IOError:
+ print >> sys.stderr, "WARNING: Could not read cpu information"
+ return devices
+
+
+def get_board_devs():
+ """
+ Return a list of board devices
+ """
+ devices = []
+ attributes = {}
+ device = {}
+ machine = get_machine_type()
+ if machine in ('i686', 'x86_64'):
+ try:
+ description = read_file('/sys/class/dmi/id/board_name') or None
+ vendor = read_file('/sys/class/dmi/id/board_vendor') or None
+ version = read_file('/sys/class/dmi/id/board_version') or None
+ if description:
+ device['description'] = description.strip()
+ if vendor:
+ attributes['vendor'] = vendor.strip()
+ if version:
+ attributes['version'] = version.strip()
+ except IOError:
+ print >> sys.stderr, "WARNING: Could not read board information"
+ return devices
+ elif machine.startswith('arm'):
+ try:
+ cpuinfo = read_file("/proc/cpuinfo")
+ if cpuinfo is None:
+ return devices
+ pattern = re.compile("^Hardware\s*:\s*(?P<description>.+)$", re.M)
+ match = pattern.search(cpuinfo)
+ if match is None:
+ return devices
+ device['description'] = match.group('description')
+ except IOError:
+ print >> sys.stderr, "WARNING: Could not read board information"
+ return devices
+ else:
+ return devices
+ if attributes:
+ device['attributes'] = attributes
+ device['device_type'] = 'device.board'
+ devices.append(device)
+ return devices
+
+
+def get_mem_devs():
+ """ Return a list of memory devices
+
+ This returns up to two items, one for physical RAM and another for swap
+ """
+ pattern = re.compile('^(?P<key>.+?)\s*:\s*(?P<value>.+) kB$', re.M)
+
+ devices = []
+ try:
+ meminfo = read_file("/proc/meminfo")
+ for match in pattern.finditer(meminfo):
+ key, value = match.groups()
+ if key not in ('MemTotal', 'SwapTotal'):
+ continue
+ capacity = int(value) << 10 # Kernel reports in 2^10 units
+ if capacity == 0:
+ continue
+ if key == 'MemTotal':
+ kind = 'RAM'
+ else:
+ kind = 'swap'
+ description = "{capacity}MiB of {kind}".format(
+ capacity=capacity >> 20, kind=kind)
+ device = {}
+ device['description'] = description
+ device['attributes'] = {'capacity': str(capacity), 'kind': kind}
+ device['device_type'] = "device.mem"
+ devices.append(device)
+ except IOError:
+ print >> sys.stderr, "WARNING: Could not read memory information"
+ return devices
+
+
+def get_usb_devs():
+ """
+ Return a list of usb devices
+ """
+ pattern = re.compile(
+ "^Bus \d{3} Device \d{3}: ID (?P<vendor_id>[0-9a-f]{4}):"
+ "(?P<product_id>[0-9a-f]{4}) (?P<description>.*)$")
+ devices = []
+ try:
+ for line in Popen('lsusb', stdout=PIPE).communicate()[0].splitlines():
+ match = pattern.match(line)
+ if match:
+ vendor_id, product_id, description = match.groups()
+ attributes = {}
+ device = {}
+ attributes['vendor_id'] = int(vendor_id, 16)
+ attributes['product_id'] = int(product_id, 16)
+ device['attributes'] = attributes
+ device['description'] = description
+ device['device_type'] = 'device.usb'
+ devices.append(device)
+ except OSError:
+ print >> sys.stderr, "WARNING: Could not read usb device information, \
+unable to run lsusb, please install usbutils package"
+ return devices
+
+
+def get_hardware_context():
+ """
+ Return a dict with all of the hardware profile information gathered
+ """
+ hardware_context = {}
+ devices = []
+ devices.extend(get_cpu_devs())
+ devices.extend(get_board_devs())
+ devices.extend(get_mem_devs())
+ devices.extend(get_usb_devs())
+ hardware_context['devices'] = devices
+ return hardware_context
=== added file 'lava_test/core/installers.py'
@@ -0,0 +1,105 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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 hashlib
+import os
+
+from lava_test.api.delegates import ITestInstaller
+from lava_test.extcmd import ExternalCommandWithDelegate
+from lava_test.utils import geturl
+
+
+class TestInstaller(ITestInstaller):
+ """
+ Base class for defining an installer object.
+
+ This class can be used as-is for simple installers, or extended
+ for more advanced functionality.
+
+ :ivar steps:
+ List of steps to be executed in a shell
+
+ :ivar deps:
+ List of Debian or Ubuntu packages to apt-get install before running the
+ steps.
+
+ :ivar url:
+ Location from which the test suite should be downloaded.
+
+ :ivar md5:
+ The md5sum to check the integrety of the download
+ """
+ def __init__(self, steps=None, deps=None, url=None, md5=None, **kwargs):
+ self.steps = steps or []
+ self.deps = deps or []
+ self.url = url
+ self.md5 = md5
+
+ def __repr__(self):
+ return "<%s steps=%r deps=%r url=%r md5=%r>" % (
+ self.__class__.__name__,
+ self.steps, self.deps, self.url, self.md5)
+
+ def _run_shell_cmd(self, cmd, observer):
+ if observer: observer.about_to_run_shell_command(cmd)
+ extcmd = ExternalCommandWithDelegate(observer)
+ returncode = extcmd.check_call(cmd, shell=True)
+ if observer: observer.did_run_shell_command(cmd, returncode)
+
+ def _installdeps(self, observer):
+ if self.deps:
+ if observer: observer.about_to_install_packages(self.deps)
+ # XXX: Possible point of target-specific package installation
+ cmd = "sudo apt-get install -y " + " ".join(self.deps)
+ self._run_shell_cmd(cmd, observer)
+ if observer: observer.did_install_packages(self.deps)
+
+ def _download(self, observer):
+ """
+ Download the file specified by the url and check the md5.
+
+ Returns the path and filename if successful, otherwise return None
+ """
+ if not self.url:
+ return
+ if observer: observer.about_to_download_file(self.url)
+ filename = geturl(self.url)
+ # If the file does not exist, then the download was not
+ # successful
+ if not os.path.exists(filename):
+ raise RuntimeError(
+ "Failed to download %r" % self.url)
+ if observer: observer.did_download_file(self.url)
+ if self.md5:
+ checkmd5 = hashlib.md5()
+ with open(filename, 'rb') as fd:
+ data = fd.read(0x10000)
+ while data:
+ checkmd5.update(data)
+ data = fd.read(0x10000)
+ if checkmd5.hexdigest() != self.md5:
+ raise RuntimeError(
+ "md5sum mismatch of file %r, got %s expected %s" % (
+ filename, checkmd5.hexdigest(), self.md5))
+ return filename
+
+ def _runsteps(self, observer):
+ for cmd in self.steps:
+ self._run_shell_cmd(cmd, observer)
+
+ def install(self, observer=None):
+ self._installdeps(observer)
+ self._download(observer)
+ self._runsteps(observer)
=== added file 'lava_test/core/loader.py'
@@ -0,0 +1,83 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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/>.
+
+from __future__ import absolute_import
+from lava_test.core.config import get_config
+
+class TestLoader(object):
+ """
+ Test loader.
+
+ Encapsulates LAVA Test's knowledge of available tests.
+
+ Test can be loaded by name with
+ :meth:`lava_test.core.loader.TestLoader.__getitem__()`. Test can also be
+ listed by :meth:`lava_test.core.loader.TestLoader.get_providers()` and then
+ iterating over tests returned by each provider.
+ """
+
+ def __init__(self, config):
+ self._config = config
+
+ def get_providers(self):
+ """
+ Return a generator of available providers
+ """
+ import pkg_resources
+ for provider_info in self._config.registry.get("providers", []):
+ entry_point_name = provider_info.get("entry_point")
+ module_name, attrs = entry_point_name.split(':', 1)
+ attrs = attrs.split('.')
+ try:
+ entry_point = pkg_resources.EntryPoint(
+ entry_point_name, module_name, attrs,
+ dist=pkg_resources.get_distribution("lava-test"))
+ provider_cls = entry_point.load()
+ provider = provider_cls(provider_info.get("config", {}))
+ yield provider
+ except pkg_resources.DistributionNotFound:
+ raise RuntimeError(
+ "lava-test is not properly set up."
+ " Please read the README file")
+ except ImportError, err:
+ print "Couldn't load module : %s . Maybe configuration needs to be updated" % module_name
+ print "The configuration is stored at %s" %(get_config().configdir)
+
+
+
+ def __getitem__(self, test_id):
+ """
+ Lookup a test with the specified test_id
+ """
+ for provider in self.get_providers():
+ try:
+ return provider[test_id]
+ except KeyError:
+ pass
+ raise KeyError(test_id)
+
+ def get_test_by_name(self, test_id):
+ """
+ Lookup a test with the specified name
+
+ .. deprecated:: 0.2
+ Use __getitem__ instead
+ """
+ for provider in self.get_providers():
+ try:
+ return provider[test_id]
+ except KeyError:
+ pass
+ raise ValueError("No such test %r" % test_id)
=== added file 'lava_test/core/parsers.py'
@@ -0,0 +1,147 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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 decimal
+import os
+import re
+
+from lava_test.api.delegates import ITestParser
+
+
+class TestParser(ITestParser):
+ """
+ Base class for defining a test parser
+
+ This class can be used as-is for simple results parsers, but will likely
+ need to be extended slightly for many. If used as it is, the parse()
+ method should be called while already in the results directory and assumes
+ that a file for test output will exist called testoutput.log.
+
+ :ivar pattern:
+ regexp pattern to identify important elements of test output For
+ example: If your testoutput had lines that look like: "test01: PASS"
+ then you could use a pattern like this:
+ "^(?P<testid>\w+):\W+(?P<result>\w+)" This would result in
+ identifying "test01" as testid and "PASS" as result. Once parse()
+ has been called, self.results.test_results[] contains a list of
+ dicts of all the key,value pairs found for each test result.
+
+ :ivar fixupdict:
+ Dict of strings to convert test results to standard strings For
+ example: if you want to standardize on having pass/fail results in
+ lower case, but your test outputs them in upper case, you could use a
+ fixupdict of something like: {'PASS':'pass','FAIL':'fail'}
+
+ :ivar appendall:
+ Append a dict to the test_results entry for each result.
+ For example: if you would like to add units="MB/s" to each result:
+ appendall={'units':'MB/s'}
+
+ :ivar results:
+ Dictionary of data that was scrubbed from the log file for this test
+ run. Most notably it contains the test_results array.
+ """
+ def __init__(self, pattern=None, fixupdict=None, appendall={}):
+ if pattern is not None:
+ try:
+ re.compile(pattern)
+ except Exception as ex:
+ raise ValueError(
+ "Invalid regular expression %r: %s", pattern, ex)
+ self._results = {'test_results': []}
+ self.pattern = pattern
+ self.fixupdict = fixupdict
+ self.appendall = appendall
+
+ def __repr__(self):
+ return "<%s pattern=%r fixupdict=%r appendall=%r>" % (
+ self.__class__.__name__,
+ self.pattern, self.fixupdict, self.appendall)
+
+ @property
+ def results(self):
+ return self._results
+
+ def parse(self, artifacts):
+ if os.path.exists(artifacts.stdout_pathname):
+ return self.parse_pathname(
+ artifacts.stdout_pathname,
+ os.path.relpath(artifacts.stdout_pathname,
+ artifacts.results_dir))
+ if os.path.exists(artifacts.stderr_pathname):
+ return self.parse_pathname(
+ artifacts.stderr_pathname,
+ os.path.relpath(artifacts.stderr_pathname,
+ artifacts.results_dir))
+
+ def parse_pathname(self, pathname, relative_pathname=None):
+ with open(pathname, 'rt') as stream:
+ for lineno, line in enumerate(stream, 1):
+ match = re.search(self.pattern, line)
+ if not match:
+ continue
+ data = match.groupdict()
+ data["log_filename"] = relative_pathname or pathname
+ data["log_lineno"] = lineno
+ self._results['test_results'].append(
+ self.analyze_test_result(data))
+ return self.results
+
+ @property
+ def badchars(self):
+ return "[^a-zA-Z0-9\._-]"
+
+ def analyze_test_result(self, data):
+ """
+ Analyze sigle match (typically single line) and convert it into a
+ proper test result object.
+
+ Currently this method does the following transformations:
+ * measurement is converted to decimal if present
+ * test_case_id is rewritten to strip badchars
+ * test_case_id is rewritten to convert spaces to underscores
+ * result is transformed using fixuptdict, if defined
+ * appendall information is added, if defined
+ """
+ if 'measurement' in data:
+ try:
+ data['measurement'] = decimal.Decimal(data['measurement'])
+ except decimal.InvalidOperation:
+ del data['measurement']
+ if 'test_case_id' in data:
+ data['test_case_id'] = re.sub(self.badchars, "",
+ data['test_case_id'])
+ data['test_case_id'] = data['test_case_id'].replace(" ", "_")
+ if 'result' in data and self.fixupdict:
+ data['result'] = self.fixupdict[data['result']]
+ if self.appendall:
+ data.update(self.appendall)
+ return data
+
+
+class NativeTestParser(ITestParser):
+ """
+ Unfinished native test parser.
+
+ This was meant to be a pass-through for tests that directly create bundles
+ """
+ def __init__(self, test_def):
+ self.test_def = test_def
+
+ def parse(self, artifacts):
+ raise NotImplementedError()
+
+ def results(self):
+ raise NotImplementedError()
=== added file 'lava_test/core/providers.py'
@@ -0,0 +1,165 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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/>.
+
+from lava_test.api.core import ITestProvider
+from lava_test.core.config import get_config
+from lava_test.core.tests import DeclarativeTest
+from lava_test.utils import Cache
+
+
+class BuiltInProvider(ITestProvider):
+ """
+ Test provider that provides tests shipped in the Lava-Test source tree
+ """
+
+ _builtin_tests = [
+ 'glmemperf',
+ 'gmpbench',
+ 'gtkperf',
+ 'ltp',
+ 'posixtestsuite',
+ 'pwrmgmt',
+ 'stream',
+ 'tiobench',
+ 'x11perf',
+ ]
+
+ def __init__(self, config):
+ pass
+
+ @property
+ def description(self):
+ return "Tests built directly into LAVA Test:"
+
+ def __iter__(self):
+ return iter(self._builtin_tests)
+
+ def __getitem__(self, test_id):
+ if test_id not in self._builtin_tests:
+ raise KeyError(test_id)
+ module = __import__("lava_test.test_definitions.%s" % test_id,
+ fromlist=[''])
+ return module.testobj
+
+
+class PkgResourcesProvider(ITestProvider):
+ """
+ Test provider that provides tests declared in pkg_resources working_set
+
+ By default it looks at the 'lava_test.test_definitions' name space but it can
+ be changed with custom 'namespace' configuration entry.
+ """
+
+ def __init__(self, config):
+ self._config = config
+
+ @property
+ def namespace(self):
+ return self._config.get("namespace", "lava_test.test_definitions")
+
+ @property
+ def description(self):
+ return ("Tests provided by installed python packages"
+ " (from namespace {0}):").format(self.namespace)
+
+ def __iter__(self):
+ from pkg_resources import working_set
+ for entry_point in working_set.iter_entry_points(self.namespace):
+ yield entry_point.name
+
+ def __getitem__(self, test_name):
+ from pkg_resources import working_set
+ for entry_point in working_set.iter_entry_points(self.namespace,
+ test_name):
+ return entry_point.load().testobj
+ raise KeyError(test_name)
+
+
+class RegistryProvider(ITestProvider):
+ """
+ Test provider that provides declarative tests listed in the test registry.
+ """
+ def __init__(self, config):
+ self._config = config
+ self._cache = None
+
+ @property
+ def entries(self):
+ """
+ List of URLs to DeclarativeTest description files
+ """
+ return self._config.get("entries", [])
+
+ @property
+ def description(self):
+ return "Tests provided by LAVA Test registry:"
+
+ @classmethod
+ def register_remote_test(self, test_url):
+ config = get_config() # This is a different config object from
+ # self._config
+ provider_config = config.get_provider_config(
+ "lava_test.core.providers:RegistryProvider")
+ if "entries" not in provider_config:
+ provider_config["entries"] = []
+ if test_url not in provider_config["entries"]:
+ provider_config["entries"].append(test_url)
+ config._save_registry()
+ else:
+ raise ValueError("This test is already registered")
+
+ @classmethod
+ def unregister_remote_test(self, test_url):
+ config = get_config() # This is a different config object from
+ # self._config
+ provider_config = config.get_provider_config(
+ "lava_test.core.providers:RegistryProvider")
+ if "entries" not in provider_config:
+ provider_config["entries"] = []
+ if test_url in provider_config["entries"]:
+ provider_config["entries"].remove(test_url)
+ config._save_registry()
+ else:
+ raise ValueError("This test is not registered")
+
+ def _load_remote_test(self, test_url):
+ """
+ Load DeclarativeTest from a (possibly cached copy of) test_url
+ """
+ cache = Cache.get_instance()
+ with cache.open_cached_url(test_url) as stream:
+ return DeclarativeTest.load_from_stream(stream)
+
+ def _fill_cache(self):
+ """
+ Fill the cache of all remote tests
+ """
+ if self._cache is not None:
+ return
+ self._cache = {}
+ for test_url in self.entries:
+ test = self._load_remote_test(test_url)
+ if test.test_id in self._cache:
+ raise ValueError("Duplicate test %s declared" % test.test_id)
+ self._cache[test.test_id] = test
+
+ def __iter__(self):
+ self._fill_cache()
+ for test_id in self._cache.iterkeys():
+ yield test_id
+
+ def __getitem__(self, test_id):
+ self._fill_cache()
+ return self._cache[test_id]
=== added file 'lava_test/core/runners.py'
@@ -0,0 +1,66 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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 datetime
+
+from lava_test.api.delegates import ITestRunner
+from lava_test.extcmd import (DisplayDelegate, ExternalCommandWithDelegate)
+
+
+class TestRunner(ITestRunner):
+ """
+ Base class for defining an test runner object.
+
+ This class can be used as-is for simple execution with the expectation that
+ the run() method will be called from the directory where the test was
+ installed. Steps, if used, should handle changing directories from there to
+ the directory where the test was extracted if necessary. This class can
+ also be extended for more advanced functionality.
+
+ :ivar steps:
+ list of shell commands to execute
+ """
+ def __init__(self, steps=None):
+ self.steps = steps or []
+ self.testoutput = [] # XXX: is this still used?
+
+ def __repr__(self):
+ return "<%s steps=%r>" % (self.__class__.__name__, self.steps)
+
+ def _run_lava_test_steps(self, artifacts, observer):
+ stdout = open(artifacts.stdout_pathname, 'at')
+ stderr = open(artifacts.stderr_pathname, 'at')
+ delegate = DisplayDelegate(stdout, stderr, observer)
+ extcmd = ExternalCommandWithDelegate(delegate)
+ try:
+ for cmd in self.steps:
+ if observer: observer.about_to_run_shell_command(cmd)
+ returncode = extcmd.call(cmd, shell=True)
+ if observer: observer.did_run_shell_command(cmd, returncode)
+ finally:
+ stdout.close()
+ stderr.close()
+
+ def run(self, artifacts, observer=None):
+ """
+ Run the test program by executing steps in sequence.
+
+ .. seealso::
+
+ :meth:`~lava_test.api.delegates.TestRunner.run`
+ """
+ self.starttime = datetime.datetime.utcnow()
+ self._run_lava_test_steps(artifacts, observer)
+ self.endtime = datetime.datetime.utcnow()
=== added file 'lava_test/core/swprofile.py'
@@ -0,0 +1,72 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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 apt
+
+from lava_test.utils import read_file
+
+
+def get_packages(apt_cache=None):
+ """ Get information about the packages installed
+
+ apt_cache - if not provided, this will be read from the system
+ """
+ if apt_cache == None:
+ apt_cache = apt.Cache()
+ packages = []
+ for apt_pkg in apt_cache:
+ if hasattr(apt_pkg, 'is_installed'):
+ is_installed = apt_pkg.is_installed
+ else:
+ is_installed = apt_pkg.isInstalled # old style API
+ if is_installed:
+ pkg = {
+ "name": apt_pkg.name,
+ "version": apt_pkg.installed.version}
+ packages.append(pkg)
+ return packages
+
+
+def get_software_context(apt_cache=None, lsb_information=None):
+ """ Return dict used for storing software_context information
+
+ test_id - Unique identifier for this test
+ time_check - whether or not a check was performed to see if
+ the time on the system was synced with a time server
+ apt_cache - if not provided, this will be read from the system
+ lsb_information - if not provided, this will be read from the system
+ """
+ software_context = {}
+ software_context['image'] = get_image(lsb_information)
+ software_context['packages'] = get_packages(apt_cache)
+ return software_context
+
+
+def get_image(lsb_information=None):
+ """ Get information about the image we are running
+
+ If /etc/buildstamp exists, get the image id from that. Otherwise
+ just use the lsb-release description for a rough idea.
+ """
+ try:
+ buildstamp = read_file("/etc/buildstamp")
+ name = buildstamp.splitlines()[1]
+ except IOError:
+ import lsb_release
+
+ if lsb_information == None:
+ lsb_information = lsb_release.get_distro_information()
+ name = lsb_information['DESCRIPTION']
+ return {"name": name}
=== added file 'lava_test/core/tests.py'
@@ -0,0 +1,166 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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/>.
+
+from __future__ import absolute_import
+
+import json
+import logging
+import os
+import shutil
+
+from lava_test.api.core import ITest
+from lava_test.core.artifacts import TestArtifacts
+from lava_test.core.config import get_config
+from lava_test.core.installers import TestInstaller
+from lava_test.core.parsers import TestParser, NativeTestParser
+from lava_test.core.runners import TestRunner
+from lava_test.utils import changed_directory
+
+
+class Test(ITest):
+ """
+ Reusable class for defining tests.
+
+ This class uses composition instead of inheritance. You should be able to
+ customize the parts you care about by providing delegate objects. This
+ class can be used by test definition files to create an object that
+ contains the building blocks for installing tests, running them, and
+ parsing the results.
+
+ :ivar test_id:
+ Name of the test or test suite
+ :ivar test_version:
+ Version of the test or test suite
+ :ivar installer:
+ ITestInstaller instance to use
+ :ivar runner:
+ ITestRunner instance to use
+ :ivar parser:
+ ITestParser instance to use
+ """
+
+ def __init__(self, test_id, test_version=None,
+ installer=None, runner=None, parser=None):
+ self._test_id = test_id
+ self._test_version = test_version
+ # Delegate objects
+ self.installer = installer
+ self.runner = runner
+ self.parser = parser
+ # Config instance
+ self._config = get_config()
+
+ def __repr__(self):
+ return ("<%s test_id=%r test_version=%r installer=%r runner=%r"
+ " parser=%r>") % (
+ self.__class__.__name__, self.test_id, self.test_version,
+ self.installer, self.runner, self.parser)
+
+ @property
+ def test_id(self):
+ """
+ Return the ID of the test.
+ """
+ return self._test_id
+
+ @property
+ def test_version(self):
+ """
+ Return the version of the test
+ """
+ return self._test_version
+
+ @property
+ def install_dir(self):
+ """
+ Pathname of a directory with binary and data files installed by the
+ test.
+
+ .. versionadded:: 0.2
+ """
+ return os.path.join(self._config.installdir, self.test_id)
+
+ @property
+ def is_installed(self):
+ return os.path.exists(self.install_dir)
+
+ def install(self, observer=None):
+ if self.is_installed:
+ raise RuntimeError(
+ "%s is already installed" % self.test_id)
+ if not self.installer:
+ raise RuntimeError(
+ "no installer defined for '%s'" % self.test_id)
+ with changed_directory(self.install_dir):
+ try:
+ logging.debug(
+ "Invoking %r.install(...)", self.installer)
+ self.installer.install(observer)
+ except:
+ self.uninstall()
+ raise
+
+ def uninstall(self):
+ logging.debug("Removing test %r", self.test_id)
+ if os.path.exists(self.install_dir):
+ shutil.rmtree(self.install_dir)
+
+ def run(self, observer=None):
+ if not self.runner:
+ raise RuntimeError(
+ "no test runner defined for '%s'" % self.test_id)
+ artifacts = TestArtifacts.allocate(self.test_id, self._config)
+ with changed_directory(self.install_dir):
+ logging.debug(
+ "Invoking %r.run_and_store_artifacts(...)",
+ self.runner, observer)
+ self.runner.run(artifacts, observer)
+ return artifacts
+
+ def parse(self, artifacts):
+ if self.parser:
+ logging.debug("Invoking %r.parse()", self.parser)
+ with changed_directory(artifacts.results_dir, False):
+ self.parser.parse(artifacts)
+ return self.parser.results
+
+
+class DeclarativeTest(Test):
+ """
+ Declaretive ITest implementation.
+
+ Declarative test is like :class:`lava_test.core.tests.Test` but cannot
+ contain any python code and is completely encapsulated in a .json file.
+
+ The idea is to write .json files that assemble a Test instance using
+ readily-available TestInstaller, TestRunner and TestParser subclasses.
+ """
+
+ def __init__(self, about):
+ self.about = about
+ super(DeclarativeTest, self).__init__(self.about.get('test_id'))
+ self.installer = TestInstaller(**self.about.get('install', {}))
+ self.runner = TestRunner(**self.about.get('run', {}))
+ if self.about.get('parse', {}).get('native', False) is True:
+ self.parser = NativeTestParser(self)
+ else:
+ self.parser = TestParser(**self.about.get('parse', {}))
+
+ @classmethod
+ def load_from_stream(cls, stream):
+ return cls(json.load(stream))
+
+ def save_to_stream(self, stream):
+ json.dumps(self.about, stream, indent="2")
=== added file 'lava_test/extcmd.py'
@@ -0,0 +1,108 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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/>.
+
+from Queue import Queue
+import subprocess
+import sys
+import threading
+try:
+ import posix
+except ImportError:
+ posix = None
+
+
+class ExternalCommand(object):
+
+ def _popen(self, *args, **kwargs):
+ if posix:
+ kwargs['close_fds'] = True
+ return subprocess.Popen(*args, **kwargs)
+
+ def call(self, *args, **kwargs):
+ proc = self._popen(*args, **kwargs)
+ proc.wait()
+ return proc.returncode
+
+ def check_call(self, *args, **kwargs):
+ returncode = self.call(*args, **kwargs)
+ if returncode != 0:
+ raise subprocess.CalledProcessError(
+ returncode, kwargs.get("args") or args[0])
+ return returncode
+
+
+class ExternalCommandWithDelegate(ExternalCommand):
+
+ def __init__(self, delegate):
+ self._queue = Queue()
+ self._delegate = delegate
+
+ def _read_stream(self, stream, stream_name):
+ for line in iter(stream.readline, ''):
+ cmd = (stream_name, line)
+ self._queue.put(cmd)
+
+ def _drain_queue(self):
+ while True:
+ args = self._queue.get()
+ if args is None:
+ break
+ self._delegate.display_subprocess_output(*args)
+
+ def call(self, *args, **kwargs):
+ kwargs['stdout'] = subprocess.PIPE
+ kwargs['stderr'] = subprocess.PIPE
+ proc = self._popen(*args, **kwargs)
+ stdout_reader = threading.Thread(
+ target=self._read_stream, args=(proc.stdout, "stdout"))
+ stderr_reader = threading.Thread(
+ target=self._read_stream, args=(proc.stderr, "stderr"))
+ ui_printer = threading.Thread(
+ target=self._drain_queue)
+
+ ui_printer.start()
+ stdout_reader.start()
+ stderr_reader.start()
+ try:
+ proc.wait()
+ except KeyboardInterrupt:
+ proc.kill()
+ finally:
+ stdout_reader.join()
+ stderr_reader.join()
+ self._queue.put(None)
+ ui_printer.join()
+ return proc.returncode
+
+
+class DisplayDelegate(object):
+ """
+ Delegate for displaying command output.
+
+ Perfect companion for ExternalCommandWithDelegate.
+ """
+
+ def __init__(self, stdout=None, stderr=None, chain=None):
+ self.stdout = stdout or sys.stdout
+ self.stderr = stderr or sys.stderr
+ self.chain = chain
+
+ def display_subprocess_output(self, stream_name, line):
+ if stream_name == 'stdout':
+ self.stdout.write(line)
+ elif stream_name == 'stderr':
+ self.stderr.write(line)
+ if self.chain:
+ self.chain.display_subprocess_output(stream_name, line)
=== modified file 'lava_test/main.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2010 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
@@ -13,22 +13,39 @@
# 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 abrek.command
-
-
-def main(argv):
- argv = argv[1:]
- if not argv:
- argv = ['help']
- cmd = argv.pop(0)
- cmd_func = abrek.command.get_command(cmd)
- if not cmd_func:
- print "command '%s' not found" % cmd
- return 1
- return cmd_func.main(argv)
+import logging
+import logging.config
+
+from lava_test.core.config import get_config
+
+from lava_tool.dispatcher import LavaDispatcher, run_with_dispatcher_class
+
+
+class LAVATestDispatcher(LavaDispatcher):
+
+ toolname = 'lava_test'
+ description = """
+ LAVA Test wrapper framework
+ """
+ epilog = """
+ Please report all bugs using the Launchpad bug tracker:
+ http://bugs.launchpad.net/lava-test/+filebug
+ """
+
+
+def main():
+
+ logging_config_file = get_config().get_logging_config_file()
+
+ if logging_config_file != None:
+ logging.config.fileConfig(logging_config_file)
+
+ run_with_dispatcher_class(LAVATestDispatcher)
+
if __name__ == '__main__':
import os
import sys
- exit_code = main(sys.argv)
- sys.exit(exit_code)
\ No newline at end of file
+ arg_only = sys.argv
+ arg_only.remove(arg_only[0])
+ LAVATestDispatcher().dispatch(arg_only)
\ No newline at end of file
=== modified file 'lava_test/test_definitions/bootchart.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2011 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
@@ -13,16 +13,19 @@
# 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 abrek.testdef
+from lava_test.core.installers import TestInstaller
+from lava_test.core.parsers import TestParser
+from lava_test.core.runners import TestRunner
+from lava_test.core.tests import Test
INSTALLSTEPS = ['bzr branch lp:~linaro-foundations/lava-test/bootchartscript']
DEPS = ['bootchart', 'pybootchartgui', 'bzr']
RUNSTEPS = ['./bootchartscript/bootchartscript.sh']
PATTERN = "^(?P<test_case_id>\w+):\W+(?P<measurement>\d+\.\d+)"
-bootchartinst = abrek.testdef.AbrekTestInstaller(INSTALLSTEPS, deps=DEPS)
-bootchartrun = abrek.testdef.AbrekTestRunner(RUNSTEPS)
-bootchartparser = abrek.testdef.AbrekTestParser(PATTERN,
+bootchartinst = TestInstaller(INSTALLSTEPS, deps=DEPS)
+bootchartrun = TestRunner(RUNSTEPS)
+bootchartparser = TestParser(PATTERN,
appendall={'units':'sec', 'result':'pass'})
-testobj = abrek.testdef.AbrekTest(testname="bootchart", installer=bootchartinst,
+testobj = Test(testname="bootchart", installer=bootchartinst,
runner=bootchartrun, parser=bootchartparser)
=== modified file 'lava_test/test_definitions/firefox.py'
@@ -13,16 +13,21 @@
# 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 abrek.testdef
+
+from lava_test.core.installers import TestInstaller
+from lava_test.core.parsers import TestParser
+from lava_test.core.runners import TestRunner
+from lava_test.core.tests import Test
+
INSTALLSTEPS = ['git clone git://github.com/janimo/firefox-startup-timing.git']
DEPS = ['firefox', 'git-core', 'gcalctool']
RUNSTEPS = ['cd firefox-startup-timing; ./firefox_startup_timing.sh']
PATTERN = "^(?P<test_case_id>\w+):(?P<measurement>\d+)"
-firefoxinst = abrek.testdef.AbrekTestInstaller(INSTALLSTEPS, deps=DEPS)
-firefoxrun = abrek.testdef.AbrekTestRunner(RUNSTEPS)
-firefoxparser = abrek.testdef.AbrekTestParser(PATTERN,
+firefoxinst = TestInstaller(INSTALLSTEPS, deps=DEPS)
+firefoxrun = TestRunner(RUNSTEPS)
+firefoxparser = TestParser(PATTERN,
appendall={'units':'ms', 'result':'pass'})
-testobj = abrek.testdef.AbrekTest(testname="firefox", installer=firefoxinst,
+testobj = Test(test_id="firefox", installer=firefoxinst,
runner=firefoxrun, parser=firefoxparser)
=== modified file 'lava_test/test_definitions/glmemperf.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2010 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
@@ -13,16 +13,21 @@
# 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 abrek.testdef
+
+from lava_test.core.installers import TestInstaller
+from lava_test.core.parsers import TestParser
+from lava_test.core.runners import TestRunner
+from lava_test.core.tests import Test
+
RUNSTEPS = ["glmemperf -e shmimage"]
PATTERN = "^(?P<test_case_id>\w+):\W+(?P<measurement>\d+) fps"
-inst = abrek.testdef.AbrekTestInstaller(deps=["glmemperf"])
-run = abrek.testdef.AbrekTestRunner(RUNSTEPS)
-parse = abrek.testdef.AbrekTestParser(PATTERN,
+inst = TestInstaller(deps=["glmemperf"])
+run = TestRunner(RUNSTEPS)
+parse = TestParser(PATTERN,
appendall={'units':'fps',
'result':'pass'})
-testobj = abrek.testdef.AbrekTest(testname="glmemperf", installer=inst,
+testobj = Test(test_id="glmemperf", installer=inst,
runner=run, parser=parse)
=== modified file 'lava_test/test_definitions/gmpbench.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2010 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
@@ -21,7 +21,12 @@
"""
-import abrek.testdef
+
+from lava_test.core.installers import TestInstaller
+from lava_test.core.parsers import TestParser
+from lava_test.core.runners import TestRunner
+from lava_test.core.tests import Test
+
VERSION='0.2'
URL="ftp://ftp.gmplib.org/pub/misc/gmpbench-%s.tar.bz2" %(VERSION)
@@ -36,10 +41,10 @@
PATTERN = "\s*(?P<test_case_id>GMPbench\.*\w*\.*\w*):?\s*"\
"(?P<measurement>\d+.\d+)"
-gmpbenchinst = abrek.testdef.AbrekTestInstaller(INSTALLSTEPS, deps=DEPS,
+gmpbenchinst = TestInstaller(INSTALLSTEPS, deps=DEPS,
url=URL)
-gmpbenchrun = abrek.testdef.AbrekTestRunner(RUNSTEPS)
-gmpbenchparser = abrek.testdef.AbrekTestParser(PATTERN,
+gmpbenchrun = TestRunner(RUNSTEPS)
+gmpbenchparser = TestParser(PATTERN,
appendall={'units':'operations/s', 'result':'pass'})
-testobj = abrek.testdef.AbrekTest(testname="gmpbench", installer=gmpbenchinst,
+testobj = Test(test_id="gmpbench", installer=gmpbenchinst,
runner=gmpbenchrun, parser=gmpbenchparser)
=== modified file 'lava_test/test_definitions/gtkperf.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2010 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
@@ -14,14 +14,19 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import re
-import abrek.testdef
+
+from lava_test.core.installers import TestInstaller
+from lava_test.core.parsers import TestParser
+from lava_test.core.runners import TestRunner
+from lava_test.core.tests import Test
+
# Run tests automatically, 500 repetitions each
gtkperf_options = "-a -c 500"
RUNSTEPS = ["LANG=C gtkperf %s" % gtkperf_options]
-class GtkTestParser(abrek.testdef.AbrekTestParser):
+class GtkTestParser(TestParser):
def parse(self):
PAT1 = "^(?P<test_case_id>\w+) - (?P<subtest>\w*\W*\w*) - time:\W+(?P<measurement>\d+\.\d+)"
PAT2 = "^(?P<test_case_id>\w+) - time:\W+(?P<measurement>\d+\.\d+)"
@@ -46,8 +51,8 @@
self.fixmeasurements()
parse = GtkTestParser()
-inst = abrek.testdef.AbrekTestInstaller(deps=["gtkperf"])
-run = abrek.testdef.AbrekTestRunner(RUNSTEPS)
+inst = TestInstaller(deps=["gtkperf"])
+run = TestRunner(RUNSTEPS)
-testobj = abrek.testdef.AbrekTest(testname="gtkperf", installer=inst,
+testobj = Test(test_id="gtkperf", installer=inst,
runner=run, parser=parse)
=== modified file 'lava_test/test_definitions/ltp.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2010 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
@@ -15,7 +15,12 @@
import re
-import abrek.testdef
+
+from lava_test.core.installers import TestInstaller
+from lava_test.core.parsers import TestParser
+from lava_test.core.runners import TestRunner
+from lava_test.core.tests import Test
+
VERSION="20100831"
URL='http://downloads.sourceforge.net/project/ltp/LTP Source/ltp-%s/ltp-full-%s.bz2' % (VERSION, VERSION)
@@ -44,7 +49,7 @@
"TWARN":"unknown"}
-class LTPParser(abrek.testdef.AbrekTestParser):
+class LTPParser(TestParser):
def parse(self):
filename = "testoutput.log"
pat = re.compile(self.pattern)
@@ -67,10 +72,10 @@
self.fixids()
-ltpinst = abrek.testdef.AbrekTestInstaller(INSTALLSTEPS, deps=DEPS, url=URL,
+ltpinst = TestInstaller(INSTALLSTEPS, deps=DEPS, url=URL,
md5=MD5)
-ltprun = abrek.testdef.AbrekTestRunner(RUNSTEPS)
+ltprun = TestRunner(RUNSTEPS)
ltpparser = LTPParser(PATTERN, fixupdict = FIXUPS)
-testobj = abrek.testdef.AbrekTest(testname="ltp", version=VERSION,
+testobj = Test(test_id="ltp", test_version=VERSION,
installer=ltpinst, runner=ltprun,
parser=ltpparser)
=== modified file 'lava_test/test_definitions/peacekeeper.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2010 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
@@ -20,7 +20,12 @@
http://clients.futuremark.com/peacekeeper/index.action
"""
-import abrek.testdef
+
+from lava_test.core.installers import TestInstaller
+from lava_test.core.parsers import TestParser
+from lava_test.core.runners import TestRunner
+from lava_test.core.tests import Test
+
import os
curdir = os.path.realpath(os.path.dirname(__file__))
@@ -29,13 +34,13 @@
RUNSTEPS = ['python peacekeeper_runner.py firefox']
DEPS = ['python-ldtp','firefox']
-my_installer = abrek.testdef.AbrekTestInstaller(INSTALLSTEPS, deps=DEPS)
-my_runner = abrek.testdef.AbrekTestRunner(RUNSTEPS)
+my_installer = TestInstaller(INSTALLSTEPS, deps=DEPS)
+my_runner = TestRunner(RUNSTEPS)
PATTERN = "^(?P<result>\w+): Score = (?P<measurement>\d+)"
-my_parser = abrek.testdef.AbrekTestParser(PATTERN,
+my_parser = TestParser(PATTERN,
appendall={'units':'point'})
-testobj = abrek.testdef.AbrekTest(testname="peacekeeper", installer=my_installer,
+testobj = Test(test_id="peacekeeper", installer=my_installer,
runner=my_runner, parser=my_parser)
=== modified file 'lava_test/test_definitions/posixtestsuite.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2010 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
@@ -23,7 +23,12 @@
"""
import re
-import abrek.testdef
+
+from lava_test.core.installers import TestInstaller
+from lava_test.core.parsers import TestParser
+from lava_test.core.runners import TestRunner
+from lava_test.core.tests import Test
+
VERSION="20100831"
URL= "http://downloads.sourceforge.net/project/ltp/LTP Source/ltp-%s/"\
@@ -45,7 +50,7 @@
}
-class PosixParser(abrek.testdef.AbrekTestParser):
+class PosixParser(TestParser):
def parse(self):
filename = "testoutput.log"
pat = re.compile(self.pattern)
@@ -60,10 +65,10 @@
if self.fixupdict:
self.fixresults(self.fixupdict)
-posix_inst = abrek.testdef.AbrekTestInstaller(INSTALLSTEPS, deps=DEPS,
+posix_inst = TestInstaller(INSTALLSTEPS, deps=DEPS,
url=URL, md5=MD5)
-posix_run = abrek.testdef.AbrekTestRunner(RUNSTEPS)
+posix_run = TestRunner(RUNSTEPS)
posixparser = PosixParser(PATTERN, fixupdict = FIXUPS)
-testobj = abrek.testdef.AbrekTest(testname="posixtestsuite", version=VERSION,
+testobj = Test(test_id="posixtestsuite", test_version=VERSION,
installer=posix_inst, runner=posix_run,
parser=posixparser)
=== modified file 'lava_test/test_definitions/pwrmgmt.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2010 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
@@ -13,33 +13,29 @@
# 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 abrek.testdef
-
-
-INSTALLSTEPS = ['git clone git://git.linaro.org/tools/pm-qa.git',
+
+from lava_test.core.installers import TestInstaller
+from lava_test.core.parsers import TestParser
+from lava_test.core.runners import TestRunner
+from lava_test.core.tests import Test
+
+
+
+INSTALLSTEPS = ['git clone git://git.linaro.org/people/torez/pm-qa.git',
'cd pm-qa && make clean && make all']
-RUNSTEPS = ['cd pm-qa && make check']
-DEPS = ['git-core', 'make', 'linux-libc-dev']
-
-pwrmgmtinst = abrek.testdef.AbrekTestInstaller(INSTALLSTEPS, deps=DEPS)
-pwrmgmtrun = abrek.testdef.AbrekTestRunner(RUNSTEPS)
-
-# test case name is before ":" , the test log is between ":" and "...", the result is after "..."
-# Each test case is separated with a test description beginning with "#"
-
-# Example:
-####
-#### cpufreq_02:
-#### test the cpufreq framework is available for governor
-####
-#cpufreq_02.0/cpu0: checking scaling_available_governors exists... pass
-#cpufreq_02.1/cpu0: checking scaling_governor exists... pass
-#cpufreq_02.0/cpu1: checking scaling_available_governors exists... pass
-#cpufreq_02.1/cpu1: checking scaling_governor exists... pass
-
-PATTERN = "^(?P<test_case_id>[\w/\.]+):\s+(?P<message>.+)\.\.\.\s+(?P<result>\w+)"
-
-pwrmgmtparser = abrek.testdef.AbrekTestParser(PATTERN)
-
-testobj = abrek.testdef.AbrekTest(testname="pwrmgmt", installer=pwrmgmtinst,
+RUNSTEPS = ['cd pm-qa && awk -f testcases.awk run_template']
+DEPS = ['git-core', 'make', 'alsa-utils', 'pulseaudio-utils', 'lame', 'festival', 'wget']
+
+pwrmgmtinst = TestInstaller(INSTALLSTEPS, deps=DEPS)
+pwrmgmtrun = TestRunner(RUNSTEPS)
+
+# test case name is between "pm-qa-" and ":" and results and/or
+# measurements are rest of the line
+PATTERN = "^pm-qa-(?P<test_case_id>\w+):\s+(?P<message>.*)"
+
+
+pwrmgmtparser = TestParser(PATTERN,
+ appendall={'result':'pass'})
+
+testobj = Test(test_id="pwrmgmt", installer=pwrmgmtinst,
runner=pwrmgmtrun, parser=pwrmgmtparser)
=== modified file 'lava_test/test_definitions/pybench.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2010 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
@@ -20,7 +20,12 @@
standardized way to measure the performance of Python implementations.
"""
-import abrek.testdef
+
+from lava_test.core.installers import TestInstaller
+from lava_test.core.parsers import TestParser
+from lava_test.core.runners import TestRunner
+from lava_test.core.tests import Test
+
VERSION='r27'
URL="http://svn.python.org/projects/python/tags/%s/Tools/pybench/" %(VERSION)
@@ -29,8 +34,8 @@
RUNSTEPS = ['python pybench/pybench.py']
DEPS = ['subversion']
-my_installer = abrek.testdef.AbrekTestInstaller(INSTALLSTEPS, deps=DEPS)
-my_runner = abrek.testdef.AbrekTestRunner(RUNSTEPS)
+my_installer = TestInstaller(INSTALLSTEPS, deps=DEPS)
+my_runner = TestRunner(RUNSTEPS)
# test case name is first column and measurement is average column
#
@@ -40,9 +45,9 @@
PATTERN = "^\s+(?P<test_case_id>\w+):\s+(\d+)ms\s+(?P<measurement>\d+)ms"
-my_parser = abrek.testdef.AbrekTestParser(PATTERN,
+my_parser = TestParser(PATTERN,
appendall={'units':'ms',
'result':'pass'})
-testobj = abrek.testdef.AbrekTest(testname="pybench", installer=my_installer,
+testobj = Test(test_id="pybench", installer=my_installer,
runner=my_runner, parser=my_parser)
=== modified file 'lava_test/test_definitions/smem.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2011 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
@@ -13,15 +13,19 @@
# 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 abrek.testdef
+from lava_test.core.installers import TestInstaller
+from lava_test.core.parsers import TestParser
+from lava_test.core.runners import TestRunner
+from lava_test.core.tests import Test
+
RUNSTEPS = ['smem -w | tail -n 3']
PATTERN = "^(?P<test_case_id>(\w+\s)+)\s\s+(?P<measurement>\d+)"
DEPS = ['smem']
-smeminst = abrek.testdef.AbrekTestInstaller(deps=DEPS)
-smemrun = abrek.testdef.AbrekTestRunner(RUNSTEPS)
-smemparser = abrek.testdef.AbrekTestParser(PATTERN,
+smeminst = TestInstaller(deps=DEPS)
+smemrun = TestRunner(RUNSTEPS)
+smemparser = TestParser(PATTERN,
appendall={'units':'KB', 'result':'pass'})
-testobj = abrek.testdef.AbrekTest(testname="smem", installer=smeminst,
+testobj = Test(testname="smem", installer=smeminst,
runner=smemrun, parser=smemparser)
=== modified file 'lava_test/test_definitions/stream.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2010 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
@@ -13,7 +13,10 @@
# 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 abrek.testdef
+from lava_test.core.installers import TestInstaller
+from lava_test.core.parsers import TestParser
+from lava_test.core.runners import TestRunner
+from lava_test.core.tests import Test
URL="http://www.cs.virginia.edu/stream/FTP/Code/stream.c"
INSTALLSTEPS = ['cc stream.c -O2 -fopenmp -o stream']
@@ -21,9 +24,11 @@
RUNSTEPS = ['./stream']
PATTERN = "^(?P<test_case_id>\w+):\W+(?P<measurement>\d+\.\d+)"
-streaminst = abrek.testdef.AbrekTestInstaller(INSTALLSTEPS, deps=DEPS, url=URL)
-streamrun = abrek.testdef.AbrekTestRunner(RUNSTEPS)
-streamparser = abrek.testdef.AbrekTestParser(PATTERN,
+streaminst = TestInstaller(INSTALLSTEPS, deps=DEPS, url=URL)
+streamrun = TestRunner(RUNSTEPS)
+streamparser = TestParser(PATTERN,
appendall={'units':'MB/s', 'result':'pass'})
-testobj = abrek.testdef.AbrekTest(testname="stream", installer=streaminst,
+testobj = Test(test_id="stream", installer=streaminst,
runner=streamrun, parser=streamparser)
+
+
=== modified file 'lava_test/test_definitions/tiobench.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2010 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
@@ -21,7 +21,12 @@
random read, sequential write, and random write.
"""
import re
-import abrek.testdef
+
+from lava_test.core.installers import TestInstaller
+from lava_test.core.parsers import TestParser
+from lava_test.core.runners import TestRunner
+from lava_test.core.tests import Test
+
VERSION="0.3.3"
URL="http://prdownloads.sourceforge.net/tiobench/tiobench-%s.tar.gz" %(VERSION)
@@ -33,7 +38,7 @@
"--numruns=2" % (VERSION)]
-class TIObenchTestParser(abrek.testdef.AbrekTestParser):
+class TIObenchTestParser(TestParser):
def parse(self):
# Pattern to match the test case name
pattern1="(?P<test_id>^(Sequential|Random) (Writes|Reads))"
@@ -63,9 +68,9 @@
self.appendtoall(self.appendall)
self.fixmeasurements()
-tiobench_inst = abrek.testdef.AbrekTestInstaller(INSTALLSTEPS, url=URL,
+tiobench_inst = TestInstaller(INSTALLSTEPS, url=URL,
md5=MD5)
-tiobench_run = abrek.testdef.AbrekTestRunner(RUNSTEPS)
+tiobench_run = TestRunner(RUNSTEPS)
parse = TIObenchTestParser(appendall={'units':'MB/s', 'result':'pass'})
-testobj = abrek.testdef.AbrekTest(testname="tiobench", version=VERSION,
+testobj = Test(test_id="tiobench", test_version=VERSION,
installer=tiobench_inst, runner=tiobench_run, parser=parse)
=== modified file 'lava_test/test_definitions/x11perf.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2010 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
@@ -14,7 +14,12 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
-import abrek.testdef
+
+from lava_test.core.installers import TestInstaller
+from lava_test.core.parsers import TestParser
+from lava_test.core.runners import TestRunner
+from lava_test.core.tests import Test
+
x11perf_options = "-repeat 3"
@@ -44,11 +49,11 @@
RUNSTEPS = ["x11perf %s %s" % (x11perf_options, " ".join(x11perf_tests))]
PATTERN = "trep @.*\(\W*(?P<measurement>\d+.\d+)/sec\):\W+(?P<test_case_id>.+)"
-inst = abrek.testdef.AbrekTestInstaller(deps=["x11-apps"])
-run = abrek.testdef.AbrekTestRunner(RUNSTEPS)
-parse = abrek.testdef.AbrekTestParser(PATTERN,
+inst = TestInstaller(deps=["x11-apps"])
+run = TestRunner(RUNSTEPS)
+parse = TestParser(PATTERN,
appendall={'units':'reps/s',
'result':'pass'})
-testobj = abrek.testdef.AbrekTest(testname="x11perf", installer=inst,
+testobj = Test(test_id="x11perf", installer=inst,
runner=run, parser=parse)
=== modified file 'lava_test/test_definitions/xrestop.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2010 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
@@ -13,15 +13,18 @@
# 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 abrek.testdef
+from lava_test.core.installers import TestInstaller
+from lava_test.core.parsers import TestParser
+from lava_test.core.runners import TestRunner
+from lava_test.core.tests import Test
INSTALLSTEPS = ['bzr branch lp:~linaro-foundations/lava-test/xrestopscript']
RUNSTEPS = ["./xrestopscript/xrestop.sh"]
PATTERN = "^(?P<test_case_id>\w+):\W+(?P<measurement>\d+)"
-xrestopinst = abrek.testdef.AbrekTestInstaller(INSTALLSTEPS, deps=["xrestop"])
-xrestoprun = abrek.testdef.AbrekTestRunner(RUNSTEPS)
-xrestopparser = abrek.testdef.AbrekTestParser(PATTERN,
+xrestopinst = TestInstaller(INSTALLSTEPS, deps=["xrestop"])
+xrestoprun = TestRunner(RUNSTEPS)
+xrestopparser = TestParser(PATTERN,
appendall={'units':'KB', 'result':'pass'})
-testobj = abrek.testdef.AbrekTest(testname="xrestop", installer=xrestopinst,
+testobj = Test(testname="xrestop", installer=xrestopinst,
runner=xrestoprun, parser=xrestopparser)
=== modified file 'lava_test/utils.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2010 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
@@ -13,10 +13,11 @@
# 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 contextlib
+import hashlib
+import logging
import os
import shutil
-import subprocess
-import sys
import urllib2
import urlparse
@@ -25,25 +26,6 @@
_fake_machine = None
-class Tee(file):
- """ A file-like object that optionally mimics tee functionality.
-
- By default, output will go to both stdout and the file specified.
- Optionally, quiet=True can be used to mute the output to stdout.
- """
- def __init__(self, *args, **kwargs):
- try:
- self.quiet = kwargs.pop('quiet')
- except KeyError:
- self.quiet = False
- super(Tee, self).__init__(*args, **kwargs)
-
- def write(self, data):
- super(Tee, self).write(data)
- if self.quiet is False:
- sys.stdout.write(data)
-
-
def geturl(url, path=""):
urlpath = urlparse.urlsplit(url).path
filename = os.path.basename(urlpath)
@@ -75,9 +57,8 @@
if _fake_paths is not None:
if path in _fake_paths:
path = _fake_paths[path]
- with open(path) as fd:
- data = fd.read()
- return data
+ with open(path, 'rb') as stream:
+ return stream.read()
def fake_file(path, data=None, newpath=None):
@@ -119,21 +100,6 @@
_fake_machine = None
-def run_and_log(cmd, fd, quiet=False):
- """
- Run a command and log the output to fd
- """
- proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,
- stderr=subprocess.STDOUT, shell=True)
- while proc.returncode == None:
- proc.poll()
- data = proc.stdout.readline()
- fd.write(data)
- if quiet is False:
- sys.stdout.write(data)
- return proc.returncode
-
-
def get_machine_type():
"""
Return the machine type
@@ -142,3 +108,123 @@
if _fake_machine is None:
return os.uname()[-1]
return _fake_machine
+
+
+def mkdir_p(dirname):
+ if not os.path.exists(dirname):
+ os.makedirs(dirname)
+
+
+@contextlib.contextmanager
+def changed_directory(dirname, make_if_needed=True):
+ """
+ A context manager for running a piece of code in another
+ directory. The directory is created if needed (by default, can
+ be changed with make_if_needed).
+ """
+ orig_dir = os.getcwd()
+ if make_if_needed:
+ mkdir_p(dirname)
+ logging.info("Changing directory to %r", dirname)
+ os.chdir(dirname)
+ try:
+ yield
+ finally:
+ logging.info("Changing directory to %r", orig_dir)
+ os.chdir(orig_dir)
+
+
+def merge_dict(merge_into, merge_from):
+ """
+ Merge two dictionaries recursively:
+
+ 1) Simple values are overwritten with a logging.warning() message
+ 2) Lists are appended
+ 3) Dictionaries are merged recursively
+ """
+ assert isinstance(merge_into, dict)
+ assert isinstance(merge_from, dict)
+ for key in merge_from.iterkeys():
+ if key in merge_into:
+ if (isinstance(merge_from[key], dict)
+ and isinstance(merge_into[key], dict)):
+ merge_dict(merge_into[key], merge_from[key])
+ elif (isinstance(merge_from[key], list)
+ and isinstance(merge_into[key], list)):
+ merge_into[key].extend(merge_from[key])
+ else:
+ logging.warning(
+ "Overwriting existing value of %r:"
+ "%r overwritten with %r",
+ key, merge_into[key], merge_from[key])
+ merge_into[key] = merge_from[key]
+ else:
+ merge_into[key] = merge_from[key]
+
+
+class Cache(object):
+ """
+ Simple open-cached-URL class
+ """
+
+ _instance = None
+
+ def __init__(self):
+ home = os.environ.get('HOME', '/')
+ basecache = os.environ.get('XDG_CACHE_HOME',
+ os.path.join(home, '.cache'))
+ self.cache_dir = os.path.join(basecache, 'lava_test')
+
+ @classmethod
+ def get_instance(cls):
+ if cls._instance is None:
+ cls._instance = cls()
+ return cls._instance
+
+ def open_cached(self, key, mode="r"):
+ """
+ Acts like open() but the pathname is relative to the
+ lava_test-specific cache directory.
+ """
+ if "w" in mode and not os.path.exists(self.cache_dir):
+ os.makedirs(self.cache_dir)
+ if os.path.isabs(key):
+ raise ValueError("key cannot be an absolute path")
+ try:
+ stream = open(os.path.join(self.cache_dir, key), mode)
+ yield stream
+ finally:
+ stream.close()
+
+ def _key_for_url(self, url):
+ return hashlib.sha1(url).hexdigest()
+
+ def _refresh_url_cache(self, key, url):
+ with contextlib.nested(
+ contextlib.closing(urllib2.urlopen(url)),
+ self.open_cached(key, "wb")) as (in_stream, out_stream):
+ out_stream.write(in_stream.read())
+
+ @contextlib.contextmanager
+ def open_cached_url(self, url):
+ """
+ Like urlopen.open() but the content may be cached.
+ """
+ # Do not cache local files, this is not what users would expect
+
+ # workaround - not using cache at all.
+ # TODO: fix this and use the cache
+ # if url.startswith("file://"):
+ if True:
+ stream = urllib2.urlopen(url)
+ else:
+ key = self._key_for_url(url)
+ try:
+ stream = self.open_cached(key, "rb")
+ except IOError:
+ self._refresh_url_cache(key, url)
+ stream = self.open_cached(key, "rb")
+ try:
+ yield stream
+ finally:
+ stream.close()
=== modified file 'setup.py'
@@ -16,21 +16,34 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from setuptools import setup, find_packages
-from abrek import __version__ as version
setup(
name='lava-test',
- version=version,
+ version=":versiontools:lava_test:",
author='Linaro Validation Team',
author_email='linaro-dev@lists.linaro.org',
url='https://launchpad.net/lava-test',
- description='Lava test execution framework',
+ description='LAVA test execution framework',
long_description=open("README").read(),
packages=find_packages(exclude=['tests']),
license="GNU GPLv3",
test_suite='tests.test_suite',
- scripts = ['bin/lava-test'],
+ entry_points="""
+ [console_scripts]
+ lava-test=lava_test.main:main
+ [lava_test.commands]
+ version=lava_test.commands:version
+ list-tests=lava_test.commands:list_tests
+ list-installed=lava_test.commands:list_installed
+ install=lava_test.commands:install
+ uninstall=lava_test.commands:uninstall
+ run=lava_test.commands:run
+ parse=lava_test.commands:parse
+ show=lava_test.commands:show
+ register_test=lava_test.commands:register_test
+ unregister_test=lava_test.commands:unregister_test
+ """,
classifiers=[
"Development Status :: 3 - Alpha",
"Intended Audience :: Developers",
@@ -39,6 +52,13 @@
"Programming Language :: Python :: 2.6",
"Topic :: Software Development :: Testing",
],
+ install_requires=[
+ 'lava-tool >= 0.2',
+ 'versiontools >= 1.4',
+ 'linaro_dashboard_bundle',
+ ],
+ setup_requires=[
+ 'versiontools >= 1.4'
+ ],
zip_safe=False,
include_package_data=True)
-
=== modified file 'tests/__init__.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2010 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
@@ -16,16 +16,12 @@
import unittest
def test_suite():
- module_names = ['tests.test_abrekcmd',
- 'tests.test_abrektest',
- 'tests.test_abrektestinstaller',
- 'tests.test_abrektestparser',
- 'tests.test_abrektestrunner',
- 'tests.test_builtins',
- 'tests.test_dashboard',
+ module_names = ['tests.test_lavatest_commands',
+ 'tests.test_lavatest_test',
+ 'tests.test_lavatest_testinstaller',
+ 'tests.test_lavatest_testparser',
+ 'tests.test_lavatest_testrunner',
'tests.test_hwprofile',
- 'tests.test_main',
- 'tests.test_results',
'tests.test_swprofile']
loader = unittest.TestLoader()
suite = loader.loadTestsFromNames(module_names)
=== modified file 'tests/fixtures.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2010 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
=== modified file 'tests/imposters.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2010 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
@@ -19,7 +19,7 @@
import tempfile
import StringIO
-from abrek.config import set_config
+from lava_test.core.config import set_config
class OutputImposter(object):
@@ -43,16 +43,17 @@
self.installdir = os.path.join(basedir, "install")
self.resultsdir = os.path.join(basedir, "results")
self.registry = {
- "format": "Abrek Test Registry 1.0 Experimental",
+ "format": "LAVA Test Test Registry 1.0",
"providers": [
{
- "entry_point": "abrek.providers:BuiltInProvider",
- },
- {
- "entry_point": "abrek.providers:PkgResourcesProvider",
- },
- {
- "entry_point": "abrek.providers:RegistryProvider",
+ "entry_point": "lava_test.core.providers:BuiltInProvider"
+ },
+ {
+ "entry_point": "lava_test.core.providers:PkgResourcesProvider",
+ "config": {"namespace": "lava_test.test_definitions" }
+ },
+ {
+ "entry_point": "lava_test.core.providers:RegistryProvider",
"config": {
"entries": []
}
=== removed file 'tests/test_abrekcmd.py'
@@ -1,137 +0,0 @@
-# Copyright (c) 2010 Linaro
-#
-# 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 unittest
-from optparse import make_option
-from abrek.command import (
- AbrekCmd,
- AbrekCmdWithSubcommands,
- get_command,
- get_all_cmds,
- )
-
-
-class testAbrekCmd(unittest.TestCase):
- def test_empty_run(self):
- cmd = AbrekCmd()
- self.assertRaises(NotImplementedError, cmd.run)
-
- def test_name(self):
- class cmd_test_name(AbrekCmd):
- pass
- cmd = cmd_test_name()
- self.assertEqual("test-name", cmd.name())
-
- def test_help(self):
- class cmd_test_help(AbrekCmd):
- """Test Help"""
- pass
- expected_str = 'Usage: lava-test test-help\n\nOptions:\n -h, ' + \
- '--help show this help message and exit\n\n' + \
- 'Description:\nTest Help'
- cmd = cmd_test_help()
- self.assertEqual(expected_str, cmd.help())
-
- def test_no_help(self):
- class cmd_test_no_help(AbrekCmd):
- pass
- expected_str = 'Usage: lava-test test-no-help\n\nOptions:\n -h, ' + \
- '--help show this help message and exit'
- cmd = cmd_test_no_help()
- self.assertEqual(expected_str, cmd.help())
-
- def test_get_command(self):
- cmd = get_command("install")
- self.assertTrue(isinstance(cmd, AbrekCmd))
-
- def test_get_all_cmds(self):
- cmds = get_all_cmds()
- self.assertTrue("install" in cmds)
-
- def test_arglist(self):
- expected_str = 'Usage: lava-test arglist FOO'
- class cmd_arglist(AbrekCmd):
- arglist = ['*foo']
- pass
- cmd = cmd_arglist()
- self.assertTrue(expected_str in cmd.help())
-
- def test_options(self):
- expected_str = '-b BAR, --bar=BAR'
- class cmd_options(AbrekCmd):
- options = [make_option("-b", "--bar", dest="bar")]
- pass
- cmd = cmd_options()
- self.assertTrue(expected_str in cmd.help())
-
- def test_subcmds(self):
- expected_str = 'Available sub-commands:\n foo'
-
- class cmd_test_subcmds(AbrekCmdWithSubcommands):
- """Help for test-subcmds."""
- class cmd_foo(AbrekCmd):
- pass
- cmd = cmd_test_subcmds()
- self.assertTrue(
- expected_str in cmd.help()
- and 'Help for test-subcmds.' in cmd.help())
-
- def test_subcmds_run(self):
- expected_str = "subcmd test str"
-
- class cmd_test_subcmds(AbrekCmdWithSubcommands):
- class cmd_foo(AbrekCmd):
- def run(self):
- return expected_str
- cmd = cmd_test_subcmds()
- argv = ['foo']
- self.assertEqual(expected_str, cmd.main(argv))
-
- def test_subcmds_name(self):
- expected_str = "subcmd test str"
-
- class cmd_test_subcmds(AbrekCmdWithSubcommands):
- class cmd_foo(AbrekCmd):
- def run(self):
- return expected_str
- cmd = cmd_test_subcmds().get_subcommand('foo')
- self.assertEqual('test-subcmds foo', cmd.name())
-
- def test_subcmds_help(self):
- expected_str = "subcmd test str"
-
- class cmd_test_subcmds(AbrekCmdWithSubcommands):
- class cmd_foo(AbrekCmd):
- """Help for foo."""
- def run(self):
- return expected_str
- cmd = cmd_test_subcmds().get_subcommand('foo')
- self.assertTrue(
- 'test-subcmds foo' in cmd.help()
- and 'Help for foo.' in cmd.help())
-
- def test_subcmd_strip_argv(self):
- """
- Make sure that the argv list is stripped after calling the subcmd
- """
-
- class cmd_test_subcmds(AbrekCmdWithSubcommands):
- class cmd_foo(AbrekCmd):
- def main(self, argv):
- return len(argv)
- cmd = cmd_test_subcmds()
- argv = ['foo']
- self.assertEqual(0, cmd.main(argv))
-
=== removed file 'tests/test_abrektest.py'
@@ -1,48 +0,0 @@
-# Copyright (c) 2010 Linaro
-#
-# 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 re
-
-from abrek.testdef import AbrekTest, AbrekTestInstaller, AbrekTestRunner
-from imposters import OutputImposter, ConfigImposter
-from fixtures import TestCaseWithFixtures
-
-def maketest(name="foo", version="", installer=None, runner=None, parser=None):
- if installer is None:
- installer = makeinstaller()
- return AbrekTest(name, version, installer, runner, parser)
-
-def makerunner(**kwargs):
- return AbrekTestRunner(**kwargs)
-
-def makeinstaller(**kwargs):
- return AbrekTestInstaller(**kwargs)
-
-class AbrekTestConfigOutput(TestCaseWithFixtures):
- def setUp(self):
- super(AbrekTestConfigOutput, self).setUp()
- self.config = self.add_fixture(ConfigImposter())
- self.out = self.add_fixture(OutputImposter())
-
- def test_run(self):
- testrunner = makerunner(steps=["echo foo"])
- test = maketest(name="foo", runner=testrunner)
- test.install()
- test.run()
- self.assertEqual("foo", self.out.getvalue().splitlines()[0])
- completion_message = self.out.getvalue().splitlines()[1]
- completion_pattern = "ABREK TEST RUN COMPLETE: Result id is 'foo\d+\.0'"
- self.assertTrue(re.match(completion_pattern, completion_message))
-
=== removed file 'tests/test_abrektestinstaller.py'
@@ -1,60 +0,0 @@
-# Copyright (c) 2010 Linaro
-#
-# 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 hashlib
-import os
-import shutil
-import tempfile
-import unittest
-
-from abrek.testdef import AbrekTestInstaller
-
-
-class testAbrekTestInstaller(unittest.TestCase):
- def setUp(self):
- self.origdir = os.path.abspath(os.curdir)
- self.tmpdir = tempfile.mkdtemp()
- self.filename = os.path.abspath(__file__)
- os.chdir(self.tmpdir)
-
- def tearDown(self):
- os.chdir(self.origdir)
- shutil.rmtree(self.tmpdir)
-
- def makeinstaller(self,**kwargs):
- return AbrekTestInstaller(**kwargs)
-
- def test_bad_download(self):
- url = "file:///xxxyyyzzz"
- installer = self.makeinstaller(url=url)
- self.assertRaises(RuntimeError, installer._download)
-
- def test_bad_md5(self):
- url = "file://%s" % self.filename
- installer = self.makeinstaller(url=url, md5='foo')
- self.assertRaises(RuntimeError, installer._download)
-
- def test_good_md5(self):
- url = "file://%s" % self.filename
- md5 = hashlib.md5(file(self.filename).read()).hexdigest()
- installer = self.makeinstaller(url=url, md5=md5)
- location = installer._download()
- self.assertTrue(os.path.exists(location))
-
- def test_runsteps(self):
- steps = ["echo test > foo"]
- installer = self.makeinstaller(steps=steps)
- installer._runsteps()
- self.assertTrue(os.path.exists("./foo"))
=== removed file 'tests/test_abrektestparser.py'
@@ -1,65 +0,0 @@
-# Copyright (c) 2010 Linaro
-#
-# 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 shutil
-import tempfile
-import unittest
-
-from abrek.testdef import AbrekTestParser
-
-
-class testAbrekTestParser(unittest.TestCase):
- def setUp(self):
- self.origdir = os.path.abspath(os.curdir)
- self.tmpdir = tempfile.mkdtemp()
- self.filename = os.path.abspath(__file__)
- os.chdir(self.tmpdir)
-
- def tearDown(self):
- os.chdir(self.origdir)
- shutil.rmtree(self.tmpdir)
-
- def makeparser(self, *args, **kwargs):
- return AbrekTestParser(*args, **kwargs)
-
- def writeoutputlog(self, str):
- with open("testoutput.log", "a") as fd:
- fd.write(str)
-
- def test_parse(self):
- pattern = "^(?P<testid>\w+):\W+(?P<result>\w+)"
- self.writeoutputlog("test001: pass")
- parser = self.makeparser(pattern)
- parser.parse()
- self.assertTrue(parser.results["test_results"][0]["testid"] == "test001" and
- parser.results["test_results"][0]["result"] == "pass")
-
- def test_fixupdict(self):
- pattern = "^(?P<testid>\w+):\W+(?P<result>\w+)"
- fixup = {"pass":"PASS"}
- self.writeoutputlog("test001: pass")
- parser = self.makeparser(pattern, fixupdict=fixup)
- parser.parse()
- self.assertEquals("PASS", parser.results["test_results"][0]["result"])
-
- def test_appendall(self):
- pattern = "^(?P<testid>\w+):\W+(?P<result>\w+)"
- append = {"units":"foo/s"}
- self.writeoutputlog("test001: pass")
- parser = self.makeparser(pattern, appendall=append)
- parser.parse()
- self.assertEqual("foo/s", parser.results["test_results"][0]["units"])
-
=== removed file 'tests/test_abrektestrunner.py'
@@ -1,104 +0,0 @@
-# Copyright (c) 2010 Linaro
-#
-# 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 shutil
-import tempfile
-import unittest
-from datetime import datetime
-
-from abrek.testdef import AbrekTestRunner
-from imposters import OutputImposter
-from fixtures import TestCaseWithFixtures
-
-def makerunner(**kwargs):
- return AbrekTestRunner(**kwargs)
-
-class testAbrekTestRunner(unittest.TestCase):
- def setUp(self):
- self.origdir = os.path.abspath(os.curdir)
- self.tmpdir = tempfile.mkdtemp()
- self.filename = os.path.abspath(__file__)
- os.chdir(self.tmpdir)
-
- def tearDown(self):
- os.chdir(self.origdir)
- shutil.rmtree(self.tmpdir)
-
- def test_starttime(self):
- runner = makerunner()
- runner.run(self.tmpdir)
- self.assertTrue(isinstance(runner.starttime, datetime))
-
- def test_endtime(self):
- runner = makerunner()
- runner.run(self.tmpdir)
- self.assertTrue(isinstance(runner.endtime, datetime))
-
- def test_timediff(self):
- steps = ['sleep 2']
- runner = makerunner(steps=steps)
- runner.run(self.tmpdir)
- self.assertNotEqual(runner.starttime, runner.endtime)
-
- def test_runsteps(self):
- steps = ["echo test > foo"]
- runner = makerunner(steps=steps)
- runner._runsteps(self.tmpdir)
- self.assertTrue(os.path.exists("./foo"))
-
- def test_logoutput(self):
- steps = ["echo test > foo"]
- runner = makerunner(steps=steps)
- runner._runsteps(self.tmpdir)
- self.assertTrue(os.path.exists("./testoutput.log"))
-
-class testAbrekTestRunnerVerbosity(TestCaseWithFixtures):
- def setUp(self):
- super(testAbrekTestRunnerVerbosity, self).setUp()
- self.origdir = os.path.abspath(os.curdir)
- self.tmpdir = tempfile.mkdtemp()
- self.filename = os.path.abspath(__file__)
- os.chdir(self.tmpdir)
- self.out = self.add_fixture(OutputImposter())
-
- def tearDown(self):
- super(testAbrekTestRunnerVerbosity, self).tearDown()
- os.chdir(self.origdir)
- shutil.rmtree(self.tmpdir)
-
- def test_runsteps_quiet_true(self):
- steps = ["echo test"]
- runner = makerunner(steps=steps)
- runner._runsteps(self.tmpdir, quiet=True)
- self.assertEqual("", self.out.getvalue().strip())
-
- def test_runsteps_quiet_false(self):
- steps = ["echo test"]
- runner = makerunner(steps=steps)
- runner._runsteps(self.tmpdir, quiet=False)
- self.assertEqual("test", self.out.getvalue().strip())
-
- def test_run_quiet_true(self):
- steps = ["echo test"]
- runner = makerunner(steps=steps)
- runner.run(self.tmpdir, quiet=True)
- self.assertEqual("", self.out.getvalue().strip())
-
- def test_run_quiet_false(self):
- steps = ["echo test"]
- runner = makerunner(steps=steps)
- runner.run(self.tmpdir, quiet=False)
- self.assertEqual("test", self.out.getvalue().strip())
=== removed file 'tests/test_builtins.py'
@@ -1,66 +0,0 @@
-# Copyright (c) 2010 Linaro
-#
-# 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 abrek.builtins
-from imposters import ConfigImposter, OutputImposter
-from fixtures import TestCaseWithFixtures
-
-
-class ListKnown(TestCaseWithFixtures):
- def test_list_tests(self):
- out = self.add_fixture(OutputImposter())
- cmd = abrek.builtins.cmd_list_tests()
- cmd.run()
- self.assertTrue("stream" in out.getvalue())
-
-class ListInstalled(TestCaseWithFixtures):
- def test_list_installed(self):
- config = self.add_fixture(ConfigImposter())
- out = self.add_fixture(OutputImposter())
- test_name="test_list_installed000"
- os.makedirs(os.path.join(config.installdir, test_name))
- cmd = abrek.builtins.cmd_list_installed()
- cmd.run()
- self.assertTrue(test_name in out.getvalue())
-
-class TestHelp(TestCaseWithFixtures):
-
- def test_command_help(self):
- out = self.add_fixture(OutputImposter())
- abrek.builtins.cmd_help().main(['results'])
- self.assertEqual(
- abrek.results.cmd_results().help() + '\n', out.getvalue())
-
- def test_subcommand_help(self):
- out = self.add_fixture(OutputImposter())
- abrek.builtins.cmd_help().main(['results', 'list'])
- self.assertEqual(
- abrek.results.cmd_results().get_subcommand('list').help() + '\n',
- out.getvalue())
-
- def test_bad_command(self):
- out = self.add_fixture(OutputImposter())
- abrek.builtins.cmd_help().main(['foo'])
- self.assertEqual(
- "No command found for 'foo'\n", out.getvalue())
-
- def test_bad_subcommand(self):
- out = self.add_fixture(OutputImposter())
- abrek.builtins.cmd_help().main(['results', 'foo'])
- self.assertEqual(
- "No sub-command of 'results' found for 'foo'\n",
- out.getvalue())
=== removed file 'tests/test_dashboard.py'
@@ -1,213 +0,0 @@
-# Copyright (c) 2010 Linaro
-#
-# 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 json
-import os
-from uuid import uuid4
-from abrek.dashboard import (
- DashboardConfig,
- cmd_dashboard,
- )
-from imposters import ConfigImposter, OutputImposter
-from fixtures import TestCaseWithFixtures
-
-
-class SetupTests(TestCaseWithFixtures):
- def setUp(self):
- super(SetupTests, self).setUp()
- self.config = self.add_fixture(ConfigImposter())
-
- def test_dashboard_setup(self):
- host, user, passwd = setup_dashboard()
- conf = DashboardConfig()
- self.assertEqual(host, conf.host)
- self.assertEqual(user, conf.user)
- self.assertEqual(passwd, conf.password)
-
-
-class SetupOutputTests(TestCaseWithFixtures):
- def setUp(self):
- super(SetupOutputTests, self).setUp()
- self.out = self.add_fixture(OutputImposter())
-
- def test_dashboard_setup_noserver(self):
- errmsg = "You must specify a server"
- cmd = cmd_dashboard.cmd_setup()
- self.assertRaises(SystemExit, cmd.main, argv=[])
- self.assertEqual(errmsg, self.out.getvalue().strip())
-
-
-class BundleOutputTests(TestCaseWithFixtures):
- def setUp(self):
- super(BundleOutputTests, self).setUp()
- self.out = self.add_fixture(OutputImposter())
-
- def test_dashboard_bundle_badresult(self):
- errmsg = "Result directory not found"
- cmd = cmd_dashboard.cmd_bundle()
- self.assertRaises(SystemExit, cmd.main, argv=['badresult'])
- self.assertEqual(errmsg, self.out.getvalue().strip())
-
- def test_dashboard_bundle_noresult(self):
- errmsg = "You must specify a result"
- cmd = cmd_dashboard.cmd_bundle()
- self.assertRaises(SystemExit, cmd.main, argv=[])
- self.assertEqual(errmsg, self.out.getvalue().strip())
-
-class BundleConfigOutputTests(TestCaseWithFixtures):
- def setUp(self):
- super(BundleConfigOutputTests, self).setUp()
- self.config = self.add_fixture(ConfigImposter())
- self.out = self.add_fixture(OutputImposter())
-
- def test_dashboard_bundle_good(self):
- cmd = cmd_dashboard.cmd_bundle()
- (testname, testuuid) = make_stream_result(self.config)
- expected_dict = {
- "format": "Dashboard Bundle Format 1.2",
- "test_runs": [{
- "analyzer_assigned_date": "2010-10-10T00:00:00Z",
- "analyzer_assigned_uuid": testuuid,
- "hardware_context": {
- "devices": []
- },
- "software_context": {},
- "test_id": "stream",
- "test_results": [{
- "measurement": 1111.1111,
- "result": "pass",
- "test_case_id": "Copy",
- "units": "MB/s",
- "log_filename": "testoutput.log",
- "log_lineno": 3
- },
- {
- "measurement": 2222.2222,
- "result": "pass",
- "test_case_id": "Scale",
- "units": "MB/s",
- "log_filename": "testoutput.log",
- "log_lineno": 4
- },
- {
- "measurement": 3333.3333,
- "result": "pass",
- "test_case_id": "Add",
- "units": "MB/s",
- "log_filename": "testoutput.log",
- "log_lineno": 5
- },
- {
- "measurement": 4444.4444,
- "result": "pass",
- "test_case_id": "Triad",
- "units": "MB/s",
- "log_filename": "testoutput.log",
- "log_lineno": 6
- }],
- "time_check_performed": False,
- "attachments": [
- {
- "mime_type": "text/plain",
- "pathname": "testoutput.log",
- "content": "CkZ1bmN0aW9uICAgICAgUmF0ZSAoTUIvcykgICBBdmcgdGltZSAgICAgTWluIHRpbWUgICAgIE1heCB0aW1lCkNvcHk6ICAgICAgICAxMTExLjExMTEgICAgICAgMC4wMTgwICAgICAgIDAuMDExMiAgICAgICAwLjAyNDIKU2NhbGU6ICAgICAgIDIyMjIuMjIyMiAgICAgICAwLjAxOTggICAgICAgMC4wMTIyICAgICAgIDAuMDI0MwpBZGQ6ICAgICAgICAgMzMzMy4zMzMzICAgICAgIDAuMDIwMSAgICAgICAwLjAxNzYgICAgICAgMC4wMjIzClRyaWFkOiAgICAgICA0NDQ0LjQ0NDQgICAgICAgMC4wMTk3ICAgICAgIDAuMDEzOCAgICAgICAwLjAyMjMK"
- }]
- }]
- }
- cmd.main(argv=[testname])
- returned_dict = json.loads(self.out.getvalue())
- self.assertEqual(expected_dict, returned_dict)
-
-
-class PutConfigOutputTests(TestCaseWithFixtures):
- def setUp(self):
- super(PutConfigOutputTests, self).setUp()
- self.config = self.add_fixture(ConfigImposter())
- self.out = self.add_fixture(OutputImposter())
-
- def test_put_nosetup(self):
- testname, testuuid = make_stream_result(self.config)
- errmsg = "Error connecting to server, please run 'abrek dashboard " \
- "setup [host]'"
- args = ["put", "somestream", testname]
- cmd = cmd_dashboard()
- self.assertRaises(SystemExit, cmd.main, argv=args)
- self.assertEqual(errmsg, self.out.getvalue().strip())
-
- def test_put_badhost(self):
- testname, testuuid = make_stream_result(self.config)
- host, user, passwd = setup_dashboard(host = "http://badhost.foo")
- errmsg = "Unable to connect to host: [Errno -2] Name or service " \
- "not known"
- args = ["put", "somestream", testname]
- cmd = cmd_dashboard()
- self.assertRaises(SystemExit, cmd.main, argv=args)
- self.assertEqual(errmsg, self.out.getvalue().strip())
-
-
-class PutOutputTests(TestCaseWithFixtures):
- def setUp(self):
- super(PutOutputTests, self).setUp()
- self.out = self.add_fixture(OutputImposter())
-
- def test_put_noargs(self):
- errmsg = "You must specify a stream and a result"
- cmd = cmd_dashboard()
- self.assertRaises(SystemExit, cmd.main, argv=["put"])
- self.assertEqual(errmsg, self.out.getvalue().strip())
-
-
-def make_stream_result(config):
- """
- Make a fake set of test results for the stream test
- """
- testname = "stream000"
- testuuid = str(uuid4())
- testdata_data = """
-{
-"format": "Dashboard Bundle Format 1.2",
-"test_runs": [{
- "analyzer_assigned_date": "2010-10-10T00:00:00Z",
- "analyzer_assigned_uuid": "%s",
- "hardware_context": {
- "devices": []
- },
- "software_context": {},
- "test_id": "stream",
- "time_check_performed": false,
- "test_results": []
- }]
-}
-""" % testuuid
- testoutput_data = """
-Function Rate (MB/s) Avg time Min time Max time
-Copy: 1111.1111 0.0180 0.0112 0.0242
-Scale: 2222.2222 0.0198 0.0122 0.0243
-Add: 3333.3333 0.0201 0.0176 0.0223
-Triad: 4444.4444 0.0197 0.0138 0.0223
-"""
- result_dir = os.path.join(config.resultsdir, testname)
- os.makedirs(result_dir)
- with open(os.path.join(result_dir, "testdata.json"), "w") as fd:
- fd.write(testdata_data)
- with open(os.path.join(result_dir, "testoutput.log"), "w") as fd:
- fd.write(testoutput_data)
- return (testname, testuuid)
-
-def setup_dashboard(host="http://localhost:8080", user="foo", passwd="baz"):
- args = ["setup", host, "-u", user, "-p", passwd]
- cmd = cmd_dashboard()
- cmd.main(argv=args)
- return host, user, passwd
=== modified file 'tests/test_hwprofile.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2010 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
@@ -16,10 +16,10 @@
import os
import unittest
-import abrek.hwprofile
-from abrek.utils import fake_file, clear_fakes, fake_machine, clear_fake_machine
-from imposters import OutputImposter
-from fixtures import TestCaseWithFixtures
+import lava_test.core.hwprofile
+from lava_test.utils import fake_file, clear_fakes, fake_machine, clear_fake_machine
+from tests.imposters import OutputImposter
+from tests.fixtures import TestCaseWithFixtures
class AptCache:
@@ -81,7 +81,7 @@
def test_get_cpu_devs_arm(self):
fake_file('/proc/cpuinfo', ARM_CPUINFO_FILE)
fake_machine('arm')
- devs = abrek.hwprofile.get_cpu_devs()
+ devs = lava_test.core.hwprofile.get_cpu_devs()
clear_fake_machine()
cpuinfo = {
'attributes': {
@@ -113,7 +113,7 @@
'vendor': 'YYYYYYY'},
'description': 'XXXXXXX',
'device_type': 'device.board'}
- devs = abrek.hwprofile.get_board_devs()
+ devs = lava_test.core.hwprofile.get_board_devs()
clear_fake_machine()
self.assertEqual(boardinfo, devs[0])
@@ -123,13 +123,13 @@
boardinfo = {
'description': 'OMAP3 Beagle Board',
'device_type': 'device.board'}
- devs = abrek.hwprofile.get_board_devs()
+ devs = lava_test.core.hwprofile.get_board_devs()
clear_fake_machine()
self.assertEqual(boardinfo, devs[0])
def test_get_mem_devs(self):
fake_file('/proc/meminfo', FAKE_MEMINFO_FILE)
- devs = abrek.hwprofile.get_mem_devs()
+ devs = lava_test.core.hwprofile.get_mem_devs()
meminfo = {
'attributes': {
'kind': 'RAM',
@@ -139,7 +139,7 @@
self.assertEqual(meminfo, devs[0])
def test_get_usb_devs(self):
- devs = abrek.hwprofile.get_usb_devs()
+ devs = lava_test.core.hwprofile.get_usb_devs()
self.assertEqual('device.usb', devs[0]['device_type'])
@@ -156,7 +156,7 @@
def test_bad_cpuinfo(self):
errmsg = "WARNING: Could not read cpu information\n"
fake_file('/proc/cpuinfo', newpath='/foo/bar')
- devs = abrek.hwprofile.get_cpu_devs()
+ devs = lava_test.core.hwprofile.get_cpu_devs()
self.assertEqual([], devs)
self.assertEqual(errmsg, self.out.getvalue())
@@ -165,7 +165,7 @@
errmsg = "WARNING: Could not read board information\n"
fake_file('/sys/class/dmi/id/board_name', newpath='/foo/bar')
fake_file('/proc/cpuinfo', newpath='/foo/bar')
- devs = abrek.hwprofile.get_board_devs()
+ devs = lava_test.core.hwprofile.get_board_devs()
clear_fake_machine()
self.assertEqual([], devs)
self.assertEqual(errmsg, self.out.getvalue())
@@ -173,7 +173,7 @@
def test_bad_meminfo(self):
errmsg = "WARNING: Could not read memory information\n"
fake_file('/proc/meminfo', newpath='/foo/bar')
- devs = abrek.hwprofile.get_mem_devs()
+ devs = lava_test.core.hwprofile.get_mem_devs()
self.assertEqual([], devs)
self.assertEqual(errmsg, self.out.getvalue())
=== added file 'tests/test_lavatest_commands.py'
@@ -0,0 +1,62 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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, re
+
+from tests.imposters import ConfigImposter, OutputImposter
+from tests.fixtures import TestCaseWithFixtures
+from lava_test.main import LAVATestDispatcher
+
+class LavaTestCommandTestCase(TestCaseWithFixtures):
+ def setUp(self):
+ self.config = self.add_fixture(ConfigImposter())
+ self.out = self.add_fixture(OutputImposter())
+
+ def _runLavaTest(self, cmds):
+ LAVATestDispatcher().dispatch(cmds)
+
+class BadCommand(LavaTestCommandTestCase):
+ def test_bad_cmd(self):
+ # Running an unknown command that does not exist of a command that does
+ # gives a nice error message.
+ errmsg = "invalid choice: 'results'"
+
+ self.assertRaises(SystemExit, LAVATestDispatcher().dispatch, ['results', 'foo'])
+ self.assertNotEqual(None, re.search(errmsg, self.out.getvalue()), re.MULTILINE)
+ self.assertTrue(errmsg in self.out.getvalue())
+
+class ListKnown(LavaTestCommandTestCase):
+ def test_list_tests(self):
+ self._runLavaTest(['list-tests'])
+ self.assertTrue("stream" in self.out.getvalue())
+
+class ListInstalled(LavaTestCommandTestCase):
+ def test_list_installed(self):
+ # test_name must be in the BuiltInProvider._builtin_tests
+ test_name="ltp"
+ os.makedirs(os.path.join(self.config.installdir, test_name))
+ self._runLavaTest(['list-installed'])
+ self.assertTrue(test_name in self.out.getvalue())
+
+ def test_run_command_test_not_exist(self):
+ self._runLavaTest(['run','abc'])
+ self.assertTrue("There is no test with the specified ID" in self.out.getvalue())
+
+class TestHelp(LavaTestCommandTestCase):
+
+ def test_command_help(self):
+ self.assertRaises(SystemExit, LAVATestDispatcher().dispatch, ['--help'])
+ self.assertTrue("--help" in self.out.getvalue())
+
=== added file 'tests/test_lavatest_test.py'
@@ -0,0 +1,55 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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 re
+
+from lava_test.core.installers import TestInstaller
+from lava_test.core.tests import Test
+from lava_test.core.runners import TestRunner
+from tests.imposters import OutputImposter, ConfigImposter
+from tests.fixtures import TestCaseWithFixtures
+
+def maketest(name="foo", version="", installer=None, runner=None, parser=None):
+ if installer is None:
+ installer = makeinstaller()
+ return Test(name, version, installer, runner, parser)
+
+def makerunner(**kwargs):
+ return TestRunner(**kwargs)
+
+def makeinstaller(**kwargs):
+ return TestInstaller(**kwargs)
+
+class TestConfigOutput(TestCaseWithFixtures):
+ def setUp(self):
+ super(TestConfigOutput, self).setUp()
+ self.config = self.add_fixture(ConfigImposter())
+
+ def test_run(self):
+ testrunner = makerunner(steps=["echo foo"])
+ test = maketest(name="foo", runner=testrunner)
+
+ self.assertFalse(test.is_installed)
+ test.install()
+ self.assertTrue(test.is_installed)
+ artifacts = test.run()
+ std_out = open(artifacts.stdout_pathname).read()
+ self.assertTrue("foo" in std_out)
+
+ result_id_pattern = "foo\.\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z"
+ self.assertTrue(re.match(result_id_pattern, artifacts.result_id))
+
+ test.uninstall()
+ self.assertFalse(test.is_installed)
=== added file 'tests/test_lavatest_testinstaller.py'
@@ -0,0 +1,63 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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 hashlib
+import os
+import shutil
+import tempfile
+import unittest
+
+from lava_test.core.installers import TestInstaller
+
+
+class testTestInstaller(unittest.TestCase):
+ def setUp(self):
+ self.origdir = os.path.abspath(os.curdir)
+ self.tmpdir = tempfile.mkdtemp()
+ self.filename = os.path.abspath(__file__)
+ os.chdir(self.tmpdir)
+
+ def tearDown(self):
+ os.chdir(self.origdir)
+ shutil.rmtree(self.tmpdir)
+
+ def makeinstaller(self,**kwargs):
+ return TestInstaller(**kwargs)
+
+ def test_bad_download(self):
+ url = "file:///xxxyyyzzz"
+ installer = self.makeinstaller(url=url)
+ self.assertRaises(RuntimeError, installer._download, None)
+
+ def test_bad_md5(self):
+ url = "file://%s" % self.filename
+ installer = self.makeinstaller(url=url, md5='foo')
+ self.assertRaises(RuntimeError, installer._download, None)
+
+ def test_good_md5(self):
+ url = "file://%s" % self.filename
+ md5 = hashlib.md5(file(self.filename).read()).hexdigest()
+ installer = self.makeinstaller(url=url, md5=md5)
+ location = installer._download(observer=None)
+ self.assertTrue(os.path.exists(location))
+
+ def test_runsteps(self):
+ self.assertFalse(os.path.exists("./foo"))
+ steps = ["echo test > foo"]
+ installer = self.makeinstaller(steps=steps)
+ installer._runsteps(observer=None)
+ self.assertTrue(os.path.exists("./foo"))
+ self.assertTrue("test" in open("./foo").read())
+
=== added file 'tests/test_lavatest_testparser.py'
@@ -0,0 +1,70 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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 shutil
+import tempfile
+import unittest
+
+from lava_test.core.parsers import TestParser
+from tests.fixtures import TestCaseWithFixtures
+from lava_test.core.artifacts import TestArtifacts
+from tests.imposters import OutputImposter, ConfigImposter
+
+class testTestParser(TestCaseWithFixtures):
+ def setUp(self):
+ self.config = self.add_fixture(ConfigImposter())
+ self.out = self.add_fixture(OutputImposter())
+ self.origdir = os.path.abspath(os.curdir)
+ self.filename = os.path.abspath(__file__)
+ self.test_id = "ABC"
+ self.artifacts = TestArtifacts.allocate(self.test_id, self.config)
+ os.chdir(self.artifacts.results_dir)
+
+ def tearDown(self):
+ os.chdir(self.origdir)
+ shutil.rmtree(self.artifacts.results_dir)
+
+ def makeparser(self, *args, **kwargs):
+ return TestParser(*args, **kwargs)
+
+ def writeoutputlog(self, str):
+ with open("testoutput.log", "a") as fd:
+ fd.write(str)
+
+ def test_parse(self):
+ pattern = "^(?P<testid>\w+):\W+(?P<result>\w+)"
+ self.writeoutputlog("test001: pass")
+ parser = self.makeparser(pattern)
+ parser.parse(self.artifacts)
+ self.assertTrue(parser.results["test_results"][0]["testid"] == "test001" and
+ parser.results["test_results"][0]["result"] == "pass")
+
+ def test_fixupdict(self):
+ pattern = "^(?P<testid>\w+):\W+(?P<result>\w+)"
+ fixup = {"pass":"PASS"}
+ self.writeoutputlog("test001: pass")
+ parser = self.makeparser(pattern, fixupdict=fixup)
+ parser.parse(self.artifacts)
+ self.assertEquals("PASS", parser.results["test_results"][0]["result"])
+
+ def test_appendall(self):
+ pattern = "^(?P<testid>\w+):\W+(?P<result>\w+)"
+ append = {"units":"foo/s"}
+ self.writeoutputlog("test001: pass")
+ parser = self.makeparser(pattern, appendall=append)
+ parser.parse(self.artifacts)
+ self.assertEqual("foo/s", parser.results["test_results"][0]["units"])
+
=== added file 'tests/test_lavatest_testrunner.py'
@@ -0,0 +1,73 @@
+# Copyright (c) 2010, 2011 Linaro
+#
+# 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 shutil
+import tempfile
+import unittest
+from datetime import datetime
+
+from lava_test.core.runners import TestRunner
+from lava_test.core.artifacts import TestArtifacts
+from tests.imposters import OutputImposter, ConfigImposter
+from tests.fixtures import TestCaseWithFixtures
+
+def makerunner(**kwargs):
+ return TestRunner(**kwargs)
+
+class testTestRunner(TestCaseWithFixtures):
+ def setUp(self):
+ self.config = self.add_fixture(ConfigImposter())
+ self.out = self.add_fixture(OutputImposter())
+ self.origdir = os.path.abspath(os.curdir)
+ self.tmpdir = tempfile.mkdtemp()
+ self.filename = os.path.abspath(__file__)
+ os.chdir(self.tmpdir)
+ self.test_id = "ABC"
+ self.artifacts = TestArtifacts.allocate(self.test_id, self.config)
+
+ def tearDown(self):
+ os.chdir(self.origdir)
+ shutil.rmtree(self.tmpdir)
+
+ def test_starttime(self):
+ runner = makerunner()
+ runner.run(self.artifacts)
+ self.assertTrue(isinstance(runner.starttime, datetime))
+
+ def test_endtime(self):
+ runner = makerunner()
+ runner.run(self.artifacts)
+ self.assertTrue(isinstance(runner.endtime, datetime))
+
+ def test_timediff(self):
+ steps = ['sleep 2']
+ runner = makerunner(steps=steps)
+ runner.run(self.artifacts)
+ self.assertNotEqual(runner.starttime, runner.endtime)
+
+ def test_runsteps(self):
+ steps = ["echo test > foo"]
+ runner = makerunner(steps=steps)
+ runner._run_lava_test_steps(self.artifacts, observer = None)
+ self.assertTrue(os.path.exists("./foo"))
+
+ def test_logoutput(self):
+ steps = ["echo test > foo"]
+ runner = makerunner(steps=steps)
+ runner._run_lava_test_steps(self.artifacts, observer = None)
+ self.assertTrue(os.path.exists(self.artifacts.stdout_pathname))
+ self.assertTrue(os.path.exists(self.artifacts.stderr_pathname))
+
=== removed file 'tests/test_main.py'
@@ -1,32 +0,0 @@
-# Copyright (c) 2010 Linaro
-#
-# 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/>.
-
-from abrek.main import main
-from imposters import OutputImposter
-from fixtures import TestCaseWithFixtures
-
-
-class testMain(TestCaseWithFixtures):
- def setUp(self):
- super(testMain, self).setUp()
- self.out = self.add_fixture(OutputImposter())
-
- def test_bad_subcmd(self):
- # Running a subcommand that does not exist of a command that does
- # gives a nice error message.
- errmsg = "'foo' not found as a sub-command of 'results'"
- main(['./abrek', 'results', 'foo'])
- self.assertEqual(errmsg, self.out.getvalue().splitlines()[0])
-
=== removed file 'tests/test_results.py'
@@ -1,117 +0,0 @@
-# Copyright (c) 2010 Linaro
-#
-# 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
-
-from abrek.results import cmd_results
-from abrek.utils import write_file
-from imposters import ConfigImposter, OutputImposter
-from fixtures import TestCaseWithFixtures
-
-
-class ResultsTests(TestCaseWithFixtures):
- def setUp(self):
- super(ResultsTests, self).setUp()
- self.config = self.add_fixture(ConfigImposter())
- self.out = self.add_fixture(OutputImposter())
-
- def test_results_list(self):
- result_name = "test_results_list000"
- os.makedirs(os.path.join(self.config.resultsdir, result_name))
- cmd = cmd_results.cmd_list()
- cmd.run()
- self.assertTrue(result_name in self.out.getvalue())
-
- def test_results_list_nodir(self):
- errmsg = "No results found"
- cmd = cmd_results.cmd_list()
- cmd.run()
- self.assertTrue(errmsg in self.out.getvalue())
-
- def test_results_show(self):
- result_name = "test_results_show000"
- result_output = "test result output"
- result_dir = os.path.join(self.config.resultsdir, result_name)
- os.makedirs(result_dir)
- outputfile = os.path.join(result_dir, 'testoutput.log')
- write_file(result_output, outputfile)
- cmd = cmd_results.cmd_show()
- cmd.main(argv=[result_name])
- self.assertEqual(result_output, self.out.getvalue().strip())
-
- def test_results_show_noarg(self):
- errmsg = "please specify the name of the result dir"
- cmd = cmd_results.cmd_show()
- self.assertRaises(SystemExit, cmd.main, argv=[])
- self.assertEqual(errmsg, self.out.getvalue().strip())
-
- def test_results_show_nodir(self):
- testname = "foo"
- errmsg = "No result found for '%s'" % testname
- cmd = cmd_results.cmd_show()
- self.assertRaises(SystemExit, cmd.main, argv=[testname])
- self.assertEqual(errmsg, self.out.getvalue().strip())
-
- def test_results_remove(self):
- result_name = "test_results_remove000"
- result_dir = os.path.join(self.config.resultsdir, result_name)
- os.makedirs(result_dir)
- cmd = cmd_results.cmd_remove()
- cmd.main(argv=[result_name, '-f'])
- self.assertFalse(os.path.exists(result_dir))
-
- def test_results_remove_noarg(self):
- errmsg = "please specify the name of the result dir"
- cmd = cmd_results.cmd_remove()
- self.assertRaises(SystemExit, cmd.main, argv=[])
- self.assertEqual(errmsg, self.out.getvalue().strip())
-
- def test_results_remove_nodir(self):
- testname = "foo"
- errmsg = "No result found for '%s'" % testname
- cmd = cmd_results.cmd_remove()
- self.assertRaises(SystemExit, cmd.main, argv=[testname])
- self.assertEqual(errmsg, self.out.getvalue().strip())
-
- def test_results_rename(self):
- result_src = "test_results_old"
- result_dest = "test_results_new"
- result_srcdir = os.path.join(self.config.resultsdir, result_src)
- result_destdir = os.path.join(self.config.resultsdir, result_dest)
- os.makedirs(result_srcdir)
- cmd = cmd_results.cmd_rename()
- cmd.main(argv=[result_src, result_dest])
- self.assertFalse(os.path.exists(result_srcdir))
- self.assertTrue(os.path.exists(result_destdir))
-
- def test_results_rename_badsrc(self):
- errmsg = "Result directory not found"
- result_src = "test_results_old"
- result_dest = "test_results_new"
- cmd = cmd_results.cmd_rename()
- self.assertRaises(SystemExit, cmd.main, argv=[result_src, result_dest])
- self.assertEqual(errmsg, self.out.getvalue().strip())
-
- def test_results_rename_baddest(self):
- errmsg = "Destination result name already exists"
- result_src = "test_results_old"
- result_dest = "test_results_new"
- result_srcdir = os.path.join(self.config.resultsdir, result_src)
- result_destdir = os.path.join(self.config.resultsdir, result_dest)
- os.makedirs(result_srcdir)
- os.makedirs(result_destdir)
- cmd = cmd_results.cmd_rename()
- self.assertRaises(SystemExit, cmd.main, argv=[result_src, result_dest])
- self.assertEqual(errmsg, self.out.getvalue().strip())
=== modified file 'tests/test_swprofile.py'
@@ -1,4 +1,4 @@
-# Copyright (c) 2010 Linaro
+# Copyright (c) 2010, 2011 Linaro
#
# 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
@@ -16,8 +16,8 @@
import os
import unittest
-import abrek.swprofile
-from abrek.utils import fake_file, clear_fakes
+import lava_test.core.swprofile
+from lava_test.utils import fake_file, clear_fakes
class Version:
@@ -49,7 +49,7 @@
cache = self.cache
if info == None:
info = self.lsb_information
- return abrek.swprofile.get_software_context(apt_cache=cache,
+ return lava_test.core.swprofile.get_software_context(apt_cache=cache,
lsb_information=info)
def test_pkg_name(self):