ske087/rdm-card-reader
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove the test suite

Browse Source 
The tests are not part of what ships to a device, and nothing in
pyproject.toml, the deploy scripts or the docs referenced them.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
scheianu
2026-10-05 21:20:19 +03:00
parent c0a354b98e
commit a54b79e488
15 changed files with 0 additions and 2684 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/__init__.py
View File
tests/conftest.py
-8
View File
@@ -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)
tests/test_bootstrap.py
-201
View File
@@ -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)
tests/test_bridge.py
-501
View File
@@ -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
tests/test_card.py
-111
View File
@@ -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
tests/test_cli.py
-93
View File
@@ -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
tests/test_deploy.py
-199
View File
@@ -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)
)
tests/test_e2e_bridge.py
-177
View File
@@ -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())
tests/test_e2e_led.py
-209
View File
@@ -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())
tests/test_e2e_pty.py
-165
View File
@@ -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()
tests/test_led.py
-306
View File
@@ -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()
tests/test_ports.py
-228
View File
@@ -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
tests/test_protocol.py
-108
View File
@@ -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)
tests/test_reader.py
-149
View File
@@ -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)
tests/test_wheels.py
-229
View File
@@ -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)