Changeset View
Changeset View
Standalone View
Standalone View
tests/atf_python/atf_pytest.py
- This file was added.
| import types | |||||||||
| from typing import Any | |||||||||
| from typing import Dict | |||||||||
| from typing import List | |||||||||
| from typing import NamedTuple | |||||||||
| from typing import Tuple | |||||||||
ngie: Is intended to work with 3.6+ instead of supporting (soon to be) 3.9+ out of the box? 3.9+… | |||||||||
Done Inline ActionsI understand and value importance of typing - it indeed greatly helps to avoid weird failures in larger projects. melifaro: I understand and value importance of typing - it indeed greatly helps to avoid weird failures… | |||||||||
| import pytest | |||||||||
| class ATFCleanupItem(pytest.Item): | |||||||||
| def runtest(self): | |||||||||
| """Runs cleanup procedure for the test instead of the test""" | |||||||||
| instance = self.parent.cls() | |||||||||
| instance.cleanup(self.nodeid) | |||||||||
| def setup_method_noop(self, method): | |||||||||
| """Overrides runtest setup method""" | |||||||||
| pass | |||||||||
| def teardown_method_noop(self, method): | |||||||||
| """Overrides runtest teardown method""" | |||||||||
| pass | |||||||||
| class ATFTestObj(object): | |||||||||
| def __init__(self, obj, has_cleanup): | |||||||||
| # Use nodeid without name to properly name class-derived tests | |||||||||
| self.ident = obj.nodeid.split("::", 1)[1] | |||||||||
| self.description = self._get_test_description(obj) | |||||||||
| self.has_cleanup = has_cleanup | |||||||||
| self.obj = obj | |||||||||
| def _get_test_description(self, obj): | |||||||||
| """Returns first non-empty line from func docstring or func name""" | |||||||||
| docstr = obj.function.__doc__ | |||||||||
| if docstr: | |||||||||
| for line in docstr.split("\n"): | |||||||||
| if line: | |||||||||
| return line | |||||||||
| return obj.name | |||||||||
| def _convert_marks(self, obj) -> Dict[str, Any]: | |||||||||
| wj_func = lambda x: " ".join(x) # noqa: E731 | |||||||||
Lint: PEP8 E731 do not assign a lambda expression, use a def Lint: PEP8 E731: do not assign a lambda expression, use a def | |||||||||
| _map: Dict[str, Dict] = { | |||||||||
Lint: PEP8 E701 multiple statements on one line (colon) Lint: PEP8 E701: multiple statements on one line (colon) | |||||||||
| "require_user": {"name": "require.user"}, | |||||||||
| "require_arch": {"name": "require.arch", "fmt": wj_func}, | |||||||||
| "require_diskspace": {"name": "require.diskspace"}, | |||||||||
| "require_files": {"name": "require.files", "fmt": wj_func}, | |||||||||
| "require_machine": {"name": "require.machine", "fmt": wj_func}, | |||||||||
| "require_memory": {"name": "require.memory"}, | |||||||||
| "require_progs": {"name": "require.progs", "fmt": wj_func}, | |||||||||
| "timeout": {}, | |||||||||
| } | |||||||||
| ret = {} | |||||||||
| for mark in obj.iter_markers(): | |||||||||
| if mark.name in _map: | |||||||||
| name = _map[mark.name].get("name", mark.name) | |||||||||
| if "fmt" in _map[mark.name]: | |||||||||
| val = _map[mark.name]["fmt"](mark.args[0]) | |||||||||
| else: | |||||||||
| val = mark.args[0] | |||||||||
| ret[name] = val | |||||||||
| return ret | |||||||||
| def as_lines(self) -> List[str]: | |||||||||
| """Output test definition in ATF-specific format""" | |||||||||
| ret = [] | |||||||||
| ret.append("ident: {}".format(self.ident)) | |||||||||
| ret.append("descr: {}".format(self._get_test_description(self.obj))) | |||||||||
| if self.has_cleanup: | |||||||||
| ret.append("has.cleanup: true") | |||||||||
| for key, value in self._convert_marks(self.obj).items(): | |||||||||
| ret.append("{}: {}".format(key, value)) | |||||||||
| return ret | |||||||||
| class ATFHandler(object): | |||||||||
| class ReportState(NamedTuple): | |||||||||
| state: str | |||||||||
Lint: PEP8 E701 multiple statements on one line (colon) Lint: PEP8 E701: multiple statements on one line (colon) | |||||||||
Not Done Inline ActionsAny is a bit loose. ngie: `Any` is a bit loose.
Could you please create a simple container object to handle this instead? | |||||||||
Done Inline ActionsDoes it look better now? melifaro: Does it look better now? | |||||||||
| reason: str | |||||||||
Lint: PEP8 E701 multiple statements on one line (colon) Lint: PEP8 E701: multiple statements on one line (colon) | |||||||||
| def __init__(self): | |||||||||
| self._tests_state_map: Dict[str, ReportStatus] = {} | |||||||||
Lint: PEP8 E701 multiple statements on one line (colon) Lint: PEP8 E701: multiple statements on one line (colon) | |||||||||
| def override_runtest(self, obj): | |||||||||
| # Override basic runtest command | |||||||||
| obj.runtest = types.MethodType(ATFCleanupItem.runtest, obj) | |||||||||
| # Override class setup/teardown | |||||||||
| obj.parent.cls.setup_method = ATFCleanupItem.setup_method_noop | |||||||||
| obj.parent.cls.teardown_method = ATFCleanupItem.teardown_method_noop | |||||||||
| def get_object_cleanup_class(self, obj): | |||||||||
| if hasattr(obj, "parent") and obj.parent is not None: | |||||||||
| if hasattr(obj.parent, "cls") and obj.parent.cls is not None: | |||||||||
| if hasattr(obj.parent.cls, "cleanup"): | |||||||||
| return obj.parent.cls | |||||||||
| return None | |||||||||
| def has_object_cleanup(self, obj): | |||||||||
| return self.get_object_cleanup_class(obj) is not None | |||||||||
| def list_tests(self, tests: List[str]): | |||||||||
| print('Content-Type: application/X-atf-tp; version="1"') | |||||||||
| print() | |||||||||
| for test_obj in tests: | |||||||||
| has_cleanup = self.has_object_cleanup(test_obj) | |||||||||
| atf_test = ATFTestObj(test_obj, has_cleanup) | |||||||||
| for line in atf_test.as_lines(): | |||||||||
| print(line) | |||||||||
| print() | |||||||||
| def set_report_state(self, test_name: str, state: str, reason: str): | |||||||||
| self._tests_state_map[test_name] = self.ReportState(state, reason) | |||||||||
| def _extract_report_reason(self, report): | |||||||||
| data = report.longrepr | |||||||||
| if data is None: | |||||||||
| return None | |||||||||
| if isinstance(data, Tuple): | |||||||||
| # ('/path/to/test.py', 23, 'Skipped: unable to test') | |||||||||
| reason = data[2] | |||||||||
| for prefix in "Skipped: ": | |||||||||
| if reason.startswith(prefix): | |||||||||
| reason = reason[len(prefix):] | |||||||||
| return reason | |||||||||
| else: | |||||||||
| # string/ traceback / exception report. Capture the last line | |||||||||
| return str(data).split("\n")[-1] | |||||||||
| return None | |||||||||
| def add_report(self, report): | |||||||||
| # MAP pytest report state to the atf-desired state | |||||||||
| # | |||||||||
| # ATF test states: | |||||||||
| # (1) expected_death, (2) expected_exit, (3) expected_failure | |||||||||
| # (4) expected_signal, (5) expected_timeout, (6) passed | |||||||||
| # (7) skipped, (8) failed | |||||||||
| # | |||||||||
| # Note that ATF don't have the concept of "soft xfail" - xpass | |||||||||
| # is a failure. It also calls teardown routine in a separate | |||||||||
| # process, thus teardown states (pytest-only) are handled as | |||||||||
| # body continuation. | |||||||||
| # (stage, state, wasxfail) | |||||||||
| # Just a passing test: WANT: passed | |||||||||
| # GOT: (setup, passed, F), (call, passed, F), (teardown, passed, F) | |||||||||
| # | |||||||||
| # Failing body test: WHAT: failed | |||||||||
| # GOT: (setup, passed, F), (call, failed, F), (teardown, passed, F) | |||||||||
| # | |||||||||
| # pytest.skip test decorator: WANT: skipped | |||||||||
| # GOT: (setup,skipped, False), (teardown, passed, False) | |||||||||
| # | |||||||||
| # pytest.skip call inside test function: WANT: skipped | |||||||||
| # GOT: (setup, passed, F), (call, skipped, F), (teardown,passed, F) | |||||||||
| # | |||||||||
| # mark.xfail decorator+pytest.xfail: WANT: expected_failure | |||||||||
| # GOT: (setup, passed, F), (call, skipped, T), (teardown, passed, F) | |||||||||
| # | |||||||||
| # mark.xfail decorator+pass: WANT: failed | |||||||||
| # GOT: (setup, passed, F), (call, passed, T), (teardown, passed, F) | |||||||||
| test_name = report.location[2] | |||||||||
| stage = report.when | |||||||||
| state = report.outcome | |||||||||
| reason = self._extract_report_reason(report) | |||||||||
| # We don't care about strict xfail - it gets translated to False | |||||||||
| if stage == "setup": | |||||||||
| if state in ("skipped", "failed"): | |||||||||
| # failed init -> failed test, skipped setup -> xskip | |||||||||
| # for the whole test | |||||||||
| self.set_report_state(test_name, state, reason) | |||||||||
Done Inline ActionsDRY potential: this statement is called 3 times the same way. ngie: DRY potential: this statement is called 3 times the same way. | |||||||||
| elif stage == "call": | |||||||||
| # "call" stage shouldn't matter if setup failed | |||||||||
| if test_name in self._tests_state_map: | |||||||||
| if self._tests_state_map[test_name].state == "failed": | |||||||||
| return | |||||||||
| if state == "failed": | |||||||||
| # Record failure & override "skipped" state | |||||||||
| self.set_report_state(test_name, state, reason) | |||||||||
| elif state == "skipped": | |||||||||
| if hasattr(reason, "wasxfail"): | |||||||||
| # xfail() called in the test body | |||||||||
| state = "expected_failure" | |||||||||
| else: | |||||||||
| # skip inside the body | |||||||||
| pass | |||||||||
| self.set_report_state(test_name, state, reason) | |||||||||
| elif state == "passed": | |||||||||
| if hasattr(reason, "wasxfail"): | |||||||||
| # the test was expected to fail but didn't | |||||||||
| # mark as hard failure | |||||||||
| state = "failed" | |||||||||
| self.set_report_state(test_name, state, reason) | |||||||||
| elif stage == "teardown": | |||||||||
| if state == "failed": | |||||||||
| # teardown should be empty, as the cleanup | |||||||||
| # procedures should be implemented as a separate | |||||||||
| # function/method, so mark teardown failure as | |||||||||
Done Inline Actions
This is a much better idiom. It avoids creating a list by walking self._tests_state_map, constructing a temporary list and grabbing the 0th element, by just using an iterator object to grab the first element (and relying on dict.keys() to be iterated over by default with dict.__iter__()). ngie: This is a much better idiom. It avoids creating a list by walking `self._tests_state_map`… | |||||||||
| # global failure | |||||||||
| self.set_report_state(test_name, state, reason) | |||||||||
| def write_report(self, path): | |||||||||
| if self._tests_state_map: | |||||||||
| # If we're executing in ATF mode, there has to be just one test | |||||||||
Not Done Inline Actionsline is always guaranteed to be a UTF-8 (or whatever the default encoding is) str? ngie: `line` is always guaranteed to be a UTF-8 (or whatever the default encoding is) `str`? | |||||||||
Done Inline ActionsWell, it should be normally the case, but I'm not sure if one need to bother to address weird corner cases - ATF runner should address that. melifaro: Well, it __should__ be normally the case, but I'm not sure if one need to bother to address… | |||||||||
| # Anyway, deterministically pick the first one | |||||||||
| first_test_name = next(iter(self._tests_state_map)) | |||||||||
| test = self._tests_state_map[first_test_name] | |||||||||
| if test.state == "passed": | |||||||||
| line = test.state | |||||||||
| else: | |||||||||
| line = "{}: {}".format(test.state, test.reason) | |||||||||
| with open(path, mode="w") as f: | |||||||||
| print(line, file=f) | |||||||||
Is intended to work with 3.6+ instead of supporting (soon to be) 3.9+ out of the box? 3.9+ supports native type hints through the generic data types.