cc2-spoolman-bridge/tests/test_web.py

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 ("/", "/log", "/tags", "/settings"):
        r = c.get(path)
        assert r.status_code == 200 and "CC2" in r.text


def test_web_app_files(tmp_path):
    c, _ = make_client(tmp_path)
    m = c.get("/manifest.webmanifest")
    assert m.status_code == 200 and m.json()["display"] == "standalone"
    for icon in m.json()["icons"]:
        assert c.get(icon["src"]).status_code == 200
    assert c.get("/static/icon-180.png").headers["content-type"] == "image/png"


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/printers/p1/ssh/hostkey", json={"fingerprint": "SHA256:forged"}, headers=H)
    assert r.status_code == 400
    assert c.post("/api/printers/nope/ssh/hostkey", json={"fingerprint": "x"}, headers=H).status_code == 404


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["printers"]) == 1 and len(st["printers"][0]["slots"]) == 4 and st["dry_run"] is True
    assert c.post("/api/printers/p1/cc2/enable", json={}, headers=H).status_code == 400
    assert c.post("/api/printers/p1/cc2/enable", json={"confirm": True}, headers=H).json()["dry_run"] is True
    assert c.post("/api/printers/p1/slots/7/release", headers=H).status_code == 404
    assert c.post("/api/printers/zz/slots/1/release", headers=H).status_code == 404


def test_printers_editable(tmp_path):
    c, _ = make_client(tmp_path)
    s = c.get("/api/settings").json()
    printers = s["printers"] + [{"name": "Links", "host": "10.10.2.235", "cc2_url": "http://10.10.2.235:8081"}]
    r = c.post("/api/settings", json={"printers": printers}, headers=H)
    assert r.status_code == 200
    got = r.json()["printers"]
    assert [p["name"] for p in got] == ["CC2", "Links"] and got[1]["id"].startswith("p") and got[1]["ssh_port"] == 22
    dup = c.post("/api/settings", json={"printers": got + [{"name": "links", "host": "x", "cc2_url": "http://x"}]}, headers=H)
    assert dup.status_code == 400
    bad = c.post("/api/settings", json={"location_slot": "Slot {n}"}, headers=H)
    assert bad.status_code == 400  # two printers need {printer}


def test_create_spools_from_existing_filament(tmp_path):
    c, bridge = make_client(tmp_path)
    sm = bridge.spoolman_factory(None)
    cat = c.get("/api/spoolman/catalog").json()
    assert cat["next_id"] == 6 and cat["filaments"][0]["name"] == "PLA+ Black"
    body = {"filament_id": 1, "count": 2, "weight": 1000, "spool_weight": 130}
    assert c.post("/api/spoolman/create", json=body, headers=H).status_code == 409  # dry-run
    assert not [x for x in sm.calls if x[0].startswith("create")]
    c.post("/api/settings", json={"dry_run": False}, headers=H)
    r = c.post("/api/spoolman/create", json=body, headers=H)
    assert r.status_code == 200 and r.json()["spools"] == [{"id": 6, "label": "SUNLU006"}, {"id": 7, "label": "SUNLU007"}]
    assert sm.db[7]["lot_nr"] == "SUNLU007" and sm.db[7]["location"] == "Lager" and sm.db[7]["initial_weight"] == 1000
    assert ("create_spool", {"filament_id": 1, "location": "Lager", "initial_weight": 1000.0, "spool_weight": 130.0}) in sm.calls


def test_create_spool_with_new_filament_and_vendor(tmp_path):
    c, bridge = make_client(tmp_path)
    sm = bridge.spoolman_factory(None)
    c.post("/api/settings", json={"dry_run": False}, headers=H)
    body = {"vendor": "Elegoo", "name": "PLA Matte Black", "material": "PLA Matte", "color": "#1a1a1a", "density": 1.24,
            "diameter": 1.75, "weight": 1000, "spool_weight": 154, "extruder_temp": 210, "bed_temp": 60, "count": 1}
    r = c.post("/api/spoolman/create", json=body, headers=H)
    assert r.status_code == 200 and r.json()["spools"][0]["label"] == "ELEGOO006"
    fil = [x[1] for x in sm.calls if x[0] == "create_filament"][0]
    assert fil["color_hex"] == "1A1A1A" and fil["settings_extruder_temp"] == 210 and fil["vendor_id"] == 2
    assert ("create_vendor", {"name": "Elegoo", "empty_spool_weight": 154.0}) in sm.calls
    bad = dict(body, color="red")
    assert c.post("/api/spoolman/create", json=bad, headers=H).status_code == 400
    assert c.post("/api/spoolman/create", json=dict(body, count=50), headers=H).status_code == 400