|
3 | 3 |
|
4 | 4 | import pytest |
5 | 5 |
|
| 6 | +from .logging import CapturedRequestLog |
| 7 | + |
6 | 8 | # Fixed timestamp for deterministic tests for asserting on message contents |
7 | 9 | FAKE_TIMESTAMP = 1755750946.721395 |
8 | 10 |
|
9 | 11 |
|
10 | | -class CapturedRequestLog: |
11 | | - """Log of requests and responses for snapshot assertions. |
12 | | -
|
13 | | - The log captures the raw bytes of each request and response along with |
14 | | - a label indicating the direction of the message. |
15 | | - """ |
16 | | - |
17 | | - def __init__(self) -> None: |
18 | | - """Initialize the request log.""" |
19 | | - self.entries: list[tuple[str, bytes]] = [] |
20 | | - |
21 | | - def add_log_entry(self, label: str, data: bytes) -> None: |
22 | | - """Add a request entry.""" |
23 | | - self.entries.append((label, data)) |
24 | | - |
25 | | - def __repr__(self): |
26 | | - """Return a string representation of the log entries. |
27 | | -
|
28 | | - This assumes that the client will behave in a request-response manner, |
29 | | - so each request is followed by a response. If a test uses non-deterministic |
30 | | - message order, this may not be accurate and the test would need to decode |
31 | | - the raw messages and remove any ordering assumptions. |
32 | | - """ |
33 | | - lines = [] |
34 | | - for label, data in self.entries: |
35 | | - lines.append(label) |
36 | | - lines.extend(self._hexdump(data)) |
37 | | - return "\n".join(lines) |
38 | | - |
39 | | - def _hexdump(self, data: bytes, bytes_per_line: int = 16) -> list[str]: |
40 | | - """Print a hexdump of the given bytes object in a tcpdump/hexdump -C style. |
41 | | -
|
42 | | - This makes the packets easier to read and compare in test snapshots. |
43 | | -
|
44 | | - Args: |
45 | | - data: The bytes object to print. |
46 | | - bytes_per_line: The number of bytes to display per line (default is 16). |
47 | | - """ |
48 | | - |
49 | | - # Use '.' for non-printable characters (ASCII < 32 or > 126) |
50 | | - def to_printable_ascii(byte_val): |
51 | | - return chr(byte_val) if 32 <= byte_val <= 126 else "." |
52 | | - |
53 | | - offset = 0 |
54 | | - lines = [] |
55 | | - while offset < len(data): |
56 | | - chunk = data[offset : offset + bytes_per_line] |
57 | | - # Format the hex values, space-padded to ensure alignment |
58 | | - hex_values = " ".join(f"{byte:02x}" for byte in chunk) |
59 | | - # Pad hex string to a fixed width so ASCII column lines up |
60 | | - # 3 chars per byte ('xx ') for a full line of 16 bytes |
61 | | - padded_hex = f"{hex_values:<{bytes_per_line * 3}}" |
62 | | - # Format the ASCII values |
63 | | - ascii_values = "".join(to_printable_ascii(byte) for byte in chunk) |
64 | | - lines.append(f"{offset:08x} {padded_hex} |{ascii_values}|") |
65 | | - offset += bytes_per_line |
66 | | - return lines |
67 | | - |
68 | | - |
69 | 12 | @pytest.fixture |
70 | 13 | def deterministic_message_fixtures() -> Generator[None, None, None]: |
71 | 14 | """Fixture to use predictable get_next_int and timestamp values for each test. |
|
0 commit comments