Jonas MenzelandClaude Opus 5.5
d03668e3c5
- A roll that ran out is set to 0 g in Spoolman (like a weigh-in) instead of booking CC2 Control's delta, so a mirror whose count was off cannot book too much or too little. - A tray that printed while CC2 Control still asked which spool it holds (untagged roll) shows the waiting millimetres on the status page and gets one log warning after the print. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014GcEk5tkfFSYD6NTF66NkC
828 lines
31 KiB
Python
828 lines
31 KiB
Python
import json
|
||
|
||
import pytest
|
||
|
||
from app.bridge import Bridge, cc2_spool_fields, material_fits
|
||
from app.sshsource import KeyStore
|
||
from app.storage import SettingsStore
|
||
|
||
from .fakes import FakeCC2, FakeReader, FakeSpoolman, channel, record
|
||
|
||
|
||
class Clock:
|
||
def __init__(self):
|
||
self.t = 1_800_000_000.0
|
||
|
||
def __call__(self):
|
||
return self.t
|
||
|
||
def tick(self, s):
|
||
self.t += s
|
||
|
||
|
||
@pytest.fixture
|
||
def env(tmp_path):
|
||
return make_env(tmp_path)
|
||
|
||
|
||
PRINTERS = [
|
||
{"id": "p1", "name": "Rechts", "host": "10.0.0.1", "cc2_url": "http://10.0.0.1:8081"},
|
||
{"id": "p2", "name": "Links", "host": "10.0.0.2", "cc2_url": "http://10.0.0.2:8081"},
|
||
]
|
||
|
||
|
||
def make_env(tmp_path, dry_run=False, reader=None, cc2=None, spoolman=None, clock=None, printers=1,
|
||
readers=None, cc2s=None):
|
||
settings = SettingsStore(tmp_path, env={})
|
||
settings.save({"printers": PRINTERS[:printers], "dry_run": dry_run, "settle_seconds": 30, "empty_grace": 20})
|
||
readers = readers or {"p1": reader or FakeReader(), "p2": FakeReader()}
|
||
cc2s = cc2s or {"p1": cc2 or FakeCC2(), "p2": FakeCC2()}
|
||
sm = spoolman or FakeSpoolman()
|
||
clock = clock or Clock()
|
||
bridge = Bridge(tmp_path, settings, KeyStore(tmp_path), reader_factory=lambda p: readers[p.id],
|
||
cc2_factory=lambda p: cc2s[p.id], spoolman_factory=lambda s: sm, clock=clock)
|
||
return {"bridge": bridge, "reader": readers["p1"], "cc2": cc2s["p1"], "reader2": readers["p2"],
|
||
"cc2_2": cc2s["p2"], "readers": readers, "cc2s": cc2s, "sm": sm, "clock": clock,
|
||
"settings": settings, "path": tmp_path}
|
||
|
||
|
||
def slots(e, pid="p1"):
|
||
return e["bridge"].state.printer(pid)["slots"]
|
||
|
||
|
||
def booked(e, pid="p1"):
|
||
return e["bridge"].state.printer(pid)["booked"]
|
||
|
||
|
||
def insert(e, slot, spool_id, pid="p1", **kw):
|
||
"""Put a tagged spool into a slot: tray loaded and the record carries the id."""
|
||
e["cc2s"][pid].slots[slot]["printer"] = {"status": 1}
|
||
e["cc2s"][pid].slots[slot]["question"] = "inserted"
|
||
e["readers"][pid].record[f"channels{slot}"] = channel(10000 + spool_id, **kw)
|
||
|
||
|
||
def poll(e, n=1, step=5):
|
||
for _ in range(n):
|
||
e["bridge"].poll_once()
|
||
e["clock"].tick(step)
|
||
|
||
|
||
def test_scan_creates_assigns_and_sets_location(env):
|
||
env["sm"].add(5, vendor="Sunlu", material="PLA+", color="1E40AF", remaining=812.5)
|
||
insert(env, 1, 5)
|
||
poll(env)
|
||
cc2 = env["cc2"]
|
||
saves = [c for c in cc2.calls if c[0] == "save"]
|
||
assert len(saves) == 1
|
||
f = saves[0][1]
|
||
assert f["name"] == "SUNLU005" and f["brand"] == "Sunlu" and f["material"] == "PLA+"
|
||
assert f["color"] == "#1E40AF" and f["note"] == "spoolman:5" and f["remaining"] == 812.5
|
||
assert f["net"] == 1000 and f["tare"] == 150 and f["diameter"] == 1.75 and f["density"] == 1.24
|
||
assert ("assign", 1, "s0001") in cc2.calls
|
||
assert cc2.slots[1]["spool"] == "s0001" and cc2.slots[1]["question"] == ""
|
||
assert ("location", 5, "Rechts Slot 2") in env["sm"].calls
|
||
poll(env, 3)
|
||
# nothing repeated
|
||
assert len([c for c in cc2.calls if c[0] in ("save", "assign")]) == 2
|
||
assert len([c for c in env["sm"].calls if c[0] == "location"]) == 1
|
||
|
||
|
||
def test_existing_mirror_is_reused(env):
|
||
env["sm"].add(5)
|
||
env["cc2"].save_spool({"name": "SUNLU005", "material": "PLA+", "color": "#000000", "note": "spoolman:5"})
|
||
env["cc2"].calls.clear()
|
||
insert(env, 0, 5)
|
||
poll(env)
|
||
assert [c for c in env["cc2"].calls if c[0] == "save"] == []
|
||
assert ("assign", 0, "s0001") in env["cc2"].calls
|
||
|
||
|
||
def test_unknown_spool_id_is_ignored(env):
|
||
insert(env, 0, 42)
|
||
poll(env)
|
||
assert env["cc2"].calls == []
|
||
assert slots(env) == {}
|
||
assert any("42 existiert nicht" in e["msg"] for e in env["bridge"].log.recent())
|
||
|
||
|
||
def test_real_weight_tags_are_ignored(env):
|
||
env["reader"].record["channels0"] = channel(1000)
|
||
env["cc2"].slots[0]["printer"] = {"status": 1}
|
||
poll(env, 2)
|
||
assert env["cc2"].calls == [] and slots(env) == {}
|
||
|
||
|
||
def test_mapping_survives_weight_falling_back_to_1000(env):
|
||
env["sm"].add(5)
|
||
insert(env, 2, 5)
|
||
poll(env)
|
||
# touchscreen edit: weight falls back, CC2 asks again ("changed") with no spool in the slot
|
||
env["reader"].record["channels2"] = channel(1000)
|
||
env["cc2"].slots[2]["spool"] = ""
|
||
env["cc2"].slots[2]["question"] = "changed"
|
||
poll(env, 10)
|
||
m = slots(env)["2"]
|
||
assert m["spool"] == 5
|
||
assert env["sm"].db[5]["location"] == "Rechts Slot 3"
|
||
|
||
|
||
def test_mapping_survives_restart(env):
|
||
env["sm"].add(5)
|
||
insert(env, 0, 5)
|
||
poll(env)
|
||
env["reader"].record["channels0"] = channel(1000)
|
||
poll(env)
|
||
e2 = make_env(env["path"], reader=env["reader"], cc2=env["cc2"], spoolman=env["sm"], clock=env["clock"])
|
||
poll(e2, 3)
|
||
assert slots(e2)["0"]["spool"] == 5
|
||
assert len([c for c in env["cc2"].calls if c[0] == "save"]) == 1
|
||
|
||
|
||
def test_empty_slot_releases_to_storage_after_grace(env):
|
||
env["sm"].add(5)
|
||
insert(env, 0, 5)
|
||
poll(env)
|
||
env["cc2"].slots[0]["printer"] = {"status": 0}
|
||
poll(env, 3, step=5) # 0, 5, 10 s
|
||
assert "0" in slots(env)
|
||
poll(env, 3, step=5) # >= 20 s
|
||
assert "0" not in slots(env)
|
||
assert env["sm"].db[5]["location"] == "Lager"
|
||
|
||
|
||
def test_cc2_returning_spool_to_storage_releases(env):
|
||
env["sm"].add(5)
|
||
insert(env, 0, 5)
|
||
poll(env, 2)
|
||
env["cc2"].slots[0]["spool"] = "" # CC2 gave it back to storage (no question)
|
||
poll(env)
|
||
assert "0" not in slots(env)
|
||
assert env["sm"].db[5]["location"] == "Lager"
|
||
|
||
|
||
def test_new_scan_with_other_id_replaces_mapping(env):
|
||
env["sm"].add(5)
|
||
env["sm"].add(8, color="F88D36")
|
||
insert(env, 0, 5)
|
||
poll(env, 2)
|
||
insert(env, 0, 8, color="0xF88D36")
|
||
poll(env, 2)
|
||
assert slots(env)["0"]["spool"] == 8
|
||
assert env["sm"].db[5]["location"] == "Lager"
|
||
assert env["sm"].db[8]["location"] == "Rechts Slot 1"
|
||
assert env["cc2"].slots[0]["spool"] == "s0002"
|
||
|
||
|
||
def test_spool_moved_to_other_slot(env):
|
||
env["sm"].add(5)
|
||
insert(env, 0, 5)
|
||
poll(env, 2)
|
||
env["reader"].record["channels0"] = channel("")
|
||
insert(env, 3, 5)
|
||
poll(env, 2)
|
||
sl = slots(env)
|
||
assert "0" not in sl and sl["3"]["spool"] == 5
|
||
assert env["sm"].db[5]["location"] == "Rechts Slot 4"
|
||
|
||
|
||
def test_consumption_booked_after_print_once(env):
|
||
env["sm"].add(5, remaining=800)
|
||
insert(env, 0, 5)
|
||
poll(env)
|
||
cc2 = env["cc2"]
|
||
cc2.job = {"active": True}
|
||
for _ in range(10):
|
||
cc2.consume("s0001", 1.25)
|
||
poll(env)
|
||
assert not [c for c in env["sm"].calls if c[0] == "use"] # not during the print
|
||
cc2.job = {"active": False}
|
||
poll(env, 3, step=5)
|
||
assert not [c for c in env["sm"].calls if c[0] == "use"] # waits for settle_seconds
|
||
poll(env, 6, step=5)
|
||
uses = [c for c in env["sm"].calls if c[0] == "use"]
|
||
assert uses == [("use", 5, 12.5)]
|
||
poll(env, 5)
|
||
assert [c for c in env["sm"].calls if c[0] == "use"] == uses
|
||
|
||
|
||
def test_no_double_booking_after_restart(env):
|
||
env["sm"].add(5, remaining=800)
|
||
insert(env, 0, 5)
|
||
poll(env)
|
||
env["cc2"].consume("s0001", 3.25)
|
||
poll(env, 10)
|
||
assert [c for c in env["sm"].calls if c[0] == "use"] == [("use", 5, 3.25)]
|
||
e2 = make_env(env["path"], reader=env["reader"], cc2=env["cc2"], spoolman=env["sm"], clock=env["clock"])
|
||
poll(e2, 10)
|
||
assert [c for c in env["sm"].calls if c[0] == "use"] == [("use", 5, 3.25)]
|
||
env["cc2"].consume("s0001", 1.0)
|
||
poll(e2, 10)
|
||
assert [c for c in env["sm"].calls if c[0] == "use"] == [("use", 5, 3.25), ("use", 5, 1.0)]
|
||
|
||
|
||
def test_interrupted_booking_is_not_repeated(env):
|
||
env["sm"].add(5, remaining=800)
|
||
insert(env, 0, 5)
|
||
poll(env)
|
||
env["cc2"].consume("s0001", 4.0)
|
||
env["sm"].fail_use_after_apply = True # PUT reached Spoolman, answer lost
|
||
poll(env, 10)
|
||
env["sm"].fail_use_after_apply = False
|
||
e2 = make_env(env["path"], reader=env["reader"], cc2=env["cc2"], spoolman=env["sm"], clock=env["clock"])
|
||
poll(e2, 10)
|
||
assert [c for c in env["sm"].calls if c[0] == "use"] == [("use", 5, 4.0)]
|
||
assert env["sm"].db[5]["used_weight"] == 4.0
|
||
b = booked(e2)["s0001"]
|
||
assert b["pending"] is None and b["remaining"] == 796.0
|
||
|
||
|
||
def test_interrupted_booking_retried_when_not_applied(env):
|
||
env["sm"].add(5, remaining=800)
|
||
insert(env, 0, 5)
|
||
poll(env)
|
||
env["cc2"].consume("s0001", 4.0)
|
||
poll(env, 7) # settles, but Spoolman is down when booking
|
||
env["sm"].fail = True
|
||
poll(env, 3)
|
||
env["sm"].fail = False
|
||
poll(env, 3)
|
||
assert [c for c in env["sm"].calls if c[0] == "use"] == [("use", 5, 4.0)]
|
||
|
||
|
||
def test_negative_change_logged_not_booked(env):
|
||
env["sm"].add(5, remaining=800)
|
||
insert(env, 0, 5)
|
||
poll(env)
|
||
env["cc2"].consume("s0001", -20) # weigh-in raised the remaining weight
|
||
poll(env, 10)
|
||
assert not [c for c in env["sm"].calls if c[0] == "use"]
|
||
assert any("gestiegen" in e["msg"] for e in env["bridge"].log.recent())
|
||
env["cc2"].consume("s0001", 5)
|
||
poll(env, 10)
|
||
assert [c for c in env["sm"].calls if c[0] == "use"] == [("use", 5, 5.0)]
|
||
|
||
|
||
def test_dry_run_writes_nothing(tmp_path):
|
||
e = make_env(tmp_path, dry_run=True)
|
||
e["sm"].add(5)
|
||
e["cc2"].save_spool({"name": "X", "material": "PLA", "color": "#000000", "note": "spoolman:7"})
|
||
e["sm"].add(7)
|
||
e["cc2"].calls.clear()
|
||
insert(e, 0, 5)
|
||
poll(e, 3)
|
||
e["cc2"].consume("s0001", 10)
|
||
poll(e, 20)
|
||
assert e["cc2"].calls == []
|
||
assert e["sm"].calls == []
|
||
msgs = [x["msg"] for x in e["bridge"].log.recent(500) if x["level"] == "dry"]
|
||
assert any("Spule SUNLU005 anlegen" in m for m in msgs)
|
||
assert any("Ort „Rechts Slot 1“, Drucker Rechts, Slot 1" in m for m in msgs)
|
||
assert any("Zusatzfeld" in m for m in msgs)
|
||
assert any("10.00 g" in m for m in msgs)
|
||
# logged once, not every cycle
|
||
assert len([m for m in msgs if "10.00 g" in m]) == 1
|
||
# switching dry-run off applies what is pending
|
||
e["settings"].save({"dry_run": False})
|
||
poll(e, 10)
|
||
assert ("use", 7, 10.0) in e["sm"].calls
|
||
assert e["sm"].db[5]["location"] == "Rechts Slot 1"
|
||
assert e["cc2"].slots[0]["spool"] == "s0002"
|
||
|
||
|
||
def test_outages_do_not_lose_scans(env):
|
||
env["sm"].add(5)
|
||
env["sm"].fail = True
|
||
insert(env, 0, 5)
|
||
poll(env, 2)
|
||
assert slots(env) == {}
|
||
assert env["bridge"].state.printer("p1")["scans"]
|
||
env["sm"].fail = False
|
||
poll(env)
|
||
assert slots(env)["0"]["spool"] == 5
|
||
|
||
|
||
def test_cc2_outage_retries_assignment(env):
|
||
env["sm"].add(5)
|
||
insert(env, 0, 5)
|
||
env["cc2"].fail = True
|
||
poll(env, 2)
|
||
env["cc2"].fail = False
|
||
poll(env, 2)
|
||
assert env["cc2"].slots[0]["spool"] == "s0001"
|
||
|
||
|
||
def test_manual_assignment_in_cc2_is_adopted(env):
|
||
env["sm"].add(9)
|
||
env["cc2"].save_spool({"name": "SUNLU009", "material": "PLA", "color": "#000000", "note": "spoolman:9"})
|
||
env["cc2"].assign(3, "s0001")
|
||
env["cc2"].slots[3]["printer"] = {"status": 1}
|
||
poll(env)
|
||
assert slots(env)["3"]["spool"] == 9
|
||
assert env["sm"].db[9]["location"] == "Rechts Slot 4"
|
||
|
||
|
||
def test_enable_command_respects_dry_run(tmp_path):
|
||
e = make_env(tmp_path, dry_run=True, cc2=FakeCC2(enabled=False))
|
||
e["bridge"].commands.put(("enable", "p1", None))
|
||
poll(e)
|
||
assert e["cc2"].calls == [] and e["bridge"].snapshot["printers"][0]["cc2_enabled"] is False
|
||
e["settings"].save({"dry_run": False})
|
||
e["bridge"].commands.put(("enable", "p1", None))
|
||
poll(e)
|
||
assert e["cc2"].calls == [("enable", True)]
|
||
|
||
|
||
def test_state_file_is_valid_json(env):
|
||
env["sm"].add(5)
|
||
insert(env, 0, 5)
|
||
poll(env)
|
||
data = json.loads((env["path"] / "state.json").read_text())
|
||
assert data["printers"]["p1"]["slots"]["0"]["spool"] == 5
|
||
assert not list(env["path"].glob(".state.json.*"))
|
||
|
||
|
||
def test_material_fits():
|
||
assert material_fits("PLA+", "PLA", "PLA+")
|
||
assert material_fits("PLA Matte", "PLA", "PLA")
|
||
assert material_fits("PLA-CF", "PLA", "")
|
||
assert not material_fits("Silk PLA", "PLA", "PLA Silk")
|
||
assert material_fits("PLA Silk", "PLA", "PLA Silk")
|
||
assert not material_fits("PETG", "PLA", "PLA")
|
||
|
||
|
||
def test_fields_use_tag_material_when_spoolman_material_does_not_fit():
|
||
sm = FakeSpoolman().add(3, material="Silk PLA", color="FFD700")
|
||
fields, notes = cc2_spool_fields(sm, channel(10003, type_="PLA", detailed="PLA Silk"))
|
||
assert fields["material"] == "PLA Silk" and notes
|
||
assert fields["color"] == "#FFD700"
|
||
|
||
|
||
# ---------- v2: several printers, Spoolman location and extra fields ----------
|
||
|
||
def test_extra_fields_created_and_kept_in_sync(env):
|
||
env["sm"].add(5)
|
||
insert(env, 1, 5)
|
||
poll(env)
|
||
assert set(env["sm"].field_defs) == {"drucker", "slot"}
|
||
assert env["sm"].field_defs["slot"]["field_type"] == "integer"
|
||
assert env["sm"].db[5]["location"] == "Rechts Slot 2"
|
||
assert env["sm"].extra(5) == {"drucker": "Rechts", "slot": 2}
|
||
env["cc2"].slots[1]["spool"] = "" # "Take out" in CC2 Control
|
||
poll(env)
|
||
assert env["sm"].db[5]["location"] == "Lager" and env["sm"].extra(5) == {}
|
||
|
||
|
||
def test_take_out_in_cc2_control_goes_to_storage(env):
|
||
env["sm"].add(5)
|
||
insert(env, 0, 5)
|
||
poll(env, 2)
|
||
env["cc2"].assign(0, None) # what the "Take out" button sends
|
||
poll(env)
|
||
assert slots(env) == {}
|
||
assert env["sm"].db[5]["location"] == "Lager"
|
||
poll(env, 20) # the unchanged tag in the record does not bring it back
|
||
assert slots(env) == {}
|
||
|
||
|
||
def test_renaming_printer_moves_location(env):
|
||
env["sm"].add(5)
|
||
insert(env, 0, 5)
|
||
poll(env)
|
||
printers = env["settings"].get().to_dict()["printers"]
|
||
printers[0]["name"] = "Centauri Carbon 2 Rechts"
|
||
env["settings"].save({"printers": printers})
|
||
poll(env)
|
||
assert env["sm"].db[5]["location"] == "Centauri Carbon 2 Rechts Slot 1"
|
||
assert env["sm"].extra(5)["drucker"] == "Centauri Carbon 2 Rechts"
|
||
assert slots(env)["0"]["spool"] == 5
|
||
|
||
|
||
def test_manual_spoolman_edits_are_corrected(env):
|
||
env["sm"].add(5)
|
||
env["sm"].add(6)
|
||
insert(env, 0, 5)
|
||
poll(env)
|
||
env["sm"].db[5]["location"] = "Regal 3" # someone moves a loaded spool by hand
|
||
env["sm"].db[5]["extra"]["slot"] = "4"
|
||
env["sm"].db[6]["location"] = "Regal 3" # a spool that is not loaded: left alone
|
||
poll(env, 13) # >= 60 s: full comparison
|
||
assert env["sm"].db[5]["location"] == "Rechts Slot 1" and env["sm"].extra(5)["slot"] == 1
|
||
assert env["sm"].db[6]["location"] == "Regal 3"
|
||
assert any("(korrigiert)" in x["msg"] for x in env["bridge"].log.recent())
|
||
|
||
|
||
def test_stale_printer_locations_go_to_storage(env):
|
||
env["sm"].add(1, location="CC2 Slot 2") # from v1 or before a state loss
|
||
env["sm"].add(2, location="Rechts Slot 3")
|
||
env["sm"].add(3, location="Lager", extra={"drucker": '"Links"', "slot": "1"})
|
||
env["sm"].add(4, location="Regal 1")
|
||
env["bridge"].state.data["written_locations"] = ["CC2 Slot 2"]
|
||
env["bridge"].state.data["written_printers"] = ["Links"] # a printer this bridge used to run
|
||
poll(env)
|
||
assert env["sm"].db[1]["location"] == "Lager"
|
||
assert env["sm"].db[2]["location"] == "Lager"
|
||
assert env["sm"].db[3]["location"] == "Lager" and env["sm"].extra(3) == {}
|
||
assert env["sm"].db[4]["location"] == "Regal 1"
|
||
assert not [c for c in env["sm"].calls if c[1] == 4]
|
||
|
||
|
||
def test_spool_moves_between_printers(tmp_path):
|
||
e = make_env(tmp_path, printers=2)
|
||
e["sm"].add(5)
|
||
insert(e, 0, 5)
|
||
poll(e, 2)
|
||
assert e["sm"].db[5]["location"] == "Rechts Slot 1"
|
||
e["readers"]["p1"].record["channels0"] = channel("") # taken out of Rechts ...
|
||
insert(e, 2, 5, pid="p2") # ... and put into Links
|
||
poll(e, 2)
|
||
assert slots(e, "p1") == {} and slots(e, "p2")["2"]["spool"] == 5
|
||
assert e["sm"].db[5]["location"] == "Links Slot 3" and e["sm"].extra(5) == {"drucker": "Links", "slot": 3}
|
||
assert e["cc2s"]["p2"].slots[2]["spool"] == "s0001"
|
||
# Rechts' CC2 Control still binds its own mirror until the tray empties: no flip-flop
|
||
e["cc2s"]["p1"].slots[0]["printer"] = {"status": 0}
|
||
poll(e, 6)
|
||
assert slots(e, "p2")["2"]["spool"] == 5 and slots(e, "p1") == {}
|
||
assert e["sm"].db[5]["location"] == "Links Slot 3"
|
||
|
||
|
||
def test_consumption_is_booked_per_printer(tmp_path):
|
||
e = make_env(tmp_path, printers=2)
|
||
e["sm"].add(5, remaining=800)
|
||
e["sm"].add(8, remaining=500)
|
||
insert(e, 0, 5)
|
||
insert(e, 1, 8, pid="p2")
|
||
poll(e)
|
||
e["cc2s"]["p1"].consume("s0001", 3.0)
|
||
e["cc2s"]["p2"].consume("s0001", 7.5) # same CC2 id in the other printer's library
|
||
poll(e, 10)
|
||
uses = sorted(c for c in e["sm"].calls if c[0] == "use")
|
||
assert uses == [("use", 5, 3.0), ("use", 8, 7.5)]
|
||
|
||
|
||
def test_one_printer_down_does_not_block_the_other(tmp_path):
|
||
e = make_env(tmp_path, printers=2)
|
||
e["sm"].add(5)
|
||
e["readers"]["p2"].fail = True
|
||
e["cc2s"]["p2"].fail = True
|
||
insert(e, 0, 5)
|
||
poll(e)
|
||
assert e["sm"].db[5]["location"] == "Rechts Slot 1"
|
||
snap = {p["id"]: p for p in e["bridge"].snapshot["printers"]}
|
||
assert snap["p2"]["errors"]["ssh"] and snap["p1"]["errors"]["ssh"] is None
|
||
|
||
|
||
def test_removed_printer_releases_its_spools(tmp_path):
|
||
e = make_env(tmp_path, printers=2)
|
||
e["sm"].add(5)
|
||
insert(e, 0, 5, pid="p2")
|
||
poll(e)
|
||
assert e["sm"].db[5]["location"] == "Links Slot 1"
|
||
e["settings"].save({"printers": PRINTERS[:1]})
|
||
poll(e)
|
||
assert "p2" not in e["bridge"].state.data["printers"]
|
||
assert e["sm"].db[5]["location"] == "Lager" and e["sm"].extra(5) == {}
|
||
|
||
|
||
def test_without_extra_fields_only_location(tmp_path):
|
||
e = make_env(tmp_path)
|
||
e["settings"].save({"spoolman_fields": False})
|
||
e["sm"].add(5)
|
||
insert(e, 0, 5)
|
||
poll(e)
|
||
assert e["sm"].db[5]["location"] == "Rechts Slot 1" and e["sm"].extra(5) == {}
|
||
assert e["sm"].field_defs == {}
|
||
|
||
|
||
def test_migration_from_v1(tmp_path):
|
||
(tmp_path / "settings.json").write_text(json.dumps({
|
||
"printer_host": "10.10.2.202", "ssh_port": 22, "ssh_user": "root", "cc2_url": "http://10.10.2.202:8081",
|
||
"spoolman_url": "http://10.10.6.10:7912", "id_offset": 10000, "poll_interval": 5.0,
|
||
"location_slot": "CC2 Slot {n}", "location_storage": "Lager", "empty_grace": 20.0,
|
||
"settle_seconds": 60.0, "dry_run": False}))
|
||
(tmp_path / "state.json").write_text(json.dumps({
|
||
"version": 1,
|
||
"slots": {"1": {"spool": 1, "name": "SUNLU001", "since": 1, "cc2": "abc", "assigned": True, "cc2_seen": True}},
|
||
"booked": {"abc": {"spoolman": 1, "remaining": 250.0, "pending": None, "last": None}},
|
||
"locations": {"1": {"want": "CC2 Slot 2", "done": "CC2 Slot 2"}},
|
||
"last_record": record("", 10001), "scans": []}))
|
||
(tmp_path / "ssh").mkdir()
|
||
(tmp_path / "ssh" / "host_key.json").write_text('{"fingerprint": "SHA256:x"}')
|
||
settings = SettingsStore(tmp_path, env={})
|
||
s = settings.get()
|
||
assert [(p.id, p.name, p.host) for p in s.printers] == [("p1", "CC2", "10.10.2.202")]
|
||
assert s.location_slot == "{printer} Slot {n}" and s.dry_run is False
|
||
assert "printer_host" not in json.loads((tmp_path / "settings.json").read_text())
|
||
|
||
reader = FakeReader(record("", 10001))
|
||
cc2 = FakeCC2()
|
||
cc2.library.append({"id": "abc", "name": "SUNLU001", "material": "PLA+", "color": "#000000",
|
||
"remaining": 250.0, "note": "spoolman:1", "archived": False})
|
||
cc2.slots[1]["spool"] = "abc"
|
||
cc2.slots[1]["printer"] = {"status": 1}
|
||
sm = FakeSpoolman()
|
||
sm.add(1, remaining=250.0, location="CC2 Slot 2")
|
||
e = make_env(tmp_path, dry_run=False, reader=reader, cc2=cc2, spoolman=sm)
|
||
poll(e)
|
||
assert slots(e)["1"]["spool"] == 1 and booked(e)["abc"]["remaining"] == 250.0
|
||
assert (tmp_path / "ssh" / "host_key_p1.json").exists()
|
||
assert sm.db[1]["location"] == "Rechts Slot 2" and sm.extra(1) == {"drucker": "Rechts", "slot": 2}
|
||
assert not [c for c in cc2.calls if c[0] in ("save", "assign")] # nothing re-created
|
||
|
||
|
||
# ---------- v3: putting rolls back, CC2 Control's log ----------
|
||
|
||
def take_out_of_tray(e, slot, pid="p1"):
|
||
"""Roll physically out: tray empty; CC2 Control gives the spool back to storage (record keeps the tag)."""
|
||
cc2 = e["cc2s"][pid]
|
||
cc2.slots[slot]["printer"] = {"status": 0}
|
||
cc2.slots[slot]["spool"] = ""
|
||
cc2.slots[slot]["question"] = ""
|
||
|
||
|
||
def put_back(e, slot, pid="p1"):
|
||
"""Roll in again without a new scan: the record is unchanged, CC2 Control asks."""
|
||
cc2 = e["cc2s"][pid]
|
||
cc2.slots[slot]["printer"] = {"status": 1}
|
||
cc2.slots[slot]["spool"] = ""
|
||
cc2.slots[slot]["question"] = "inserted"
|
||
|
||
|
||
def uses(e):
|
||
return [c for c in e["sm"].calls if c[0] == "use"]
|
||
|
||
|
||
def msgs(e):
|
||
return [x["msg"] for x in e["bridge"].log.recent()]
|
||
|
||
|
||
def test_roll_put_back_after_long_absence_is_mapped_again(env):
|
||
env["sm"].add(5)
|
||
insert(env, 1, 5)
|
||
poll(env)
|
||
take_out_of_tray(env, 1)
|
||
poll(env, 6) # > 20 s
|
||
assert slots(env) == {} and env["sm"].db[5]["location"] == "Lager"
|
||
put_back(env, 1)
|
||
poll(env)
|
||
assert slots(env)["1"]["spool"] == 5 and slots(env)["1"]["reinserted"]
|
||
assert env["cc2"].slots[1]["spool"] == "s0001" and env["cc2"].slots[1]["question"] == ""
|
||
assert env["sm"].db[5]["location"] == "Rechts Slot 2"
|
||
assert len([c for c in env["cc2"].calls if c[0] == "save"]) == 1 # same mirror
|
||
|
||
|
||
def test_put_back_then_touchscreen_edit_means_another_roll(env):
|
||
env["sm"].add(5)
|
||
insert(env, 1, 5)
|
||
poll(env)
|
||
take_out_of_tray(env, 1)
|
||
poll(env, 6)
|
||
put_back(env, 1) # really an untagged roll ...
|
||
poll(env)
|
||
assert slots(env)["1"]["spool"] == 5
|
||
env["reader"].record["channels1"] = channel(1000) # ... chosen on the touchscreen
|
||
env["cc2"].slots[1]["spool"] = ""
|
||
env["cc2"].slots[1]["question"] = "changed"
|
||
poll(env, 2)
|
||
assert slots(env) == {} and env["sm"].db[5]["location"] == "Lager"
|
||
poll(env, 10)
|
||
assert slots(env) == {} # and it does not come back
|
||
|
||
|
||
def test_untagged_roll_edited_before_the_bridge_looked(env):
|
||
env["sm"].add(5)
|
||
insert(env, 1, 5)
|
||
poll(env)
|
||
put_back(env, 1) # swapped within seconds ...
|
||
env["reader"].record["channels1"] = channel(1000) # ... and chosen on the touchscreen
|
||
poll(env, 2)
|
||
assert slots(env) == {} and env["sm"].db[5]["location"] == "Lager"
|
||
|
||
|
||
def test_short_removal_answers_cc2_again_and_counts(env):
|
||
env["sm"].add(5, remaining=800)
|
||
insert(env, 1, 5)
|
||
poll(env)
|
||
env["cc2"].slots[1]["printer"] = {"status": 0}
|
||
poll(env, 2) # 10 s out, mapping kept
|
||
put_back(env, 1)
|
||
poll(env)
|
||
assert env["cc2"].slots[1]["spool"] == "s0001" and env["cc2"].slots[1]["question"] == ""
|
||
assert slots(env)["1"]["reinserted"]
|
||
assert any("erneut zugewiesen" in m for m in msgs(env))
|
||
env["cc2"].consume("s0001", 2.0)
|
||
poll(env, 10)
|
||
assert uses(env) == [("use", 5, 2.0)]
|
||
|
||
|
||
def test_touchscreen_edit_of_scanned_roll_keeps_counting(env):
|
||
env["sm"].add(5)
|
||
insert(env, 2, 5)
|
||
poll(env)
|
||
env["reader"].record["channels2"] = channel(1000)
|
||
env["cc2"].slots[2]["spool"] = ""
|
||
env["cc2"].slots[2]["question"] = "changed"
|
||
poll(env)
|
||
assert slots(env)["2"]["spool"] == 5
|
||
assert env["cc2"].slots[2]["spool"] == "s0001" # question answered with the mapped roll
|
||
assign_calls = len([c for c in env["cc2"].calls if c[0] == "assign"])
|
||
env["cc2"].slots[2]["spool"] = ""
|
||
env["cc2"].slots[2]["question"] = "changed"
|
||
poll(env, 3, step=5) # not again within a minute
|
||
assert len([c for c in env["cc2"].calls if c[0] == "assign"]) == assign_calls
|
||
|
||
|
||
def test_take_out_with_roll_still_in_tray_books_from_cc2_log(env):
|
||
env["sm"].add(5, remaining=800)
|
||
insert(env, 0, 5)
|
||
poll(env, 2)
|
||
env["cc2"].assign(0, None) # "Take out", roll stays
|
||
poll(env)
|
||
assert slots(env) == {} and env["sm"].db[5]["location"] == "Lager"
|
||
t = env["clock"]()
|
||
env["cc2"].add_log(t + 30, "print", "", 0, -2.98, 1000.0, "/local/benchy.gcode", "complete")
|
||
env["cc2"].add_log(t + 30, "print", "", 1, -1.0, 300.0, "/local/benchy.gcode", "complete") # other tray
|
||
poll(env, 3)
|
||
expected = round(1000 * 3.141592653589793 * 0.875 ** 2 * 1.24 / 1000, 3)
|
||
assert uses(env) == [("use", 5, expected)]
|
||
assert any("nachgebucht" in m and "benchy.gcode (fertig)" in m for m in msgs(env))
|
||
e2 = make_env(env["path"], reader=env["reader"], cc2=env["cc2"], spoolman=env["sm"], clock=env["clock"])
|
||
poll(e2, 3)
|
||
assert uses(env) == [("use", 5, expected)]
|
||
|
||
|
||
def test_take_out_window_closes_when_roll_leaves(env):
|
||
env["sm"].add(5)
|
||
insert(env, 0, 5)
|
||
poll(env, 2)
|
||
env["cc2"].assign(0, None)
|
||
poll(env)
|
||
env["cc2"].slots[0]["printer"] = {"status": 0}
|
||
poll(env)
|
||
env["clock"].tick(600)
|
||
env["cc2"].add_log(env["clock"](), "print", "", 0, -3.0, 1000.0, "x.gcode", "complete")
|
||
poll(env, 2)
|
||
assert uses(env) == []
|
||
|
||
|
||
def test_old_log_entries_are_not_booked_on_first_start(env):
|
||
env["sm"].add(5)
|
||
env["cc2"].add_log(env["clock"]() - 50, "print", "", 0, -3.0, 1000.0, "old.gcode", "complete")
|
||
insert(env, 0, 5)
|
||
poll(env, 2)
|
||
env["cc2"].assign(0, None)
|
||
poll(env, 3)
|
||
assert uses(env) == []
|
||
|
||
|
||
def test_booking_names_print_and_result(env):
|
||
env["sm"].add(5, remaining=800)
|
||
insert(env, 0, 5)
|
||
poll(env)
|
||
env["cc2"].job = {"active": True}
|
||
env["cc2"].consume("s0001", 3.21)
|
||
poll(env)
|
||
env["cc2"].job = {"active": False}
|
||
env["cc2"].add_log(env["clock"](), "print", "s0001", 0, -3.21, 1076.0, "/local/benchy.gcode", "cancelled")
|
||
poll(env, 10)
|
||
assert uses(env) == [("use", 5, 3.21)]
|
||
assert any(m.endswith("gebucht – benchy.gcode (abgebrochen)") for m in msgs(env))
|
||
assert booked(env)["s0001"]["last"]["details"] == "benchy.gcode (abgebrochen)"
|
||
env["cc2"].consume("s0001", 1.0) # next booking does not repeat the old print
|
||
poll(env, 10)
|
||
assert msgs(env)[0].endswith("auf Spoolman-Spule 5 gebucht")
|
||
|
||
|
||
def test_weigh_in_in_cc2_sets_spoolman_remaining(env):
|
||
env["sm"].add(5, remaining=800)
|
||
env["sm"].db[5]["used_weight"] = 200.0
|
||
insert(env, 0, 5)
|
||
poll(env)
|
||
env["cc2"].weigh("s0001", 812.0, env["clock"]()) # more than counted
|
||
poll(env, 10)
|
||
assert uses(env) == []
|
||
assert ("used", 5, 188.0) in env["sm"].calls and env["sm"].db[5]["remaining_weight"] == 812.0
|
||
env["cc2"].consume("s0001", 2.5)
|
||
poll(env, 10)
|
||
assert uses(env) == [("use", 5, 2.5)]
|
||
assert len([c for c in env["sm"].calls if c[0] == "used"]) == 1
|
||
|
||
|
||
def test_weigh_in_lower_is_not_booked_as_use(env):
|
||
env["sm"].add(5, remaining=800)
|
||
env["sm"].db[5]["used_weight"] = 200.0
|
||
insert(env, 0, 5)
|
||
poll(env)
|
||
env["cc2"].weigh("s0001", 790.0, env["clock"]())
|
||
poll(env, 10)
|
||
assert uses(env) == [] and env["sm"].db[5]["remaining_weight"] == 790.0
|
||
|
||
|
||
def test_runout_and_backup_roll(env):
|
||
env["sm"].add(5, remaining=40)
|
||
env["sm"].add(8, remaining=1000, color="FFFFFF")
|
||
insert(env, 0, 5)
|
||
insert(env, 1, 8, color="0xFFFFFF")
|
||
poll(env)
|
||
cc2 = env["cc2"]
|
||
cc2.job = {"active": True}
|
||
cc2.consume("s0001", 35.0) # counted 5 g left when it ran out
|
||
poll(env)
|
||
t = env["clock"]()
|
||
cc2.slots[0]["printer"] = {"status": 0}
|
||
cc2.slots[0]["spool"] = ""
|
||
cc2.add_log(t, "print", "s0001", 0, -35.0, 11700.0, "vase.gcode", "runout")
|
||
cc2.add_log(t, "runout", "s0001", 0, -5.0, 0, "vase.gcode", "runout")
|
||
cc2.consume("s0001", 5.0)
|
||
cc2.consume("s0002", 12.0) # the printer goes on with the other roll
|
||
poll(env, 6)
|
||
assert uses(env) == [] # still printing
|
||
assert "0" not in slots(env) and env["sm"].db[5]["location"] == "Lager"
|
||
cc2.job = {"active": False}
|
||
cc2.add_log(env["clock"](), "print", "s0002", 1, -12.0, 4000.0, "vase.gcode", "complete")
|
||
poll(env, 10)
|
||
assert uses(env) == [("use", 8, 12.0)]
|
||
assert ("used", 5, 1000.0) in env["sm"].calls # ran out: Spoolman's rest is 0, whatever the count said
|
||
assert env["sm"].db[5]["remaining_weight"] == 0
|
||
assert any("Rolle leergelaufen" in m for m in msgs(env))
|
||
assert booked(env)["s0001"]["last"]["grams"] == 40.0
|
||
|
||
|
||
def test_runout_sets_spoolman_to_zero_even_if_the_mirror_was_off(env):
|
||
env["sm"].add(5, remaining=300)
|
||
insert(env, 0, 5)
|
||
poll(env)
|
||
cc2 = env["cc2"]
|
||
cc2.library[0]["remaining"] = 400.0 # e.g. an old mirror reused after the roll printed elsewhere
|
||
booked(env)["s0001"]["remaining"] = 400.0
|
||
cc2.job = {"active": True}
|
||
cc2.consume("s0001", 380.0)
|
||
poll(env)
|
||
cc2.slots[0]["printer"] = {"status": 0}
|
||
cc2.slots[0]["spool"] = ""
|
||
t = env["clock"]()
|
||
cc2.add_log(t, "runout", "s0001", 0, -20.0, 0, "big.gcode", "runout")
|
||
cc2.library[0]["remaining"] = 0.0
|
||
cc2.job = {"active": False}
|
||
cc2.add_log(t + 5, "print", "s0001", 0, -380.0, 127000.0, "big.gcode", "runout")
|
||
poll(env, 15)
|
||
assert uses(env) == [] and env["sm"].db[5]["remaining_weight"] == 0
|
||
|
||
|
||
def test_open_question_after_print_is_reported_once(env):
|
||
cc2 = env["cc2"]
|
||
cc2.slots[3].update({"printer": {"status": 1}, "question": "changed", "question_since": 100, "question_mm": 2605.0})
|
||
poll(env, 3)
|
||
warn = [m for m in msgs(env) if "ohne Spule gedruckt" in m]
|
||
assert len(warn) == 1 and "2605 mm" in warn[0] and "7.8 g" in warn[0]
|
||
assert env["bridge"].snapshot["printers"][0]["slots"][3]["cc2_question_mm"] == 2605.0
|
||
|
||
|
||
def test_multicolour_print_books_each_roll(env):
|
||
for sid in (5, 6, 7):
|
||
env["sm"].add(sid, remaining=900)
|
||
insert(env, 0, 5)
|
||
insert(env, 1, 6)
|
||
insert(env, 3, 7)
|
||
poll(env)
|
||
cc2 = env["cc2"]
|
||
cc2.job = {"active": True}
|
||
for _ in range(4):
|
||
cc2.consume("s0001", 1.5)
|
||
cc2.consume("s0002", 0.75)
|
||
poll(env)
|
||
cc2.job = {"active": False}
|
||
poll(env, 10)
|
||
assert sorted(uses(env)) == [("use", 5, 6.0), ("use", 6, 3.0)]
|
||
|
||
|
||
def test_job_progress_only_read_while_printing(env):
|
||
b = env["bridge"]
|
||
env["cc2"].print_info = None
|
||
poll(env)
|
||
assert b.snapshot["printers"][0]["job"] is None
|
||
env["cc2"].print_info = {"machine": {"progress": 40}, "print": {"current_layer": 12, "total_layers": 30, "remaining": 600}}
|
||
env["cc2"].job = {"active": True, "file": "/local/cube.gcode", "started": 1,
|
||
"slots": [{"mm": 0, "grams": 0, "spool": ""}, {"mm": 330, "grams": 1.234, "spool": "s0001"}]}
|
||
poll(env)
|
||
job = b.snapshot["printers"][0]["job"]
|
||
assert job["file"] == "cube.gcode" and job["grams"] == 1.23 and job["slots"] == [{"slot": 1, "grams": 1.23}]
|
||
assert job["progress"] == 40 and job["layer"] == 12 and job["layers"] == 30 and job["remaining"] == 600
|
||
|
||
|
||
def test_fresh_bridge_leaves_other_bridges_spools_alone(env):
|
||
"""A new instance (empty data volume, no printers yet) must not pull spools out of printers it never ran."""
|
||
env["settings"].save({"printers": []})
|
||
env["sm"].add(11, location="CC2 Links Slot 4", extra={"drucker": '"CC2 Links"', "slot": "4"})
|
||
env["sm"].add(1, location="CC2 Rechts Slot 2", extra={"drucker": '"CC2 Rechts"', "slot": "2"})
|
||
poll(env, 13)
|
||
assert env["sm"].db[11]["location"] == "CC2 Links Slot 4" and env["sm"].extra(11)["drucker"] == "CC2 Links"
|
||
assert env["sm"].db[1]["location"] == "CC2 Rechts Slot 2"
|
||
assert not [c for c in env["sm"].calls if c[0] in ("location", "extra")]
|
||
|
||
|
||
def test_printer_names_written_are_remembered(env):
|
||
env["sm"].add(5)
|
||
insert(env, 0, 5)
|
||
poll(env)
|
||
assert env["bridge"].state.data["written_printers"] == ["Rechts"]
|