CC2 ↔ Spoolman bridge with NFC tag generator
CI / test (push) Successful in 1m4s
CI / image (push) Successful in 1m1s

- Reads /opt/usr/canvas_record_info.json over SSH (pinned host key,
  restricted key), decodes Spoolman spool ids from the tag weight
- Mirrors spools into CC2 Control's spool library and assigns slots,
  keeps its own slot mapping, sets Spoolman locations
- Books the drop of CC2 Control's measured remaining weight to Spoolman
  after each print, idempotent across restarts and interrupted bookings
- Dry-run by default; web UI for settings, SSH key, status and log
- Tag generator for NFC Tools command chains (NTAG213/215/216)
- Tests, mock environment, Dockerfile, Gitea Actions CI

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014GcEk5tkfFSYD6NTF66NkC
This commit is contained in:
Jonas MenzelandClaude Opus 5.5 committed 2026-10-25 03:01:17 +01:00
commit bf32667f77
32 files changed
+4216

No files matched your search

+110
View File
@@ -0,0 +1,110 @@
import base64
import os
import stat
import pytest
from fastapi.testclient import TestClient
from app.bridge import Bridge
from app.sshsource import KeyStore
from app.storage import SettingsStore
from app.web import create_app
from .fakes import FakeCC2, FakeReader, FakeSpoolman
from .test_ssh import _new_key_text
H = {"X-Bridge-Request": "1"}
def make_client(tmp_path, env=None):
settings = SettingsStore(tmp_path, env={})
sm = FakeSpoolman()
sm.add(5, material="PLA+", color="1E40AF")
bridge = Bridge(tmp_path, settings, KeyStore(tmp_path), reader_factory=lambda s: FakeReader(),
cc2_factory=lambda s: FakeCC2(), spoolman_factory=lambda s: sm)
app = create_app(tmp_path, bridge=bridge, start_loop=False, env=env or {})
return TestClient(app), bridge
def test_pages_render(tmp_path):
c, _ = make_client(tmp_path)
for path in ("/", "/tags", "/settings"):
r = c.get(path)
assert r.status_code == 200 and "CC2" in r.text
def test_writes_need_marker_header(tmp_path):
c, _ = make_client(tmp_path)
assert c.post("/api/settings", json={"dry_run": False}).status_code == 403
assert c.post("/api/settings", json={"dry_run": True}, headers=H).status_code == 200
def test_settings_roundtrip_and_validation(tmp_path):
c, _ = make_client(tmp_path)
s = c.get("/api/settings").json()
assert s["dry_run"] is True and s["id_offset"] == 10000
r = c.post("/api/settings", json={"poll_interval": "7", "location_storage": "Regal"}, headers=H)
assert r.json()["poll_interval"] == 7.0
assert c.get("/api/settings").json()["location_storage"] == "Regal"
assert c.post("/api/settings", json={"location_slot": "Slot"}, headers=H).status_code == 400
assert SettingsStore(tmp_path).get().location_storage == "Regal"
def test_ssh_key_upload_never_returned(tmp_path):
c, _ = make_client(tmp_path)
text = _new_key_text()
r = c.post("/api/ssh/key", json={"key": text}, headers=H)
assert r.status_code == 200
body = r.text + c.get("/api/ssh").text + c.get("/api/status").text
assert "PRIVATE KEY" not in body and text.splitlines()[1] not in body
assert r.json()["key"]["fingerprint"].startswith("SHA256:")
path = tmp_path / "ssh" / "id_key"
assert stat.S_IMODE(os.stat(path).st_mode) == 0o600
assert c.post("/api/ssh/key", json={"key": "junk"}, headers=H).status_code == 400
assert c.delete("/api/ssh/key", headers=H).json() == {"key": None}
assert not path.exists()
def test_hostkey_only_from_last_test(tmp_path):
c, _ = make_client(tmp_path)
r = c.post("/api/ssh/hostkey", json={"fingerprint": "SHA256:forged"}, headers=H)
assert r.status_code == 400
def test_tag_endpoint(tmp_path):
c, _ = make_client(tmp_path)
r = c.post("/api/tag", json={"spool_id": 5, "color": "2850DF", "material": "PLA", "subtype": 1,
"temp_min": 190, "temp_max": 230}, headers=H)
assert r.status_code == 200
d = r.json()
assert "A2:14:2850DFFF" in d["commands"] and "A2:17:00AF2715" in d["commands"]
assert d["decoded"]["spool_id"] == 5 and d["qr_svg"].startswith("<svg")
bad = c.post("/api/tag", json={"spool_id": 5, "color": "2850DF", "material": "PLA", "subtype": 0x0100,
"temp_min": 190, "temp_max": 230}, headers=H)
assert bad.status_code == 400
def test_materials_endpoint(tmp_path):
c, _ = make_client(tmp_path)
m = c.get("/api/materials").json()
assert m["materials"]["PETG"]["code"] == "80698471" and len(m["subtypes"]) == 50 and len(m["palette"]) == 24
def test_basic_auth(tmp_path):
c, _ = make_client(tmp_path, env={"BRIDGE_USER": "admin", "BRIDGE_PASSWORD": "geheim"})
assert c.get("/").status_code == 401
assert c.get("/healthz").status_code == 200
good = {"Authorization": "Basic " + base64.b64encode(b"admin:geheim").decode()}
bad = {"Authorization": "Basic " + base64.b64encode(b"admin:nope").decode()}
assert c.get("/", headers=bad).status_code == 401
assert c.get("/", headers=good).status_code == 200
def test_status_and_commands(tmp_path):
c, bridge = make_client(tmp_path)
bridge.poll_once()
st = c.get("/api/status").json()
assert len(st["slots"]) == 4 and st["dry_run"] is True
assert c.post("/api/cc2/enable", json={}, headers=H).status_code == 400
assert c.post("/api/cc2/enable", json={"confirm": True}, headers=H).json()["dry_run"] is True
assert c.post("/api/slots/7/release", headers=H).status_code == 404