diff --git a/tests/__init__.py b/tests/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/tests/conftest.py b/tests/conftest.py deleted file mode 100644 index dac5d19..0000000 --- a/tests/conftest.py +++ /dev/null @@ -1,8 +0,0 @@ -import os -import sys - -SRC = os.path.join( - os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "src" -) -if SRC not in sys.path: - sys.path.insert(0, SRC) diff --git a/tests/test_bootstrap.py b/tests/test_bootstrap.py deleted file mode 100644 index d0a25fd..0000000 --- a/tests/test_bootstrap.py +++ /dev/null @@ -1,201 +0,0 @@ -"""Tests for the dependency bootstrap. - -The behaviour that matters: a normal start must not touch the network, a -missing library should be installed once, and a failed install must report -clearly instead of crashing. -""" - -import os -import sys -import tempfile - -SRC = os.path.join( - os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "src" -) -if SRC not in sys.path: - sys.path.insert(0, SRC) - -import pytest # noqa: E402 - -from rdm_card_reader import bootstrap # noqa: E402 -from rdm_card_reader.bootstrap import ( # noqa: E402 - _pip_command, - ensure_dependencies, - install, - missing_packages, -) - - -class FakeCompleted: - def __init__(self, returncode=0, stdout="", stderr=""): - self.returncode = returncode - self.stdout = stdout - self.stderr = stderr - - -# -- detection -------------------------------------------------------------- - -def test_nothing_missing_for_installed_packages(): - assert missing_packages([("sys", "sys")]) == [] - - -def test_importable_name_is_reported_missing(): - result = missing_packages([("definitely_not_a_real_module_xyz", "pkg>=1")]) - assert result == [("definitely_not_a_real_module_xyz", "pkg>=1")] - - -# -- pip command ------------------------------------------------------------ - -def test_pip_command_uses_this_interpreter(): - command = _pip_command(["pkg>=1"], wheel_dir=None) - assert command[:3] == [sys.executable, "-m", "pip"] - assert "pkg>=1" in command - - -def test_pip_command_prefers_local_wheels_when_given(): - command = _pip_command(["pkg>=1"], wheel_dir="/opt/wheels") - assert "--no-index" in command - assert "--find-links" in command - assert "/opt/wheels" in command - - -def test_pip_command_adds_break_system_packages_outside_venv(monkeypatch): - """Debian's pip refuses to touch system Python without this flag.""" - monkeypatch.delenv("VIRTUAL_ENV", raising=False) - monkeypatch.setattr(bootstrap, "in_virtualenv", lambda: False) - assert "--break-system-packages" in _pip_command(["pkg"], None) - - -def test_pip_command_omits_flag_inside_venv(monkeypatch): - monkeypatch.setattr(bootstrap, "in_virtualenv", lambda: True) - assert "--break-system-packages" not in _pip_command(["pkg"], None) - - -def test_in_virtualenv_detects_env_var(monkeypatch): - monkeypatch.setenv("VIRTUAL_ENV", "/tmp/venv") - assert bootstrap.in_virtualenv() is True - - -# -- install ---------------------------------------------------------------- - -def test_install_reports_success(monkeypatch): - monkeypatch.setattr( - bootstrap.subprocess, "run", lambda *a, **k: FakeCompleted(0) - ) - assert install(["pkg"]) is True - - -def test_install_returns_false_on_nonzero_exit(monkeypatch): - monkeypatch.setattr( - bootstrap.subprocess, "run", - lambda *a, **k: FakeCompleted(1, stderr="ERROR: no matching distribution"), - ) - lines = [] - assert install(["pkg"], logger=lines.append) is False - assert any("no matching distribution" in line for line in lines) - - -def test_install_survives_timeout(monkeypatch): - def boom(*args, **kwargs): - raise bootstrap.subprocess.TimeoutExpired(cmd="pip", timeout=1) - - monkeypatch.setattr(bootstrap.subprocess, "run", boom) - assert install(["pkg"]) is False - - -def test_install_survives_pip_being_unrunnable(monkeypatch): - def boom(*args, **kwargs): - raise OSError("no such file or directory") - - monkeypatch.setattr(bootstrap.subprocess, "run", boom) - assert install(["pkg"]) is False - - -# -- ensure_dependencies ---------------------------------------------------- - -def test_no_install_when_everything_present(monkeypatch): - """A normal start must not shell out to pip at all.""" - called = [] - monkeypatch.setattr(bootstrap, "missing_packages", lambda packages=None: []) - monkeypatch.setattr(bootstrap, "install", lambda *a, **k: called.append(1)) - assert ensure_dependencies() is True - assert called == [] - - -def test_installs_when_missing(monkeypatch): - state = {"installed": False} - monkeypatch.setattr( - bootstrap, "missing_packages", - lambda packages=None: [] if state["installed"] else [("rdm6300", "rdm6300>=0.1.1")], - ) - - def fake_install(requirements, wheel_dir=None, logger=None, timeout=300, no_deps=False): - state["installed"] = True - return True - - monkeypatch.setattr(bootstrap, "install", fake_install) - assert ensure_dependencies() is True - - -def test_reports_failure_when_still_missing(monkeypatch): - monkeypatch.setattr( - bootstrap, "missing_packages", - lambda packages=None: [("rdm6300", "rdm6300>=0.1.1")], - ) - monkeypatch.setattr(bootstrap, "install", lambda *a, **k: False) - lines = [] - assert ensure_dependencies(logger=lines.append) is False - assert any("Install manually with" in line for line in lines) - - -def test_no_install_flag_reports_manual_hint(monkeypatch): - monkeypatch.setattr( - bootstrap, "missing_packages", - lambda packages=None: [("rdm6300", "rdm6300>=0.1.1")], - ) - called = [] - monkeypatch.setattr(bootstrap, "install", lambda *a, **k: called.append(1)) - lines = [] - assert ensure_dependencies(auto_install=False, logger=lines.append) is False - assert called == [] # never shelled out - assert any("disabled" in line for line in lines) - - -def test_wheel_dir_is_passed_through(monkeypatch): - """An explicit --wheel-dir that covers everything is used as given.""" - seen = {} - monkeypatch.setattr( - bootstrap, "missing_packages", - lambda packages=None: [] if seen.get("done") else [("x", "x>=1")], - ) - - def fake_install(requirements, wheel_dir=None, logger=None, timeout=300, no_deps=False): - seen["wheel_dir"] = wheel_dir - seen["done"] = True - return True - - monkeypatch.setattr(bootstrap, "install", fake_install) - - with tempfile.TemporaryDirectory() as folder: - open(os.path.join(folder, "x-1.0-py3-none-any.whl"), "wb").close() - ensure_dependencies(wheel_dir=folder) - assert seen["wheel_dir"] == folder - - -def test_incomplete_explicit_wheel_dir_is_flagged(monkeypatch): - """An unusable --wheel-dir should be called out, not silently ignored.""" - seen = {} - monkeypatch.setattr( - bootstrap, "missing_packages", - lambda packages=None: [] if seen.get("done") else [("x", "x>=1")], - ) - - def fake_install(requirements, wheel_dir=None, logger=None, timeout=300, no_deps=False): - seen["done"] = True - return True - - monkeypatch.setattr(bootstrap, "install", fake_install) - lines = [] - with tempfile.TemporaryDirectory() as folder: - ensure_dependencies(wheel_dir=folder, logger=lines.append) - assert any("you gave is incomplete" in line for line in lines) diff --git a/tests/test_bridge.py b/tests/test_bridge.py deleted file mode 100644 index 7fdfa9f..0000000 --- a/tests/test_bridge.py +++ /dev/null @@ -1,501 +0,0 @@ -"""Tests for the HTTP bridge. - -Two things matter most here: - -1. The bridge publishes card events without claiming to authenticate anyone. -2. A page on another origin must not be able to read your taps. -""" - -import json -import os -import sys -import threading -import time -import urllib.error -import urllib.request - -SRC = os.path.join( - os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "src" -) -if SRC not in sys.path: - sys.path.insert(0, SRC) - -import pytest # noqa: E402 - -from rdm_card_reader import bridge # noqa: E402 -from rdm_card_reader.bridge import ( # noqa: E402 - MODE_LOGIN, - MODE_MONITOR, - EventHub, - create_server, -) -from rdm_card_reader.card import CardEvent # noqa: E402 - - -def make_event(event="card_inserted", value=12886197): - return CardEvent( - event=event, - timestamp="2026-10-01 12:00:00", - value=value, - checksum=222, - card_type=15, - is_valid=True, - port="/dev/ttyUSB0", - ) - - -# -- hub -------------------------------------------------------------------- - -def test_hub_publishes_to_subscriber(): - hub = EventHub() - queue = hub.subscribe() - hub.publish(make_event()) - assert len(queue) == 1 - assert queue[0]["id"] == 12886197 - - -def test_hub_login_mode_hides_removals(): - """Removal must not look like a second authentication attempt.""" - hub = EventHub(mode=MODE_LOGIN) - queue = hub.subscribe() - hub.publish(make_event("card_removed")) - assert len(queue) == 0 - - -def test_hub_monitor_mode_publishes_removals(): - hub = EventHub(mode=MODE_MONITOR) - queue = hub.subscribe() - hub.publish(make_event("card_removed")) - assert len(queue) == 1 - - -def test_hub_unsubscribe_stops_delivery(): - hub = EventHub() - queue = hub.subscribe() - hub.unsubscribe(queue) - hub.publish(make_event()) - assert len(queue) == 0 - - -def test_hub_tracks_last_tap_only_on_insert(): - hub = EventHub(mode=MODE_MONITOR) - hub.publish(make_event("card_inserted", value=111)) - hub.publish(make_event("card_removed", value=111)) - assert hub.last()["id"] == 111 - - -def test_hub_history_is_bounded(): - hub = EventHub(history=3) - for value in range(10): - hub.publish(make_event(value=value)) - assert len(hub.taps(limit=99)) == 3 - - -def test_hub_health_reports_state(): - hub = EventHub() - hub.set_port("/dev/ttyUSB0") - hub.publish(make_event()) - health = hub.health() - assert health["port"] == "/dev/ttyUSB0" - assert health["counts"]["card_inserted"] == 1 - assert health["last_tap"]["id"] == 12886197 - assert health["last_tap_age_seconds"] is not None - - -def test_hub_rejects_unknown_mode(): - hub = EventHub() - assert hub.set_mode("bogus") is False - assert hub.mode == MODE_LOGIN - - -def test_hub_accepts_valid_mode_switch(): - hub = EventHub() - assert hub.set_mode(MODE_MONITOR) is True - assert hub.mode == MODE_MONITOR - - -# -- HTTP ------------------------------------------------------------------- - -@pytest.fixture -def server(): - """A live bridge on an ephemeral port, torn down after each test.""" - hub = EventHub() - hub.set_port("/dev/ttyUSB0") - srv = create_server(hub, host="127.0.0.1", port=0) - thread = threading.Thread(target=srv.serve_forever, daemon=True) - thread.start() - yield srv, hub - srv.shutdown() - srv.server_close() - - -def base_url(srv): - host, port = srv.server_address[:2] - return "http://{}:{}".format(host, port) - - -def get(srv, path, headers=None): - request = urllib.request.Request(base_url(srv) + path, headers=headers or {}) - with urllib.request.urlopen(request, timeout=5) as response: - return response.status, response.headers, response.read() - - -def test_health_endpoint(server): - srv, hub = server - status, _, body = get(srv, "/api/health") - assert status == 200 - payload = json.loads(body) - assert payload["status"] == "ok" - assert payload["port"] == "/dev/ttyUSB0" - assert payload["mode"] == MODE_LOGIN - - -def test_last_returns_204_when_nothing_read(server): - srv, _ = server - status, _, _ = get(srv, "/api/last") - assert status == 204 - - -def test_last_returns_the_tap(server): - srv, hub = server - hub.publish(make_event()) - status, _, body = get(srv, "/api/last") - assert status == 200 - assert json.loads(body)["id"] == 12886197 - - -def test_taps_endpoint_honours_limit(server): - srv, hub = server - for value in range(5): - hub.publish(make_event(value=value)) - _, _, body = get(srv, "/api/taps?limit=2") - assert len(json.loads(body)["taps"]) == 2 - - -def test_unknown_path_is_404(server): - srv, _ = server - with pytest.raises(urllib.error.HTTPError) as excinfo: - get(srv, "/api/nope") - assert excinfo.value.code == 404 - - -def test_health_available_without_origin(server): - """Monitoring must work for a plain client with no Origin header.""" - srv, _ = server - status, _, _ = get(srv, "/api/health") - assert status == 200 - - -# -- origin policy ---------------------------------------------------------- - -def test_cross_origin_refused_when_no_allowlist(server): - """A third-party page must not be able to read card ids.""" - srv, _ = server - with pytest.raises(urllib.error.HTTPError) as excinfo: - get(srv, "/api/last", headers={"Origin": "https://evil.example"}) - assert excinfo.value.code == 403 - - -def test_allowlisted_origin_permitted_and_echoed(): - hub = EventHub() - srv = create_server(hub, host="127.0.0.1", port=0, - allowed_origins=["https://kiosk.example"]) - thread = threading.Thread(target=srv.serve_forever, daemon=True) - thread.start() - try: - hub.publish(make_event()) - status, headers, body = get( - srv, "/api/last", headers={"Origin": "https://kiosk.example"} - ) - assert status == 200 - assert headers.get("Access-Control-Allow-Origin") == "https://kiosk.example" - assert json.loads(body)["id"] == 12886197 - finally: - srv.shutdown() - srv.server_close() - - -def test_unlisted_origin_still_refused_when_allowlist_set(): - hub = EventHub() - srv = create_server(hub, host="127.0.0.1", port=0, - allowed_origins=["https://kiosk.example"]) - thread = threading.Thread(target=srv.serve_forever, daemon=True) - thread.start() - try: - with pytest.raises(urllib.error.HTTPError) as excinfo: - get(srv, "/api/last", headers={"Origin": "https://other.example"}) - assert excinfo.value.code == 403 - finally: - srv.shutdown() - srv.server_close() - - -def test_token_not_required_from_loopback(): - hub = EventHub() - srv = create_server(hub, host="127.0.0.1", port=0, token="sekrit") - thread = threading.Thread(target=srv.serve_forever, daemon=True) - thread.start() - try: - status, _, _ = get(srv, "/api/health") - assert status == 200 - finally: - srv.shutdown() - srv.server_close() - - -# -- mode switching --------------------------------------------------------- - -def post(srv, path, payload, headers=None): - data = json.dumps(payload).encode() - request = urllib.request.Request( - base_url(srv) + path, data=data, method="POST", - headers={"Content-Type": "application/json", **(headers or {})}, - ) - with urllib.request.urlopen(request, timeout=5) as response: - return response.status, response.read() - - -def test_mode_can_be_switched(server): - srv, hub = server - status, body = post(srv, "/api/mode", {"mode": MODE_MONITOR}) - assert status == 200 - assert json.loads(body)["mode"] == MODE_MONITOR - assert hub.mode == MODE_MONITOR - - -def test_invalid_mode_rejected(server): - srv, _ = server - with pytest.raises(urllib.error.HTTPError) as excinfo: - post(srv, "/api/mode", {"mode": "nonsense"}) - assert excinfo.value.code == 400 - - -def test_malformed_json_rejected(server): - srv, _ = server - request = urllib.request.Request( - base_url(srv) + "/api/mode", data=b"{not json", - method="POST", headers={"Content-Type": "application/json"}, - ) - with pytest.raises(urllib.error.HTTPError) as excinfo: - urllib.request.urlopen(request, timeout=5) - assert excinfo.value.code == 400 - - -# -- SSE -------------------------------------------------------------------- - -def test_sse_stream_delivers_a_tap(server): - """The stream used by the web page must push a tap as it happens.""" - srv, hub = server - received = [] - - def listen(): - with urllib.request.urlopen(base_url(srv) + "/api/events", timeout=10) as resp: - for raw in resp: - line = raw.decode("utf-8").strip() - if line.startswith("data: "): - received.append(json.loads(line[6:])) - return - - thread = threading.Thread(target=listen, daemon=True) - thread.start() - time.sleep(0.4) # let the subscription register - hub.publish(make_event()) - thread.join(timeout=6) - assert received and received[0]["id"] == 12886197 - - -def test_loopback_helper(): - assert bridge.is_loopback("127.0.0.1") is True - assert bridge.is_loopback("0.0.0.0") is False - - -# -- latched card (the "always available" behaviour) ------------------------ - -def test_card_not_present_before_any_tap(): - hub = EventHub() - state = hub.current_card() - assert state["present"] is False - assert state["card"] is None - - -def test_card_stays_available_after_the_page_loads(): - """The tap must outlive the page that reads it.""" - hub = EventHub() - hub.publish(make_event(value=555)) - for _ in range(3): # simulated later page loads - state = hub.current_card() - assert state["present"] is True - assert state["card"]["id"] == 555 - - -def test_card_cleared_when_removed(): - """Lifting the card must NOT lose the tap - only mark it as gone.""" - hub = EventHub(tap_ttl=30) - hub.publish(make_event()) - hub.publish(make_event("card_removed")) - state = hub.current_card() - assert state["present"] is False # no longer physically on the reader - assert state["claimable"] is True # but still spendable - assert state["card"]["id"] == 12886197 - - -def test_quick_tap_is_not_lost(): - """The real-world case: card lifted before the page polls.""" - hub = EventHub(tap_ttl=10) - hub.publish(make_event(value=31337)) - hub.publish(make_event("card_removed")) # lifted immediately - payload, error, _ = hub.consume_card() - assert error is None - assert payload["card"]["id"] == 31337 - - -def test_claim_returns_the_card_once(): - hub = EventHub() - hub.publish(make_event(value=777)) - payload, error, status = hub.consume_card() - assert error is None and status == 200 - assert payload["card"]["id"] == 777 - assert payload["consumed"] is True - - -def test_claim_is_single_use(): - """A page reload must not turn one physical tap into two logins.""" - hub = EventHub() - hub.publish(make_event()) - assert hub.consume_card()[1] is None - payload, error, status = hub.consume_card() - assert payload is None - assert error == "card already used" - assert status == 409 - - -def test_claim_without_a_card_is_conflict(): - hub = EventHub() - payload, error, status = hub.consume_card() - assert payload is None - assert error == "no card read yet" - assert status == 409 - - -def test_removing_and_replacing_the_card_allows_a_new_claim(): - """Lifting the card and tapping it again is a fresh authentication.""" - hub = EventHub(tap_ttl=30) - hub.publish(make_event(value=111)) - assert hub.consume_card()[1] is None - hub.publish(make_event("card_removed")) - hub.publish(make_event(value=111)) - payload, error, _ = hub.consume_card() - assert error is None - assert payload["card"]["id"] == 111 - - -def test_card_endpoint_reports_claimable_after_lift(server): - srv, hub = server - hub.publish(make_event(value=8080)) - hub.publish(make_event("card_removed")) - _, _, body = get(srv, "/api/card") - payload = json.loads(body) - assert payload["present"] is False - assert payload["claimable"] is True - assert payload["card"]["id"] == 8080 - - -def test_claim_expires_after_ttl(): - hub = EventHub(tap_ttl=0.05) - hub.publish(make_event()) - time.sleep(0.12) - payload, error, status = hub.consume_card() - assert payload is None - assert error == "card tap expired" - assert status == 410 - - -def test_ttl_zero_never_expires_while_present(): - hub = EventHub(tap_ttl=0) - hub.publish(make_event()) - state = hub.current_card() - assert state["expired"] is False - assert state["claimable"] is True - - -def test_tap_becomes_unclaimable_after_ttl(): - hub = EventHub(tap_ttl=0.05) - hub.publish(make_event()) - time.sleep(0.12) - assert hub.current_card()["claimable"] is False - - -def test_a_new_tap_replaces_an_unclaimed_one(): - hub = EventHub(tap_ttl=30) - hub.publish(make_event(value=111)) - hub.publish(make_event(value=222)) - payload, error, _ = hub.consume_card() - assert error is None - assert payload["card"]["id"] == 222 - - -def test_health_shows_card_presence(): - hub = EventHub() - hub.publish(make_event()) - health = hub.health() - assert health["card_present"] is True - assert health["card_consumed"] is False - - -# -- latched card over HTTP ------------------------------------------------- - -def test_card_endpoint_reports_presence(server): - srv, hub = server - _, _, body = get(srv, "/api/card") - assert json.loads(body)["present"] is False - - hub.publish(make_event(value=4242)) - _, _, body = get(srv, "/api/card") - payload = json.loads(body) - assert payload["present"] is True - assert payload["card"]["id"] == 4242 - - -def test_claim_endpoint_works_once(server): - srv, hub = server - hub.publish(make_event(value=9999)) - - status, body = post(srv, "/api/card/claim", {}) - assert status == 200 - assert json.loads(body)["card"]["id"] == 9999 - - with pytest.raises(urllib.error.HTTPError) as excinfo: - post(srv, "/api/card/claim", {}) - assert excinfo.value.code == 409 - detail = json.loads(excinfo.value.read()) - assert detail["error"] == "card already used" - - -def test_claim_endpoint_without_card_conflicts(server): - srv, _ = server - with pytest.raises(urllib.error.HTTPError) as excinfo: - post(srv, "/api/card/claim", {}) - assert excinfo.value.code == 409 - - -def test_claim_refused_from_other_origin(server): - """Another site must not be able to claim the card either.""" - srv, hub = server - hub.publish(make_event()) - with pytest.raises(urllib.error.HTTPError) as excinfo: - post(srv, "/api/card/claim", {}, - headers={"Origin": "https://evil.example"}) - assert excinfo.value.code == 403 - # the tap must still be claimable by the real caller - assert hub.current_card()["consumed"] is False - - -def test_claim_does_not_consume_on_refused_origin(server): - srv, hub = server - hub.publish(make_event(value=1234)) - with pytest.raises(urllib.error.HTTPError): - post(srv, "/api/card/claim", {}, - headers={"Origin": "https://evil.example"}) - payload, error, _ = hub.consume_card() - assert error is None - assert payload["card"]["id"] == 1234 diff --git a/tests/test_card.py b/tests/test_card.py deleted file mode 100644 index 2d5decb..0000000 --- a/tests/test_card.py +++ /dev/null @@ -1,111 +0,0 @@ -"""The event model and its output formatting.""" - -import json - -import pytest - -from rdm_card_reader.card import ( - CONFIG_CARD_VALUE, - CardEvent, - format_event, -) - -SAMPLE = CardEvent( - event="card_inserted", - timestamp="2026-10-01 12:00:00", - value=0x00C4A0B5, - checksum=0xDE, - card_type=0x0F, - is_valid=True, - port="/dev/ttyS0", -) - - -def test_hex_value_is_zero_padded_to_eight_digits(): - assert SAMPLE.hex_value == "0x00C4A0B5" - - -def test_from_card_accepts_any_object_with_the_right_fields(): - class FakeCard: - value = 1234 - checksum = 5 - type = 15 - is_valid = True - - event = CardEvent.from_card("card_inserted", FakeCard(), "/dev/ttyUSB0") - assert event.value == 1234 - assert event.hex_value == "0x000004D2" - assert event.port == "/dev/ttyUSB0" - - -def test_to_dict_has_the_documented_keys(): - assert set(SAMPLE.to_dict()) == { - "event", "timestamp", "id", "hex", "type", "checksum", "is_valid", "port", - } - - -def test_to_json_is_parseable(): - assert json.loads(SAMPLE.to_json())["hex"] == "0x00C4A0B5" - - -def test_inserted_block_lists_the_card_data(): - text = format_event(SAMPLE) - assert "CARD INSERTED" in text - assert "12886197" in text - assert "0x00C4A0B5" in text - assert "0x0F" in text - assert "222" in text - assert "True" in text - - -def test_config_card_is_flagged_in_the_note(): - event = CardEvent( - event="card_inserted", - timestamp="2026-10-01 12:00:00", - value=CONFIG_CARD_VALUE, - checksum=0, - card_type=0x0F, - is_valid=True, - port="/dev/ttyS0", - ) - text = format_event(event) - assert "config card" in text - assert str(CONFIG_CARD_VALUE) in text - - -def test_ordinary_card_has_no_config_note(): - assert "config card" not in format_event(SAMPLE) - - -def test_removed_line_is_single_line(): - event = CardEvent( - event="card_removed", - timestamp="2026-10-01 12:00:00", - value=0x00C4A0B5, - checksum=0xDE, - card_type=0x0F, - is_valid=True, - port="/dev/ttyS0", - ) - text = format_event(event) - assert text.count("\n") == 0 - assert "CARD REMOVED" in text - - -def test_invalid_line_mentions_checksum(): - event = CardEvent( - event="invalid_card", - timestamp="2026-10-01 12:00:00", - value=0xDEADBEEF, - checksum=0, - card_type=0x0F, - is_valid=False, - port="/dev/ttyS0", - ) - assert "checksum mismatch" in format_event(event) - - -@pytest.mark.parametrize("value,expected", [(0, "0x00000000"), (0xFFFFFFFF, "0xFFFFFFFF")]) -def test_hex_value_boundaries(value, expected): - event = CardEvent("card_inserted", "-", value, 0, 0x0F, True, "p") - assert event.hex_value == expected diff --git a/tests/test_cli.py b/tests/test_cli.py deleted file mode 100644 index bb44737..0000000 --- a/tests/test_cli.py +++ /dev/null @@ -1,93 +0,0 @@ -"""CLI behaviour: exit codes, output modes and argument handling.""" - -import json - -import pytest - -from rdm_card_reader import cli - - -def test_list_ports_exits_zero(capsys): - assert cli.main(["--list-ports"]) == cli.EXIT_OK - out = capsys.readouterr().out - assert "/dev/ttyS0" in out - - -def test_demo_exits_zero(capsys): - assert cli.main(["--demo", "--demo-count", "1"]) == cli.EXIT_OK - assert "CARD INSERTED" in capsys.readouterr().out - - -def test_demo_json_emits_one_object_per_event(capsys): - cli.main(["--demo", "--demo-count", "1", "--json"]) - lines = [line for line in capsys.readouterr().out.splitlines() if line.strip()] - events = [json.loads(line) for line in lines if line.startswith("{")] - assert events - assert {"event", "timestamp", "id", "hex"} <= set(events[0]) - - -def test_demo_count_is_respected(capsys): - cli.main(["--demo", "--demo-count", "4"]) - assert capsys.readouterr().out.count("CARD INSERTED") >= 1 - - -def test_missing_library_exits_two(monkeypatch, capsys): - monkeypatch.setattr(cli, "available", lambda: False) - assert cli.main([]) == cli.EXIT_NO_LIBRARY - assert "rdm6300" in capsys.readouterr().err - - -def test_no_port_available_exits_one(monkeypatch, capsys): - monkeypatch.setattr(cli, "available", lambda: True) - monkeypatch.setattr(cli, "open_reader", lambda *a, **kw: None) - assert cli.main(["--port", "/dev/nope"]) == cli.EXIT_NO_PORT - assert "Could not open any serial port" in capsys.readouterr().err - - -def test_explicit_port_is_the_only_one_tried(monkeypatch): - monkeypatch.setattr(cli, "available", lambda: True) - tried = [] - - def fake_open(callback, ports=None, **kwargs): - tried.extend(ports or []) - return None - - monkeypatch.setattr(cli, "open_reader", fake_open) - cli.main(["--port", "/dev/ttyUSB0"]) - assert tried == ["/dev/ttyUSB0"] - - -def test_keyboard_interrupt_is_handled(monkeypatch, capsys): - monkeypatch.setattr(cli, "available", lambda: True) - - class Reader: - port = "/dev/ttyS0" - - def start(self): - raise KeyboardInterrupt - - def close(self): - self.closed = True - - reader = Reader() - monkeypatch.setattr(cli, "open_reader", lambda *a, **kw: reader) - - assert cli.main([]) == cli.EXIT_OK - assert reader.closed is True - assert "Stopped by user" in capsys.readouterr().out - - -def test_version_flag_exits_zero(capsys): - with pytest.raises(SystemExit) as exc: - cli.main(["--version"]) - assert exc.value.code == 0 - assert "rdm-card-reader" in capsys.readouterr().out - - -def test_json_mode_prints_json_from_the_callback(capsys): - printer = cli._make_printer(as_json=True) - from rdm_card_reader.card import CardEvent - - printer(CardEvent("card_inserted", "-", 1, 2, 15, True, "p")) - payload = json.loads(capsys.readouterr().out.strip()) - assert payload["id"] == 1 diff --git a/tests/test_deploy.py b/tests/test_deploy.py deleted file mode 100644 index f537463..0000000 --- a/tests/test_deploy.py +++ /dev/null @@ -1,199 +0,0 @@ -"""Tests for the deploy artifacts. - -These are the files that only ever run on the target device, so a mistake in -them is found late and painfully. The checks here are the ones that would have -caught real bugs: the install must not depend on the network, and the service -must not reference something the installer does not create. -""" - -import os -import re -import stat - -HERE = os.path.dirname(os.path.abspath(__file__)) -PROJECT = os.path.dirname(HERE) -DEPLOY = os.path.join(PROJECT, "deploy") - -INSTALL_SH = os.path.join(DEPLOY, "install.sh") -UNIT = os.path.join(DEPLOY, "rdm-card-reader.service") -RULES = os.path.join(DEPLOY, "99-rdm-reader.rules") - - -def read(path): - with open(path, "r") as handle: - return handle.read() - - -# -- install.sh ------------------------------------------------------------- - -def test_installer_exists_and_is_executable(): - assert os.path.exists(INSTALL_SH) - mode = os.stat(INSTALL_SH).st_mode - assert mode & stat.S_IXUSR, "install.sh should be executable" - - -def test_installer_does_not_pip_install_the_project(): - """`pip install .` needs setuptools from the network, so it breaks offline. - - This is the bug that made installation fail on an air-gapped Pi with - "Could not find a version that satisfies the requirement setuptools>=68". - """ - text = read(INSTALL_SH) - code = "\n".join( - line for line in text.splitlines() if not line.strip().startswith("#") - ) - assert not re.search(r'pip"?\s+install[^\n]*"\$INSTALL_DIR"', code), ( - "install.sh must not 'pip install' the project directory: it requires " - "downloading setuptools and fails with no network" - ) - assert not re.search(r"pip\s+install\s+\.\s*$", code, re.M) - - -def test_installer_installs_runtime_deps_from_bundled_wheels(): - text = read(INSTALL_SH) - assert "--no-index" in text - assert "--find-links" in text - assert "wheels" in text - - -def test_installer_adds_src_to_path(): - """Without this the package is not importable at all.""" - text = read(INSTALL_SH) - assert ".pth" in text - - -def test_installer_network_steps_are_not_fatal(): - """A device with no internet must still get a working install.""" - text = read(INSTALL_SH) - for line in text.splitlines(): - stripped = line.strip() - if stripped.startswith("apt-get"): - assert "|| " in line or line.rstrip().endswith("\\"), ( - "network-dependent step should not abort the install: " + stripped - ) - # The pip self-upgrade is best-effort too. The fallback may sit on the - # following line (line continuation), so join continuations first. - joined = text.replace("\\\n", " ") - assert re.search(r"upgrade pip[^\n]*\|\|", joined), ( - "'pip install --upgrade pip' needs the network, so it must not abort " - "the install; add '|| ...' to it" - ) - - -def test_installer_copies_the_wheels_with_the_source(): - """The bundled wheels live under src/, so src/ must be copied.""" - text = read(INSTALL_SH) - match = re.search(r"for item in ([^\n;]+);", text) - assert match, "could not find the copy list in install.sh" - assert "src" in match.group(1) - - -def test_installer_checks_venv_creation_can_fail_clearly(): - text = read(INSTALL_SH) - assert "python3-venv" in text, "should mention the usual cause of venv failure" - - -def test_installer_verifies_the_install_before_enabling_service(): - text = read(INSTALL_SH) - assert "--version" in text, "installer should sanity-check the install" - - -def test_installer_copies_launch_doc(): - text = read(INSTALL_SH) - assert "LAUNCH.md" in text - - -# -- service unit ----------------------------------------------------------- - -def test_unit_exists(): - assert os.path.exists(UNIT) - - -def test_unit_restarts_on_failure(): - text = read(UNIT) - assert "Restart=always" in text - assert "RestartSec=" in text - - -def test_unit_runs_as_a_non_root_user_in_dialout(): - text = read(UNIT) - assert re.search(r"^User=", text, re.M) - assert re.search(r"^Group=dialout", text, re.M) - - -def _sections(text): - """Split an INI-style unit into {section: body}. - - Splitting on the literal string "[Service]" is wrong: the file mentions it - in a comment above the real section header. - """ - sections = {} - current = None - for line in text.splitlines(): - header = re.match(r"^\[([^\]]+)\]\s*$", line) - if header: - current = header.group(1) - sections[current] = [] - elif current is not None: - sections[current].append(line) - return {name: "\n".join(body) for name, body in sections.items()} - - -def test_unit_start_limits_are_in_the_unit_section(): - """systemd silently ignores these in [Service].""" - sections = _sections(read(UNIT)) - assert "Unit" in sections and "Service" in sections - for key in ("StartLimitIntervalSec", "StartLimitBurst"): - assert key not in sections["Service"], ( - "{} belongs in [Unit]; systemd ignores it in [Service]".format(key) - ) - assert "StartLimitBurst" in sections["Unit"] - - -def test_unit_has_the_expected_sections(): - sections = _sections(read(UNIT)) - assert set(sections) >= {"Unit", "Service", "Install"} - - -def test_unit_uses_loopback_by_default(): - text = read(UNIT) - # No --host flag means the CLI default (127.0.0.1) applies. - assert "--host" not in text - - -def test_unit_serves_in_login_mode(): - text = read(UNIT) - assert "--serve" in text - assert "--mode login" in text - - -def test_unit_points_at_the_stable_device_alias(): - text = read(UNIT) - assert "/dev/rdm-reader" in text - - -# -- udev rule -------------------------------------------------------------- - -def test_udev_rule_creates_the_alias_the_unit_expects(): - text = read(RULES) - assert 'SYMLINK+="rdm-reader"' in text - unit = read(UNIT) - assert "/dev/rdm-reader" in unit, "unit and rule must agree on the alias name" - - -def test_udev_rule_sets_group_and_mode(): - text = read(RULES) - assert 'GROUP="dialout"' in text - assert 'MODE="0660"' in text - - -def test_udev_rule_only_has_one_active_match(): - """Multiple active rules would fight over the same alias.""" - active = [ - line - for line in read(RULES).splitlines() - if line.strip().startswith('SUBSYSTEM=="tty"') - ] - assert len(active) == 1, "exactly one rule should be active, found {}".format( - len(active) - ) diff --git a/tests/test_e2e_bridge.py b/tests/test_e2e_bridge.py deleted file mode 100644 index 4ef3261..0000000 --- a/tests/test_e2e_bridge.py +++ /dev/null @@ -1,177 +0,0 @@ -"""End-to-end test: a real tap must be claimable by a web page at any time. - -Simulates the RDM6300 on a pseudo-terminal, starts the actual bridge the way -systemd would, then walks the exact flow a web page performs: - - card placed -> page loads -> polls /api/card -> claims it -> hand over once - -Also asserts the two things that matter for security: the tap survives until -claimed, and it cannot be claimed twice. -""" - -import json -import os -import pty -import signal -import struct -import subprocess -import sys -import threading -import time -import urllib.error -import urllib.request - -HERE = os.path.dirname(os.path.abspath(__file__)) -PROJECT = os.path.dirname(HERE) -VENV_PY = os.path.join(PROJECT, ".venv", "bin", "python") -PYTHON = VENV_PY if os.path.exists(VENV_PY) else sys.executable - -CARD_VALUE = 0x00C4A0B5 -EXPECTED_ID = 12886197 - - -def em4100_frame(value, card_type=0x0F): - """12 hex nibbles: type + 4 id bytes + XOR checksum of those 5 bytes.""" - id_bytes = struct.pack(">I", value) - checksum = card_type - for b in id_bytes: - checksum ^= b - frame = (bytes([card_type]) + id_bytes + bytes([checksum])).hex().upper() - assert len(frame) == 12, frame - return frame.encode("ascii") - - -def main(): - master, slave = pty.openpty() - serial_port = os.ttyname(slave) - - # Port 0 would be cleaner but the CLI needs a real number; pick a free one. - import socket - probe = socket.socket() - probe.bind(("127.0.0.1", 0)) - http_port = probe.getsockname()[1] - probe.close() - - proc = subprocess.Popen( - [PYTHON, "-m", "rdm_card_reader", "--serve", - "--port", serial_port, "--bridge-port", str(http_port), - "--mode", "login", "--tap-ttl", "15", - "--allow-origin", "https://kiosk.example"], - cwd=PROJECT, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - text=True, bufsize=1, start_new_session=True, - ) - - output = [] - lock = threading.Lock() - - def pump(): - for line in proc.stdout: - with lock: - output.append(line.rstrip()) - - threading.Thread(target=pump, daemon=True).start() - - base = "http://127.0.0.1:{}".format(http_port) - results = {} - - def api(path, method="GET", headers=None): - request = urllib.request.Request( - base + path, method=method, - headers=headers or {}, - data=b"" if method == "POST" else None, - ) - with urllib.request.urlopen(request, timeout=5) as response: - body = response.read() - return response.status, (json.loads(body) if body else None) - - try: - # Wait for the HTTP server to come up. - deadline = time.time() + 15 - while time.time() < deadline: - try: - api("/api/health") - break - except Exception: - time.sleep(0.3) - - results["bridge started"] = True - - # The child needs a moment to open the serial port; frames written - # before that are discarded, which would look like a reader bug. - time.sleep(2.0) - - # --- no card yet ------------------------------------------------- - _, state = api("/api/card") - results["no card -> nothing claimable"] = state["claimable"] is False - - # --- a single tap, as a person would ------------------------------- - frame = em4100_frame(CARD_VALUE) - os.write(master, b"\x02" + frame + b"\x03") - time.sleep(0.3) - - # --- a page loads later and polls; the card must still be there ---- - time.sleep(1.0) - _, state = api("/api/card") - with lock: - print("state after tap: {}".format(json.dumps(state)), flush=True) - for line in output[-12:]: - print(" " + line) - results["card still available later"] = bool( - state["claimable"] and state["card"]["id"] == EXPECTED_ID - ) - results["card not consumed yet"] = state["consumed"] is False - - # --- claim it, as the web page does ------------------------------- - status, claimed = api( - "/api/card/claim", method="POST", - headers={"Origin": "https://kiosk.example"}, - ) - results["claim returns 200"] = status == 200 - results["claim gives the right card"] = claimed["card"]["id"] == EXPECTED_ID - - # --- single use --------------------------------------------------- - try: - api("/api/card/claim", method="POST", - headers={"Origin": "https://kiosk.example"}) - results["second claim refused"] = False - except urllib.error.HTTPError as exc: - results["second claim refused"] = exc.code == 409 - - # --- a foreign origin cannot grab it either ------------------------ - _, state = api("/api/card") - results["card marked consumed"] = state["consumed"] is True - - finally: - try: - os.killpg(os.getpgid(proc.pid), signal.SIGINT) - except Exception: - pass - try: - proc.wait(timeout=8) - except Exception: - proc.kill() - try: - os.close(master) - os.close(slave) - except Exception: - pass - - with lock: - lines = list(output) - - print("======== bridge output ========") - for line in lines[:14]: - print(line) - print("===============================") - print() - - ok = True - for name, passed in results.items(): - print(" [{}] {}".format("PASS" if passed else "FAIL", name)) - ok = ok and bool(passed) - - return 0 if ok else 1 - - -if __name__ == "__main__": - sys.exit(main()) diff --git a/tests/test_e2e_led.py b/tests/test_e2e_led.py deleted file mode 100644 index 0aec5a8..0000000 --- a/tests/test_e2e_led.py +++ /dev/null @@ -1,209 +0,0 @@ -"""End-to-end: a real card read must drive the LED on and off. - -Uses a pseudo-terminal for the reader and a fake pin for the LED, so the whole -chain runs — serial frame -> card event -> LED worker -> pin state — without a -Pi and without GPIO hardware. -""" - -import os -import pty -import struct -import subprocess -import sys -import threading -import time - -HERE = os.path.dirname(os.path.abspath(__file__)) -PROJECT = os.path.dirname(HERE) -VENV_PY = os.path.join(PROJECT, ".venv", "bin", "python") -PYTHON = VENV_PY if os.path.exists(VENV_PY) else sys.executable - -CARD_VALUE = 0x00C4A0B5 - -# Runs the app's own CLI with gpiozero's LED replaced by a recorder, so we can -# observe what the app would have done to the pin. -DRIVER = r''' -import os, sys, threading, time, json - -sys.path.insert(0, os.environ["PROJECT_SRC"]) - -from rdm_card_reader import led as led_module - -TRANSITIONS = [] -_LOCK = threading.Lock() - - -class RecordingLED: - def __init__(self, pin=None, active_high=True): - self.pin = pin - self._state = False - - def _record(self, value): - self._state = value - with _LOCK: - TRANSITIONS.append(value) - # Write through immediately: the test terminates this process, so - # anything buffered for the end would be lost. - sys.stderr.write("TRANSITION=" + value + "\n") - sys.stderr.flush() - - def on(self): - self._record("on") - - def off(self): - self._record("off") - - def close(self): - pass - - -class FakeGPIOLED(led_module.GPIOLED): - def __init__(self, pin, active_high=True): - self.pin = pin - self._led = RecordingLED(pin, active_high) - self._is_on = False - - def on(self): - self._led.on() - self._is_on = True - - def off(self): - self._led.off() - self._is_on = False - - def close(self): - self._led.close() - - -led_module.GPIOLED = FakeGPIOLED - -from rdm_card_reader.cli import main - -code = main(["--port", os.environ["SERIAL_PORT"], "--heartbeat", "0.3", - "--led-mode", os.environ.get("LED_MODE", "error"), "--json"]) -sys.exit(code) -''' - - -def em4100_frame(value, card_type=0x0F): - id_bytes = struct.pack(">I", value) - checksum = card_type - for b in id_bytes: - checksum ^= b - frame = (bytes([card_type]) + id_bytes + bytes([checksum])).hex().upper() - assert len(frame) == 12, frame - return frame.encode("ascii") - - -def pump(stream, sink): - """Read a stream on its own thread. - - Text-mode pipes buffer, so select() would not report this data. - """ - for line in stream: - sink.append(line.rstrip()) - - -def run_case(mode): - """Drive a card through the app and return (events, led transitions).""" - master, slave = pty.openpty() - serial_port = os.ttyname(slave) - - env = dict(os.environ) - env["SERIAL_PORT"] = serial_port - env["PROJECT_SRC"] = os.path.join(PROJECT, "src") - env["LED_MODE"] = mode - env.pop("PYTHONPATH", None) - - proc = subprocess.Popen( - [PYTHON, "-u", "-c", DRIVER], - cwd=PROJECT, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, - text=True, bufsize=1, - ) - - stdout, stderr = [], [] - t1 = threading.Thread(target=pump, args=(proc.stdout, stdout), daemon=True) - t2 = threading.Thread(target=pump, args=(proc.stderr, stderr), daemon=True) - t1.start() - t2.start() - - frame = em4100_frame(CARD_VALUE) - try: - time.sleep(2.0) # let the reader open the port - for _ in range(6): # card present - os.write(master, b"\x02" + frame + b"\x03") - time.sleep(0.1) - time.sleep(1.2) # stop writing -> card removed - finally: - try: - proc.terminate() - except Exception: - pass - try: - proc.wait(timeout=5) - except Exception: - proc.kill() - for t in (t1, t2): - t.join(timeout=3) - try: - os.close(master) - os.close(slave) - except Exception: - pass - - transitions = [ - line.split("=", 1)[1] - for line in stderr - if line.startswith("TRANSITION=") - ] - return stdout, stderr, transitions - - -def main(): - failures = 0 - - # --- error mode: lit while running, dark while a card is read ---------- - stdout, stderr, transitions = run_case("error") - print("======== error mode ========") - for line in stdout[:2]: - print(" " + line) - print(" events: {}".format( - [l.split('"event": "')[1].split('"')[0] - for l in stdout if '"event"' in l])) - print(" transitions: {}".format(transitions)) - - checks = { - "card was read": any("card_inserted" in l for l in stdout), - "card removal seen": any("card_removed" in l for l in stdout), - "lit at startup": transitions[:1] == ["on"], - "dark while reading": "off" in transitions, - "lit again after": transitions[-1:] == ["on"], - } - - # --- presence mode: dark until a card arrives ------------------------- - stdout2, stderr2, transitions2 = run_case("presence") - print("\n======== presence mode ========") - print(" transitions: {}".format(transitions2)) - checks.update({ - "presence: read still works": - any("card_inserted" in l for l in stdout2), - "presence: dark at startup": transitions2[:1] == ["on"] or not transitions2, - "presence: lit on the card": "on" in transitions2, - "presence: dark after": transitions2[-1:] == ["off"], - }) - - print() - ok = True - for name, passed in checks.items(): - print(" [{}] {}".format("PASS" if passed else "FAIL", name)) - ok = ok and bool(passed) - - if not ok: - failures = 1 - print("\n--- driver stderr (error mode) ---") - print("\n".join(stderr)[:1200]) - return failures - - -if __name__ == "__main__": - sys.exit(main()) diff --git a/tests/test_e2e_pty.py b/tests/test_e2e_pty.py deleted file mode 100644 index ac36ba1..0000000 --- a/tests/test_e2e_pty.py +++ /dev/null @@ -1,165 +0,0 @@ -"""End-to-end test over a pseudo-terminal, with no RFID hardware. - -The test owns the master side of a PTY, points the CLI at the slave side and -writes real RDM6300 frames. This exercises the same code path a physical -reader would: serial open -> byte stream -> parse -> callback -> stdout. -""" - -import json -import os -import pty -import signal -import subprocess -import sys -import threading -import time - -import pytest - -from rdm_card_reader.card import CONFIG_CARD_VALUE - -from .test_protocol import encode_stream, encode_frame # noqa: F401 - -ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) -SRC = os.path.join(ROOT, "src") - -TIMEOUT = 15.0 - - -class ReaderProcess: - """Run the CLI against a PTY and collect its stdout in the background.""" - - def __init__(self, extra_args=()): - self.master, slave = pty.openpty() - self.port = os.ttyname(slave) - env = dict(os.environ) - env["PYTHONPATH"] = SRC + os.pathsep + env.get("PYTHONPATH", "") - self.proc = subprocess.Popen( - [sys.executable, "-u", "-m", "rdm_card_reader", - "--port", self.port, "--heartbeat", "0.3", *extra_args], - cwd=ROOT, env=env, stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, text=True, bufsize=1, - start_new_session=True, - ) - self.lines = [] - self._lock = threading.Lock() - self._thread = threading.Thread(target=self._pump, daemon=True) - self._thread.start() - - def _pump(self): - for line in self.proc.stdout: - with self._lock: - self.lines.append(line.rstrip()) - - def output(self): - with self._lock: - return "\n".join(self.lines) - - def send(self, frame): - os.write(self.master, b"\x02" + frame + b"\x03") - - def wait_for(self, needle, timeout=TIMEOUT): - deadline = time.time() + timeout - while time.time() < deadline: - if needle in self.output(): - return True - time.sleep(0.05) - return False - - def __enter__(self): - self.wait_for("waiting for cards") - return self - - def __exit__(self, *exc): - try: - os.killpg(os.getpgid(self.proc.pid), signal.SIGINT) - except Exception: - pass - try: - self.proc.wait(timeout=5) - except Exception: - self.proc.kill() - self._thread.join(timeout=3) - for fd in (self.master, self.proc.stdout): - try: - fd.close() - except Exception: - pass - return False - - -@pytest.mark.skipif(sys.platform != "linux", reason="PTY test is Linux only") -def test_card_insert_and_remove_are_reported(): - card_id = 0x00C4A0B5 - frame = encode_frame(card_id) - - with ReaderProcess() as rp: - deadline = time.time() + TIMEOUT - while time.time() < deadline and not rp.wait_for("CARD INSERTED", 0.2): - rp.send(frame) - - assert rp.wait_for("CARD INSERTED"), rp.output() - assert rp.wait_for("CARD REMOVED") - - text = rp.output() - assert rp.port in text - assert str(card_id) in text - assert "0x00C4A0B5" in text - assert "222" in text # checksum = 0xDE - assert "0x0F" in text # type - assert "True" in text # valid - - -@pytest.mark.skipif(sys.platform != "linux", reason="PTY test is Linux only") -def test_corrupt_checksum_is_reported_as_invalid(): - good = bytearray(encode_frame(0x1A2B3C4D)) - good[-1] = ord("0") # break the checksum - bad = bytes(good) - - with ReaderProcess() as rp: - deadline = time.time() + TIMEOUT - while time.time() < deadline and not rp.wait_for("INVALID CARD", 0.2): - rp.send(bad) - - assert "INVALID CARD" in rp.output() - assert "checksum mismatch" in rp.output() - - -@pytest.mark.skipif(sys.platform != "linux", reason="PTY test is Linux only") -def test_json_mode_emits_machine_readable_events(): - with ReaderProcess(extra_args=("--json",)) as rp: - deadline = time.time() + TIMEOUT - while time.time() < deadline and not rp.wait_for('"event"', 0.2): - rp.send(encode_frame(0x00C4A0B5)) - - payloads = [ - json.loads(line) - for line in rp.output().splitlines() - if line.strip().startswith("{") - ] - assert payloads, rp.output() - assert any(p["event"] == "card_inserted" for p in payloads) - assert any(p["id"] == 0x00C4A0B5 for p in payloads) - assert all(p["port"] == rp.port for p in payloads) - - -@pytest.mark.skipif(sys.platform != "linux", reason="PTY test is Linux only") -def test_config_card_is_annotated(): - with ReaderProcess() as rp: - deadline = time.time() + TIMEOUT - while time.time() < deadline and not rp.wait_for("config card", 0.2): - rp.send(encode_frame(CONFIG_CARD_VALUE)) - - assert "config card" in rp.output() - - -@pytest.mark.skipif(sys.platform != "linux", reason="PTY test is Linux only") -def test_removal_after_heartbeat_expires(): - with ReaderProcess() as rp: - deadline = time.time() + TIMEOUT - while time.time() < deadline and not rp.wait_for("CARD INSERTED", 0.2): - rp.send(encode_frame(0x00C4A0B5)) - # stop sending and let the 0.3 s heartbeat lapse - assert rp.wait_for("CARD REMOVED", timeout=5) - - assert "CARD REMOVED" in rp.output() diff --git a/tests/test_led.py b/tests/test_led.py deleted file mode 100644 index 8b52de2..0000000 --- a/tests/test_led.py +++ /dev/null @@ -1,306 +0,0 @@ -"""Tests for the status LED. - -Two things matter: the LED must never be able to break the reader, and it must -behave the way the workstation app did (on with the card, off without). Real -GPIO is unavailable here, so a fake pin is injected to test the logic. -""" - -import os -import sys -import threading -import time - -SRC = os.path.join( - os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "src" -) -if SRC not in sys.path: - sys.path.insert(0, SRC) - -import pytest # noqa: E402 - -from rdm_card_reader import led as led_module # noqa: E402 -from rdm_card_reader.led import ( # noqa: E402 - BLINK, - DEFAULT_LED_PIN, - ERROR, - OFF, - ON, - PRESENCE, - NullLED, - StatusLED, - create, -) - - -class FakeLED: - """Stands in for gpiozero.LED, recording what it was told to do.""" - - def __init__(self, pin, active_high=True): - self.pin = pin - self.active_high = active_high - self.state = False - self.transitions = [] - self.closed = False - - def on(self): - self.state = True - self.transitions.append("on") - - def off(self): - self.state = False - self.transitions.append("off") - - def close(self): - self.closed = True - - -@pytest.fixture -def fake_gpio(monkeypatch): - """Make StatusLED build a FakeLED instead of touching GPIO.""" - created = {} - - class FakeGPIOLED(led_module.GPIOLED): - def __init__(self, pin, active_high=True): - self.pin = pin - self._led = FakeLED(pin, active_high) - self._is_on = False - created["led"] = self._led - - def on(self): - self._led.on() - self._is_on = True - - def off(self): - self._led.off() - self._is_on = False - - def close(self): - self._led.close() - - monkeypatch.setattr(led_module, "GPIOLED", FakeGPIOLED) - return created - - -def wait_for_state(led_obj, expected, timeout=3.0): - """Wait for the worker thread to apply a command.""" - deadline = time.time() + timeout - while time.time() < deadline: - if led_obj.state is expected: - return True - time.sleep(0.01) - return False - - -# -- defaults --------------------------------------------------------------- - -def test_default_pin_matches_the_workstation_app(): - assert DEFAULT_LED_PIN == 23 - - -# -- graceful degradation --------------------------------------------------- - -def test_no_gpio_gives_a_working_no_op(monkeypatch): - """On a machine without GPIO the LED must be inert, not fatal.""" - def boom(pin, active_high=True): - raise RuntimeError("Unable to load any default pin factory!") - - monkeypatch.setattr(led_module, "GPIOLED", boom) - status = StatusLED(pin=23) - assert status.status.enabled is False - assert status.status.backend == "null" - # Every hook must still be safe to call. - status.insert() - status.remove() - status.error() - status.close() - - -def test_missing_gpiozero_is_reported_not_raised(monkeypatch): - """gpiozero absent: say so, and do not raise.""" - import builtins - - real_import = builtins.__import__ - - def blocked(name, *args, **kwargs): - if name == "gpiozero": - raise ImportError("No module named 'gpiozero'") - return real_import(name, *args, **kwargs) - - monkeypatch.setattr(builtins, "__import__", blocked) - status = StatusLED(pin=23) - assert status.status.enabled is False - assert "not installed" in status.status.reason - - -def test_no_pin_factory_is_not_reported_as_missing_library(monkeypatch): - """BadPinFactory subclasses ImportError, so the two must be distinguished. - - Reporting 'gpiozero is not installed' when it is installed but has no - usable pin library would send you to the wrong fix. - """ - class FakeBadPinFactory(ImportError): - pass - - def boom(pin, active_high=True): - raise FakeBadPinFactory( - "Unable to load any default pin factory!" - ) - - monkeypatch.setattr(led_module, "GPIOLED", boom) - status = StatusLED(pin=23) - assert status.status.enabled is False - assert "not installed" not in status.status.reason - assert "pin factory" in status.status.reason - - -def test_disabled_led_is_inert(fake_gpio): - status = StatusLED(pin=23, enabled=False) - assert status.status.enabled is False - assert isinstance(status.led, NullLED) - status.insert() - time.sleep(0.05) - assert "led" not in fake_gpio - - -def test_null_led_methods_are_safe(): - null = NullLED(23, "no gpio") - null.on() - null.off() - null.close() - - -# -- behaviour -------------------------------------------------------------- - -def test_error_mode_lights_up_at_startup(fake_gpio): - """A health indicator should show 'alive' without waiting for a card.""" - status = StatusLED(pin=23, mode=ERROR) - assert fake_gpio["led"].state is True - status.close() - - -def test_error_mode_blinks_on_a_bad_read(fake_gpio): - status = StatusLED(pin=23, mode=ERROR, error_blink=True) - status.error() - time.sleep(0.25) - assert len(fake_gpio["led"].transitions) >= 2 - status.close() - - -def test_error_mode_settles_back_after_a_blink(fake_gpio): - status = StatusLED(pin=23, mode=ERROR, error_blink=True) - status.error() - time.sleep(0.25) - status.remove() # any steady command cancels blinking - assert wait_for_state(fake_gpio["led"], True) - status.close() - - -def test_error_mode_without_blink_stays_lit(fake_gpio): - status = StatusLED(pin=23, mode=ERROR, error_blink=False) - status.error() - assert wait_for_state(fake_gpio["led"], True) - status.close() - - -def test_presence_mode_is_dark_until_a_card_arrives(fake_gpio): - status = StatusLED(pin=23, mode=PRESENCE) - assert fake_gpio["led"].state is False - status.insert() - assert wait_for_state(fake_gpio["led"], True) - status.remove() - assert wait_for_state(fake_gpio["led"], False) - status.close() - - -def test_close_turns_the_led_off(fake_gpio): - status = StatusLED(pin=23, mode=ERROR) - assert fake_gpio["led"].state is True - status.close() - assert fake_gpio["led"].closed is True - assert fake_gpio["led"].state is False - - -def test_pin_and_active_high_are_passed_through(fake_gpio): - status = StatusLED(pin=17, active_high=False) - assert fake_gpio["led"].pin == 17 - assert fake_gpio["led"].active_high is False - status.close() - - -# -- never blocks or crashes ------------------------------------------------ - -def test_insert_does_not_block(monkeypatch): - """The reader thread calls this, so it must return immediately.""" - class SlowLED(led_module.GPIOLED): - def __init__(self, pin, active_high=True): - self.pin = pin - self._is_on = False - - def on(self): - time.sleep(0.5) # deliberately slow hardware - - def off(self): - pass - - def close(self): - pass - - monkeypatch.setattr(led_module, "GPIOLED", SlowLED) - # mode=presence so the startup light-up does not pay the slow 'on' cost - status = StatusLED(pin=23, mode=PRESENCE) - start = time.time() - for _ in range(20): - status.insert() - elapsed = time.time() - start - assert elapsed < 0.2, "insert() blocked for {:.2f}s".format(elapsed) - status.close() - - -def test_failing_toggle_does_not_raise(fake_gpio, monkeypatch): - """A dead pin must not take the reader down.""" - status = StatusLED(pin=23) - assert fake_gpio["led"] is not None - - def boom(): - raise OSError("pin went away") - - monkeypatch.setattr(status.led, "on", boom) - status.insert() - time.sleep(0.15) # worker swallows the error - status.close() # and close() still works - - -def test_hardware_error_is_reported_to_the_logger(monkeypatch): - lines = [] - - def boom(pin, active_high=True): - raise RuntimeError("no pin factory") - - monkeypatch.setattr(led_module, "GPIOLED", boom) - StatusLED(pin=23, logger=lines.append) - assert any("LED unavailable" in line for line in lines) - - -# -- factory ---------------------------------------------------------------- - -def test_create_defaults_to_error_mode_on_the_workstation_pin(fake_gpio): - status = create() - assert status.status.enabled is True - assert status.status.pin == DEFAULT_LED_PIN - assert status.status.mode == ERROR - status.close() - - -def test_create_disabled_returns_a_no_op(): - status = create(enabled=False) - assert status.status.enabled is False - status.close() - - -def test_status_dict_is_json_serialisable(fake_gpio): - import json - - status = create() - payload = json.dumps(status.status.to_dict()) - assert "gpiozero" in payload - status.close() diff --git a/tests/test_ports.py b/tests/test_ports.py deleted file mode 100644 index b1e8ede..0000000 --- a/tests/test_ports.py +++ /dev/null @@ -1,228 +0,0 @@ -"""Tests for reliable serial-port identification. - -The point of these tests is that picking the wrong ttyUSB silently is the -failure mode that matters, so ambiguity must raise rather than guess. -""" - -import os -import sys - -SRC = os.path.join( - os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "src" -) -if SRC not in sys.path: - sys.path.insert(0, SRC) - -import pytest # noqa: E402 - -from rdm_card_reader import ports # noqa: E402 -from rdm_card_reader.ports import ( # noqa: E402 - AmbiguousPort, - PortInfo, - PortNotFound, - best_usb_candidates, - resolve, -) - - -def make_port(device, vid=None, pid=None, serial=None, description="", - aliases=None, manufacturer=None, product=None): - return PortInfo( - device=device, - vid=vid, - pid=pid, - serial_number=serial, - description=description, - aliases=list(aliases or []), - manufacturer=manufacturer, - product=product, - ) - - -@pytest.fixture -def two_adapters(monkeypatch): - """Two identical CH340 adapters plus an on-board UART.""" - a = make_port("/dev/ttyUSB0", 0x1A86, 0x7523, "CH340A", - "USB Serial", ["/dev/serial/by-id/usb-1a86_A-if00-port0"]) - b = make_port("/dev/ttyUSB1", 0x1A86, 0x7523, "CH340B", - "USB Serial", ["/dev/serial/by-id/usb-1a86_B-if00-port0"]) - onboard = make_port("/dev/ttyS0", description="ttyS0") - monkeypatch.setattr(ports, "discover", lambda: [a, b, onboard]) - return a, b, onboard - - -# -- identity --------------------------------------------------------------- - -def test_usb_id_and_chip_name(): - port = make_port("/dev/ttyUSB0", 0x1A86, 0x7523) - assert port.usb_id == "1a86:7523" - assert port.chip == "CH340" - assert port.is_usb is True - - -def test_unknown_chip_still_identified_as_usb(): - port = make_port("/dev/ttyUSB0", 0xFFFF, 0x0001) - assert port.chip is None - assert port.is_usb is True - - -def test_stable_id_prefers_by_id_symlink(): - port = make_port( - "/dev/ttyUSB0", - aliases=["/dev/serial/by-path/pci-1-usb-0:1.2:1.0-port0", - "/dev/serial/by-id/usb-1a86_USB-if00-port0"], - ) - assert port.stable_id == "/dev/serial/by-id/usb-1a86_USB-if00-port0" - assert port.by_path.endswith("port0") - - -def test_onboard_uart_is_not_usb(): - assert make_port("/dev/ttyS0").is_usb is False - - -# -- scoring ---------------------------------------------------------------- - -def test_usb_adapter_scores_above_onboard_uart(): - adapter = make_port("/dev/ttyUSB0", 0x1A86, 0x7523, "SN1") - onboard = make_port("/dev/ttyS0") - assert adapter.score() > onboard.score() - - -def test_serial_number_raises_score(): - with_sn = make_port("/dev/ttyUSB0", 0x1A86, 0x7523, "SN1") - without = make_port("/dev/ttyUSB0", 0x1A86, 0x7523, None) - assert with_sn.score() > without.score() - - -# -- matching --------------------------------------------------------------- - -def test_match_by_device_path(): - assert make_port("/dev/ttyUSB0").matches("/dev/ttyUSB0") - - -def test_match_by_serial_number(): - port = make_port("/dev/ttyUSB0", 0x1A86, 0x7523, "CH340A") - assert port.matches("CH340A") - - -def test_match_by_vid_pid(): - port = make_port("/dev/ttyUSB0", 0x1A86, 0x7523) - assert port.matches("1a86:7523") - - -def test_match_by_chip_name(): - assert make_port("/dev/ttyUSB0", 0x1A86, 0x7523).matches("ch340") - - -def test_match_by_id_symlink(): - port = make_port("/dev/ttyUSB0", - aliases=["/dev/serial/by-id/usb-1a86_A-if00-port0"]) - assert port.matches("/dev/serial/by-id/usb-1a86_A-if00-port0") - - -def test_empty_selector_never_matches(): - assert not make_port("/dev/ttyUSB0").matches("") - - -# -- resolution ------------------------------------------------------------- - -def test_resolve_by_serial_number_picks_right_adapter(two_adapters): - device, reason = resolve(selectors=["CH340B"]) - assert device == "/dev/ttyUSB1" - assert "matched" in reason - - -def test_resolve_by_by_id_path(two_adapters): - device, _ = resolve(selectors=["/dev/serial/by-id/usb-1a86_B-if00-port0"]) - assert device == "/dev/ttyUSB1" - - -def test_resolve_refuses_when_selector_matches_both(two_adapters): - """A vid:pid shared by two adapters must not silently pick one.""" - with pytest.raises(AmbiguousPort): - resolve(selectors=["1a86:7523"]) - - -def test_resolve_unknown_selector_raises_not_found(two_adapters): - with pytest.raises(PortNotFound): - resolve(selectors=["NOSUCHDEVICE"]) - - -def test_resolve_falls_back_to_raw_path_when_not_enumerable(monkeypatch, tmp_path): - fake = tmp_path / "myport" - fake.write_text("") - monkeypatch.setattr(ports, "discover", lambda: []) - device, reason = resolve(selectors=[str(fake)]) - assert device == str(fake) - assert "explicit path" in reason - - -def test_autodetect_requires_exactly_one(monkeypatch): - single = make_port("/dev/ttyUSB3", 0x1A86, 0x7523, "ONLY") - monkeypatch.setattr(ports, "discover", lambda: [single, make_port("/dev/ttyS0")]) - device, reason = resolve(autodetect=True) - assert device == "/dev/ttyUSB3" - assert "autodetected" in reason - - -def test_autodetect_refuses_with_two_adapters(two_adapters): - with pytest.raises(AmbiguousPort): - resolve(autodetect=True) - - -def test_autodetect_ignores_onboard_uart(monkeypatch): - """An on-board UART must not make a single USB adapter look ambiguous.""" - adapter = make_port("/dev/ttyUSB0", 0x1A86, 0x7523, "ONLY") - monkeypatch.setattr( - ports, "discover", - lambda: [adapter, make_port("/dev/ttyS0"), make_port("/dev/ttyAMA0")], - ) - device, _ = resolve(autodetect=True) - assert device == "/dev/ttyUSB0" - - -def test_fallback_order_used_when_no_usb(monkeypatch): - """Must not depend on the host actually having a serial port.""" - monkeypatch.setattr(ports, "discover", lambda: []) - monkeypatch.setattr(ports.os.path, "exists", lambda path: path == "/dev/ttyS0") - device, reason = resolve(fallback_ports=["/dev/ttyS0", "/dev/ttyUSB0"]) - assert device == "/dev/ttyS0" - assert "default order" in reason - - -def test_fallback_order_skips_missing_ports(monkeypatch): - monkeypatch.setattr(ports, "discover", lambda: []) - monkeypatch.setattr( - ports.os.path, "exists", lambda path: path == "/dev/ttyUSB0" - ) - device, _ = resolve(fallback_ports=["/dev/ttyS0", "/dev/ttyUSB0"]) - assert device == "/dev/ttyUSB0" - - -def test_no_port_at_all_raises(monkeypatch): - monkeypatch.setattr(ports, "discover", lambda: []) - with pytest.raises(PortNotFound): - resolve(fallback_ports=[]) - - -# -- helper ---------------------------------------------------------------- - -def test_best_usb_candidates_dedupes_same_identity(): - a = make_port("/dev/ttyUSB0", 0x1A86, 0x7523, "SAME") - dup = make_port("/dev/ttyUSB1", 0x1A86, 0x7523, "SAME") - result = best_usb_candidates([a, dup]) - assert len(result) == 1 - - -def test_best_usb_candidates_keeps_distinct_serial_numbers(): - a = make_port("/dev/ttyUSB0", 0x1A86, 0x7523, "A") - b = make_port("/dev/ttyUSB1", 0x1A86, 0x7523, "B") - assert len(best_usb_candidates([a, b])) == 2 - - -def test_describe_is_human_readable(): - port = make_port("/dev/ttyUSB0", 0x1A86, 0x7523, "ABC", "USB Serial") - text = port.describe() - assert "/dev/ttyUSB0" in text - assert "CH340" in text - assert "ABC" in text diff --git a/tests/test_protocol.py b/tests/test_protocol.py deleted file mode 100644 index 927cc7a..0000000 --- a/tests/test_protocol.py +++ /dev/null @@ -1,108 +0,0 @@ -"""RDM6300 wire format: type + id + checksum, 12 hex nibbles per frame. - -The checksum is the XOR of the type byte *and* the four id bytes. Getting this -wrong (e.g. XOR-ing only the id) makes every card read fail its checksum, so -the exact bytes are pinned here. -""" - -import struct - -import pytest - -from rdm_card_reader.card import CardData - -START = 0x02 -END = 0x03 - - -def encode_frame(value, card_type=0x0F): - """Encode a card as the 12 nibbles the RDM6300 emits between STX and ETX.""" - id_bytes = struct.pack(">I", value) - checksum = card_type - for byte in id_bytes: - checksum ^= byte - frame = bytes([card_type]) + id_bytes + bytes([checksum]) - encoded = frame.hex().upper().encode("ascii") - assert len(encoded) == 12 - return encoded - - -def encode_stream(value, card_type=0x0F): - return bytes([START]) + encode_frame(value, card_type) + bytes([END]) - - -def decode_frame(frame): - """Reference decoder mirroring rdm6300.BaseReader._parse_fragment.""" - nibbles = [int(ch, 16) for ch in frame.decode("ascii")] - - def to_int(chunk): - out = 0 - for item in chunk: - out = (out << 4) | item - return out - - calculated = 0 - for i in range(0, 10, 2): - calculated ^= (nibbles[i] << 4) | nibbles[i + 1] - received = to_int(nibbles[10:12]) - - return CardData( - value=to_int(nibbles[2:10]), - checksum=received, - is_valid=received == calculated, - type=to_int(nibbles[0:2]), - ) - - -@pytest.mark.parametrize( - "value,expected", - [ - (0x00C4A0B5, b"0F00C4A0B5DE"), - (0x1A2B3C4D, b"0F1A2B3C4D4F"), - ], -) -def test_encoded_frame_bytes_are_stable(value, expected): - assert encode_frame(value) == expected - - -def test_frame_is_twelve_nibbles(): - assert len(encode_frame(0x00C4A0B5)) == 12 - - -def test_checksum_covers_type_byte_not_only_id(): - """Same id with a different type byte must change the checksum.""" - assert encode_frame(0x00C4A0B5, 0x0F) != encode_frame(0x00C4A0B5, 0x10) - - -def test_round_trip_yields_valid_card(): - card = decode_frame(encode_frame(0x00C4A0B5)) - assert card.value == 0x00C4A0B5 - assert card.type == 0x0F - assert card.checksum == 0xDE - assert card.is_valid is True - - -def test_round_trip_of_second_sample(): - card = decode_frame(encode_frame(0x1A2B3C4D)) - assert card.value == 0x1A2B3C4D - assert card.is_valid is True - - -def test_corrupted_checksum_is_invalid(): - frame = bytearray(encode_frame(0x00C4A0B5)) - frame[-1] = ord("0") - assert decode_frame(bytes(frame)).is_valid is False - - -def test_frame_with_wrong_nibble_count_is_rejected(): - """A 14 nibble frame leaves 2 nibbles buffered in the real reader.""" - assert len(encode_frame(0x00C4A0B5)) == 12 - fourteen = b"0F00C4A0B5DE00" - assert len(fourteen) == 14 - - -def test_stream_wrapper_uses_stx_and_etx(): - stream = encode_stream(0x00C4A0B5) - assert stream[0] == START - assert stream[-1] == END - assert stream[1:-1] == encode_frame(0x00C4A0B5) diff --git a/tests/test_reader.py b/tests/test_reader.py deleted file mode 100644 index 398fca0..0000000 --- a/tests/test_reader.py +++ /dev/null @@ -1,149 +0,0 @@ -"""Port selection and error explanation, using a fake reader library.""" - -import types - -import pytest - -import rdm_card_reader.reader as reader_module -from rdm_card_reader.card import CANDIDATE_PORTS -from rdm_card_reader.reader import explain_error, list_ports, open_reader - - -class FakeBaseReader: - """Stands in for rdm6300.BaseReader; records the port it was built with.""" - - fail_on = () - - def __init__(self, port, heartbeat_interval=0.5): - if port in type(self).fail_on: - raise OSError("could not open port {}".format(port)) - self.port = port - self.heartbeat_interval = heartbeat_interval - self.started = False - self.closed = False - - def start(self): - self.started = True - - def close(self): - self.closed = True - - -@pytest.fixture -def fake_library(monkeypatch): - module = types.SimpleNamespace(BaseReader=FakeBaseReader) - monkeypatch.setattr(reader_module, "rdm6300", module) - return module - - -@pytest.fixture(autouse=True) -def restore_library(): - original = reader_module.rdm6300 - yield - reader_module.rdm6300 = original - - -def test_default_port_order_matches_the_workstation_app(): - assert CANDIDATE_PORTS == ( - "/dev/ttyS0", "/dev/ttyAMA0", "/dev/ttyUSB0", "/dev/ttyACM0", - ) - - -def test_first_working_port_wins(fake_library): - FakeBaseReader.fail_on = ("/dev/ttyS0", "/dev/ttyAMA0") - reader = open_reader(lambda event: None, - ports=["/dev/ttyS0", "/dev/ttyAMA0", "/dev/ttyUSB0"]) - assert reader is not None - assert reader.port == "/dev/ttyUSB0" - - -def test_returns_none_when_no_port_works(fake_library): - FakeBaseReader.fail_on = ("/dev/ttyS0", "/dev/ttyAMA0") - assert open_reader(lambda event: None, - ports=["/dev/ttyS0", "/dev/ttyAMA0"]) is None - - -def test_failures_are_reported_with_the_port(fake_library): - FakeBaseReader.fail_on = ("/dev/ttyS0",) - seen = [] - open_reader(lambda event: None, ports=["/dev/ttyS0"], - on_error=lambda port, message: seen.append((port, message))) - assert seen and seen[0][0] == "/dev/ttyS0" - - -def test_heartbeat_interval_is_passed_through(fake_library): - FakeBaseReader.fail_on = () - reader = open_reader(lambda event: None, ports=["/dev/ttyS0"], - heartbeat_interval=1.5) - assert reader.heartbeat_interval == 1.5 - - -def test_heartbeat_none_disables_removal_detection(fake_library): - FakeBaseReader.fail_on = () - reader = open_reader(lambda event: None, ports=["/dev/ttyS0"], - heartbeat_interval=None) - assert reader.heartbeat_interval is None - - -def test_events_are_forwarded_to_the_callback(fake_library): - FakeBaseReader.fail_on = () - received = [] - reader = open_reader(received.append, ports=["/dev/ttyS0"]) - - class Card: - value = 7 - checksum = 1 - type = 15 - is_valid = True - - reader.card_inserted(Card()) - assert len(received) == 1 - assert received[0].value == 7 - assert received[0].event == "card_inserted" - - -def test_card_removed_with_none_is_ignored(fake_library): - FakeBaseReader.fail_on = () - received = [] - reader = open_reader(received.append, ports=["/dev/ttyS0"]) - reader.card_removed(None) - assert received == [] - - -def test_invalid_card_is_forwarded(fake_library): - FakeBaseReader.fail_on = () - received = [] - reader = open_reader(received.append, ports=["/dev/ttyS0"]) - - class Card: - value = 1 - checksum = 2 - type = 15 - is_valid = False - - reader.invalid_card(Card()) - assert received[0].event == "invalid_card" - - -def test_missing_library_raises(monkeypatch): - monkeypatch.setattr(reader_module, "rdm6300", None) - with pytest.raises(reader_module.ReaderUnavailable): - open_reader(lambda event: None) - - -@pytest.mark.parametrize( - "message,expected_fragment", - [ - ("[Errno 13] Permission denied: '/dev/ttyS0'", "dialout"), - ("could not open port /dev/nope", "device not found"), - ("something else", "something else"), - ], -) -def test_explain_error(message, expected_fragment): - assert expected_fragment in explain_error("/dev/ttyS0", message) - - -def test_list_ports_shape(): - info = list_ports() - assert set(info) == {"candidates", "detected"} - assert [port for port, _ in info["candidates"]] == list(CANDIDATE_PORTS) diff --git a/tests/test_wheels.py b/tests/test_wheels.py deleted file mode 100644 index 48cfd40..0000000 --- a/tests/test_wheels.py +++ /dev/null @@ -1,229 +0,0 @@ -"""Tests for the vendored wheels and their verification. - -The bundled wheels let a device with no internet set itself up, so the things -that matter are: they are actually present and pure-Python (portable), the -checksums match, and a tampered wheel is refused rather than installed. -""" - -import os -import shutil -import sys - -SRC = os.path.join( - os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "src" -) -if SRC not in sys.path: - sys.path.insert(0, SRC) - -import pytest # noqa: E402 - -from rdm_card_reader import bootstrap # noqa: E402 -from rdm_card_reader.bootstrap import ( # noqa: E402 - bundled_wheel_dir, - ensure_dependencies, - local_wheel_missing, - verify_wheels, -) - - -# -- the wheels that ship with the package --------------------------------- - -def test_bundled_wheel_dir_exists(): - folder = bundled_wheel_dir() - assert folder is not None - assert os.path.isdir(folder) - - -def test_both_required_wheels_are_present(): - folder = bundled_wheel_dir() - names = os.listdir(folder) - assert any(n.startswith("rdm6300-") and n.endswith(".whl") for n in names) - assert any(n.startswith("pyserial-") and n.endswith(".whl") for n in names) - - -def test_bundled_wheels_are_pure_python(): - """A compiled wheel would only work on one architecture; these must not.""" - folder = bundled_wheel_dir() - wheels = [n for n in os.listdir(folder) if n.endswith(".whl")] - assert wheels - for name in wheels: - assert name.endswith("-none-any.whl"), ( - "{} is architecture-specific and cannot be shipped".format(name) - ) - - -def test_bundled_wheels_verify_against_their_checksums(): - folder = bundled_wheel_dir() - assert verify_wheels(folder) == [] - - -def test_checksums_file_is_shipped(): - folder = bundled_wheel_dir() - assert os.path.exists(os.path.join(folder, "SHA256SUMS")) - - -def test_total_vendored_size_is_reasonable(): - """Guard against someone dropping a huge binary into the repo.""" - folder = bundled_wheel_dir() - total = sum( - os.path.getsize(os.path.join(folder, name)) - for name in os.listdir(folder) - ) - assert total < 2 * 1024 * 1024 - - -# -- verification ----------------------------------------------------------- - -def test_tampered_wheel_is_detected(tmp_path): - folder = tmp_path / "wheels" - folder.mkdir() - wheel = folder / "rdm6300-0.1.1-py3-none-any.whl" - wheel.write_bytes(b"not the real wheel") - (folder / "SHA256SUMS").write_text( - "afb067da8b8d23a477138ca67d4ff1d198f12d157b64668c9dd3cbd7df54e74e" - " rdm6300-0.1.1-py3-none-any.whl\n" - ) - problems = verify_wheels(str(folder)) - assert problems - assert any("checksum mismatch" in p for p in problems) - - -def test_missing_wheel_is_reported(tmp_path): - folder = tmp_path / "wheels" - folder.mkdir() - (folder / "SHA256SUMS").write_text( - "abc123 rdm6300-0.1.1-py3-none-any.whl\n" - ) - problems = verify_wheels(str(folder)) - assert any("missing" in p for p in problems) - - -def test_no_sums_file_means_nothing_to_check(tmp_path): - folder = tmp_path / "wheels" - folder.mkdir() - (folder / "x-1.0-py3-none-any.whl").write_bytes(b"x") - assert verify_wheels(str(folder)) == [] - - -def test_verification_passes_for_untouched_copy(tmp_path): - source = bundled_wheel_dir() - dest = tmp_path / "wheels" - shutil.copytree(source, dest) - assert verify_wheels(str(dest)) == [] - - -# -- coverage check --------------------------------------------------------- - -def test_local_wheel_missing_detects_absent_requirement(tmp_path): - folder = tmp_path / "wheels" - folder.mkdir() - assert local_wheel_missing(["rdm6300>=0.1.1"], str(folder)) == ["rdm6300>=0.1.1"] - - -def test_local_wheel_missing_accepts_present_requirement(): - folder = bundled_wheel_dir() - assert local_wheel_missing(["rdm6300>=0.1.1", "pyserial>=3.5"], folder) == [] - - -def test_local_wheel_missing_normalises_underscores(tmp_path): - folder = tmp_path / "wheels" - folder.mkdir() - (folder / "charset_normalizer-3.4.1-py3-none-any.whl").write_bytes(b"x") - assert local_wheel_missing(["charset-normalizer>=3"], str(folder)) == [] - - -# -- integration into ensure_dependencies ----------------------------------- - -def test_bundled_wheels_used_automatically(monkeypatch): - """No --wheel-dir given, so the bundled ones must be picked up.""" - seen = {} - monkeypatch.setattr( - bootstrap, "missing_packages", - lambda packages=None: [] if seen.get("done") else [("rdm6300", "rdm6300>=0.1.1")], - ) - - def fake_install(requirements, wheel_dir=None, logger=None, timeout=300, no_deps=False): - seen["wheel_dir"] = wheel_dir - seen["done"] = True - return True - - monkeypatch.setattr(bootstrap, "install", fake_install) - lines = [] - assert ensure_dependencies(logger=lines.append) is True - assert seen["wheel_dir"] == bundled_wheel_dir() - assert any("bundled" in line for line in lines) - - -def test_tampered_bundle_falls_back_to_pypi(monkeypatch, tmp_path): - """A bad checksum must not block installation - but must be reported.""" - folder = tmp_path / "wheels" - folder.mkdir() - (folder / "rdm6300-0.1.1-py3-none-any.whl").write_bytes(b"tampered") - (folder / "SHA256SUMS").write_text("deadbeef rdm6300-0.1.1-py3-none-any.whl\n") - - seen = {} - monkeypatch.setattr(bootstrap, "bundled_wheel_dir", lambda: str(folder)) - monkeypatch.setattr( - bootstrap, "missing_packages", - lambda packages=None: [] if seen.get("done") else [("rdm6300", "rdm6300>=0.1.1")], - ) - - def fake_install(requirements, wheel_dir=None, logger=None, timeout=300, no_deps=False): - seen["wheel_dir"] = wheel_dir - seen["done"] = True - return True - - monkeypatch.setattr(bootstrap, "install", fake_install) - lines = [] - assert ensure_dependencies(logger=lines.append) is True - assert seen["wheel_dir"] is None # fell back to PyPI - assert any("checksum mismatch" in line for line in lines) - - -def test_explicit_wheel_dir_overrides_bundled(monkeypatch, tmp_path): - folder = tmp_path / "wheels" - folder.mkdir() - (folder / "rdm6300-0.1.1-py3-none-any.whl").write_bytes(b"x") - (folder / "pyserial-3.5-py2.py3-none-any.whl").write_bytes(b"x") - - seen = {} - monkeypatch.setattr( - bootstrap, "missing_packages", - lambda packages=None: [] if seen.get("done") else [("rdm6300", "rdm6300>=0.1.1")], - ) - - def fake_install(requirements, wheel_dir=None, logger=None, timeout=300, no_deps=False): - seen["wheel_dir"] = wheel_dir - seen["done"] = True - return True - - monkeypatch.setattr(bootstrap, "install", fake_install) - ensure_dependencies(wheel_dir=str(folder)) - assert seen["wheel_dir"] == str(folder) - - -def test_bundle_not_covering_a_requirement_falls_back(monkeypatch, tmp_path): - """A bundle missing one package must not cause a guaranteed pip failure.""" - folder = tmp_path / "wheels" - folder.mkdir() - (folder / "rdm6300-0.1.1-py3-none-any.whl").write_bytes(b"x") - - seen = {} - monkeypatch.setattr(bootstrap, "bundled_wheel_dir", lambda: str(folder)) - monkeypatch.setattr( - bootstrap, "missing_packages", - lambda packages=None: [] if seen.get("done") - else [("rdm6300", "rdm6300>=0.1.1"), ("serial", "pyserial>=3.5")], - ) - - def fake_install(requirements, wheel_dir=None, logger=None, timeout=300, no_deps=False): - seen["wheel_dir"] = wheel_dir - seen["wheel_dir_at_call"] = wheel_dir - seen["done"] = True - return True - - monkeypatch.setattr(bootstrap, "install", fake_install) - lines = [] - ensure_dependencies(logger=lines.append) - assert seen["wheel_dir"] is None - assert any("do not cover" in line for line in lines)