|
| 1 | +import traceback |
1 | 2 | from dataclasses import dataclass |
2 | 3 | from dataclasses import field |
3 | | - |
4 | | -import pytest |
| 4 | +from pathlib import Path |
5 | 5 |
|
6 | 6 | from statemachine import State |
7 | 7 | from statemachine import StateMachine |
8 | 8 | from statemachine.event import Event |
| 9 | +from statemachine.io.scxml.processor import SCXMLProcessor |
9 | 10 |
|
10 | 11 | """ |
11 | 12 | Test cases as defined by W3C SCXML Test Suite |
|
18 | 19 | """ # noqa: E501 |
19 | 20 |
|
20 | 21 |
|
21 | | -@dataclass(frozen=True, unsafe_hash=True) |
| 22 | +@dataclass(frozen=True, unsafe_hash=True, kw_only=True) |
| 23 | +class DebugEvent: |
| 24 | + source: str |
| 25 | + event: str |
| 26 | + data: str |
| 27 | + target: str |
| 28 | + |
| 29 | + |
| 30 | +@dataclass(frozen=True, unsafe_hash=True, kw_only=True) |
22 | 31 | class DebugListener: |
23 | | - events: list = field(default_factory=list) |
| 32 | + events: list[DebugEvent] = field(default_factory=list) |
24 | 33 |
|
25 | 34 | def on_transition(self, event: Event, source: State, target: State, event_data): |
26 | 35 | self.events.append( |
27 | | - ( |
28 | | - f"{source and source.id}", |
29 | | - f"{event and event.id}", |
30 | | - f"{event_data.trigger_data.kwargs}", |
31 | | - f"{target.id}", |
| 36 | + DebugEvent( |
| 37 | + source=f"{source and source.id}", |
| 38 | + event=f"{event and event.id}", |
| 39 | + data=f"{event_data.trigger_data.kwargs}", |
| 40 | + target=f"{target.id}", |
32 | 41 | ) |
33 | 42 | ) |
34 | 43 |
|
35 | 44 |
|
36 | | -@pytest.mark.scxml() |
37 | | -def test_scxml_usecase(testcase_path, processor): |
| 45 | +@dataclass(kw_only=True) |
| 46 | +class FailedMark: |
| 47 | + reason: str |
| 48 | + events: list[DebugEvent] |
| 49 | + is_assertion_error: bool |
| 50 | + exception: Exception |
| 51 | + logs: str |
| 52 | + configuration: list[str] = field(default_factory=list) |
| 53 | + |
| 54 | + @staticmethod |
| 55 | + def _get_header(report: str) -> str: |
| 56 | + header_end_index = report.find("---") |
| 57 | + return report[:header_end_index] |
| 58 | + |
| 59 | + def write_fail_markdown(self, testcase_path: Path): |
| 60 | + fail_file_path = testcase_path.with_suffix(".fail.md") |
| 61 | + if not self.is_assertion_error: |
| 62 | + exception_traceback = "".join( |
| 63 | + traceback.format_exception( |
| 64 | + type(self.exception), self.exception, self.exception.__traceback__ |
| 65 | + ) |
| 66 | + ) |
| 67 | + else: |
| 68 | + exception_traceback = "Assertion of the testcase failed." |
| 69 | + |
| 70 | + report = f"""# Testcase: {testcase_path.stem} |
| 71 | +
|
| 72 | +{self.reason} |
| 73 | +
|
| 74 | +Final configuration: `{self.configuration if self.configuration else 'No configuration'}` |
| 75 | +
|
| 76 | +--- |
| 77 | +
|
| 78 | +## Logs |
| 79 | +```py |
| 80 | +{self.logs if self.logs else 'No logs'} |
| 81 | +``` |
| 82 | +
|
| 83 | +## "On transition" events |
| 84 | +```py |
| 85 | +{'\n'.join(map(repr, self.events)) if self.events else 'No events'} |
| 86 | +``` |
| 87 | +
|
| 88 | +## Traceback |
| 89 | +```py |
| 90 | +{exception_traceback} |
| 91 | +``` |
| 92 | +""" |
| 93 | + |
| 94 | + if fail_file_path.exists(): |
| 95 | + last_report = fail_file_path.read_text() |
| 96 | + |
| 97 | + if self._get_header(report) == self._get_header(last_report): |
| 98 | + return |
| 99 | + |
| 100 | + with fail_file_path.open("w") as fail_file: |
| 101 | + fail_file.write(report) |
| 102 | + |
| 103 | + |
| 104 | +def test_scxml_usecase(testcase_path: Path, caplog): |
38 | 105 | # from statemachine.contrib.diagram import DotGraphMachine |
39 | 106 |
|
40 | 107 | # DotGraphMachine(sm_class).get_graph().write_png( |
41 | 108 | # testcase_path.parent / f"{testcase_path.stem}.png" |
42 | 109 | # ) |
43 | | - debug = DebugListener() |
44 | | - sm = processor.start(listeners=[debug]) |
45 | | - assert isinstance(sm, StateMachine) |
46 | | - assert sm.current_state.id == "pass", debug |
| 110 | + sm: "StateMachine | None" = None |
| 111 | + try: |
| 112 | + debug = DebugListener() |
| 113 | + processor = SCXMLProcessor() |
| 114 | + processor.parse_scxml_file(testcase_path) |
| 115 | + |
| 116 | + sm = processor.start(listeners=[debug]) |
| 117 | + assert isinstance(sm, StateMachine) |
| 118 | + assert "pass" in {s.id for s in sm.configuration}, debug |
| 119 | + except Exception as e: |
| 120 | + # Import necessary module |
| 121 | + reason = f"{e.__class__.__name__}: {e.__class__.__doc__}" |
| 122 | + is_assertion_error = isinstance(e, AssertionError) |
| 123 | + fail_mark = FailedMark( |
| 124 | + reason=reason, |
| 125 | + is_assertion_error=is_assertion_error, |
| 126 | + events=debug.events, |
| 127 | + exception=e, |
| 128 | + logs=caplog.text, |
| 129 | + configuration=[s.id for s in sm.configuration] if sm else [], |
| 130 | + ) |
| 131 | + fail_mark.write_fail_markdown(testcase_path) |
| 132 | + raise |
0 commit comments