Skip to content

Commit

Permalink
Store logged test data as base64 strings in JSON Lines
Browse files Browse the repository at this point in the history
  • Loading branch information
@bessman
bessman committed May 22, 2024
1 parent 78edda8 commit 2b047d7
Show file tree
Hide file tree
Showing 3 changed files with 145 additions and 74 deletions.
4 changes: 2 additions & 2 deletions README.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ Next:
4. Disconnect the device.
5. Run `pytest --replay my_serial_app.py`. The test will pass!

The logged traffic will be stored in JSON files in the same directory as your test files, and will have the same names as the test files except with a .json extension instead of .py. For example, if your project layout is:
The logged traffic will be stored as JSON Lines, with one file per test file and one line per test, in the same directory as your test files. The files will have the same names as the test files except with a .jsonl extension instead of .py. For example, if your project layout is:

```shell
├── src
Expand All @@ -54,7 +54,7 @@ The logged traffic will be stored in JSON files in the same directory as your te
│ ├── test_myproject.py
```

Then after running `pytest --record`, the test/ directory will contain a new file, test_myproject.json, containing the recorded serial traffic from the tests.
Then after running `pytest --record`, the test/ directory will contain a new file, test_myproject.jsonl, containing the recorded serial traffic from the tests.

## Why

Expand Down
102 changes: 59 additions & 43 deletions src/pytest_reserial/reserial.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,15 +2,16 @@

from __future__ import annotations

import base64
import json
from enum import IntEnum
from pathlib import Path
from typing import Callable, Dict, Generator, List, Tuple
from typing import Callable, Dict, Iterator, Literal, Tuple

import pytest
from serial import Serial # type: ignore[import-untyped]

