|
| 1 | +import pytest |
| 2 | +from allure_commons.utils import now |
| 3 | +from allure_commons.model2 import Label, Link |
| 4 | +from allure_commons.model2 import Status |
| 5 | + |
| 6 | +from allure_commons.types import LabelType, AttachmentType |
| 7 | +from allure_commons.utils import platform_label |
| 8 | +from allure_commons.utils import host_tag, thread_tag |
| 9 | +from .utils import get_uuid |
| 10 | +from .utils import get_step_name |
| 11 | +from .utils import get_status_details |
| 12 | +from allure_commons.model2 import StatusDetails |
| 13 | +from functools import partial, reduce |
| 14 | +from allure_commons.lifecycle import AllureLifecycle |
| 15 | +from .utils import get_full_name |
| 16 | + |
| 17 | + |
| 18 | +class PytestBDDListener(object): |
| 19 | + def __init__(self): |
| 20 | + self.lifecycle = AllureLifecycle() |
| 21 | + self.host = host_tag() |
| 22 | + self.thread = thread_tag() |
| 23 | + |
| 24 | + def _scenario_finalizer(self, scenario): |
| 25 | + for step in scenario.steps: |
| 26 | + step_uuid = get_uuid(str(id(step))) |
| 27 | + with self.lifecycle.update_step(uuid=step_uuid) as step_result: |
| 28 | + if step_result: |
| 29 | + step_result.status = Status.SKIPPED |
| 30 | + self.lifecycle.stop_step(uuid=step_uuid) |
| 31 | + |
| 32 | + @pytest.hookimpl |
| 33 | + def pytest_bdd_before_scenario(self, request, feature, scenario): |
| 34 | + uuid = get_uuid(request.node.nodeid) |
| 35 | + full_name = get_full_name(feature, scenario) |
| 36 | + with self.lifecycle.schedule_test_case(uuid=uuid) as test_result: |
| 37 | + test_result.fullName = full_name |
| 38 | + test_result.start = now() |
| 39 | + test_result.labels.append(Label(name=LabelType.HOST, value=self.host)) |
| 40 | + test_result.labels.append(Label(name=LabelType.THREAD, value=self.thread)) |
| 41 | + test_result.labels.append(Label(name=LabelType.FRAMEWORK, value="pytest-bdd")) |
| 42 | + test_result.labels.append(Label(name=LabelType.LANGUAGE, value=platform_label())) |
| 43 | + test_result.labels.append(Label(name=LabelType.FEATURE, value=feature.name)) |
| 44 | + |
| 45 | + finalizer = partial(self._scenario_finalizer, scenario) |
| 46 | + request.node.addfinalizer(finalizer) |
| 47 | + |
| 48 | + @pytest.hookimpl |
| 49 | + def pytest_bdd_after_scenario(self, request, feature, scenario): |
| 50 | + uuid = get_uuid(request.node.nodeid) |
| 51 | + with self.lifecycle.update_test_case(uuid=uuid) as test_result: |
| 52 | + test_result.stop = now() |
| 53 | + |
| 54 | + @pytest.hookimpl |
| 55 | + def pytest_bdd_before_step_call(self, request, feature, scenario, step, step_func, step_func_args): |
| 56 | + parent_uuid = get_uuid(request.node.nodeid) |
| 57 | + uuid = get_uuid(str(id(step))) |
| 58 | + with self.lifecycle.start_step(parent_uuid=parent_uuid, uuid=uuid) as step_result: |
| 59 | + step_result.name = get_step_name(step) |
| 60 | + |
| 61 | + @pytest.hookimpl |
| 62 | + def pytest_bdd_after_step(self, request, feature, scenario, step, step_func, step_func_args): |
| 63 | + uuid = get_uuid(str(id(step))) |
| 64 | + with self.lifecycle.update_step(uuid=uuid) as step_result: |
| 65 | + step_result.status = Status.PASSED |
| 66 | + self.lifecycle.stop_step(uuid=uuid) |
| 67 | + |
| 68 | + @pytest.hookimpl |
| 69 | + def pytest_bdd_step_error(self, request, feature, scenario, step, step_func, step_func_args, exception): |
| 70 | + uuid = get_uuid(str(id(step))) |
| 71 | + with self.lifecycle.update_step(uuid=uuid) as step_result: |
| 72 | + step_result.status = Status.FAILED |
| 73 | + step_result.statusDetails = get_status_details(exception) |
| 74 | + self.lifecycle.stop_step(uuid=uuid) |
| 75 | + |
| 76 | + @pytest.hookimpl |
| 77 | + def pytest_bdd_step_func_lookup_error(self, request, feature, scenario, step, exception): |
| 78 | + uuid = get_uuid(str(id(step))) |
| 79 | + with self.lifecycle.update_step(uuid=uuid) as step_result: |
| 80 | + step_result.status = Status.BROKEN |
| 81 | + self.lifecycle.stop_step(uuid=uuid) |
| 82 | + |
| 83 | + @pytest.hookimpl(hookwrapper=True) |
| 84 | + def pytest_runtest_makereport(self, item, call): |
| 85 | + report = (yield).get_result() |
| 86 | + |
| 87 | + status = reduce(lambda final_status, current_status: final_status or getattr(report, current_status, None), |
| 88 | + ["failed", "passed", "skipped"]) |
| 89 | + |
| 90 | + status_details = StatusDetails( |
| 91 | + message=call.excinfo.exconly(), |
| 92 | + trace=report.longreprtext) if call.excinfo else None |
| 93 | + |
| 94 | + uuid = get_uuid(report.nodeid) |
| 95 | + with self.lifecycle.update_test_case(uuid=uuid) as test_result: |
| 96 | + |
| 97 | + if test_result and report.when == "setup": |
| 98 | + test_result.status = status |
| 99 | + test_result.statusDetails = status_details |
| 100 | + |
| 101 | + if report.when == "call" and test_result: |
| 102 | + if test_result.status not in [Status.PASSED, Status.FAILED]: |
| 103 | + test_result.status = status |
| 104 | + test_result.statusDetails = status_details |
| 105 | + |
| 106 | + if report.when == "teardown" and test_result: |
| 107 | + if test_result.status == Status.PASSED and status != Status.PASSED: |
| 108 | + test_result.status = status |
| 109 | + test_result.statusDetails = status_details |
| 110 | + |
| 111 | + if report.when == 'teardown': |
| 112 | + self.lifecycle.write_test_case(uuid=uuid) |
0 commit comments