Bug 788842 - Mirror mozbase to m-c, r=jhammel

This commit is contained in:
Jonathan Griffin 2012-09-07 17:58:39 -07:00
Родитель fff44eb48a
Коммит d8e0dcfd02
40 изменённых файлов: 1545 добавлений и 601 удалений

Просмотреть файл

@ -10,15 +10,14 @@ What ManifestDestiny gives you:
are just dicts with some keys. For example, a test with no
user-specified metadata looks like this:
[{'path':
'/home/jhammel/mozmill/src/ManifestDestiny/manifestdestiny/tests/testToolbar/testBackForwardButtons.js',
'name': 'testToolbar/testBackForwardButtons.js', 'here':
'/home/jhammel/mozmill/src/ManifestDestiny/manifestdestiny/tests',
'manifest': '/home/jhammel/mozmill/src/ManifestDestiny/manifestdestiny/tests',}]
[{'expected': 'pass',
'path': '/home/mozilla/mozmill/src/ManifestDestiny/manifestdestiny/tests/testToolbar/testBackForwardButtons.js',
'relpath': 'testToolbar/testBackForwardButtons.js',
'name': 'testBackForwardButtons.js',
'here': '/home/mozilla/mozmill/src/ManifestDestiny/manifestdestiny/tests',
'manifest': '/home/mozilla/mozmill/src/ManifestDestiny/manifestdestiny/tests/manifest.ini',}]
The keys displayed here (path, name, here, and manifest) are reserved
keys for ManifestDestiny and any consuming APIs. You can add
additional key, value metadata to each test.
The keys displayed here (path, relpath, name, here, and manifest) are reserved keys for ManifestDestiny and any consuming APIs. You can add additional key, value metadata to each test.
# Why have test manifests?
@ -110,8 +109,8 @@ Manifest Destiny gives tests as a list of dictionaries (in python
terms).
* path: full path to the test
* name: short name of the test; this is the (usually) relative path
specified in the section name
* relpath: relative path starting from the root manifest location
* name: file name of the test
* here: the parent directory of the manifest
* manifest: the path to the manifest containing the test

Просмотреть файл