TrafficLog = Dict[str, List[int]]
TrafficLog = Dict[Literal["rx", "tx"], bytes]
PatchMethods = Tuple[
Callable[[Serial, int], bytes],
Callable[[Serial, bytes], int],
Expand Down Expand Up @@ -60,7 +61,7 @@ def reconfigure_port_patch(
def reserial(
monkeypatch: pytest.MonkeyPatch,
request: pytest.FixtureRequest,
) -> Generator[None, None, None]:
) -> Iterator[None]:
"""Record or replay serial traffic.
Raises
Expand All @@ -72,9 +73,9 @@ def reserial(
replay = request.config.getoption("--replay")
mode = Mode(replay | record << 1)

logpath = Path(request.path).parent / (Path(request.path).stem + ".json")
testname = request.node.name
log = get_traffic_log(mode, logpath, testname)
log_path = Path(request.path).parent / (Path(request.path).stem + ".jsonl")
test_name = request.node.name
log = get_traffic_log(mode, log_path, test_name)

read_patch, write_patch, open_patch, close_patch = get_patched_methods(mode, log)
monkeypatch.setattr(Serial, "read", read_patch)
Expand All @@ -86,7 +87,7 @@ def reserial(
yield

if mode == Mode.RECORD:
write_log(log, logpath, testname)
write_log(log, log_path, test_name)
return

if log["rx"] or log["tx"]:
Expand All @@ -98,16 +99,16 @@ def reserial(
pytest.fail(msg)


def get_traffic_log(mode: Mode, logpath: Path, testname: str) -> TrafficLog:
def get_traffic_log(mode: Mode, log_path: Path, test_name: str) -> TrafficLog:
"""Load recorded traffic (replay) or create an empty log (record).
Parameters
----------
mode : Mode
The requested mode of operation, i.e. `REPLAY`, `RECORD`, or `DONT_PATCH`.
logpath: str
log_path: str
The name of the file where recorded traffic is logged.
testname: str
test_name: str
The name of the currently running test, which is used as a key in the log file.
Returns
Expand All @@ -120,17 +121,25 @@ def get_traffic_log(mode: Mode, logpath: Path, testname: str) -> TrafficLog:
------
ValueError
If both '--replay' and '--record' were specified.
If no the test has no recorded traffic .
"""
if mode == Mode.INVALID:
msg = "Choose one of 'replay' or 'record', not both"
raise ValueError(msg)

log: TrafficLog = {"rx": [], "tx": []}
log: TrafficLog = {"rx": b"", "tx": b""}

if mode == Mode.REPLAY:
with Path.open(logpath) as logfile:
logs = json.load(logfile)
log = logs[testname]
with Path.open(log_path) as fin:
for line in fin:
if log := json.loads(line).get(test_name):
break
else:
msg = f"No recorded traffic for test: {test_name}"
raise ValueError(msg)

log["rx"] = base64.b64decode(log["rx"])
log["tx"] = base64.b64decode(log["tx"])

return log

Expand Down Expand Up @@ -200,12 +209,12 @@ def replay_write(
_pytest.outcomes.Failed
If written data does not match recorded data.
"""
if list(data) == log["tx"][: len(data)]:
if data == log["tx"][: len(data)]:
log["tx"] = log["tx"][len(data) :]
else:
msg = (
"Written data does not match recorded data: "
f"{list(data)} != {log['tx'][: len(data)]}"
f"{data!r} != {log['tx'][: len(data)]!r}"
)
pytest.fail(msg)

Expand Down Expand Up @@ -267,7 +276,7 @@ def record_write(self: Serial, data: bytes) -> int:
Monkeypatch this method over Serial.write to record traffic. Parameters and
return values are identical to Serial.write.
"""
log["tx"] += list(data)
log["tx"] += data
written: int = real_write(self, data)
return written

Expand All @@ -278,46 +287,53 @@ def record_read(self: Serial, size: int = 1) -> bytes:
return values are identical to Serial.read.
"""
data: bytes = real_read(self, size)
log["rx"] += list(data)
log["rx"] += data
return data

return record_read, record_write, Serial.open, Serial.close


def write_log(
log: TrafficLog,
logpath: Path,
testname: str,
log_path: Path,
test_name: str,
) -> None:
"""Write recorded traffic to log file.
Parameters
----------
log: dict[str, list[int]]
Dictionary holding recorded traffic.
logpath: str
log_path: str
The name of the file where recorded traffic is logged.
testname: str
test_name: str
The name of the currently running test, which is used as a key in the log file.
"""
try:
# If the file exists, read its contents.
with Path.open(logpath) as logfile:
logs = json.load(logfile)
except FileNotFoundError:
logs = {}

logs[testname] = log
logs_str = json.dumps(logs, indent=4)

# Print traffic logs on a single line if jsbeautifier is available.
try:
import jsbeautifier # type: ignore[import-untyped]

logs_str = jsbeautifier.beautify(logs_str)
except ImportError:
pass

# Wipe the file if it exists, or create a new file if it doesn't.
with Path.open(logpath, mode="w") as logfile:
logfile.write(logs_str)
# Make sure log file exists.
log_path.touch()
# Write new data to temporary file.
tmp_path = Path(test_name + "_" + hex(abs(hash(test_name))).lstrip("0x"))

with log_path.open("r") as fin, tmp_path.open("w") as fout:
seen = False
rx = base64.b64encode(bytes(log["rx"])).decode("ascii")
tx = base64.b64encode(bytes(log["tx"])).decode("ascii")
new_line = json.dumps({test_name: {"rx": rx, "tx": tx}}) + "\n"

# Recorded traffic is stored as JSON Lines. Parse one line at a time.
for old_line in fin:
test_data = json.loads(old_line)

# Each record contains a single key.
if test_name in test_data:
fout.write(new_line)
seen = True
continue

fout.write(old_line)

if not seen:
fout.write(new_line)

# Overwrite old log file.
Path(tmp_path).rename(log_path)
113 changes: 84 additions & 29 deletions tests/test_reserial.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,37 +3,31 @@
from serial import Serial

TEST_RX = b"\x01"
TEST_RX_ENC = "AQ=="
TEST_TX = b"\x02"
TEST_TX_ENC = "Ag=="
TEST_FILE = f"""
import serial
def test_reserial(reserial):
s = serial.Serial(port="/dev/ttyUSB0")
s.write({TEST_TX})
assert s.read() == {TEST_RX}
s.write({TEST_TX!r})
assert s.read() == {TEST_RX!r}
def test_reserial2(reserial):
s = serial.Serial(port="/dev/ttyUSB0")
s.write({TEST_TX})
assert s.read() == {TEST_RX}
s.write({TEST_TX!r})
assert s.read() == {TEST_RX!r}
"""
TEST_FILE_BAD_TX = f"""
import serial
def test_reserial(reserial):
s = serial.Serial(port="/dev/ttyUSB0")
s.write({TEST_RX})
assert s.read() == {TEST_RX}
"""
TEST_JSON = f"""
{{
"test_reserial": {{
"rx": {list(TEST_RX)},
"tx": {list(TEST_TX)}
}},
"test_reserial2": {{
"rx": {list(TEST_RX)},
"tx": {list(TEST_TX)}
}}
}}
"""
import serial
def test_reserial(reserial):
s = serial.Serial(port="/dev/ttyUSB0")
s.write({TEST_RX!r})
assert s.read() == {TEST_RX!r}
"""
TEST_JSONL = (
f'{{"test_reserial": {{"rx": "{TEST_RX_ENC}", "tx": "{TEST_TX_ENC}"}}}}\n'
f'{{"test_reserial2": {{"rx": "{TEST_RX_ENC}", "tx": "{TEST_TX_ENC}"}}}}\n'
)


def test_record(monkeypatch, pytester):
Expand All @@ -57,15 +51,63 @@ def patch_close(self: Serial) -> None:
monkeypatch.setattr(Serial, "close", patch_close)
result = pytester.runpytest("--record")

with open("test_record.json", "r") as f:
recording = json.load(f)
with open("test_record.jsonl", "r") as f:
recording = [json.loads(line) for line in f.readlines()]

expected = [json.loads(line) for line in TEST_JSONL.splitlines()]

assert recording == expected
assert result.ret == 0


def test_update_existing(monkeypatch, pytester):
pytester.makefile(".jsonl", test_update_existing=TEST_JSONL)
pytester.makepyfile(
f"""
import serial
def test_reserial(reserial):
s = serial.Serial(port="/dev/ttyUSB0")
s.write({2 * TEST_TX})
assert s.read() == {2 * TEST_RX}
"""
)

def patch_write(self: Serial, data: bytes) -> int:
return len(data)

def patch_read(self: Serial, size: int = 1) -> bytes:
return 2 * TEST_RX

def patch_open(self: Serial) -> None:
self.is_open = True

def patch_close(self: Serial) -> None:
self.is_open = False

monkeypatch.setattr(Serial, "write", patch_write)
monkeypatch.setattr(Serial, "read", patch_read)
monkeypatch.setattr(Serial, "open", patch_open)
monkeypatch.setattr(Serial, "close", patch_close)
result = pytester.runpytest("--record")

with open("test_update_existing.jsonl", "r") as f:
recording = [json.loads(line) for line in f.readlines()]

assert recording == json.loads(TEST_JSON)
expected_rx_enc = "AQE="
expected_tx_enc = "AgI="
expected_jsonl = (
f'{{"test_reserial": {{"rx": "{expected_rx_enc}", "tx": "{expected_tx_enc}"}}}}\n'
f'{{"test_reserial2": {{"rx": "{TEST_RX_ENC}", "tx": "{TEST_TX_ENC}"}}}}\n'
)

expected = [json.loads(line) for line in expected_jsonl.splitlines()]

assert recording == expected
assert result.ret == 0


def test_replay(pytester):
pytester.makefile(".json", test_replay=TEST_JSON)
pytester.makefile(".jsonl", test_replay=TEST_JSONL)
pytester.makepyfile(TEST_FILE)
result = pytester.runpytest("--replay")
assert result.ret == 0
Expand All @@ -91,7 +133,7 @@ def test_invalid_option(pytester):


def test_bad_tx(pytester):
pytester.makefile(".json", test_bad_tx=TEST_JSON)
pytester.makefile(".jsonl", test_bad_tx=TEST_JSONL)
pytester.makepyfile(TEST_FILE_BAD_TX)
result = pytester.runpytest("--replay")
result.assert_outcomes(errors=1, failed=1)
Expand All @@ -111,8 +153,8 @@ def test_help_message(pytester):

def test_change_settings(pytester):
pytester.makefile(
".json",
test_change_settings='{"test_reserial": {"tx": [], "rx": []}}',
".jsonl",
test_change_settings='{"test_reserial": {"tx": "", "rx": ""}}',
)
pytester.makepyfile(
"""
Expand All @@ -124,3 +166,16 @@ def test_reserial(reserial):
)
result = pytester.runpytest("--replay")
assert result.ret == 0


def test_no_traffic_for_test(pytester):
pytester.makefile(".jsonl", test_no_traffic_for_test=TEST_JSONL)
pytester.makepyfile(
"""
import serial
def test_reserial3(reserial):
pass
"""
)
result = pytester.runpytest("--replay")
result.assert_outcomes(errors=1)

0 comments on commit 2b047d7

Please sign in to comment.