import aexpect import glob import json import os import re import shutil import signal import sys import tempfile import time import xml.dom.minidom import zipfile import unittest import psutil from io import BytesIO try: from lxml import etree SCHEMA_CAPABLE = True except ImportError: SCHEMA_CAPABLE = False from avocado.core import exit_codes from avocado.utils import astring from avocado.utils import genio from avocado.utils import process from avocado.utils import script from avocado.utils import path as utils_path from .. import AVOCADO, BASEDIR, python_module_available LOCAL_IMPORT_TEST_CONTENTS = ''' from avocado import Test from mylib import hello class LocalImportTest(Test): def test(self): self.log.info(hello()) ''' UNSUPPORTED_STATUS_TEST_CONTENTS = ''' from avocado import Test class FakeStatusTest(Test): def run_avocado(self): super(FakeStatusTest, self).run_avocado() # Please do NOT ever use this, it's for unittesting only. self._Test__status = 'not supported' def test(self): pass ''' INVALID_PYTHON_TEST = ''' from avocado import Test class MyTest(Test): non_existing_variable_causing_crash def test_my_name(self): pass ''' VALID_PYTHON_TEST_WITH_TAGS = ''' from avocado import Test class MyTest(Test): def test(self): """ :avocado: tags=BIG_TAG_NAME """ pass ''' REPORTS_STATUS_AND_HANG = ''' from avocado import Test import time class MyTest(Test): def test(self): self.runner_queue.put({"running": False}) time.sleep(70) ''' DIE_WITHOUT_REPORTING_STATUS = ''' from avocado import Test import os import signal class MyTest(Test): def test(self): os.kill(os.getpid(), signal.SIGKILL) ''' RAISE_CUSTOM_PATH_EXCEPTION_CONTENT = '''import os import sys from avocado import Test class SharedLibTest(Test): def test(self): sys.path.append(os.path.join(os.path.dirname(__file__), "shared_lib")) from mylib import CancelExc raise CancelExc("This should not crash on unpickling in runner") ''' TEST_OTHER_LOGGERS_CONTENT = ''' import logging from avocado import Test class My(Test): def test(self): logging.getLogger("some.other.logger").info("SHOULD BE ON debug.log") ''' def probe_binary(binary): try: return utils_path.find_command(binary) except utils_path.CmdNotFoundError: return None TRUE_CMD = probe_binary('true') CC_BINARY = probe_binary('cc') # On macOS, the default GNU core-utils installation (brew) # installs the gnu utility versions with a g prefix. It still has the # BSD versions of the core utilities installed on their expected paths # but their behavior and flags are in most cases different. GNU_ECHO_BINARY = probe_binary('echo') if GNU_ECHO_BINARY is not None: if probe_binary('man') is not None: echo_cmd = 'man %s' % os.path.basename(GNU_ECHO_BINARY) echo_manpage = process.run(echo_cmd, env={'LANG': 'C'}, encoding='ascii').stdout if b'-e' not in echo_manpage: GNU_ECHO_BINARY = probe_binary('gecho') READ_BINARY = probe_binary('read') SLEEP_BINARY = probe_binary('sleep') class RunnerOperationTest(unittest.TestCase): def setUp(self): self.tmpdir = tempfile.mkdtemp(prefix='avocado_' + __name__) os.chdir(BASEDIR) def test_show_version(self): result = process.run('%s -v' % AVOCADO, ignore_status=True) self.assertEqual(result.exit_status, 0) if sys.version_info[0] == 3: content = result.stdout_text else: content = result.stderr_text self.assertTrue(re.match(r"^Avocado \d+\.\d+$", content), "Version string does not match 'Avocado \\d\\.\\d:'\n" "%r" % (content)) def test_alternate_config_datadir(self): """ Uses the "--config" flag to check custom configuration is applied Even on the more complex data_dir module, which adds extra checks to what is set on the plain settings module. """ base_dir = os.path.join(self.tmpdir, 'datadir_base') os.mkdir(base_dir) mapping = {'base_dir': base_dir, 'test_dir': os.path.join(base_dir, 'test'), 'data_dir': os.path.join(base_dir, 'data'), 'logs_dir': os.path.join(base_dir, 'logs')} config = '[datadir.paths]\n' for key, value in mapping.items(): if not os.path.isdir(value): os.mkdir(value) config += "%s = %s\n" % (key, value) fd, config_file = tempfile.mkstemp(dir=self.tmpdir) os.write(fd, config.encode()) os.close(fd) cmd = '%s --config %s config --datadir' % (AVOCADO, config_file) result = process.run(cmd) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) self.assertIn(' base ' + mapping['base_dir'], result.stdout_text) self.assertIn(' data ' + mapping['data_dir'], result.stdout_text) self.assertIn(' logs ' + mapping['logs_dir'], result.stdout_text) def test_runner_all_ok(self): cmd_line = ('%s run --sysinfo=off --job-results-dir %s ' 'passtest.py passtest.py' % (AVOCADO, self.tmpdir)) process.run(cmd_line) # Also check whether jobdata contains correct parameter paths variants = open(os.path.join(self.tmpdir, "latest", "jobdata", "variants.json")).read() self.assertIn('["/run/*"]', variants, "paths stored in jobdata " "does not contains [\"/run/*\"]\n%s" % variants) def test_runner_failfast(self): cmd_line = ('%s run --sysinfo=off --job-results-dir %s ' 'passtest.py failtest.py passtest.py --failfast on' % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) self.assertIn(b'Interrupting job (failfast).', result.stdout) self.assertIn(b'PASS 1 | ERROR 0 | FAIL 1 | SKIP 1', result.stdout) expected_rc = exit_codes.AVOCADO_TESTS_FAIL | exit_codes.AVOCADO_JOB_INTERRUPTED self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) def test_runner_ignore_missing_references_one_missing(self): cmd_line = ('%s run --sysinfo=off --job-results-dir %s ' 'passtest.py badtest.py --ignore-missing-references on' % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) self.assertIn(b"Unable to resolve reference(s) 'badtest.py'", result.stderr) self.assertIn(b'PASS 1 | ERROR 0 | FAIL 0 | SKIP 0', result.stdout) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) def test_runner_ignore_missing_references_all_missing(self): cmd_line = ('%s run --sysinfo=off --job-results-dir %s ' 'badtest.py badtest2.py --ignore-missing-references on' % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) self.assertIn(b"Unable to resolve reference(s) 'badtest.py', 'badtest2.py'", result.stderr) self.assertEqual(b'', result.stdout) expected_rc = exit_codes.AVOCADO_JOB_FAIL self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) def test_runner_test_with_local_imports(self): mylib = script.TemporaryScript( 'mylib.py', "def hello():\n return 'Hello world'", 'avocado_simpletest_functional') mylib.save() mytest = script.Script( os.path.join(os.path.dirname(mylib.path), 'test_local_imports.py'), LOCAL_IMPORT_TEST_CONTENTS) mytest.save() cmd_line = ("%s run --sysinfo=off --job-results-dir %s " "%s" % (AVOCADO, self.tmpdir, mytest)) process.run(cmd_line) def test_unsupported_status(self): with script.TemporaryScript("fake_status.py", UNSUPPORTED_STATUS_TEST_CONTENTS, "avocado_unsupported_status") as tst: res = process.run("%s run --sysinfo=off --job-results-dir %s %s" " --json -" % (AVOCADO, self.tmpdir, tst), ignore_status=True) self.assertEqual(res.exit_status, exit_codes.AVOCADO_TESTS_FAIL) results = json.loads(res.stdout_text) self.assertEqual(results["tests"][0]["status"], "ERROR", "%s != %s\n%s" % (results["tests"][0]["status"], "ERROR", res)) self.assertIn("Runner error occurred: Test reports unsupported", results["tests"][0]["fail_reason"]) @unittest.skipIf(int(os.environ.get("AVOCADO_CHECK_LEVEL", 0)) < 1, "Skipping test that take a long time to run, are " "resource intensive or time sensitve") def test_hanged_test_with_status(self): """ Check that avocado handles hanged tests properly """ with script.TemporaryScript("report_status_and_hang.py", REPORTS_STATUS_AND_HANG, "hanged_test_with_status") as tst: res = process.run("%s run --sysinfo=off --job-results-dir %s %s " "--json - --job-timeout 1" % (AVOCADO, self.tmpdir, tst), ignore_status=True) self.assertEqual(res.exit_status, exit_codes.AVOCADO_TESTS_FAIL) results = json.loads(res.stdout_text) self.assertEqual(results["tests"][0]["status"], "ERROR", "%s != %s\n%s" % (results["tests"][0]["status"], "ERROR", res)) self.assertIn("Test reported status but did not finish", results["tests"][0]["fail_reason"]) # Currently it should finish up to 1s after the job-timeout # but the prep and postprocess could take a bit longer on # some environments, so let's just check it does not take # > 60s, which is the deadline for force-finishing the test. self.assertLess(res.duration, 55, "Test execution took too long, " "which is likely because the hanged test was not " "interrupted. Results:\n%s" % res) def test_no_status_reported(self): with script.TemporaryScript("die_without_reporting_status.py", DIE_WITHOUT_REPORTING_STATUS, "no_status_reported") as tst: res = process.run("%s run --sysinfo=off --job-results-dir %s %s " "--json -" % (AVOCADO, self.tmpdir, tst), ignore_status=True) self.assertEqual(res.exit_status, exit_codes.AVOCADO_TESTS_FAIL) results = json.loads(res.stdout_text) self.assertEqual(results["tests"][0]["status"], "ERROR", "%s != %s\n%s" % (results["tests"][0]["status"], "ERROR", res)) self.assertIn("Test died without reporting the status", results["tests"][0]["fail_reason"]) def test_runner_tests_fail(self): cmd_line = ('%s run --sysinfo=off --job-results-dir %s passtest.py ' 'failtest.py passtest.py' % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_TESTS_FAIL self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) def test_runner_nonexistent_test(self): cmd_line = ('%s run --sysinfo=off --job-results-dir ' '%s bogustest' % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_JOB_FAIL unexpected_rc = exit_codes.AVOCADO_FAIL self.assertNotEqual(result.exit_status, unexpected_rc, "Avocado crashed (rc %d):\n%s" % (unexpected_rc, result)) self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) def test_runner_doublefail(self): cmd_line = ('%s run --sysinfo=off --job-results-dir %s ' '--xunit - doublefail.py' % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_TESTS_FAIL unexpected_rc = exit_codes.AVOCADO_FAIL self.assertNotEqual(result.exit_status, unexpected_rc, "Avocado crashed (rc %d):\n%s" % (unexpected_rc, result)) self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) self.assertIn(b"TestError: Failing during tearDown. Yay!", result.stdout, "Cleanup exception not printed to log output") self.assertIn(b"TestFail: This test is supposed to fail", result.stdout, "Test did not fail with action exception:\n%s" % result.stdout) def test_uncaught_exception(self): cmd_line = ("%s run --sysinfo=off --job-results-dir %s " "--json - uncaught_exception.py" % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_TESTS_FAIL self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) self.assertIn(b'"status": "ERROR"', result.stdout) def test_fail_on_exception(self): cmd_line = ("%s run --sysinfo=off --job-results-dir %s " "--json - fail_on_exception.py" % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_TESTS_FAIL self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) self.assertIn(b'"status": "FAIL"', result.stdout) def test_assert_raises(self): cmd_line = ("%s run --sysinfo=off --job-results-dir %s " "-- assert.py" % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_TESTS_FAIL self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) self.assertIn(b'Assert.test_assert_raises: PASS', result.stdout) self.assertIn(b'Assert.test_fails_to_raise: FAIL', result.stdout) self.assertIn(b'PASS 1 | ERROR 0 | FAIL 1 ', result.stdout) def test_exception_not_in_path(self): os.mkdir(os.path.join(self.tmpdir, "shared_lib")) mylib = script.Script(os.path.join(self.tmpdir, "shared_lib", "mylib.py"), "from avocado import TestCancel\n\n" "class CancelExc(TestCancel):\n" " pass") mylib.save() mytest = script.Script(os.path.join(self.tmpdir, "mytest.py"), RAISE_CUSTOM_PATH_EXCEPTION_CONTENT) mytest.save() result = process.run("%s --show test run --sysinfo=off " "--job-results-dir %s %s" % (AVOCADO, self.tmpdir, mytest)) self.assertIn(b"mytest.py:SharedLibTest.test -> CancelExc: This " b"should not crash on unpickling in runner", result.stdout) self.assertNotIn(b"Failed to read queue", result.stdout) def test_runner_timeout(self): cmd_line = ('%s run --sysinfo=off --job-results-dir %s ' '--xunit - timeouttest.py' % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) output = result.stdout expected_rc = exit_codes.AVOCADO_JOB_INTERRUPTED unexpected_rc = exit_codes.AVOCADO_FAIL self.assertNotEqual(result.exit_status, unexpected_rc, "Avocado crashed (rc %d):\n%s" % (unexpected_rc, result)) self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) self.assertIn(b"Runner error occurred: Timeout reached", output, "Timeout reached message not found in the output:\n%s" % output) # Ensure no test aborted error messages show up self.assertNotIn(b"TestAbortError: Test aborted unexpectedly", output) @unittest.skipIf(int(os.environ.get("AVOCADO_CHECK_LEVEL", 0)) < 2, "Skipping test that take a long time to run, are " "resource intensive or time sensitve") def test_runner_abort(self): cmd_line = ('%s run --sysinfo=off --job-results-dir %s ' '--xunit - abort.py' % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) excerpt = b'Test died without reporting the status.' expected_rc = exit_codes.AVOCADO_TESTS_FAIL unexpected_rc = exit_codes.AVOCADO_FAIL self.assertNotEqual(result.exit_status, unexpected_rc, "Avocado crashed (rc %d):\n%s" % (unexpected_rc, result)) self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) self.assertIn(excerpt, result.stdout) def test_silent_output(self): cmd_line = ('%s --silent run --sysinfo=off --job-results-dir %s ' 'passtest.py' % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) self.assertEqual(result.exit_status, exit_codes.AVOCADO_ALL_OK) self.assertEqual(result.stdout, b'') def test_empty_args_list(self): cmd_line = AVOCADO result = process.run(cmd_line, ignore_status=True) self.assertEqual(result.exit_status, exit_codes.AVOCADO_FAIL) if sys.version_info[0] == 3: exp = b'avocado: error: the following arguments are required' else: exp = b'error: too few arguments' self.assertIn(exp, result.stderr) def test_empty_test_list(self): cmd_line = '%s run --sysinfo=off --job-results-dir %s' % (AVOCADO, self.tmpdir) result = process.run(cmd_line, ignore_status=True) self.assertEqual(result.exit_status, exit_codes.AVOCADO_JOB_FAIL) self.assertIn(b'No test references provided nor any other arguments ' b'resolved into tests', result.stderr) def test_not_found(self): cmd_line = ('%s run --sysinfo=off --job-results-dir %s sbrubles' % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) self.assertEqual(result.exit_status, exit_codes.AVOCADO_JOB_FAIL) self.assertIn(b'Unable to resolve reference', result.stderr) self.assertNotIn(b'Unable to resolve reference', result.stdout) def test_invalid_unique_id(self): cmd_line = ('%s run --sysinfo=off --job-results-dir %s --force-job-id ' 'foobar passtest.py' % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) self.assertNotEqual(result.exit_status, exit_codes.AVOCADO_ALL_OK) self.assertIn(b'needs to be a 40 digit hex', result.stderr) self.assertNotIn(b'needs to be a 40 digit hex', result.stdout) def test_valid_unique_id(self): cmd_line = ('%s run --job-results-dir %s --sysinfo=off ' '--force-job-id 975de258ac05ce5e490648dec4753657b7ccc7d1 ' 'passtest.py' % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) self.assertEqual(result.exit_status, exit_codes.AVOCADO_ALL_OK) self.assertNotIn(b'needs to be a 40 digit hex', result.stderr) self.assertIn(b'PASS', result.stdout) def test_automatic_unique_id(self): cmd_line = ('%s run --job-results-dir %s --sysinfo=off ' 'passtest.py --json -' % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) self.assertEqual(result.exit_status, exit_codes.AVOCADO_ALL_OK) r = json.loads(result.stdout_text) int(r['job_id'], 16) # it's an hex number self.assertEqual(len(r['job_id']), 40) @unittest.skipIf(int(os.environ.get("AVOCADO_CHECK_LEVEL", 0)) < 2, "Skipping test that take a long time to run, are " "resource intensive or time sensitve") def test_early_latest_result(self): """ Tests that the `latest` link to the latest job results is created early """ cmd_line = ('%s run --sysinfo=off --job-results-dir %s ' 'examples/tests/passtest.py' % (AVOCADO, self.tmpdir)) avocado_process = process.SubProcess(cmd_line) try: avocado_process.start() link = os.path.join(self.tmpdir, 'latest') for _ in range(0, 50): time.sleep(0.1) if os.path.exists(link) and os.path.islink(link): avocado_process.wait() break self.assertTrue(os.path.exists(link)) self.assertTrue(os.path.islink(link)) finally: avocado_process.wait() def test_dry_run(self): cmd = ("%s run --sysinfo=off --dry-run --dry-run-no-cleanup --json - " "--mux-inject foo:1 bar:2 baz:3 foo:foo:a " "foo:bar:b foo:baz:c bar:bar:bar " "-- passtest.py failtest.py gendata.py " % AVOCADO) result = json.loads(process.run(cmd).stdout_text) debuglog = result['debuglog'] log = genio.read_file(debuglog) # Remove the result dir shutil.rmtree(os.path.dirname(os.path.dirname(debuglog))) self.assertIn(tempfile.gettempdir(), debuglog) # Use tmp dir, not default location self.assertEqual(result['job_id'], u'0' * 40) # Check if all tests were skipped self.assertEqual(result['cancel'], 4) for i in range(4): test = result['tests'][i] self.assertEqual(test['fail_reason'], u'Test cancelled due to --dry-run') # Check if all params are listed # The "/:bar ==> 2 is in the tree, but not in any leave so inaccessible # from test. for line in ("/:foo ==> 1", "/:baz ==> 3", "/foo:foo ==> a", "/foo:bar ==> b", "/foo:baz ==> c", "/bar:bar ==> bar"): self.assertEqual(log.count(line), 4, "Avocado log count for param '%s' not as expected:\n%s" % (line, log)) def test_invalid_python(self): test = script.make_script(os.path.join(self.tmpdir, 'test.py'), INVALID_PYTHON_TEST) cmd_line = ('%s --show test run --sysinfo=off ' '--job-results-dir %s %s') % (AVOCADO, self.tmpdir, test) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_TESTS_FAIL self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) self.assertIn('1-%s:MyTest.test_my_name -> TestError' % test, result.stdout_text) @unittest.skipIf(not READ_BINARY, "read binary not available.") @unittest.skipIf(int(os.environ.get("AVOCADO_CHECK_LEVEL", 0)) < 1, "Skipping test that take a long time to run, are " "resource intensive or time sensitve") def test_read(self): cmd = "%s run --sysinfo=off --job-results-dir %%s %%s" % AVOCADO cmd %= (self.tmpdir, READ_BINARY) result = process.run(cmd, timeout=10, ignore_status=True) self.assertLess(result.duration, 8, "Duration longer than expected." "\n%s" % result) self.assertEqual(result.exit_status, 1, "Expected exit status is 1\n%s" % result) def test_runner_test_parameters(self): cmd_line = ('%s --show=test run --sysinfo=off --job-results-dir %s ' '-p "sleep_length=0.01" -- sleeptest.py ' % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) self.assertIn(b"PARAMS (key=sleep_length, path=*, default=1) => '0.01'", result.stdout) self.assertIn(b"Sleeping for 0.01 seconds", result.stdout) def test_other_loggers(self): with script.TemporaryScript( 'mytest.py', TEST_OTHER_LOGGERS_CONTENT, 'avocado_functional_test_other_loggers') as mytest: cmd_line = ('%s run --sysinfo=off --job-results-dir %s ' '-- %s' % (AVOCADO, self.tmpdir, mytest)) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) test_log_dir = glob.glob(os.path.join(self.tmpdir, 'job-*', 'test-results', '1-*'))[0] test_log_path = os.path.join(test_log_dir, 'debug.log') with open(test_log_path, 'rb') as test_log: self.assertIn(b'SHOULD BE ON debug.log', test_log.read()) def tearDown(self): shutil.rmtree(self.tmpdir) class RunnerHumanOutputTest(unittest.TestCase): def setUp(self): self.tmpdir = tempfile.mkdtemp(prefix='avocado_' + __name__) os.chdir(BASEDIR) def test_output_pass(self): cmd_line = ('%s run --sysinfo=off --job-results-dir %s ' 'passtest.py' % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) self.assertIn(b'passtest.py:PassTest.test: PASS', result.stdout) def test_output_fail(self): cmd_line = ('%s run --sysinfo=off --job-results-dir %s ' 'failtest.py' % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_TESTS_FAIL self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) self.assertIn(b'failtest.py:FailTest.test: FAIL', result.stdout) def test_output_error(self): cmd_line = ('%s run --sysinfo=off --job-results-dir %s ' 'errortest.py' % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_TESTS_FAIL self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) self.assertIn(b'errortest.py:ErrorTest.test: ERROR', result.stdout) def test_output_cancel(self): cmd_line = ('%s run --sysinfo=off --job-results-dir %s ' 'cancelonsetup.py' % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) self.assertIn(b'PASS 0 | ERROR 0 | FAIL 0 | SKIP 0 | WARN 0 | ' b'INTERRUPT 0 | CANCEL 1', result.stdout) @unittest.skipIf(not GNU_ECHO_BINARY, 'GNU style echo binary not available') def test_ugly_echo_cmd(self): cmd_line = ('%s run --external-runner "%s -ne" ' '"foo\\\\\\n\\\'\\\\\\"\\\\\\nbar/baz" --job-results-dir %s' ' --sysinfo=off --show-job-log' % (AVOCADO, GNU_ECHO_BINARY, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %s:\n%s" % (expected_rc, result)) self.assertIn(b'[stdout] foo', result.stdout, result) self.assertIn(b'[stdout] \'"', result.stdout, result) self.assertIn(b'[stdout] bar/baz', result.stdout, result) self.assertIn(b'PASS 1-foo\\\\n\\\'\\"\\\\nbar/baz', result.stdout, result) # logdir name should escape special chars (/) test_dirs = glob.glob(os.path.join(self.tmpdir, 'latest', 'test-results', '*')) self.assertEqual(len(test_dirs), 1, "There are multiple directories in" " test-results dir, but only one test was executed: " "%s" % (test_dirs)) self.assertEqual(os.path.basename(test_dirs[0]), "1-foo__n_'____nbar_baz") def test_replay_skip_skipped(self): cmd = ("%s run --job-results-dir %s --json - " "cancelonsetup.py" % (AVOCADO, self.tmpdir)) result = process.run(cmd) result = json.loads(result.stdout_text) jobid = str(result["job_id"]) cmd = ("%s run --job-results-dir %s --replay %s " "--replay-test-status PASS" % (AVOCADO, self.tmpdir, jobid)) process.run(cmd) def tearDown(self): shutil.rmtree(self.tmpdir) class RunnerSimpleTest(unittest.TestCase): def setUp(self): self.tmpdir = tempfile.mkdtemp(prefix='avocado_' + __name__) self.pass_script = script.TemporaryScript( u'\u00e1 \u00e9 \u00ed \u00f3 \u00fa', "#!/bin/sh\ntrue", 'avocado_simpletest_functional') self.pass_script.save() self.fail_script = script.TemporaryScript('avocado_fail.sh', "#!/bin/sh\nfalse", 'avocado_simpletest_' 'functional') self.fail_script.save() os.chdir(BASEDIR) def test_simpletest_pass(self): cmd_line = ('%s run --job-results-dir %s --sysinfo=off' ' "%s"' % (AVOCADO, self.tmpdir, self.pass_script.path)) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) def test_simpletest_fail(self): cmd_line = ('%s run --job-results-dir %s --sysinfo=off' ' %s' % (AVOCADO, self.tmpdir, self.fail_script.path)) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_TESTS_FAIL self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) @unittest.skipIf(int(os.environ.get("AVOCADO_CHECK_LEVEL", 0)) < 2, "Skipping test that take a long time to run, are " "resource intensive or time sensitve") def test_runner_onehundred_fail_timing(self): """ We can be pretty sure that a failtest should return immediately. Let's run 100 of them and assure they not take more than 30 seconds to run. Notice: on a current machine this takes about 0.12s, so 30 seconds is considered to be pretty safe here. """ one_hundred = 'failtest.py ' * 100 cmd_line = ('%s run --job-results-dir %s --sysinfo=off %s' % (AVOCADO, self.tmpdir, one_hundred)) initial_time = time.time() result = process.run(cmd_line, ignore_status=True) actual_time = time.time() - initial_time self.assertLess(actual_time, 30.0) expected_rc = exit_codes.AVOCADO_TESTS_FAIL self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) @unittest.skipIf(int(os.environ.get("AVOCADO_CHECK_LEVEL", 0)) < 2, "Skipping test that take a long time to run, are " "resource intensive or time sensitve") def test_runner_sleep_fail_sleep_timing(self): """ Sleeptest is supposed to take 1 second, let's make a sandwich of 100 failtests and check the test runner timing. """ sleep_fail_sleep = ('sleeptest.py ' + 'failtest.py ' * 100 + 'sleeptest.py') cmd_line = ('%s run --job-results-dir %s --sysinfo=off %s' % (AVOCADO, self.tmpdir, sleep_fail_sleep)) initial_time = time.time() result = process.run(cmd_line, ignore_status=True) actual_time = time.time() - initial_time self.assertLess(actual_time, 33.0) expected_rc = exit_codes.AVOCADO_TESTS_FAIL self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) def test_simplewarning(self): """ simplewarning.sh uses the avocado-bash-utils """ # simplewarning.sh calls "avocado" without specifying a path # let's add the path that was defined at the global module # scope here os.environ['PATH'] += ":" + os.path.dirname(AVOCADO) # simplewarning.sh calls "avocado exec-path" which hasn't # access to an installed location for the libexec scripts os.environ['PATH'] += ":" + os.path.join(BASEDIR, 'libexec') cmd_line = ('%s run --job-results-dir %s --sysinfo=off ' 'examples/tests/simplewarning.sh --show-job-log' % (AVOCADO, self.tmpdir)) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %s:\n%s" % (expected_rc, result)) self.assertIn(b'DEBUG| Debug message', result.stdout, result) self.assertIn(b'INFO | Info message', result.stdout, result) self.assertIn(b'WARN | Warning message (should cause this test to ' b'finish with warning)', result.stdout, result) self.assertIn(b'ERROR| Error message (ordinary message not changing ' b'the results)', result.stdout, result) self.assertIn(b'Test passed but there were warnings', result.stdout, result) @unittest.skipIf(not GNU_ECHO_BINARY, "Uses echo as test") def test_fs_unfriendly_run(self): os.chdir(BASEDIR) commands_path = os.path.join(self.tmpdir, "commands") script.make_script(commands_path, "echo '\"\\/|?*<>'") config_path = os.path.join(self.tmpdir, "config.conf") script.make_script(config_path, "[sysinfo.collectibles]\ncommands = %s" % commands_path) cmd_line = ("%s --show all --config %s run --job-results-dir %s " "--sysinfo=on --external-runner %s -- \"'\\\"\\/|?*<>'\"" % (AVOCADO, config_path, self.tmpdir, GNU_ECHO_BINARY)) process.run(cmd_line) self.assertTrue(os.path.exists(os.path.join(self.tmpdir, "latest", "test-results", "1-\'________\'/"))) self.assertTrue(os.path.exists(os.path.join(self.tmpdir, "latest", "sysinfo", "pre", "echo \'________\'"))) if python_module_available('avocado-framework-plugin-result-html'): with open(os.path.join(self.tmpdir, "latest", "results.html")) as html_res: html_results = html_res.read() # test results should replace odd chars with "_" # HTML could contain either the literal char, or an entity reference test1_href = (os.path.join("test-results", "1-'________'") in html_results or os.path.join("test-results", "1-'________'") in html_results) self.assertTrue(test1_href) # sysinfo replaces "_" with " " sysinfo = ("echo '________'" in html_results or "echo '________'" in html_results) self.assertTrue(sysinfo) def test_non_absolute_path(self): avocado_path = os.path.join(BASEDIR, 'scripts', 'avocado') test_base_dir = os.path.dirname(self.pass_script.path) os.chdir(test_base_dir) test_file_name = os.path.basename(self.pass_script.path) cmd_line = ('%s %s run --job-results-dir %s --sysinfo=off' ' "%s"' % (sys.executable, avocado_path, self.tmpdir, test_file_name)) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) @unittest.skipIf(not SLEEP_BINARY, 'sleep binary not available') @unittest.skipIf(int(os.environ.get("AVOCADO_CHECK_LEVEL", 0)) < 2, "Skipping test that take a long time to run, are " "resource intensive or time sensitve") def test_kill_stopped_sleep(self): proc = aexpect.Expect("%s run 60 --job-results-dir %s " "--external-runner %s --sysinfo=off " "--job-timeout 3" % (AVOCADO, self.tmpdir, SLEEP_BINARY)) proc.read_until_output_matches([r"\(1/1\)"], timeout=3, internal_timeout=0.01) # We need pid of the avocado process, not the shell executing it avocado_shell = psutil.Process(proc.get_pid()) avocado_proc = avocado_shell.children()[0] pid = avocado_proc.pid os.kill(pid, signal.SIGTSTP) # This freezes the process # The deadline is 3s timeout + 10s test postprocess before kill + # 10s reserve for additional steps (still below 60s) deadline = time.time() + 20 while time.time() < deadline: if not proc.is_alive(): break time.sleep(0.1) else: proc.kill(signal.SIGKILL) self.fail("Avocado process still alive 17s after " "job-timeout:\n%s" % proc.get_output()) output = proc.get_output() self.assertIn("ctrl+z pressed, stopping test", output, "SIGTSTP " "message not in the output, test was probably not " "stopped.") self.assertIn("TIME", output, "TIME not in the output, avocado " "probably died unexpectadly") self.assertEqual(proc.get_status(), 8, "Avocado did not finish with " "1.") sleep_dir = astring.string_to_safe_path("1-60") debug_log_path = os.path.join(self.tmpdir, "latest", "test-results", sleep_dir, "debug.log") debug_log = genio.read_file(debug_log_path) self.assertIn("Runner error occurred: Timeout reached", debug_log, "Runner error occurred: Timeout reached message not " "in the test's debug.log:\n%s" % debug_log) self.assertNotIn("Traceback (most recent", debug_log, "Traceback " "present in the test's debug.log file, but it was " "suppose to be stopped and unable to produce it.\n" "%s" % debug_log) def tearDown(self): self.pass_script.remove() self.fail_script.remove() shutil.rmtree(self.tmpdir) class RunnerSimpleTestStatus(unittest.TestCase): def setUp(self): self.tmpdir = tempfile.mkdtemp(prefix='avocado_' + __name__) self.config_file = script.TemporaryScript('avocado.conf', "[simpletests.status]\n" "warn_regex = ^WARN$\n" "skip_regex = ^SKIP$\n" "skip_location = stdout\n") self.config_file.save() os.chdir(BASEDIR) def test_simpletest_status(self): # Multi-line warning in STDERR should by default be handled warn_script = script.TemporaryScript('avocado_warn.sh', '#!/bin/sh\n' '>&2 echo -e "\\n\\nWARN\\n"', 'avocado_simpletest_' 'functional') warn_script.save() cmd_line = ('%s --config %s run --job-results-dir %s --sysinfo=off' ' %s --json -' % (AVOCADO, self.config_file.path, self.tmpdir, warn_script.path)) result = process.run(cmd_line, ignore_status=True) json_results = json.loads(result.stdout_text) self.assertEqual(json_results['tests'][0]['status'], 'WARN') warn_script.remove() # Skip in STDOUT should be handled because of config skip_script = script.TemporaryScript('avocado_skip.sh', "#!/bin/sh\necho SKIP", 'avocado_simpletest_' 'functional') skip_script.save() cmd_line = ('%s --config %s run --job-results-dir %s --sysinfo=off' ' %s --json -' % (AVOCADO, self.config_file.path, self.tmpdir, skip_script.path)) result = process.run(cmd_line, ignore_status=True) json_results = json.loads(result.stdout_text) self.assertEqual(json_results['tests'][0]['status'], 'SKIP') skip_script.remove() # STDERR skip should not be handled skip2_script = script.TemporaryScript('avocado_skip.sh', "#!/bin/sh\n>&2 echo SKIP", 'avocado_simpletest_' 'functional') skip2_script.save() cmd_line = ('%s --config %s run --job-results-dir %s --sysinfo=off' ' %s --json -' % (AVOCADO, self.config_file.path, self.tmpdir, skip2_script.path)) result = process.run(cmd_line, ignore_status=True) json_results = json.loads(result.stdout_text) self.assertEqual(json_results['tests'][0]['status'], 'PASS') skip2_script.remove() def tearDown(self): self.config_file.remove() shutil.rmtree(self.tmpdir) class ExternalRunnerTest(unittest.TestCase): def setUp(self): self.tmpdir = tempfile.mkdtemp(prefix='avocado_' + __name__) self.pass_script = script.TemporaryScript( 'pass', "exit 0", 'avocado_externalrunner_functional') self.pass_script.save() self.fail_script = script.TemporaryScript( 'fail', "exit 1", 'avocado_externalrunner_functional') self.fail_script.save() os.chdir(BASEDIR) def test_externalrunner_pass(self): cmd_line = ('%s run --job-results-dir %s --sysinfo=off ' '--external-runner=/bin/sh %s' % (AVOCADO, self.tmpdir, self.pass_script.path)) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) def test_externalrunner_fail(self): cmd_line = ('%s run --job-results-dir %s --sysinfo=off ' '--external-runner=/bin/sh %s' % (AVOCADO, self.tmpdir, self.fail_script.path)) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_TESTS_FAIL self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) def test_externalrunner_chdir_no_testdir(self): cmd_line = ('%s run --job-results-dir %s --sysinfo=off ' '--external-runner=/bin/sh --external-runner-chdir=test %s' % (AVOCADO, self.tmpdir, self.pass_script.path)) result = process.run(cmd_line, ignore_status=True) expected_output = (b'Option "--external-runner-chdir=test" requires ' b'"--external-runner-testdir" to be set') self.assertIn(expected_output, result.stderr) expected_rc = exit_codes.AVOCADO_JOB_FAIL self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) def test_externalrunner_chdir_runner_relative(self): avocado_abs = " ".join([os.path.abspath(_) for _ in AVOCADO.split(" ")]) pass_abs = os.path.abspath(self.pass_script.path) os.chdir('/') cmd_line = ('%s run --job-results-dir %s --sysinfo=off ' '--external-runner=bin/sh --external-runner-chdir=runner -- %s' % (avocado_abs, self.tmpdir, pass_abs)) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) def test_externalrunner_no_url(self): cmd_line = ('%s run --job-results-dir %s --sysinfo=off ' '--external-runner=%s' % (AVOCADO, self.tmpdir, TRUE_CMD)) result = process.run(cmd_line, ignore_status=True) expected_output = (b'No test references provided nor any other ' b'arguments resolved into tests') self.assertIn(expected_output, result.stderr) expected_rc = exit_codes.AVOCADO_JOB_FAIL self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) def tearDown(self): self.pass_script.remove() self.fail_script.remove() shutil.rmtree(self.tmpdir) class AbsPluginsTest(object): def setUp(self): self.base_outputdir = tempfile.mkdtemp(prefix='avocado_' + __name__) os.chdir(BASEDIR) def tearDown(self): shutil.rmtree(self.base_outputdir) class PluginsTest(AbsPluginsTest, unittest.TestCase): def test_sysinfo_plugin(self): cmd_line = '%s sysinfo %s' % (AVOCADO, self.base_outputdir) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) sysinfo_files = os.listdir(self.base_outputdir) self.assertGreater(len(sysinfo_files), 0, "Empty sysinfo files dir") def test_list_plugin(self): cmd_line = '%s list' % AVOCADO result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) self.assertNotIn(b'No tests were found on current tests dir', result.stdout) def test_list_error_output(self): cmd_line = '%s list sbrubles' % AVOCADO result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_FAIL self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) self.assertIn(b"Unable to resolve reference", result.stderr) def test_list_no_file_loader(self): cmd_line = ("%s list --loaders external --verbose -- " "this-wont-be-matched" % AVOCADO) result = process.run(cmd_line, ignore_status=True) self.assertEqual(result.exit_status, exit_codes.AVOCADO_ALL_OK, "Avocado did not return rc %d:\n%s" % (exit_codes.AVOCADO_ALL_OK, result)) exp = (b"Type Test Tag(s)\n" b"MISSING this-wont-be-matched\n\n" b"TEST TYPES SUMMARY\n" b"==================\n" b"EXTERNAL: 0\n" b"MISSING: 1\n") self.assertEqual(exp, result.stdout, "Stdout mismatch:\n%s\n\n%s" % (exp, result)) def test_list_verbose_tags(self): """ Runs list verbosely and check for tag related output """ test = script.make_script(os.path.join(self.base_outputdir, 'test.py'), VALID_PYTHON_TEST_WITH_TAGS) cmd_line = ("%s list --loaders file --verbose %s" % (AVOCADO, test)) result = process.run(cmd_line, ignore_status=True) self.assertEqual(result.exit_status, exit_codes.AVOCADO_ALL_OK, "Avocado did not return rc %d:\n%s" % (exit_codes.AVOCADO_ALL_OK, result)) stdout_lines = result.stdout_text.splitlines() self.assertIn("Tag(s)", stdout_lines[0]) full_test_name = "%s:MyTest.test" % test self.assertEqual("INSTRUMENTED %s BIG_TAG_NAME" % full_test_name, stdout_lines[1]) self.assertIn("TEST TYPES SUMMARY", stdout_lines) self.assertIn("INSTRUMENTED: 1", stdout_lines) self.assertIn("TEST TAGS SUMMARY", stdout_lines) self.assertEqual("BIG_TAG_NAME: 1", stdout_lines[-1]) def test_plugin_list(self): cmd_line = '%s plugins' % AVOCADO result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) if sys.version_info[:2] >= (2, 7, 0): self.assertNotIn(b'Disabled', result.stdout) def test_config_plugin(self): cmd_line = '%s config --paginator off' % AVOCADO result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) self.assertNotIn(b'Disabled', result.stdout) def test_config_plugin_datadir(self): cmd_line = '%s config --datadir --paginator off' % AVOCADO result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) self.assertNotIn(b'Disabled', result.stdout) def test_disable_plugin(self): cmd_line = '%s plugins' % AVOCADO result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) self.assertIn(b"Collect system information", result.stdout) config_content = "[plugins]\ndisable=['cli.cmd.sysinfo',]" config = script.TemporaryScript("disable_sysinfo_cmd.conf", config_content) with config: cmd_line = '%s --config %s plugins' % (AVOCADO, config) result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) self.assertNotIn(b"Collect system information", result.stdout) def test_plugin_order(self): """ Tests plugin order by configuration file First it checks if html, json, xunit and zip_archive plugins are enabled. Then it runs a test with zip_archive running first, which means the html, json and xunit output files do not make into the archive. Then it runs with zip_archive set to run last, which means the html, json and xunit output files *do* make into the archive. """ def run_config(config_path): cmd = ('%s --config %s run passtest.py --archive ' '--job-results-dir %s --sysinfo=off' % (AVOCADO, config_path, self.base_outputdir)) result = process.run(cmd, ignore_status=True) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) result_plugins = ["json", "xunit", "zip_archive"] result_outputs = ["results.json", "results.xml"] if python_module_available('avocado-framework-plugin-result-html'): result_plugins.append("html") result_outputs.append("results.html") cmd_line = '%s plugins' % AVOCADO result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) for result_plugin in result_plugins: self.assertIn(result_plugin, result.stdout_text) config_content_zip_first = "[plugins.result]\norder=['zip_archive']" config_zip_first = script.TemporaryScript("zip_first.conf", config_content_zip_first) with config_zip_first: run_config(config_zip_first) archives = glob.glob(os.path.join(self.base_outputdir, '*.zip')) self.assertEqual(len(archives), 1, "ZIP Archive not generated") zip_file = zipfile.ZipFile(archives[0], 'r') zip_file_list = zip_file.namelist() for result_output in result_outputs: self.assertNotIn(result_output, zip_file_list) os.unlink(archives[0]) config_content_zip_last = ("[plugins.result]\norder=['html', 'json'," "'xunit', 'non_existing_plugin_is_ignored'" ",'zip_archive']") config_zip_last = script.TemporaryScript("zip_last.conf", config_content_zip_last) with config_zip_last: run_config(config_zip_last) archives = glob.glob(os.path.join(self.base_outputdir, '*.zip')) self.assertEqual(len(archives), 1, "ZIP Archive not generated") zip_file = zipfile.ZipFile(archives[0], 'r') zip_file_list = zip_file.namelist() for result_output in result_outputs: self.assertIn(result_output, zip_file_list) def test_Namespace_object_has_no_attribute(self): cmd_line = '%s plugins' % AVOCADO result = process.run(cmd_line, ignore_status=True) expected_rc = exit_codes.AVOCADO_ALL_OK self.assertEqual(result.exit_status, expected_rc, "Avocado did not return rc %d:\n%s" % (expected_rc, result)) self.assertNotIn(b"'Namespace' object has no attribute", result.stderr) class ParseXMLError(Exception): pass class PluginsXunitTest(AbsPluginsTest, unittest.TestCase): @unittest.skipUnless(SCHEMA_CAPABLE, 'Unable to validate schema due to missing lxml.etree library') def setUp(self): self.tmpdir = tempfile.mkdtemp(prefix='avocado_' + __name__) junit_xsd = os.path.join(os.path.dirname(__file__), os.path.pardir, ".data", 'jenkins-junit.xsd') self.junit = os.path.abspath(junit_xsd) super(PluginsXunitTest, self).setUp() def run_and_check(self, testname, e_rc, e_ntests, e_nerrors, e_nnotfound, e_nfailures, e_nskip): # pylint: disable=W0613 cmd_line = ('%s run --job-results-dir %s --sysinfo=off' ' --xunit - %s' % (AVOCADO, self.tmpdir, testname)) result = process.run(cmd_line, ignore_status=True) xml_output = result.stdout self.assertEqual(result.exit_status, e_rc, "Avocado did not return rc %d:\n%s" % (e_rc, result)) try: xunit_doc = xml.dom.minidom.parseString(xml_output) except Exception as detail: raise ParseXMLError("Failed to parse content: %s\n%s" % (detail, xml_output)) with open(self.junit, 'rb') as f: xmlschema = etree.XMLSchema(etree.parse(f)) # pylint: disable=I1101 # pylint: disable=I1101 self.assertTrue(xmlschema.validate(etree.parse(BytesIO(xml_output))), "Failed to validate against %s, message:\n%s" % (self.junit, xmlschema.error_log.filter_from_errors())) testsuite_list = xunit_doc.getElementsByTagName('testsuite') self.assertEqual(len(testsuite_list), 1, 'More than one testsuite tag') testsuite_tag = testsuite_list[0] self.assertEqual(len(testsuite_tag.attributes), 7, 'The testsuite tag does not have 7 attributes. ' 'XML:\n%s' % xml_output) n_tests = int(testsuite_tag.attributes['tests'].value) self.assertEqual(n_tests, e_ntests, "Unexpected number of executed tests, " "XML:\n%s" % xml_output) n_errors = int(testsuite_tag.attributes['errors'].value) self.assertEqual(n_errors, e_nerrors, "Unexpected number of test errors, " "XML:\n%s" % xml_output) n_failures = int(testsuite_tag.attributes['failures'].value) self.assertEqual(n_failures, e_nfailures, "Unexpected number of test failures, " "XML:\n%s" % xml_output) n_skip = int(testsuite_tag.attributes['skipped'].value) self.assertEqual(n_skip, e_nskip, "Unexpected number of test skips, " "XML:\n%s" % xml_output) def test_xunit_plugin_passtest(self): self.run_and_check('passtest.py', exit_codes.AVOCADO_ALL_OK, 1, 0, 0, 0, 0) def test_xunit_plugin_failtest(self): self.run_and_check('failtest.py', exit_codes.AVOCADO_TESTS_FAIL, 1, 0, 0, 1, 0) def test_xunit_plugin_skiponsetuptest(self): self.run_and_check('cancelonsetup.py', exit_codes.AVOCADO_ALL_OK, 1, 0, 0, 0, 1) def test_xunit_plugin_errortest(self): self.run_and_check('errortest.py', exit_codes.AVOCADO_TESTS_FAIL, 1, 1, 0, 0, 0) def tearDown(self): shutil.rmtree(self.tmpdir) super(PluginsXunitTest, self).tearDown() class ParseJSONError(Exception): pass class PluginsJSONTest(AbsPluginsTest, unittest.TestCase): def setUp(self): self.tmpdir = tempfile.mkdtemp(prefix='avocado_' + __name__) super(PluginsJSONTest, self).setUp() def run_and_check(self, testname, e_rc, e_ntests, e_nerrors, e_nfailures, e_nskip, e_ncancel=0, external_runner=None): cmd_line = ('%s run --job-results-dir %s --sysinfo=off --json - ' '--archive %s' % (AVOCADO, self.tmpdir, testname)) if external_runner is not None: cmd_line += " --external-runner '%s'" % external_runner result = process.run(cmd_line, ignore_status=True) json_output = result.stdout_text self.assertEqual(result.exit_status, e_rc, "Avocado did not return rc %d:\n%s" % (e_rc, result)) try: json_data = json.loads(json_output) except Exception as detail: raise ParseJSONError("Failed to parse content: %s\n%s" % (detail, json_output)) self.assertTrue(json_data, "Empty JSON result:\n%s" % json_output) self.assertIsInstance(json_data['tests'], list, "JSON result lacks 'tests' list") n_tests = len(json_data['tests']) self.assertEqual(n_tests, e_ntests, "Different number of expected tests") n_errors = json_data['errors'] self.assertEqual(n_errors, e_nerrors, "Different number of expected tests") n_failures = json_data['failures'] self.assertEqual(n_failures, e_nfailures, "Different number of expected tests") n_skip = json_data['skip'] self.assertEqual(n_skip, e_nskip, "Different number of skipped tests") n_cancel = json_data['cancel'] self.assertEqual(n_cancel, e_ncancel) return json_data def test_json_plugin_passtest(self): self.run_and_check('passtest.py', exit_codes.AVOCADO_ALL_OK, 1, 0, 0, 0) def test_json_plugin_failtest(self): self.run_and_check('failtest.py', exit_codes.AVOCADO_TESTS_FAIL, 1, 0, 1, 0) def test_json_plugin_skiponsetuptest(self): self.run_and_check('cancelonsetup.py', exit_codes.AVOCADO_ALL_OK, 1, 0, 0, 0, 1) def test_json_plugin_errortest(self): self.run_and_check('errortest.py', exit_codes.AVOCADO_TESTS_FAIL, 1, 1, 0, 0) @unittest.skipIf(not GNU_ECHO_BINARY, 'echo binary not available') def test_ugly_echo_cmd(self): data = self.run_and_check('"-ne foo\\\\\\n\\\'\\\\\\"\\\\\\' 'nbar/baz"', exit_codes.AVOCADO_ALL_OK, 1, 0, 0, 0, external_runner=GNU_ECHO_BINARY) # The executed test should be this self.assertEqual(data['tests'][0]['id'], '1--ne foo\\\\n\\\'\\"\\\\nbar/baz') # logdir name should escape special chars (/) self.assertEqual(os.path.basename(data['tests'][0]['logdir']), "1--ne foo__n_'____nbar_baz") def tearDown(self): shutil.rmtree(self.tmpdir) super(PluginsJSONTest, self).tearDown() if __name__ == '__main__': unittest.main()