diff --git a/python/README.md b/python/README.md index c1d70919..b04a46c8 100644 --- a/python/README.md +++ b/python/README.md @@ -24,6 +24,15 @@ Then, to then get a summary report of all the tests, run the following on anothe py.test -p ciqueue.pytest_report --queue redis://:6379?build=&retry= ``` +Workers store rendered pytest reports as compressed JSON in Redis; the reporter never loads Python objects from the queue. +Upgrade workers and the reporter together, using the same `ciqueue` and pytest versions within a build. +Legacy dill records and malformed reports are rejected with a non-zero exit status, not treated as passing tests. +After upgrading, rerun all workers with a fresh build ID; do not reuse an old build's records. + +Traceback formatting (for example, `--tb` and `--showlocals`) is determined by the worker's options. +The queue transports standard `TestReport` fields, including captured output, user properties rendered as strings, +and xfail reasons; plugin-specific report attributes are not transported. + ## Implementing a new integration The reference implementation is the minitest one (Ruby). diff --git a/python/ciqueue/_pytest/outcomes.py b/python/ciqueue/_pytest/outcomes.py deleted file mode 100644 index e31fec80..00000000 --- a/python/ciqueue/_pytest/outcomes.py +++ /dev/null @@ -1,95 +0,0 @@ -""" -This module is used for querying and altering test outcomes, and loosely -follows pytest's _pytest.outcomes.py in version 3.2 onwards. - -Much of his module exists because pytest adds the field `__module__ = 'builtins'` -to the Skipped and Failed exception classes, rendering them unserializable. -We get around this by creating our own serializable version of -these classes, which we swap in in place of the original when we want to -be put them on the redis queue. Then, we swap back in the original exception -when reading off the queue. These operations are performed by -`swap_in_serializable` and `swap_back_original`, respectively. -""" - - -from __future__ import absolute_import -import dill -try: - from _pytest import outcomes -except ImportError: - from _pytest import runner as outcomes -from _pytest._code import code -from tblib import pickling_support - -pickling_support.install() - - -class Skipped(Exception): - """placeholder for outcomes.Skipped which is not serializable""" - - -class Failed(Exception): - """placeholder for outcomes.Failed which is not serializable""" - - -class UnserializableException(Exception): - """placeholder for any Exceptions that cannnot be serialized""" - - -SERIALIZE_TYPES = {outcomes.Skipped: Skipped, - outcomes.Failed: Failed} -DESERIALIZE_TYPES = {Skipped: outcomes.Skipped, - Failed: outcomes.Failed} - - -try: - from_exc_info = code.ExceptionInfo.from_exc_info -except AttributeError: - # pytest < 7.4 - def from_exc_info(tup): - return code.ExceptionInfo(tup) - - -def swap_in_serializable(excinfo): - def pickles(excinfo): - try: - return dill.pickles(excinfo) - except BaseException: - return False - - if excinfo.type in SERIALIZE_TYPES: - cls = SERIALIZE_TYPES[excinfo.type] - tup = (cls, cls(*excinfo.value.args), excinfo.tb) - excinfo = from_exc_info(tup) - elif not pickles(excinfo): - tup = (UnserializableException, - UnserializableException( - "Actual Exception thrown on test node was %r" % - excinfo.value), - excinfo.tb) - excinfo = from_exc_info(tup) - return excinfo - - -def swap_back_original(excinfo): - if excinfo.type in DESERIALIZE_TYPES: - tipe = DESERIALIZE_TYPES[excinfo.type] - tup = (tipe, tipe(*excinfo.value.args), excinfo.tb) - return from_exc_info(tup) - return excinfo - - -def marked_xfail(item): - return hasattr(item, '_evalxfail') and item._evalxfail.istrue() # pylint: disable=protected-access - - -def failed(item): - return hasattr(item, 'error_reports') and \ - not marked_xfail(item) and \ - not all(issubclass(i['excinfo'].type, Skipped) for i in item.error_reports.values()) - - -def skipped_excinfo(item, msg): - traceback = list(item.error_reports.values())[0]['excinfo'].tb - tup = (outcomes.Skipped, outcomes.Skipped(msg), traceback) - return from_exc_info(tup) diff --git a/python/ciqueue/_pytest/reports.py b/python/ciqueue/_pytest/reports.py new file mode 100644 index 00000000..0b879cbb --- /dev/null +++ b/python/ciqueue/_pytest/reports.py @@ -0,0 +1,90 @@ +"""Transport rendered pytest reports, never exception or traceback objects.""" +import json +import math +import zlib + +import pytest +from _pytest.reports import TestReport + + +REPORT_FIELDS = { + '$report_type', 'nodeid', 'location', 'keywords', 'outcome', 'longrepr', 'when', + 'sections', 'duration', 'start', 'stop', 'user_properties', 'wasxfail', +} + + +def dumps(config, reports): + payload = {} + for when, report in reports.items(): + data = config.hook.pytest_report_to_serializable(config=config, report=report) + # Plugin-specific attributes are not part of the queue's report protocol. + payload[when] = {key: value for key, value in data.items() if key in REPORT_FIELDS} + # JUnit renders property values as text; arbitrary objects stay on the worker. + payload[when]['user_properties'] = [(name, str(value)) for name, value in report.user_properties] + return zlib.compress(json.dumps(payload, allow_nan=False).encode('utf-8')) + + +def _location(value, allow_none=False): + return (isinstance(value, list) and len(value) == 3 and + isinstance(value[0], str) and isinstance(value[2], str) and + (type(value[1]) is int or (allow_none and value[1] is None))) + + +def _pairs(value): + return isinstance(value, list) and all( + isinstance(pair, list) and len(pair) == 2 and + all(isinstance(part, str) for part in pair) for pair in value) + + +def _validate(data, when, nodeid): + if not isinstance(data, dict) or not data.keys() <= REPORT_FIELDS: + raise ValueError('unsupported report fields') + if (data.get('$report_type') != 'TestReport' or data.get('nodeid') != nodeid or + data.get('when') != when or data.get('outcome') not in ('failed', 'skipped')): + raise ValueError('invalid report identity or outcome') + if not _location(data.get('location'), allow_none=True) or not isinstance(data.get('keywords'), dict): + raise ValueError('invalid report location or keywords') + if not _pairs(data.get('sections', [])) or not _pairs(data.get('user_properties', [])): + raise ValueError('invalid report sections or properties') + for name in ('duration', 'start', 'stop'): + if name in data and (type(data[name]) not in (int, float) or not math.isfinite(data[name])): + raise ValueError('invalid report timing') + if 'wasxfail' in data and (not isinstance(data['wasxfail'], str) or data['outcome'] != 'skipped'): + raise ValueError('xfail metadata requires a skipped outcome') + longrepr = data.get('longrepr') + if isinstance(longrepr, list): + if not _location(longrepr): + raise ValueError('invalid skip representation') + data['longrepr'] = tuple(longrepr) + elif isinstance(longrepr, dict): + if not {'reprcrash', 'reprtraceback', 'sections', 'chain'} <= longrepr.keys(): + raise ValueError('invalid traceback representation') + elif not isinstance(longrepr, str): + raise ValueError('missing failure representation') + if data['outcome'] == 'skipped' and 'wasxfail' not in data and not isinstance(data['longrepr'], tuple): + raise ValueError('missing skip location') + data['location'] = tuple(data['location']) + + +def loads(config, payload, nodeid): + try: + data = json.loads(zlib.decompress(payload).decode('utf-8')) + if not isinstance(data, dict) or not data or not data.keys() <= {'setup', 'call', 'teardown'}: + raise ValueError('expected setup/call/teardown reports') + if 'setup' in data and 'call' in data: + raise ValueError('a non-passing setup cannot have a call report') + reports = {} + for when, report_data in data.items(): + _validate(report_data, when, nodeid) + report = config.hook.pytest_report_from_serializable(config=config, data=report_data) + if not isinstance(report, TestReport): + raise ValueError('expected a TestReport') + # Exercise pytest's structured traceback renderer before accepting a record. + # Malformed nested representations must fail here, not during reporting. + _ = report.longreprtext + reports[when] = report + return reports + except (ValueError, TypeError, KeyError, AttributeError, AssertionError, RuntimeError, zlib.error) as error: + raise pytest.UsageError( + 'Invalid error report for {}: {}. Expected compressed JSON reports; ' + 'upgrade workers and reporter together and use a fresh build ID.'.format(nodeid, error)) from error diff --git a/python/ciqueue/pytest.py b/python/ciqueue/pytest.py index be2231f4..8aa110c1 100644 --- a/python/ciqueue/pytest.py +++ b/python/ciqueue/pytest.py @@ -5,10 +5,8 @@ """ from __future__ import absolute_import from __future__ import print_function -import zlib from ciqueue._pytest import test_queue -from ciqueue._pytest import outcomes -import dill +from ciqueue._pytest import reports import pytest from _pytest import terminal @@ -73,19 +71,21 @@ def _get_progress(self): # pylint: disable=unused-argument terminal.TerminalReporter._get_progress_information_message = _get_progress # pylint: disable=protected-access - def record(self, item): - # if the test passed, we remove it from the errors queue - # otherwise we add it - if hasattr(item, 'error_reports'): - self.redis.hset( - self.errors_key, - test_queue.key_item(item), - zlib.compress(dill.dumps(item.error_reports))) + def record(self, item, test_failed): + # Serialize before acknowledging so encoding errors cannot lose a failure. + payload = reports.dumps(self.config, item.error_reports) if hasattr(item, 'error_reports') else None + test_name = test_queue.key_item(item) + # A late worker may replace an earlier failure only if it succeeded. + if not self.queue.acknowledge(test_name) and test_failed: + return False + if payload is not None: + self.redis.hset(self.errors_key, test_name, payload) else: - self.redis.hdel(self.errors_key, test_queue.key_item(item)) + self.redis.hdel(self.errors_key, test_name) + return True - def mark_as_skipped(self, call, item, msg): - assert call.when == 'teardown' + def mark_as_skipped(self, report, item, msg): + assert report.when == 'teardown' stats = self.terminalreporter.stats @@ -106,54 +106,48 @@ def clear_out_stats(key): if self.logxml: self.logxml.node_reporters_ordered[-1].nodes = [] - # the call is converted to a skip - call.excinfo = outcomes.skipped_excinfo(item, msg) + # Render retries locally; no exception or traceback objects go on the wire. + path, lineno, _ = item.location + report.outcome = 'skipped' + report.longrepr = (path, (lineno or 0) + 1, msg) + if hasattr(report, 'wasxfail'): + del report.wasxfail # clear out the stats like the test never happened for key in ('passed', 'error', 'failed'): clear_out_stats(key) # rollback the testsfailed number like it never happened - item.session.testsfailed -= len([v for k, v in item.error_reports.items() - if not issubclass(v['excinfo'].type, outcomes.Skipped) and k != 'teardown']) + item.session.testsfailed -= sum( + report.failed for when, report in item.error_reports.items() if when != 'teardown') # and clear out any state on the item like it never happened if hasattr(item, 'error_reports'): del item.error_reports - @pytest.hookimpl(tryfirst=True) + @pytest.hookimpl(hookwrapper=True, tryfirst=True) def pytest_runtest_makereport(self, item, call): - """This function hooks into pytest's reporting of test results, and pushes a failed test's error report - onto the redis queue. A test can fail in any of the 3 call stages: setup, test, or teardown. - This is captured by pushing a dict of {call_state: error} for each failed test.""" - if call.excinfo: - payload = call.__dict__.copy() - payload['excinfo'] = outcomes.swap_in_serializable(payload['excinfo']) - + """Record final reports after pytest has applied skip and xfail outcomes.""" + result = yield + report = result.get_result() + if not report.passed: if not hasattr(item, 'error_reports'): - item.error_reports = {call.when: payload} - else: - item.error_reports[call.when] = payload + item.error_reports = {} + item.error_reports[report.when] = report - if call.when == 'teardown': + if report.when == 'teardown': test_name = test_queue.key_item(item) - test_failed = outcomes.failed(item) + test_failed = any(report.failed for report in getattr(item, 'error_reports', {}).values()) # Only attempt to requeue if the test failed. # The method will return `False` if the test couldn't be requeued if test_failed and self.queue.requeue(test_name): - self.mark_as_skipped(call, item, "WILL_RETRY") + self.mark_as_skipped(report, item, "WILL_RETRY") self.terminalwriter.write(' WILL_RETRY ', green=True) - # If the test was already acknowledged by another worker (we timed out) - # Then we only record it if it was successful. - elif self.queue.acknowledge(test_name) or not test_failed: - self.record(item) - - # The test timed out and failed, mark it as skipped so that it doesn't - # fail the build - else: - self.mark_as_skipped(call, item, "TIMED OUT") + # Ignore a late failure if another worker already acknowledged the test. + elif not self.record(item, test_failed): + self.mark_as_skipped(report, item, "TIMED OUT") self.terminalwriter.write(' TIMED OUT ', green=True) diff --git a/python/ciqueue/pytest_report.py b/python/ciqueue/pytest_report.py index a189f542..457948e0 100644 --- a/python/ciqueue/pytest_report.py +++ b/python/ciqueue/pytest_report.py @@ -6,12 +6,9 @@ from __future__ import absolute_import from __future__ import print_function -import zlib -import dill import pytest -from _pytest import runner from ciqueue._pytest import test_queue -from ciqueue._pytest import outcomes +from ciqueue._pytest import reports def pytest_addoption(parser): @@ -45,24 +42,19 @@ def pytest_collection_modifyitems(session, config, items): # pylint: disable=un # store the errors on setup/test/teardown to item.error_reports key = test_queue.key_item(item) if key in error_reports: - item.error_reports = dill.loads(zlib.decompress(error_reports[key])) - for _, call_dict in item.error_reports.items(): - call_dict['excinfo'] = outcomes.swap_back_original(call_dict['excinfo']) + item.error_reports = reports.loads(config, error_reports[key], key) -@pytest.hookimpl(tryfirst=True) +@pytest.hookimpl(hookwrapper=True, tryfirst=True) def pytest_runtest_makereport(item, call): - """This function hooks into pytest's reporting of test results, and replaces the - result of each test's setup/runtest/teardown call with the result from the redis queue""" - - # ensure all errors should come off the error-reports queue + """Replay the worker's final outcome after local skip/xfail hooks have run.""" call.excinfo = None + result = yield if hasattr(item, 'error_reports') and call.when in item.error_reports: - call.__dict__ = item.error_reports[call.when] - - # This is needed to change the location of the failure - # to point to the item definition, otherwise it will display - # the location of where the skip exception was raised within pytest - # https://github.com/pytest-dev/pytest/blob/master/_pytest/skipping.py#L263-L269 - if call.excinfo and call.excinfo.type == runner.Skipped: - item._evalskip = True # pylint: disable=protected-access + result.force_result(item.error_reports[call.when]) + elif hasattr(item, 'error_reports') and call.when == 'teardown': + # JUnit finalizes metadata on teardown, even when only an earlier phase failed. + previous = item.error_reports.get('call') or item.error_reports['setup'] + report = result.get_result() + report.user_properties = previous.user_properties + report.sections = previous.sections diff --git a/python/setup.py b/python/setup.py index efab4573..17e5c750 100644 --- a/python/setup.py +++ b/python/setup.py @@ -37,10 +37,8 @@ def get_lua_scripts(): packages=['ciqueue', 'ciqueue._pytest'], python_requires='>=3.10', install_requires=[ - 'dill>=0.2.7', - 'pytest>=2.7', + 'pytest>=6.2.5', 'redis>=2.10.5', - 'tblib>=1.3.2', 'uritools>=2.0.0', 'future>=0.16.0' ], diff --git a/python/tests/test_pytest.py b/python/tests/test_pytest.py index c3a7e0dc..49ae524c 100644 --- a/python/tests/test_pytest.py +++ b/python/tests/test_pytest.py @@ -1,6 +1,12 @@ +import json import os +import pickle import re import subprocess +import sys +import textwrap +import xml.etree.ElementTree as ET +import zlib import redis import pytest @@ -47,8 +53,6 @@ def test_integration(self): queue = "redis://localhost:6379/0?build=foo&retry=0" output = check_output(report_cmd) assert '= 11 passed, 1 xpassed in' in output, output - assert ('integrations/pytest/test_all.py:27: message' not in output - and 'integrations/pytest/test_all.py:28: message' not in output), output def test_retries_and_junit_xml(self, tmpdir): queue = ('redis://localhost:6379/0?worker=0&build=bar&retry=0&timeout=5' @@ -65,8 +69,7 @@ def test_retries_and_junit_xml(self, tmpdir): output = check_output(cmd.format(queue, filename)) assert re.search(r'= 4 failed, 2 passed, 4 skipped, 1 xpassed, (1 warning, )?6 errors in', output), output - assert ('integrations/pytest/test_all.py:27: skipping test message' in output - or 'integrations/pytest/test_all.py:28: skipping test message' in output), output + assert re.search(r':\d+: skipping test message', output), output assert ' WILL_RETRY ' in output, output xml = open(xml_file).read() @@ -92,3 +95,140 @@ def test_flakey(self): output = check_output(report_cmd) assert '= 1 passed in' in output, output + + def test_report_rejects_executable_records(self, tmp_path): + sentinel = tmp_path / 'deserialized' + + class ExecutableRecord: + def __reduce__(self): + return os.mkdir, (str(sentinel),) + + nodeid = 'integrations/pytest/test_all.py::TestHappy::test_method' + self.redis.set('build:unsafe:master-status', 'finished') + self.redis.hset('build:unsafe:error-reports', nodeid, + zlib.compress(pickle.dumps(ExecutableRecord()))) + result = subprocess.run( + [sys.executable, '-m', 'pytest', '-p', 'ciqueue.pytest_report', + '--queue', 'redis://localhost:6379/0?build=unsafe', nodeid], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=30) + + assert not sentinel.exists(), result.stdout + assert result.returncode != 0, result.stdout + assert 'Invalid error report' in result.stdout + assert nodeid in result.stdout + + @pytest.mark.parametrize('payload', [ + b'not compressed', + zlib.compress(b'{'), + zlib.compress(b'{}'), + zlib.compress(b'{"collect": {}}'), + zlib.compress(b'{"call": {"$report_type": "CollectReport"}}'), + ]) + def test_report_rejects_invalid_records(self, payload): + nodeid = 'integrations/pytest/test_all.py::TestHappy::test_method' + self.redis.set('build:invalid:master-status', 'finished') + self.redis.hset('build:invalid:error-reports', nodeid, payload) + result = subprocess.run( + [sys.executable, '-m', 'pytest', '-p', 'ciqueue.pytest_report', + '--queue', 'redis://localhost:6379/0?build=invalid', nodeid], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=30) + + assert result.returncode == pytest.ExitCode.USAGE_ERROR, result.stdout + assert nodeid in result.stdout + + def test_xfail_outcomes_survive_retries_and_reporting(self, tmp_path): + test_file = tmp_path / 'test_outcomes.py' + test_file.write_text(textwrap.dedent('''\ + import pytest + + @pytest.mark.xfail(reason="expected", raises=ValueError) + def test_expected(): + raise ValueError("expected") + + @pytest.mark.xfail(reason="wrong exception", raises=TypeError) + def test_unexpected(): + raise ValueError("not a TypeError") + + @pytest.mark.xfail(strict=True, reason="must fail") + def test_strict(): + pass + + def test_dynamic(): + pytest.xfail("dynamic reason") + + @pytest.fixture + def cleanup(): + yield + pytest.xfail("cleanup") + + def test_cleanup(cleanup): + assert False, "call failed before cleanup" + ''')) + queue = ('redis://localhost:6379/0?worker=0&build=xfail&timeout=5' + '&max_requeues=1&requeue_tolerance=1') + worker = subprocess.run( + [sys.executable, '-m', 'pytest', '-ra', '-p', 'ciqueue.pytest', + '--queue', queue, str(test_file)], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=30) + assert worker.returncode == pytest.ExitCode.TESTS_FAILED, worker.stdout + assert '3 failed, 3 skipped, 3 xfailed' in worker.stdout + + xml_file = tmp_path / 'report.xml' + reporter = subprocess.run( + [sys.executable, '-m', 'pytest', '-ra', '-p', 'ciqueue.pytest_report', + '--queue', queue, '--junitxml', str(xml_file), str(test_file)], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=30) + assert reporter.returncode == pytest.ExitCode.TESTS_FAILED, reporter.stdout + assert '3 failed, 3 xfailed' in reporter.stdout + cases = {case.get('name'): case for case in ET.parse(xml_file).iter('testcase')} + assert cases['test_expected'].find('skipped').get('message') == 'expected' + assert cases['test_dynamic'].find('skipped').get('message') == 'dynamic reason' + assert 'not a TypeError' in cases['test_unexpected'].find('failure').text + assert '[XPASS(strict)]' in cases['test_strict'].find('failure').text + + @pytest.mark.parametrize('corruption', ['identity', 'xfail', 'phases']) + def test_report_rejects_inconsistent_records(self, corruption): + nodeid = 'integrations/pytest/test_all.py::TestSadTest::test_method' + queue = 'redis://localhost:6379/0?worker=0&build=identity&timeout=5' + worker = subprocess.run( + [sys.executable, '-m', 'pytest', '-p', 'ciqueue.pytest', '--queue', queue, nodeid], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=30) + assert worker.returncode == pytest.ExitCode.TESTS_FAILED, worker.stdout + key = 'build:identity:error-reports' + data = json.loads(zlib.decompress(self.redis.hget(key, nodeid))) + if corruption == 'identity': + data['call']['nodeid'] = 'another-test' + elif corruption == 'xfail': + data['call']['wasxfail'] = '' + else: + data['setup'] = dict(data['call'], when='setup', outcome='skipped', + longrepr=['test_all.py', 1, 'Skipped: contradictory setup']) + self.redis.hset(key, nodeid, zlib.compress(json.dumps(data).encode('utf-8'))) + reporter = subprocess.run( + [sys.executable, '-m', 'pytest', '-p', 'ciqueue.pytest_report', '--queue', queue, nodeid], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=30) + + assert reporter.returncode == pytest.ExitCode.USAGE_ERROR, reporter.stdout + assert nodeid in reporter.stdout + + def test_report_preserves_rendered_user_properties(self, tmp_path): + test_file = tmp_path / 'test_properties.py' + test_file.write_text(textwrap.dedent('''\ + from pathlib import Path + + def test_failure(record_property): + record_property("artifact", Path("output.txt")) + assert False, "original failure" + ''')) + queue = 'redis://localhost:6379/0?worker=0&build=properties&timeout=5' + for plugin in ('ciqueue.pytest', 'ciqueue.pytest_report'): + xml_file = tmp_path / (plugin + '.xml') + result = subprocess.run( + [sys.executable, '-m', 'pytest', '-p', plugin, '--queue', queue, + '--junitxml', str(xml_file), '-o', 'junit_family=xunit1', str(test_file)], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=30) + + assert result.returncode == pytest.ExitCode.TESTS_FAILED, result.stdout + case = next(ET.parse(xml_file).iter('testcase')) + assert 'original failure' in case.find('failure').text + assert case.find('properties/property').attrib == {'name': 'artifact', 'value': 'output.txt'}