@ -400,6 +400,55 @@ class ManifestParser(object):
def getRelativeRoot(self, root):
return root
def _read(self, root, filename, defaults):
# get directory of this file
here = os.path.dirname(os.path.abspath(filename))
defaults['here'] = here
# read the configuration
sections = read_ini(fp=filename, variables=defaults, strict=self.strict)
# get the tests
for section, data in sections:
# a file to include
# TODO: keep track of included file structure:
# self.manifests = {'manifest.ini': 'relative/path.ini'}
if section.startswith('include:'):
include_file = section.split('include:', 1)[-1]
include_file = normalize_path(include_file)
if not os.path.isabs(include_file):
include_file = os.path.join(self.getRelativeRoot(here), include_file)
if not os.path.exists(include_file):
if self.strict:
raise IOError("File '%s' does not exist" % include_file)
else:
continue
include_defaults = data.copy()
self._read(root, include_file, include_defaults)
continue
# otherwise an item
test = data
test['name'] = section
test['manifest'] = os.path.abspath(filename)
# determine the path
path = test.get('path', section)
relpath = path
if '://' not in path: # don't futz with URLs
path = normalize_path(path)
if not os.path.isabs(path):
path = os.path.join(here, path)
relpath = os.path.relpath(path, self.rootdir)
test['path'] = path
test['relpath'] = relpath
# append the item
self.tests.append(test)
def read(self, *filenames, **defaults):
# ensure all files exist
@ -421,44 +470,7 @@ class ManifestParser(object):
# == the directory of the first manifest given
self.rootdir = here
# read the configuration
sections = read_ini(fp=filename, variables=defaults, strict=self.strict)
# get the tests
for section, data in sections:
# a file to include
# TODO: keep track of included file structure:
# self.manifests = {'manifest.ini': 'relative/path.ini'}
if section.startswith('include:'):
include_file = section.split('include:', 1)[-1]
include_file = normalize_path(include_file)
if not os.path.isabs(include_file):
include_file = os.path.join(self.getRelativeRoot(here), include_file)
if not os.path.exists(include_file):
if self.strict:
raise IOError("File '%s' does not exist" % include_file)
else:
continue
include_defaults = data.copy()
self.read(include_file, **include_defaults)
continue
# otherwise an item
test = data
test['name'] = section
test['manifest'] = os.path.abspath(filename)
# determine the path
path = test.get('path', section)
if '://' not in path: # don't futz with URLs
path = normalize_path(path)
if not os.path.isabs(path):
path = os.path.join(here, path)
test['path'] = path
# append the item
self.tests.append(test)
self._read(here, filename, defaults)
### methods for querying manifests
@ -595,7 +607,7 @@ class ManifestParser(object):
print >> fp, '[%s]' % path
# reserved keywords:
reserved = ['path', 'name', 'here', 'manifest']
reserved = ['path', 'name', 'here', 'manifest', 'relpath']
for key in sorted(test.keys()):
if key in reserved:
continue
@ -1012,7 +1024,7 @@ def main(args=sys.argv[1:]):
# set up an option parser
usage = '%prog [options] [command] ...'
description = __doc__
description = "%s. Use `help` to display commands" % __doc__.strip()
parser = OptionParser(usage=usage, description=description)
parser.add_option('-s', '--strict', dest='strict',
action='store_true', default=False,

Просмотреть файл

@ -2,12 +2,7 @@
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
# The real details are in manifestparser.py; this is just a front-end
# BUT use this file when you want to distribute to python!
# otherwise setuptools will complain that it can't find setup.py
# and result in a useless package
from setuptools import setup, find_packages
from setuptools import setup
import sys
import os
@ -19,7 +14,7 @@ except:
description = ''
PACKAGE_NAME = "ManifestDestiny"
PACKAGE_VERSION = "0.5.4"
PACKAGE_VERSION = '0.5.5'
setup(name=PACKAGE_NAME,
version=PACKAGE_VERSION,
@ -27,17 +22,15 @@ setup(name=PACKAGE_NAME,
long_description=description,
classifiers=[], # Get strings from http://pypi.python.org/pypi?%3Aaction=list_classifiers
keywords='mozilla manifests',
author='Jeff Hammel',
author_email='jhammel@mozilla.com',
url='https://github.com/mozilla/mozbase/tree/master/manifestdestiny',
author='Mozilla Automation and Testing Team',
author_email='tools@lists.mozilla.org',
url='https://wiki.mozilla.org/Auto-tools/Projects/MozBase',
license='MPL',
zip_safe=False,
packages=find_packages(exclude=['legacy']),
install_requires=[
# -*- Extra requirements: -*-
],
packages=['manifestparser'],
install_requires=[],
entry_points="""
[console_scripts]
manifestparser = manifestparser:main
manifestparser = manifestparser.manifestparser:main
""",
)

Просмотреть файл

@ -0,0 +1,4 @@
# test manifest for mozbase tests
[test_expressionparser.py]
[test_manifestparser.py]
[test_testmanifest.py]

Просмотреть файл

@ -1,81 +0,0 @@
#!/usr/bin/env python
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
"""tests for ManifestDestiny"""
import doctest
import os
import sys
from optparse import OptionParser
def run_tests(raise_on_error=False, report_first=False):
# add results here
results = {}
# doctest arguments
directory = os.path.dirname(os.path.abspath(__file__))
extraglobs = {}
doctest_args = dict(extraglobs=extraglobs,
module_relative=False,
raise_on_error=raise_on_error)
if report_first:
doctest_args['optionflags'] = doctest.REPORT_ONLY_FIRST_FAILURE
# gather tests
directory = os.path.dirname(os.path.abspath(__file__))
tests = [ test for test in os.listdir(directory)
if test.endswith('.txt') and test.startswith('test_')]
os.chdir(directory)
# run the tests
for test in tests:
try:
results[test] = doctest.testfile(test, **doctest_args)
except doctest.DocTestFailure, failure:
raise
except doctest.UnexpectedException, failure:
raise failure.exc_info[0], failure.exc_info[1], failure.exc_info[2]
return results
def main(args=sys.argv[1:]):
# parse command line options
parser = OptionParser(description=__doc__)
parser.add_option('--raise', dest='raise_on_error',
default=False, action='store_true',
help="raise on first error")
parser.add_option('--report-first', dest='report_first',
default=False, action='store_true',
help="report the first error only (all tests will still run)")
parser.add_option('-q', '--quiet', dest='quiet',
default=False, action='store_true',
help="minimize output")
options, args = parser.parse_args(args)
quiet = options.__dict__.pop('quiet')
# run the tests
results = run_tests(**options.__dict__)
# check for failure
failed = False
for result in results.values():
if result[0]: # failure count; http://docs.python.org/library/doctest.html#basic-api
failed = True
break
if failed:
sys.exit(1) # error
if not quiet:
# print results
print "manifestparser.py: All tests pass!"
for test in sorted(results.keys()):
result = results[test]
print "%s: failed=%s, attempted=%s" % (test, result[0], result[1])
if __name__ == '__main__':
main()

Просмотреть файл

@ -0,0 +1,68 @@
#!/usr/bin/env python
import unittest
from manifestparser import parse
class ExpressionParserTest(unittest.TestCase):
"""Test the conditional expression parser."""
def test_basic(self):
self.assertEqual(parse("1"), 1)
self.assertEqual(parse("100"), 100)
self.assertEqual(parse("true"), True)
self.assertEqual(parse("false"), False)
self.assertEqual('', parse('""'))
self.assertEqual(parse('"foo bar"'), 'foo bar')
self.assertEqual(parse("'foo bar'"), 'foo bar')
self.assertEqual(parse("foo", foo=1), 1)
self.assertEqual(parse("bar", bar=True), True)
self.assertEqual(parse("abc123", abc123="xyz"), 'xyz')
def test_equality(self):
self.assertTrue(parse("true == true"))
self.assertTrue(parse("false == false"))
self.assertTrue(parse("1 == 1"))
self.assertTrue(parse("100 == 100"))
self.assertTrue(parse('"some text" == "some text"'))
self.assertTrue(parse("true != false"))
self.assertTrue(parse("1 != 2"))
self.assertTrue(parse('"text" != "other text"'))
self.assertTrue(parse("foo == true", foo=True))
self.assertTrue(parse("foo == 1", foo=1))
self.assertTrue(parse('foo == "bar"', foo='bar'))
self.assertTrue(parse("foo == bar", foo=True, bar=True))
self.assertTrue(parse("true == foo", foo=True))
self.assertTrue(parse("foo != true", foo=False))
self.assertTrue(parse("foo != 2", foo=1))
self.assertTrue(parse('foo != "bar"', foo='abc'))
self.assertTrue(parse("foo != bar", foo=True, bar=False))
self.assertTrue(parse("true != foo", foo=False))
self.assertTrue(parse("!false"))
def test_conjunctures(self):
self.assertTrue(parse("true && true"))
self.assertTrue(parse("true || false"))
self.assertFalse(parse("false || false"))
self.assertFalse(parse("true && false"))
self.assertTrue(parse("true || false && false"))
def test_parentheses(self):
self.assertTrue(parse("(true)"))
self.assertEqual(parse("(10)"), 10)
self.assertEqual(parse('("foo")'), 'foo')
self.assertEqual(parse("(foo)", foo=1), 1)
self.assertTrue(parse("(true == true)"), True)
self.assertTrue(parse("(true != false)"))
self.assertTrue(parse("(true && true)"))
self.assertTrue(parse("(true || false)"))
self.assertTrue(parse("(true && true || false)"))
self.assertFalse(parse("(true || false) && false"))
self.assertTrue(parse("(true || false) && true"))
self.assertTrue(parse("true && (true || false)"))
self.assertTrue(parse("true && (true || false)"))
self.assertTrue(parse("(true && false) || (true && (true || false))"))
if __name__ == '__main__':
unittest.main()

Просмотреть файл

@ -1,118 +0,0 @@
Test Expressionparser
=====================
Test the conditional expression parser.
Boilerplate::
>>> from manifestparser import parse
Test basic values::
>>> parse("1")
1
>>> parse("100")
100
>>> parse("true")
True
>>> parse("false")
False
>>> '' == parse('""')
True
>>> parse('"foo bar"')
'foo bar'
>>> parse("'foo bar'")
'foo bar'
>>> parse("foo", foo=1)
1
>>> parse("bar", bar=True)
True
>>> parse("abc123", abc123="xyz")
'xyz'
Test equality::
>>> parse("true == true")
True
>>> parse("false == false")
True
>>> parse("1 == 1")
True
>>> parse("100 == 100")
True
>>> parse('"some text" == "some text"')
True
>>> parse("true != false")
True
>>> parse("1 != 2")
True
>>> parse('"text" != "other text"')
True
>>> parse("foo == true", foo=True)
True
>>> parse("foo == 1", foo=1)
True
>>> parse('foo == "bar"', foo='bar')
True
>>> parse("foo == bar", foo=True, bar=True)
True
>>> parse("true == foo", foo=True)
True
>>> parse("foo != true", foo=False)
True
>>> parse("foo != 2", foo=1)
True
>>> parse('foo != "bar"', foo='abc')
True
>>> parse("foo != bar", foo=True, bar=False)
True
>>> parse("true != foo", foo=False)
True
>>> parse("!false")
True
Test conjunctions::
>>> parse("true && true")
True
>>> parse("true || false")
True
>>> parse("false || false")
False
>>> parse("true && false")
False
>>> parse("true || false && false")
True
Test parentheses::
>>> parse("(true)")
True
>>> parse("(10)")
10
>>> parse('("foo")')
'foo'
>>> parse("(foo)", foo=1)
1
>>> parse("(true == true)")
True
>>> parse("(true != false)")
True
>>> parse("(true && true)")
True
>>> parse("(true || false)")
True
>>> parse("(true && true || false)")
True
>>> parse("(true || false) && false")
False
>>> parse("(true || false) && true")
True
>>> parse("true && (true || false)")
True
>>> parse("true && (true || false)")
True
>>> parse("(true && false) || (true && (true || false))")
True

Просмотреть файл

@ -0,0 +1,219 @@
#!/usr/bin/env python
import os
import shutil
import tempfile
import unittest
from manifestparser import convert
from manifestparser import ManifestParser
from StringIO import StringIO
here = os.path.dirname(os.path.abspath(__file__))
class TestManifestparser(unittest.TestCase):
"""
Test the manifest parser
You must have ManifestDestiny installed before running these tests.
Run ``python manifestparser.py setup develop`` with setuptools installed.
"""
def test_sanity(self):
"""Ensure basic parser is sane"""
parser = ManifestParser()
mozmill_example = os.path.join(here, 'mozmill-example.ini')
parser.read(mozmill_example)
tests = parser.tests
self.assertEqual(len(tests), len(file(mozmill_example).read().strip().splitlines()))
# Ensure that capitalization and order aren't an issue:
lines = ['[%s]' % test['name'] for test in tests]
self.assertEqual(lines, file(mozmill_example).read().strip().splitlines())
# Show how you select subsets of tests:
mozmill_restart_example = os.path.join(here, 'mozmill-restart-example.ini')
parser.read(mozmill_restart_example)
restart_tests = parser.get(type='restart')
self.assertTrue(len(restart_tests) < len(parser.tests))
self.assertEqual(len(restart_tests), len(parser.get(manifest=mozmill_restart_example)))
self.assertFalse([test for test in restart_tests
if test['manifest'] != os.path.join(here, 'mozmill-restart-example.ini')])
self.assertEqual(parser.get('name', tags=['foo']),
['restartTests/testExtensionInstallUninstall/test2.js',
'restartTests/testExtensionInstallUninstall/test1.js'])
self.assertEqual(parser.get('name', foo='bar'),
['restartTests/testExtensionInstallUninstall/test2.js'])
def test_include(self):
"""Illustrate how include works"""
include_example = os.path.join(here, 'include-example.ini')
parser = ManifestParser(manifests=(include_example,))
# All of the tests should be included, in order:
self.assertEqual(parser.get('name'),
['crash-handling', 'fleem', 'flowers'])
self.assertEqual([(test['name'], os.path.basename(test['manifest'])) for test in parser.tests],
[('crash-handling', 'bar.ini'), ('fleem', 'include-example.ini'), ('flowers', 'foo.ini')])
# The manifests should be there too:
self.assertEqual(len(parser.manifests()), 3)
# We already have the root directory:
self.assertEqual(here, parser.rootdir)
# DEFAULT values should persist across includes, unless they're
# overwritten. In this example, include-example.ini sets foo=bar, but
# it's overridden to fleem in bar.ini
self.assertEqual(parser.get('name', foo='bar'),
['fleem', 'flowers'])
self.assertEqual(parser.get('name', foo='fleem'),
['crash-handling'])
# Passing parameters in the include section allows defining variables in
#the submodule scope:
self.assertEqual(parser.get('name', tags=['red']),
['flowers'])
# However, this should be overridable from the DEFAULT section in the
# included file and that overridable via the key directly connected to
# the test:
self.assertEqual(parser.get(name='flowers')[0]['blue'],
'ocean')
self.assertEqual(parser.get(name='flowers')[0]['yellow'],
'submarine')
# You can query multiple times if you need to::
flowers = parser.get(foo='bar')
self.assertEqual(len(flowers), 2)
# Using the inverse flag should invert the set of tests returned:
self.assertEqual(parser.get('name', inverse=True, tags=['red']),
['crash-handling', 'fleem'])
# All of the included tests actually exist::
self.assertEqual([i['name'] for i in parser.missing()], [])
# Write the output to a manifest:
buffer = StringIO()
parser.write(fp=buffer, global_kwargs={'foo': 'bar'})
self.assertEqual(buffer.getvalue().strip(),
'[DEFAULT]\nfoo = bar\n\n[fleem]\n\n[include/flowers]\nblue = ocean\nred = roses\nyellow = submarine')
def test_directory_to_manifest(self):
"""
Test our ability to convert a static directory structure to a
manifest.
"""
# First, stub out a directory with files in it::
def create_stub():
directory = tempfile.mkdtemp()
for i in 'foo', 'bar', 'fleem':
file(os.path.join(directory, i), 'w').write(i)
subdir = os.path.join(directory, 'subdir')
os.mkdir(subdir)
file(os.path.join(subdir, 'subfile'), 'w').write('baz')
return directory
stub = create_stub()
self.assertTrue(os.path.exists(stub) and os.path.isdir(stub))
# Make a manifest for it:
self.assertEqual(convert([stub]),
"""[bar]
[fleem]
[foo]
[subdir/subfile]""")
shutil.rmtree(stub) # cleanup
# Now do the same thing but keep the manifests in place:
stub = create_stub()
convert([stub], write='manifest.ini')
self.assertEqual(sorted(os.listdir(stub)),
['bar', 'fleem', 'foo', 'manifest.ini', 'subdir'])
parser = ManifestParser()
parser.read(os.path.join(stub, 'manifest.ini'))
self.assertEqual([i['name'] for i in parser.tests],
['subfile', 'bar', 'fleem', 'foo'])
parser = ManifestParser()
parser.read(os.path.join(stub, 'subdir', 'manifest.ini'))
self.assertEqual(len(parser.tests), 1)
self.assertEqual(parser.tests[0]['name'], 'subfile')
shutil.rmtree(stub)
def test_copy(self):
"""Test our ability to copy a set of manifests"""
tempdir = tempfile.mkdtemp()
include_example = os.path.join(here, 'include-example.ini')
manifest = ManifestParser(manifests=(include_example,))
manifest.copy(tempdir)
self.assertEqual(sorted(os.listdir(tempdir)),
['fleem', 'include', 'include-example.ini'])
self.assertEqual(sorted(os.listdir(os.path.join(tempdir, 'include'))),
['bar.ini', 'crash-handling', 'flowers', 'foo.ini'])
from_manifest = ManifestParser(manifests=(include_example,))
to_manifest = os.path.join(tempdir, 'include-example.ini')
to_manifest = ManifestParser(manifests=(to_manifest,))
self.assertEqual(to_manifest.get('name'), from_manifest.get('name'))
shutil.rmtree(tempdir)
def test_update(self):
"""
Test our ability to update tests from a manifest and a directory of
files
"""
# boilerplate
tempdir = tempfile.mkdtemp()
for i in range(10):
file(os.path.join(tempdir, str(i)), 'w').write(str(i))
# First, make a manifest:
manifest = convert([tempdir])
newtempdir = tempfile.mkdtemp()
manifest_file = os.path.join(newtempdir, 'manifest.ini')
file(manifest_file,'w').write(manifest)
manifest = ManifestParser(manifests=(manifest_file,))
self.assertEqual(manifest.get('name'),
[str(i) for i in range(10)])
# All of the tests are initially missing:
self.assertEqual([i['name'] for i in manifest.missing()],
[str(i) for i in range(10)])
# But then we copy one over:
self.assertEqual(manifest.get('name', name='1'), ['1'])
manifest.update(tempdir, name='1')
self.assertEqual(sorted(os.listdir(newtempdir)),
['1', 'manifest.ini'])
# Update that one file and copy all the "tests":
file(os.path.join(tempdir, '1'), 'w').write('secret door')
manifest.update(tempdir)
self.assertEqual(sorted(os.listdir(newtempdir)),
['0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'manifest.ini'])
self.assertEqual(file(os.path.join(newtempdir, '1')).read().strip(),
'secret door')
# clean up:
shutil.rmtree(tempdir)
shutil.rmtree(newtempdir)
def test_path_override(self):
"""You can override the path in the section too.
This shows that you can use a relative path"""
path_example = os.path.join(here, 'path-example.ini')
manifest = ManifestParser(manifests=(path_example,))
self.assertEqual(manifest.tests[0]['path'],
os.path.join(here, 'fleem'))
if __name__ == '__main__':
unittest.main()

Просмотреть файл

@ -1,217 +0,0 @@
Test the manifest parser
========================
You must have ManifestDestiny installed before running these tests.
Run ``python manifestparser.py setup develop`` with setuptools installed.
Ensure basic parser is sane::
>>> from manifestparser import ManifestParser
>>> parser = ManifestParser()
>>> parser.read('mozmill-example.ini')
>>> tests = parser.tests
>>> len(tests) == len(file('mozmill-example.ini').read().strip().splitlines())
True
Ensure that capitalization and order aren't an issue:
>>> lines = ['[%s]' % test['name'] for test in tests]
>>> lines == file('mozmill-example.ini').read().strip().splitlines()
True
Show how you select subsets of tests:
>>> parser.read('mozmill-restart-example.ini')
>>> restart_tests = parser.get(type='restart')
>>> len(restart_tests) < len(parser.tests)
True
>>> import os
>>> len(restart_tests) == len(parser.get(manifest=os.path.abspath('mozmill-restart-example.ini')))
True
>>> assert not [test for test in restart_tests if test['manifest'] != os.path.abspath('mozmill-restart-example.ini')]
>>> parser.get('name', tags=['foo'])
['restartTests/testExtensionInstallUninstall/test2.js', 'restartTests/testExtensionInstallUninstall/test1.js']
>>> parser.get('name', foo='bar')
['restartTests/testExtensionInstallUninstall/test2.js']
Illustrate how include works::
>>> parser = ManifestParser(manifests=('include-example.ini',))
All of the tests should be included, in order::
>>> parser.get('name')
['crash-handling', 'fleem', 'flowers']
>>> [(test['name'], os.path.basename(test['manifest'])) for test in parser.tests]
[('crash-handling', 'bar.ini'), ('fleem', 'include-example.ini'), ('flowers', 'foo.ini')]
The manifests should be there too::
>>> len(parser.manifests())
3
We're already in the root directory::
>>> os.getcwd() == parser.rootdir
True
DEFAULT values should persist across includes, unless they're
overwritten. In this example, include-example.ini sets foo=bar, but
its overridden to fleem in bar.ini::
>>> parser.get('name', foo='bar')
['fleem', 'flowers']
>>> parser.get('name', foo='fleem')
['crash-handling']
Passing parameters in the include section allows defining variables in
the submodule scope:
>>> parser.get('name', tags=['red'])
['flowers']
However, this should be overridable from the DEFAULT section in the
included file and that overridable via the key directly connected to
the test::
>>> parser.get(name='flowers')[0]['blue']
'ocean'
>>> parser.get(name='flowers')[0]['yellow']
'submarine'
You can query multiple times if you need to::
>>> flowers = parser.get(foo='bar')
>>> len(flowers)
2
>>> roses = parser.get(tests=flowers, red='roses')
Using the inverse flag should invert the set of tests returned::
>>> parser.get('name', inverse=True, tags=['red'])
['crash-handling', 'fleem']
All of the included tests actually exist::
>>> [i['name'] for i in parser.missing()]
[]
Write the output to a manifest:
>>> from StringIO import StringIO
>>> buffer = StringIO()
>>> parser.write(fp=buffer, global_kwargs={'foo': 'bar'})
>>> buffer.getvalue().strip()
'[DEFAULT]\nfoo = bar\n\n[fleem]\n\n[include/flowers]\nblue = ocean\nred = roses\nyellow = submarine'
Test our ability to convert a static directory structure to a
manifest. First, stub out a directory with files in it::
>>> import shutil, tempfile
>>> def create_stub():
... directory = tempfile.mkdtemp()
... for i in 'foo', 'bar', 'fleem':
... file(os.path.join(directory, i), 'w').write(i)
... subdir = os.path.join(directory, 'subdir')
... os.mkdir(subdir)
... file(os.path.join(subdir, 'subfile'), 'w').write('baz')
... return directory
>>> stub = create_stub()
>>> os.path.exists(stub) and os.path.isdir(stub)
True
Make a manifest for it::
>>> from manifestparser import convert
>>> print convert([stub])
[bar]
[fleem]
[foo]
[subdir/subfile]
>>> shutil.rmtree(stub)
Now do the same thing but keep the manifests in place::
>>> stub = create_stub()
>>> convert([stub], write='manifest.ini')
>>> sorted(os.listdir(stub))
['bar', 'fleem', 'foo', 'manifest.ini', 'subdir']
>>> parser = ManifestParser()
>>> parser.read(os.path.join(stub, 'manifest.ini'))
>>> [i['name'] for i in parser.tests]
['subfile', 'bar', 'fleem', 'foo']
>>> parser = ManifestParser()
>>> parser.read(os.path.join(stub, 'subdir', 'manifest.ini'))
>>> len(parser.tests)
1
>>> parser.tests[0]['name']
'subfile'
>>> shutil.rmtree(stub)
Test our ability to copy a set of manifests::
>>> tempdir = tempfile.mkdtemp()
>>> manifest = ManifestParser(manifests=('include-example.ini',))
>>> manifest.copy(tempdir)
>>> sorted(os.listdir(tempdir))
['fleem', 'include', 'include-example.ini']
>>> sorted(os.listdir(os.path.join(tempdir, 'include')))
['bar.ini', 'crash-handling', 'flowers', 'foo.ini']
>>> from_manifest = ManifestParser(manifests=('include-example.ini',))
>>> to_manifest = os.path.join(tempdir, 'include-example.ini')
>>> to_manifest = ManifestParser(manifests=(to_manifest,))
>>> to_manifest.get('name') == from_manifest.get('name')
True
>>> shutil.rmtree(tempdir)
Test our ability to update tests from a manifest and a directory of
files::
>>> tempdir = tempfile.mkdtemp()
>>> for i in range(10):
... file(os.path.join(tempdir, str(i)), 'w').write(str(i))
First, make a manifest::
>>> manifest = convert([tempdir])
>>> newtempdir = tempfile.mkdtemp()
>>> manifest_file = os.path.join(newtempdir, 'manifest.ini')
>>> file(manifest_file,'w').write(manifest)
>>> manifest = ManifestParser(manifests=(manifest_file,))
>>> manifest.get('name') == [str(i) for i in range(10)]
True
All of the tests are initially missing::
>>> [i['name'] for i in manifest.missing()] == [str(i) for i in range(10)]
True
But then we copy one over::
>>> manifest.get('name', name='1')
['1']
>>> manifest.update(tempdir, name='1')
>>> sorted(os.listdir(newtempdir))
['1', 'manifest.ini']
Update that one file and copy all the "tests"::
>>> file(os.path.join(tempdir, '1'), 'w').write('secret door')
>>> manifest.update(tempdir)
>>> sorted(os.listdir(newtempdir))
['0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'manifest.ini']
>>> file(os.path.join(newtempdir, '1')).read().strip()
'secret door'
Clean up::
>>> shutil.rmtree(tempdir)
>>> shutil.rmtree(newtempdir)
You can override the path in the section too. This shows that you can
use a relative path::
>>> manifest = ManifestParser(manifests=('path-example.ini',))
>>> manifest.tests[0]['path'] == os.path.abspath('fleem')
True

Просмотреть файл

@ -0,0 +1,34 @@
#!/usr/bin/env python
import os
import unittest
from manifestparser import TestManifest
here = os.path.dirname(os.path.abspath(__file__))
class TestTestManifest(unittest.TestCase):
"""Test the Test Manifest"""
def test_testmanifest(self):
# Test filtering based on platform:
filter_example = os.path.join(here, 'filter-example.ini')
manifest = TestManifest(manifests=(filter_example,))
self.assertEqual([i['name'] for i in manifest.active_tests(os='win', disabled=False, exists=False)],
['windowstest', 'fleem'])
self.assertEqual([i['name'] for i in manifest.active_tests(os='linux', disabled=False, exists=False)],
['fleem', 'linuxtest'])
# Look for existing tests. There is only one:
self.assertEqual([i['name'] for i in manifest.active_tests()],
['fleem'])
# You should be able to expect failures:
last_test = manifest.active_tests(exists=False, toolkit='gtk2')[-1]
self.assertEqual(last_test['name'], 'linuxtest')
self.assertEqual(last_test['expected'], 'pass')
last_test = manifest.active_tests(exists=False, toolkit='cocoa')[-1]
self.assertEqual(last_test['expected'], 'fail')
if __name__ == '__main__':
unittest.main()

Просмотреть файл

@ -1,32 +0,0 @@
Test the Test Manifest
======================
Boilerplate::
>>> import os
Test filtering based on platform::
>>> from manifestparser import TestManifest
>>> manifest = TestManifest(manifests=('filter-example.ini',))
>>> [i['name'] for i in manifest.active_tests(os='win', disabled=False, exists=False)]
['windowstest', 'fleem']
>>> [i['name'] for i in manifest.active_tests(os='linux', disabled=False, exists=False)]
['fleem', 'linuxtest']
Look for existing tests. There is only one::
>>> [i['name'] for i in manifest.active_tests()]
['fleem']
You should be able to expect failures::
>>> last_test = manifest.active_tests(exists=False, toolkit='gtk2')[-1]
>>> last_test['name']
'linuxtest'
>>> last_test['expected']
'pass'
>>> last_test = manifest.active_tests(exists=False, toolkit='cocoa')[-1]
>>> last_test['expected']
'fail'

Просмотреть файл

@ -0,0 +1,12 @@
# Mozcrash
Package for getting a stack trace out of processes that have crashed and left behind a minidump file using the Google Breakpad library.
## Usage example
TODO
import mozcrash
#...

Просмотреть файл

@ -0,0 +1,5 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
from mozcrash import *

Просмотреть файл

@ -0,0 +1,137 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
from __future__ import with_statement
__all__ = ['check_for_crashes']
import os, sys, glob, urllib2, tempfile, subprocess, shutil, urlparse, zipfile
import mozlog
def is_url(thing):
"""
Return True if thing looks like a URL.
"""
# We want to download URLs like http://... but not Windows paths like c:\...
return len(urlparse.urlparse(thing).scheme) >= 2
def extractall(zip, path = None):
"""
Compatibility shim for Python 2.6's ZipFile.extractall.
"""
if hasattr(zip, "extractall"):
return zip.extractall(path)
if path is None:
path = os.curdir
for name in self._zipfile.namelist():
filename = os.path.normpath(os.path.join(path, name))
if name.endswith("/"):
os.makedirs(filename)
else:
path = os.path.split(filename)[0]
if not os.path.isdir(path):
os.makedirs(path)
with open(filename, "wb") as dest:
dest.write(zip.read(name))
def check_for_crashes(dump_directory, symbols_path,
stackwalk_binary=None,
dump_save_path=None,
test_name=None):
"""
Print a stack trace for minidumps left behind by a crashing program.
Arguments:
dump_directory: The directory in which to look for minidumps.
symbols_path: The path to symbols to use for dump processing.
This can either be a path to a directory
containing Breakpad-format symbols, or a URL
to a zip file containing a set of symbols.
stackwalk_binary: The path to the minidump_stackwalk binary.
If not set, the environment variable
MINIDUMP_STACKWALK will be checked.
dump_save_path: A directory in which to copy minidump files
for safekeeping. If not set, the environment
variable MINIDUMP_SAVE_PATH will be checked.
test_name: The test name to be used in log output.
Returns True if any minidumps were found, False otherwise.
"""
log = mozlog.getLogger('mozcrash')
if stackwalk_binary is None:
stackwalk_binary = os.environ.get('MINIDUMP_STACKWALK', None)
# try to get the caller's filename if no test name is given
if test_name is None:
try:
test_name = os.path.basename(sys._getframe(1).f_code.co_filename)
except:
test_name = "unknown"
# Check preconditions
dumps = glob.glob(os.path.join(dump_directory, '*.dmp'))
if len(dumps) == 0:
return False
found_crash = False
remove_symbols = False
# If our symbols are at a remote URL, download them now
if is_url(symbols_path):
log.info("Downloading symbols from: %s", symbols_path)
remove_symbols = True
# Get the symbols and write them to a temporary zipfile
data = urllib2.urlopen(symbols_path)
symbols_file = tempfile.TemporaryFile()
symbols_file.write(data.read())
# extract symbols to a temporary directory (which we'll delete after
# processing all crashes)
symbols_path = tempfile.mkdtemp()
zfile = zipfile.ZipFile(symbols_file, 'r')
extractall(zfile, symbols_path)
zfile.close()
try:
for d in dumps:
log.info("PROCESS-CRASH | %s | application crashed (minidump found)", test_name)
log.info("Crash dump filename: %s", d)
if symbols_path and stackwalk_binary and os.path.exists(stackwalk_binary):
# run minidump_stackwalk
p = subprocess.Popen([stackwalk_binary, d, symbols_path],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
(out, err) = p.communicate()
if len(out) > 3:
# minidump_stackwalk is chatty,
# so ignore stderr when it succeeds.
print out
else:
print "stderr from minidump_stackwalk:"
print err
if p.returncode != 0:
log.error("minidump_stackwalk exited with return code %d", p.returncode)
else:
if not symbols_path:
log.warn("No symbols path given, can't process dump.")
if not stackwalk_binary:
log.warn("MINIDUMP_STACKWALK not set, can't process dump.")
elif stackwalk_binary and not os.path.exists(stackwalk_binary):
log.warn("MINIDUMP_STACKWALK binary not found: %s", stackwalk_binary)
if dump_save_path is None:
dump_save_path = os.environ.get('MINIDUMP_SAVE_PATH', None)
if dump_save_path:
shutil.move(d, dump_save_path)
log.info("Saved dump as %s", os.path.join(dump_save_path,
os.path.basename(d)))
else:
os.remove(d)
extra = os.path.splitext(d)[0] + ".extra"
if os.path.exists(extra):
os.remove(extra)
found_crash = True
finally:
if remove_symbols:
shutil.rmtree(symbols_path)
return found_crash

Просмотреть файл

@ -0,0 +1,35 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
import os
from setuptools import setup
PACKAGE_VERSION = '0.1'
# get documentation from the README
try:
here = os.path.dirname(os.path.abspath(__file__))
description = file(os.path.join(here, 'README.md')).read()
except (OSError, IOError):
description = ''
# dependencies
deps = ['']
setup(name='mozcrash',
version=PACKAGE_VERSION,
description="Package for printing stack traces from minidumps left behind by crashed processes.",
long_description=description,
classifiers=[], # Get strings from http://pypi.python.org/pypi?%3Aaction=list_classifiers
keywords='mozilla',
author='Mozilla Automation and Tools team',
author_email='tools@lists.mozilla.org',
url='https://wiki.mozilla.org/Auto-tools/Projects/MozBase',
license='MPL',
packages=['mozcrash'],
include_package_data=True,
zip_safe=False,
install_requires=deps,
)

Просмотреть файл

@ -0,0 +1 @@
[test.py]

Просмотреть файл

@ -0,0 +1,136 @@
#!/usr/bin/env python
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
import os, unittest, subprocess, tempfile, shutil, urlparse, zipfile, StringIO
import mozcrash, mozlog, mozhttpd
# Make logs go away
log = mozlog.getLogger("mozcrash", os.devnull)
def popen_factory(stdouts):
"""
Generate a class that can mock subprocess.Popen. |stdouts| is an iterable that
should return an iterable for the stdout of each process in turn.
"""
class mock_popen(object):
def __init__(self, args, *args_rest, **kwargs):
self.stdout = stdouts.next()
self.returncode = 0
def wait(self):
return 0
def communicate(self):
return (self.stdout.next(), "")
return mock_popen
class TestCrash(unittest.TestCase):
def setUp(self):
self.tempdir = tempfile.mkdtemp()
# a fake file to use as a stackwalk binary
self.stackwalk = os.path.join(self.tempdir, "stackwalk")
open(self.stackwalk, "w").write("fake binary")
self._subprocess_popen = subprocess.Popen
subprocess.Popen = popen_factory(self.next_mock_stdout())
self.stdouts = []
def tearDown(self):
subprocess.Popen = self._subprocess_popen
shutil.rmtree(self.tempdir)
def next_mock_stdout(self):
if not self.stdouts:
yield iter([])
for s in self.stdouts:
yield iter(s)
def test_nodumps(self):
"""
Test that check_for_crashes returns False if no dumps are present.
"""
self.stdouts.append(["this is some output"])
self.assertFalse(mozcrash.check_for_crashes(self.tempdir,
'symbols_path',
stackwalk_binary=self.stackwalk))
def test_simple(self):
"""
Test that check_for_crashes returns True if a dump is present.
"""
open(os.path.join(self.tempdir, "test.dmp"), "w").write("foo")
self.stdouts.append(["this is some output"])
self.assert_(mozcrash.check_for_crashes(self.tempdir,
'symbols_path',
stackwalk_binary=self.stackwalk))
def test_stackwalk_envvar(self):
"""
Test that check_for_crashes uses the MINIDUMP_STACKWALK environment var.
"""
open(os.path.join(self.tempdir, "test.dmp"), "w").write("foo")
self.stdouts.append(["this is some output"])
os.environ['MINIDUMP_STACKWALK'] = self.stackwalk
self.assert_(mozcrash.check_for_crashes(self.tempdir,
'symbols_path'))
del os.environ['MINIDUMP_STACKWALK']
def test_save_path(self):
"""
Test that dump_save_path works.
"""
open(os.path.join(self.tempdir, "test.dmp"), "w").write("foo")
save_path = os.path.join(self.tempdir, "saved")
os.mkdir(save_path)
self.stdouts.append(["this is some output"])
self.assert_(mozcrash.check_for_crashes(self.tempdir,
'symbols_path',
stackwalk_binary=self.stackwalk,
dump_save_path=save_path))
self.assert_(os.path.isfile(os.path.join(save_path, "test.dmp")))
def test_save_path_envvar(self):
"""
Test that the MINDUMP_SAVE_PATH environment variable works.
"""
open(os.path.join(self.tempdir, "test.dmp"), "w").write("foo")
save_path = os.path.join(self.tempdir, "saved")
os.mkdir(save_path)
self.stdouts.append(["this is some output"])
os.environ['MINIDUMP_SAVE_PATH'] = save_path
self.assert_(mozcrash.check_for_crashes(self.tempdir,
'symbols_path',
stackwalk_binary=self.stackwalk))
del os.environ['MINIDUMP_SAVE_PATH']
self.assert_(os.path.isfile(os.path.join(save_path, "test.dmp")))
def test_symbol_path_url(self):
"""
Test that passing a URL as symbols_path correctly fetches the URL.
"""
open(os.path.join(self.tempdir, "test.dmp"), "w").write("foo")
self.stdouts.append(["this is some output"])
def make_zipfile():
data = StringIO.StringIO()
z = zipfile.ZipFile(data, 'w')
z.writestr("symbols.txt", "abc/xyz")
z.close()
return data.getvalue()
def get_symbols(req):
headers = {}
return (200, headers, make_zipfile())
httpd = mozhttpd.MozHttpd(port=0,
urlhandlers=[{'method':'GET', 'path':'/symbols', 'function':get_symbols}])
httpd.start()
symbol_url = urlparse.urlunsplit(('http', '%s:%d' % httpd.httpd.server_address,
'/symbols','',''))
self.assert_(mozcrash.check_for_crashes(self.tempdir,
symbol_url,
stackwalk_binary=self.stackwalk))
if __name__ == '__main__':
unittest.main()

Просмотреть файл

@ -3,7 +3,7 @@
# You can obtain one at http://mozilla.org/MPL/2.0/.
import os
from setuptools import setup, find_packages
from setuptools import setup
try:
here = os.path.dirname(os.path.abspath(__file__))
@ -21,11 +21,11 @@ setup(name='mozhttpd',
long_description=description,
classifiers=[], # Get strings from http://pypi.python.org/pypi?%3Aaction=list_classifiers
keywords='mozilla',
author='Joel Maher',
author='Mozilla Automation and Testing Team',
author_email='tools@lists.mozilla.org',
url='https://github.com/mozilla/mozbase/tree/master/mozhttpd',
url='https://wiki.mozilla.org/Auto-tools/Projects/MozBase',
license='MPL',
packages=find_packages(exclude=['ez_setup', 'examples', 'tests']),
packages=['mozhttpd'],
include_package_data=True,
zip_safe=False,
install_requires=deps,

Просмотреть файл

@ -4,7 +4,7 @@
import os
from setuptools import setup, find_packages
from setuptools import setup
PACKAGE_VERSION = '0.3.3'
@ -28,11 +28,11 @@ setup(name='mozinfo',
long_description=description,
classifiers=[], # Get strings from http://pypi.python.org/pypi?%3Aaction=list_classifiers
keywords='mozilla',
author='Jeff Hammel',
author_email='jhammel@mozilla.com',
url='https://wiki.mozilla.org/Auto-tools',
author='Mozilla Automation and Testing Team',
author_email='tools@lists.mozilla.org',
url='https://wiki.mozilla.org/Auto-tools/Projects/MozBase',
license='MPL',
packages=find_packages(exclude=['legacy']),
packages=['mozinfo'],
include_package_data=True,
zip_safe=False,
install_requires=deps,

Просмотреть файл

@ -3,7 +3,7 @@
# You can obtain one at http://mozilla.org/MPL/2.0/.
import os
from setuptools import setup, find_packages
from setuptools import setup
try:
here = os.path.dirname(os.path.abspath(__file__))
@ -32,9 +32,9 @@ setup(name='mozInstall',
keywords='mozilla',
author='Mozilla Automation and Tools team',
author_email='tools@lists.mozilla.org',
url='https://github.com/mozilla/mozbase',
url='https://wiki.mozilla.org/Auto-tools/Projects/MozBase',
license='MPL 2.0',
packages=find_packages(exclude=['legacy']),
packages=['mozinstall'],
include_package_data=True,
zip_safe=False,
install_requires=deps,

Просмотреть файл

@ -4,7 +4,7 @@
import os
import sys
from setuptools import setup, find_packages
from setuptools import setup
PACKAGE_NAME = "mozlog"
PACKAGE_VERSION = "1.1"
@ -21,11 +21,11 @@ setup(name=PACKAGE_NAME,
version=PACKAGE_VERSION,
description=desc,
long_description=description,
author='Andrew Halberstadt, Mozilla',
author_email='halbersa@gmail.com',
url='http://github.com/ahal/mozbase',
author='Mozilla Automation and Testing Team',
author_email='tools@lists.mozilla.org',
url='https://wiki.mozilla.org/Auto-tools/Projects/MozBase',
license='MPL 1.1/GPL 2.0/LGPL 2.1',
packages=find_packages(exclude=['legacy']),
packages=['mozlog'],
zip_safe=False,
platforms =['Any'],
classifiers=['Development Status :: 4 - Beta',

Просмотреть файл

@ -14,24 +14,59 @@ is the central exposed API for mozprocess. `ProcessHandler` utilizes
a contained subclass of [subprocess.Popen](http://docs.python.org/library/subprocess.html),
`Process`, which does the brunt of the process management.
Basic usage:
## Basic usage
process = ProcessHandler(['command', '-line', 'arguments'],
cwd=None, # working directory for cmd; defaults to None
env={}, # environment to use for the process; defaults to os.environ
)
exit_code = process.waitForFinish(timeout=60) # seconds
process.run(timeout=60) # seconds
process.wait()
`ProcessHandler` offers several other properties and methods as part of its API:
def __init__(self,
cmd,
args=None,
cwd=None,
env=None,
ignore_children = False,
processOutputLine=(),
onTimeout=(),
onFinish=(),
**kwargs):
"""
cmd = Command to run
args = array of arguments (defaults to None)
cwd = working directory for cmd (defaults to None)
env = environment to use for the process (defaults to os.environ)
ignore_children = when True, causes system to ignore child processes,
defaults to False (which tracks child processes)
processOutputLine = handlers to process the output line
onTimeout = handlers for timeout event
kwargs = keyword args to pass directly into Popen
NOTE: Child processes will be tracked by default. If for any reason
we are unable to track child processes and ignore_children is set to False,
then we will fall back to only tracking the root process. The fallback
will be logged.
"""
@property
def timedOut(self):
"""True if the process has timed out."""
def run(self):
def run(self, timeout=None, outputTimeout=None):
"""
Starts the process. waitForFinish must be called to allow the
process to complete.
Starts the process.
If timeout is not None, the process will be allowed to continue for
that number of seconds before being killed.
If outputTimeout is not None, the process will be allowed to continue
for that number of seconds without producing any output before
being killed.
"""
def kill(self):
@ -74,16 +109,14 @@ Basic usage:
for handler in self.onFinishHandlers:
handler()
def waitForFinish(self, timeout=None, outputTimeout=None):
def wait(self, timeout=None):
"""
Handle process output until the process terminates or times out.
Waits until all output has been read and the process is
terminated.
If timeout is not None, the process will be allowed to continue for
that number of seconds before being killed.
If outputTimeout is not None, the process will be allowed to continue
for that number of seconds without producing any output before
being killed.
If timeout is not None, will return after timeout seconds.
This timeout only causes the wait function to return and
does not kill the process.
"""
See https://github.com/mozilla/mozbase/blob/master/mozprocess/mozprocess/processhandler.py
@ -99,6 +132,36 @@ the `onTimeout()` method), process completion (by overriding
`onFinish()`), and to process the command output (by overriding
`processOutputLine()`).
## Examples
In the most common case, a process_handler is created, then run followed by wait are called:
proc_handler = ProcessHandler([cmd, args])
proc_handler.run(outputTimeout=60) # will time out after 60 seconds without output
proc_handler.wait()
Often, the main thread will do other things:
proc_handler = ProcessHandler([cmd, args])
proc_handler.run(timeout=60) # will time out after 60 seconds regardless of output
do_other_work()
if proc_handler.proc.poll() is None:
proc_handler.wait()
By default output is printed to stdout, but anything is possible:
# this example writes output to both stderr and a file called 'output.log'
def some_func(line):
print >> sys.stderr, line
with open('output.log', 'a') as log:
log.write('%s\n' % line)
proc_handler = ProcessHandler([cmd, args], processOutputLine=some_func)
proc_handler.run()
proc_handler.wait()
# TODO
- Document improvements over `subprocess.Popen.kill`

Просмотреть файл

@ -591,9 +591,16 @@ falling back to not using job objects for managing child processes"""
"""the string value of the command line"""
return subprocess.list2cmdline([self.cmd] + self.args)
def run(self):
"""Starts the process. waitForFinish must be called to allow the
process to complete.
def run(self, timeout=None, outputTimeout=None):
"""
Starts the process.
If timeout is not None, the process will be allowed to continue for
that number of seconds before being killed.
If outputTimeout is not None, the process will be allowed to continue
for that number of seconds without producing any output before
being killed.
"""
self.didTimeout = False
self.startTime = datetime.now()
@ -605,6 +612,8 @@ falling back to not using job objects for managing child processes"""
ignore_children = self._ignore_children,
**self.keywordargs)
self.processOutput(timeout=timeout, outputTimeout=outputTimeout)
def kill(self):
"""
Kills the managed process and if you created the process with
@ -660,9 +669,6 @@ falling back to not using job objects for managing child processes"""
being killed.
"""
def _processOutput():
if not hasattr(self, 'proc'):
self.run()
self.didTimeout = False
logsource = self.proc.stdout
@ -684,21 +690,24 @@ falling back to not using job objects for managing child processes"""
self.onTimeout()
else:
self.onFinish()
if not hasattr(self, 'proc'):
self.run()
if not self.outThread:
self.outThread = threading.Thread(target=_processOutput)
self.outThread.daemon = True
self.outThread.start()
def waitForFinish(self, timeout=None):
def wait(self, timeout=None):
"""
Waits until all output has been read and the process is
terminated.
If timeout is not None, will return after timeout seconds.
This timeout is only for waitForFinish and doesn't affect
the didTimeout or onTimeout properties.
This timeout only causes the wait function to return and
does not kill the process.
"""
if self.outThread:
# Thread.join() blocks the main thread until outThread is finished
@ -712,6 +721,12 @@ falling back to not using job objects for managing child processes"""
return self.proc.wait()
# TODO Remove this method when consumers have been fixed
def waitForFinish(self, timeout=None):
print >> sys.stderr, "MOZPROCESS WARNING: ProcessHandler.waitForFinish() is deprecated, " \
"use ProcessHandler.wait() instead"
return self.wait(timeout=timeout)
### Private methods from here on down. Thar be dragons.
@ -754,6 +769,10 @@ falling back to not using job objects for managing child processes"""
return ('', True)
return (f.readline(), False)
@property
def pid(self):
return self.proc.pid
### default output handlers
### these should be callables that take the output line

Просмотреть файл

@ -3,9 +3,9 @@
# You can obtain one at http://mozilla.org/MPL/2.0/.
import os
from setuptools import setup, find_packages
from setuptools import setup
PACKAGE_VERSION = '0.4'
PACKAGE_VERSION = '0.7'
# take description from README
here = os.path.dirname(os.path.abspath(__file__))
@ -28,10 +28,10 @@ setup(name='mozprocess',
],
keywords='mozilla',
author='Mozilla Automation and Tools team',
author_email='tools@lists.mozilla.com',
url='https://github.com/mozilla/mozbase/tree/master/mozprocess',
author_email='tools@lists.mozilla.org',
url='https://wiki.mozilla.org/Auto-tools/Projects/MozBase',
license='MPL 2.0',
packages=find_packages(exclude=['ez_setup', 'examples', 'tests']),
packages=['mozprocess'],
include_package_data=True,
zip_safe=False,
install_requires=['mozinfo'],

Просмотреть файл

@ -24,8 +24,8 @@ def make_proclaunch(aDir):
"""
# Ideally make should take care of this, but since it doesn't - on windows,
# anyway, let's just call out both targets explicitly.
p = subprocess.call(["make", "-C", "iniparser"], cwd=aDir)
p = subprocess.call(["make"], cwd=aDir)
p = subprocess.call(["make", "-C", "iniparser"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=aDir)
p = subprocess.call(["make"],stdout=subprocess.PIPE, stderr=subprocess.PIPE ,cwd=aDir)
if sys.platform == "win32":
exepath = os.path.join(aDir, "proclaunch.exe")
else:
@ -81,8 +81,7 @@ class ProcTest1(unittest.TestCase):
p = processhandler.ProcessHandler([self.proclaunch, "process_normal_finish.ini"],
cwd=here)
p.run()
p.processOutput()
p.waitForFinish()
p.wait()
detected, output = check_for_process(self.proclaunch)
self.determine_status(detected,
@ -96,9 +95,8 @@ class ProcTest1(unittest.TestCase):
"""
p = processhandler.ProcessHandler([self.proclaunch, "process_waittimeout.ini"],
cwd=here)
p.run()
p.processOutput(timeout=10)
p.waitForFinish()
p.run(timeout=10)
p.wait()
detected, output = check_for_process(self.proclaunch)
self.determine_status(detected,
@ -114,7 +112,6 @@ class ProcTest1(unittest.TestCase):
p = processhandler.ProcessHandler([self.proclaunch, "process_normal_finish.ini"],
cwd=here)
p.run()
p.processOutput()
p.kill()
detected, output = check_for_process(self.proclaunch)

Просмотреть файл

@ -29,8 +29,8 @@ def make_proclaunch(aDir):
"""
# Ideally make should take care of this, but since it doesn't - on windows,
# anyway, let's just call out both targets explicitly.
p = subprocess.call(["make", "-C", "iniparser"], cwd=aDir)
p = subprocess.call(["make"], cwd=aDir)
p = subprocess.call(["make", "-C", "iniparser"],stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=aDir)
p = subprocess.call(["make"],stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=aDir)
if sys.platform == "win32":
exepath = os.path.join(aDir, "proclaunch.exe")
else:
@ -85,9 +85,8 @@ class ProcTest2(unittest.TestCase):
p = processhandler.ProcessHandler([self.proclaunch,
"process_waittimeout_10s.ini"],
cwd=here)
p.run()
p.processOutput(timeout=30)
p.waitForFinish()
p.run(timeout=30)
p.wait()
detected, output = check_for_process(self.proclaunch)
self.determine_status(detected,
@ -103,7 +102,7 @@ class ProcTest2(unittest.TestCase):
"process_waittimeout_10s.ini"],
cwd=here)
p.run()
p.waitForFinish()
p.wait()
detected, output = check_for_process(self.proclaunch)
self.determine_status(detected,
@ -113,7 +112,7 @@ class ProcTest2(unittest.TestCase):
def test_process_waittimeout(self):
"""
Process is started, then waitForFinish is called and times out.
Process is started, then wait is called and times out.
Process is still running and didn't timeout
"""
p = processhandler.ProcessHandler([self.proclaunch,
@ -121,8 +120,7 @@ class ProcTest2(unittest.TestCase):
cwd=here)
p.run()
p.processOutput()
p.waitForFinish(timeout=5)
p.wait(timeout=5)
detected, output = check_for_process(self.proclaunch)
self.determine_status(detected,
@ -132,6 +130,27 @@ class ProcTest2(unittest.TestCase):
True,
[])
def test_process_output_twice(self):
"""
Process is started, then processOutput is called a second time explicitly
"""
p = processhandler.ProcessHandler([self.proclaunch,
"process_waittimeout_10s.ini"],
cwd=here)
p.run()
p.processOutput(timeout=5)
p.wait()
detected, output = check_for_process(self.proclaunch)
self.determine_status(detected,
output,
p.proc.returncode,
p.didTimeout,
False,
[])
def determine_status(self,
detected=False,
output = '',

Просмотреть файл

@ -231,9 +231,6 @@ class Permissions(object):
# Open database and create table
permDB = sqlite3.connect(os.path.join(self._profileDir, "permissions.sqlite"))
cursor = permDB.cursor();
cursor.execute("PRAGMA user_version=3");
# SQL copied from
# http://mxr.mozilla.org/mozilla-central/source/extensions/cookie/nsPermissionManager.cpp
cursor.execute("""CREATE TABLE IF NOT EXISTS moz_hosts (
@ -242,9 +239,7 @@ class Permissions(object):
type TEXT,
permission INTEGER,
expireType INTEGER,
expireTime INTEGER,
appId INTEGER,
isInBrowserElement INTEGER)""")
expireTime INTEGER)""")
for location in locations:
# set the permissions
@ -255,7 +250,7 @@ class Permissions(object):
permission_type = 1
else:
permission_type = 2
cursor.execute("INSERT INTO moz_hosts values(?, ?, ?, ?, 0, 0, 0, 0)",
cursor.execute("INSERT INTO moz_hosts values(?, ?, ?, ?, 0, 0)",
(self._num_permissions, location.host, perm,
permission_type))

Просмотреть файл

@ -4,7 +4,7 @@
import os
import sys
from setuptools import setup, find_packages
from setuptools import setup
PACKAGE_VERSION = '0.4'
@ -44,10 +44,10 @@ setup(name='mozprofile',
],
keywords='mozilla',
author='Mozilla Automation and Tools team',
author_email='tools@lists.mozilla.com',
url='https://github.com/mozilla/mozbase/tree/master/mozprofile',
author_email='tools@lists.mozilla.org',
url='https://wiki.mozilla.org/Auto-tools/Projects/MozBase',
license='MPL 2.0',
packages=find_packages(exclude=['ez_setup', 'examples', 'tests']),
packages=['mozprofile'],
include_package_data=True,
zip_safe=False,
install_requires=deps,

Просмотреть файл

@ -180,10 +180,7 @@ class Runner(object):
else:
# this run uses the managed processhandler
self.process_handler = self.process_class(cmd, env=self.env, **self.kp_kwargs)
self.process_handler.run()
# start processing output from the process
self.process_handler.processOutput(timeout, outputTimeout)
self.process_handler.run(timeout, outputTimeout)
def wait(self, timeout=None):
"""
@ -195,13 +192,15 @@ class Runner(object):
"""
if self.process_handler is None:
return
if isinstance(self.process_handler, subprocess.Popen):
self.process_handler.wait()
else:
self.process_handler.waitForFinish(timeout)
if not getattr(self.process_handler.proc, 'returncode', False):
self.process_handler.wait(timeout)
if self.process_handler.proc.poll() is None:
# waitForFinish timed out
return
self.process_handler = None
def stop(self):

Просмотреть файл

@ -4,10 +4,10 @@
import os
import sys
from setuptools import setup, find_packages
from setuptools import setup
PACKAGE_NAME = "mozrunner"
PACKAGE_VERSION = '5.8'
PACKAGE_VERSION = '5.12'
desc = """Reliable start/stop/configuration of Mozilla Applications (Firefox, Thunderbird, etc.)"""
# take description from README
@ -18,7 +18,7 @@ except (OSError, IOError):
description = ''
deps = ['mozinfo == 0.3.3',
'mozprocess == 0.4',
'mozprocess == 0.7',
'mozprofile == 0.4',
]
@ -39,10 +39,10 @@ setup(name=PACKAGE_NAME,
],
keywords='mozilla',
author='Mozilla Automation and Tools team',
author_email='tools@lists.mozilla.com',
url='https://github.com/mozilla/mozbase/tree/master/mozrunner',
author_email='tools@lists.mozilla.org',
url='https://wiki.mozilla.org/Auto-tools/Projects/MozBase',
license='MPL 2.0',
packages=find_packages(exclude=['legacy']),
packages=['mozrunner'],
zip_safe=False,
install_requires = deps,
entry_points="""

Просмотреть файл

@ -0,0 +1,16 @@
# Moztest
Package for handling Mozilla test results.
## Usage example
This shows how you can create an xUnit representation of python unittest results.
from results import TestResultCollection
from output import XUnitOutput
collection = TestResultCollection.from_unittest_results(results)
out = XUnitOutput()
with open('out.xml', 'w') as f:
out.serialize(collection, f)

Просмотреть файл

@ -0,0 +1,73 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
from mozautolog import RESTfulAutologTestGroup
from base import Output, count, long_name
class AutologOutput(Output):
def __init__(self, es_server='buildbot-es.metrics.scl3.mozilla.com:9200',
rest_server='http://brasstacks.mozilla.com/autologserver',
name='moztest',
harness='moztest'):
self.es_server = es_server
self.rest_server = rest_server
def serialize(self, results_collection, file_obj):
grps = self.make_testgroups(results_collection)
for g in grps:
file_obj.write(g.serialize())
def make_testgroups(self, results_collection):
testgroups = []
for context in results_collection.contexts:
coll = results_collection.subset(lambda t: t.context == context)
passed = coll.tests_with_result('PASS')
failed = coll.tests_with_result('UNEXPECTED-FAIL')
unexpected_passes = coll.tests_with_result('UNEXPECTED-PASS')
errors = coll.tests_with_result('ERROR')
skipped = coll.tests_with_result('SKIPPED')
known_fails = coll.tests_with_result('KNOWN-FAIL')
testgroup = RESTfulAutologTestGroup(
testgroup=context.testgroup,
os=context.os,
platform=context.arch,
harness=context.harness,
server=self.es_server,
restserver=self.rest_server,
machine=context.hostname,
logfile=context.logfile,
)
testgroup.add_test_suite(
testsuite=results_collection.suite_name,
elapsedtime=coll.time_taken,
passed=count(passed),
failed=count(failed) + count(errors) + count(unexpected_passes),
todo=count(skipped) + count(known_fails),
)
testgroup.set_primary_product(
tree=context.tree,
revision=context.revision,
productname=context.product,
buildtype=context.buildtype,
)
# need to call this again since we already used the generator
for f in coll.tests_with_result('UNEXPECTED-FAIL'):
testgroup.add_test_failure(
test=long_name(f),
text='\n'.join(f.output),
status=f.result,
)
testgroups.append(testgroup)
return testgroups
def post(self, data):
msg = "Must pass in a list returned by make_testgroups."
for d in data:
assert isinstance(d, RESTfulAutologTestGroup), msg
d.submit()

Просмотреть файл

@ -0,0 +1,50 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
from __future__ import with_statement
from contextlib import closing
from StringIO import StringIO
try:
from abc import abstractmethod
except ImportError:
# abc is python 2.6+
# from https://github.com/mozilla/mozbase/blob/master/mozdevice/mozdevice/devicemanager.py
def abstractmethod(method):
line = method.func_code.co_firstlineno
filename = method.func_code.co_filename
def not_implemented(*args, **kwargs):
raise NotImplementedError('Abstract method %s at File "%s", line %s should be implemented by a concrete class' %
(repr(method), filename,line))
return not_implemented
class Output(object):
""" Abstract base class for outputting test results """
@abstractmethod
def serialize(self, results_collection, file_obj):
""" Writes the string representation of the results collection
to the given file object"""
def dump_string(self, results_collection):
""" Returns the string representation of the results collection """
with closing(StringIO()) as s:
self.serialize(results_collection, s)
return s.getvalue()
# helper functions
def count(iterable):
""" Return the count of an iterable. Useful for generators. """
c = 0
for i in iterable:
c += 1
return c
def long_name(test):
if test.test_class:
return '%s.%s' % (test.test_class, test.name)
return test.name

Просмотреть файл

@ -0,0 +1,93 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
import xml.dom.minidom as dom
from base import Output, count
from moztest.results import TestResult
class XUnitOutput(Output):
""" Class for writing xUnit formatted test results in an XML file """
def serialize(self, results_collection, file_obj):
""" Writes the xUnit formatted results to the given file object """
def _extract_xml(test_result, text='', result='Pass'):
if not isinstance(text, basestring):
text = '\n'.join(text)
cls_name = test_result.test_class
# if the test class is not already created, create it
if cls_name not in classes:
cls = doc.createElement('class')
cls.setAttribute('name', cls_name)
assembly.appendChild(cls)
classes[cls_name] = cls
t = doc.createElement('test')
t.setAttribute('name', test_result.name)
t.setAttribute('result', result)
if result == 'Fail':
f = doc.createElement('failure')
st = doc.createElement('stack-trace')
st.appendChild(doc.createTextNode(text))
f.appendChild(st)
t.appendChild(f)
elif result == 'Skip':
r = doc.createElement('reason')
msg = doc.createElement('message')
msg.appendChild(doc.createTextNode(text))
r.appendChild(msg)
t.appendChild(f)
cls = classes[cls_name]
cls.appendChild(t)
doc = dom.Document()
failed = sum([count(results_collection.tests_with_result(t))
for t in TestResult.FAIL_RESULTS])
passed = count(results_collection.tests_with_result('PASS'))
skipped = count(results_collection.tests_with_result('SKIPPED'))
assembly = doc.createElement('assembly')
assembly.setAttribute('name', results_collection.suite_name)
assembly.setAttribute('time', str(results_collection.time_taken))
assembly.setAttribute('total', str(len(results_collection)))
assembly.setAttribute('passed', str(passed))
assembly.setAttribute('failed', str(failed))
assembly.setAttribute('skipped', str(skipped))
classes = {} # str -> xml class element
for tr in results_collection.tests_with_result('ERROR'):
_extract_xml(tr, text=tr.output, result='Fail')
for tr in results_collection.tests_with_result('UNEXPECTED-FAIL'):
_extract_xml(tr, text=tr.output, result='Fail')
for tr in results_collection.tests_with_result('UNEXPECTED-PASS'):
_extract_xml(tr, text='UNEXPECTED-PASS', result='Fail')
for tr in results_collection.tests_with_result('SKIPPED'):
_extract_xml(tr, text=tr.output, result='Skip')
for tr in results_collection.tests_with_result('KNOWN-FAIL'):
_extract_xml(tr, text=tr.output, result='Pass')
for tr in results_collection.tests_with_result('PASS'):
_extract_xml(tr, result='Pass')
for cls in classes.itervalues():
assembly.appendChild(cls)
doc.appendChild(assembly)
file_obj.write(doc.toxml(encoding='utf-8'))

Просмотреть файл

@ -0,0 +1,314 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
import time
import os
import mozinfo
class TestContext(object):
""" Stores context data about the test """
attrs = ['hostname', 'arch', 'env', 'os', 'os_version', 'tree', 'revision',
'product', 'logfile', 'testgroup', 'harness', 'buildtype']
def __init__(self, hostname='localhost', tree='', revision='', product='',
logfile=None, arch='', operating_system='', testgroup='',
harness='moztest', buildtype=''):
self.hostname = hostname
self.arch = arch or mozinfo.processor
self.env = os.environ.copy()
self.os = operating_system or mozinfo.os
self.os_version = mozinfo.version
self.tree = tree
self.revision = revision
self.product = product
self.logfile = logfile
self.testgroup = testgroup
self.harness = harness
self.buildtype = buildtype
def __str__(self):
return '%s (%s, %s)' % (self.hostname, self.os, self.arch)
def __repr__(self):
return '<%s>' % self.__str__()
def __eq__(self, other):
if not isinstance(other, TestContext):
return False
diffs = [a for a in self.attrs if getattr(self, a) != getattr(other, a)]
return len(diffs) == 0
def __hash__(self):
def get(attr):
value = getattr(self, attr)
if isinstance(value, dict):
value = frozenset(value.items())
return value
return hash(frozenset([get(a) for a in self.attrs]))
class TestResult(object):
""" Stores test result data """
FAIL_RESULTS = [
'UNEXPECTED-PASS',
'UNEXPECTED-FAIL',
'ERROR',
]
COMPUTED_RESULTS = FAIL_RESULTS + [
'PASS',
'KNOWN-FAIL',
'SKIPPED',
]
POSSIBLE_RESULTS = [
'PASS',
'FAIL',
'SKIP',
'ERROR',
]
def __init__(self, name, test_class='', time_start=None, context=None,
result_expected='PASS'):
""" Create a TestResult instance.
name = name of the test that is running
test_class = the class that the test belongs to
time_start = timestamp (seconds since UNIX epoch) of when the test started
running; if not provided, defaults to the current time
! Provide 0 if you only have the duration
context = TestContext instance; can be None
result_expected = string representing the expected outcome of the test"""
msg = "Result '%s' not in possible results: %s" %\
(result_expected, ', '.join(self.POSSIBLE_RESULTS))
assert isinstance(name, basestring), "name has to be a string"
assert result_expected in self.POSSIBLE_RESULTS, msg
self.name = name
self.test_class = test_class
self.context = context
self.time_start = time_start if time_start is not None else time.time()
self.time_end = None
self._result_expected = result_expected
self._result_actual = None
self.result = None
self.filename = None
self.description = None
self.output = []
self.reason = None
def __str__(self):
return '%s | %s (%s) | %s' % (self.result or 'PENDING',
self.name, self.test_class, self.reason)
def __repr__(self):
return '<%s>' % self.__str__()
def calculate_result(self, expected, actual):
if actual == 'ERROR':
return 'ERROR'
if actual == 'SKIP':
return 'SKIPPED'
if expected == 'PASS':
if actual == 'PASS':
return 'PASS'
if actual == 'FAIL':
return 'UNEXPECTED-FAIL'
if expected == 'FAIL':
if actual == 'PASS':
return 'UNEXPECTED-PASS'
if actual == 'FAIL':
return 'KNOWN-FAIL'
# if actual is skip or error, we return at the beginning, so if we get
# here it is definitely some kind of error
return 'ERROR'
def infer_results(self, computed_result):
assert computed_result in self.COMPUTED_RESULTS
if computed_result == 'UNEXPECTED-PASS':
expected = 'FAIL'
actual = 'PASS'
elif computed_result == 'UNEXPECTED-FAIL':
expected = 'PASS'
actual = 'FAIL'
elif computed_result == 'KNOWN-FAIL':
expected = actual = 'FAIL'
elif computed_result == 'SKIPPED':
expected = actual = 'SKIP'
else:
return
self._result_expected = expected
self._result_actual = actual
def finish(self, result, time_end=None, output=None, reason=None):
""" Marks the test as finished, storing its end time and status
! Provide the duration as time_end if you only have that. """
if result in self.POSSIBLE_RESULTS:
self._result_actual = result
self.result = self.calculate_result(self._result_expected,
self._result_actual)
elif result in self.COMPUTED_RESULTS:
self.infer_results(result)
self.result = result
else:
valid = self.POSSIBLE_RESULTS + self.COMPUTED_RESULTS
msg = "Result '%s' not valid. Need one of: %s" %\
(result, ', '.join(valid))
raise ValueError(msg)
# use lists instead of multiline strings
if isinstance(output, basestring):
output = output.splitlines()
self.time_end = time_end if time_end is not None else time.time()
self.output = output or self.output
self.reason = reason
@property
def finished(self):
""" Boolean saying if the test is finished or not """
return self.result is not None
@property
def duration(self):
""" Returns the time it took for the test to finish. If the test is
not finished, returns the elapsed time so far """
if self.result is not None:
return self.time_end - self.time_start
else:
# returns the elapsed time
return time.time() - self.time_start
class TestResultCollection(list):
""" Container class that stores test results """
def __init__(self, suite_name, time_taken=0):
list.__init__(self)
self.suite_name = suite_name
self.time_taken = time_taken
def __str__(self):
return "%s (%.2fs)\n%s" % (self.suite_name, self.time_taken,
list.__str__(self))
def subset(self, predicate):
tests = self.filter(predicate)
duration = 0
sub = TestResultCollection(self.suite_name)
for t in tests:
sub.append(t)
duration += t.duration
sub.time_taken = duration
return sub
@property
def contexts(self):
""" List of unique contexts for the test results contained """
cs = [tr.context for tr in self]
return list(set(cs))
def filter(self, predicate):
""" Returns a generator of TestResults that satisfy a given predicate """
return (tr for tr in self if predicate(tr))
def tests_with_result(self, result):
""" Returns a generator of TestResults with the given result """
msg = "Result '%s' not in possible results: %s" %\
(result, ', '.join(TestResult.COMPUTED_RESULTS))
assert result in TestResult.COMPUTED_RESULTS, msg
return self.filter(lambda t: t.result == result)
@property
def tests(self):
""" Generator of all tests in the collection """
return (t for t in self)
@property
def num_failures(self):
fails = 0
for t in self:
if t.result in TestResult.FAIL_RESULTS:
fails += 1
return fails
def add_unittest_result(self, result, context=None):
""" Adds the python unittest result provided to the collection"""
def get_class(test):
return test.__class__.__module__ + '.' + test.__class__.__name__
def add_test_result(test, result_expected='PASS',
result_actual='PASS', output=''):
t = TestResult(name=str(test).split()[0], test_class=get_class(test),
time_start=0, result_expected=result_expected,
context=context)
t.finish(result_actual, time_end=0, reason=relevant_line(output),
output=output)
self.append(t)
if hasattr(result, 'time_taken'):
self.time_taken += result.time_taken
for test, output in result.errors:
add_test_result(test, result_actual='ERROR', output=output)
for test, output in result.failures:
add_test_result(test, result_actual='FAIL',
output=output)
if hasattr(result, 'unexpectedSuccesses'):
for test in result.unexpectedSuccesses:
add_test_result(test, result_expected='FAIL',
result_actual='PASS')
if hasattr(result, 'skipped'):
for test, output in result.skipped:
add_test_result(test, result_expected='SKIP',
result_actual='SKIP', output=output)
if hasattr(result, 'expectedFailures'):
for test, output in result.expectedFailures:
add_test_result(test, result_expected='FAIL',
result_actual='FAIL', output=output)
# unittest does not store these by default
if hasattr(result, 'tests_passed'):
for test in result.tests_passed:
add_test_result(test)
@classmethod
def from_unittest_results(cls, context, *results):
""" Creates a TestResultCollection containing the given python
unittest results """
if not results:
return cls('from unittest')
# all the TestResult instances share the same context
context = context or TestContext()
collection = cls('from %s' % results[0].__class__.__name__)
for result in results:
collection.add_unittest_result(result, context)
return collection
# used to get exceptions/errors from tracebacks
def relevant_line(s):
KEYWORDS = ('Error:', 'Exception:', 'error:', 'exception:')
lines = s.splitlines()
for line in lines:
for keyword in KEYWORDS:
if keyword in line:
return line
return 'N/A'

Просмотреть файл

@ -0,0 +1,39 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
import os
from setuptools import setup
PACKAGE_VERSION = '0.1'
# get documentation from the README
try:
here = os.path.dirname(os.path.abspath(__file__))
description = file(os.path.join(here, 'README.md')).read()
except (OSError, IOError):
description = ''
# dependencies
deps = ['mozinfo']
try:
import json
except ImportError:
deps.append('simplejson')
setup(name='moztest',
version=PACKAGE_VERSION,
description="Package for storing and outputting Mozilla test results",
long_description=description,
classifiers=[], # Get strings from http://pypi.python.org/pypi?%3Aaction=list_classifiers
keywords='mozilla',
author='Mozilla Automation and Tools team',
author_email='tools@lists.mozilla.org',
url='https://wiki.mozilla.org/Auto-tools/Projects/MozBase',
license='MPL',
packages=['moztest'],
include_package_data=True,
zip_safe=False,
install_requires=deps,
)

Просмотреть файл

@ -0,0 +1 @@
[test.py]

Просмотреть файл

@ -0,0 +1,55 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
import math
import time
import unittest
from moztest.results import TestContext, TestResult, TestResultCollection
class Result(unittest.TestCase):
def test_results(self):
self.assertRaises(AssertionError,
lambda: TestResult('test', result_expected='hello'))
t = TestResult('test')
self.assertRaises(ValueError, lambda: t.finish(result='good bye'))
def test_time(self):
now = time.time()
t = TestResult('test')
time.sleep(1)
t.finish('PASS')
duration = time.time() - now
self.assertTrue(math.fabs(duration - t.duration) < 1)
def test_custom_time(self):
t = TestResult('test', time_start=0)
t.finish(result='PASS', time_end=1000)
self.assertEqual(t.duration, 1000)
class Collection(unittest.TestCase):
def setUp(self):
c1 = TestContext('host1')
c2 = TestContext('host2')
c3 = TestContext('host2')
c3.os = 'B2G'
c4 = TestContext('host1')
t1 = TestResult('t1', context=c1)
t2 = TestResult('t2', context=c2)
t3 = TestResult('t3', context=c3)
t4 = TestResult('t4', context=c4)
self.collection = TestResultCollection('tests')
self.collection.extend([t1, t2, t3, t4])
def test_unique_contexts(self):
self.assertEqual(len(self.collection.contexts), 3)
if __name__ == '__main__':
unittest.main()

Просмотреть файл

@ -13,3 +13,5 @@
[include:mozprofile/tests/manifest.ini]
[include:mozhttpd/tests/manifest.ini]
[include:mozdevice/tests/manifest.ini]
[include:moztest/tests/manifest.ini]
[include:mozcrash/tests/manifest.ini]

Просмотреть файл

@ -15,6 +15,8 @@ import os
import sys
import unittest
from moztest.results import TestResultCollection
here = os.path.dirname(os.path.abspath(__file__))
def unittests(path):
@ -57,11 +59,11 @@ def main(args=sys.argv[1:]):
# run the tests
suite = unittest.TestSuite(unittestlist)
runner = unittest.TextTestRunner()
results = runner.run(suite)
runner = unittest.TextTestRunner(verbosity=2) # default=1 does not show success of unittests
results = TestResultCollection.from_unittest_results(runner.run(suite))
# exit according to results
sys.exit((results.failures or results.errors) and 1 or 0)
sys.exit(1 if results.num_failures else 0)
if __name__ == '__main__':
main()