Jonas MenzelandClaude Opus 5.5
d863287f1f
CC2 Control compares a spool's colour exactly with the tray's, which is the tag's colour from the display palette (e.g. #F72221 for Spoolman's #FF0000), and flags the slot or asks again when they differ. New copies take the tag's colour; copies in a slot are set to the tray's colour, others keep theirs. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014GcEk5tkfFSYD6NTF66NkC
1004 lines
38 KiB
Python
1004 lines
38 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, color="0x2850E0") # the tag carries the nearest colour of the display palette
|
||
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"] == "#2850E0" and f["note"] == "spoolman:5" and f["remaining"] == 812.5 # as the printer reports it
|
||
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" and "id" not in c[1]] == [] # not created again
|
||
assert ("assign", 0, "s0001") in env["cc2"].calls
|
||
# the copy made by hand lacked brand and empty weight: taken from Spoolman
|
||
assert ("save", {"id": "s0001", "brand": "SUNLU", "tare": 150.0}) 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"] == "#000000" # the tray's (tag's) colour, which CC2 Control compares with
|
||
assert cc2_spool_fields(sm, None)[0]["color"] == "#FFD700" # no tray known: Spoolman's
|
||
|
||
|
||
# ---------- 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] == "assign" or (c[0] == "save" and "id" not in c[1])] # 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 and "archivieren" in m for m in msgs(env))
|
||
assert env["sm"].db[5]["archived"] # empty: archived in Spoolman ...
|
||
assert ("delete", "s0001") in env["cc2"].calls # ... and its copy leaves CC2 Control
|
||
|
||
|
||
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"]
|
||
|
||
|
||
def test_rest_follows_spoolman_when_roll_comes_back(tmp_path):
|
||
e = make_env(tmp_path, printers=2)
|
||
for c in e["cc2s"].values():
|
||
c.clock = e["clock"]
|
||
e["sm"].add(5, remaining=800)
|
||
insert(e, 0, 5)
|
||
poll(e, 2)
|
||
take_out_of_tray(e, 0)
|
||
e["readers"]["p1"].record["channels0"] = channel("")
|
||
insert(e, 1, 5, pid="p2") # into the other printer: a fresh copy there with 800 g
|
||
poll(e, 6)
|
||
e["cc2s"]["p2"].consume("s0001", 100.0)
|
||
poll(e, 15)
|
||
assert e["sm"].db[5]["remaining_weight"] == 700
|
||
take_out_of_tray(e, 1, pid="p2")
|
||
e["readers"]["p2"].record["channels1"] = channel("")
|
||
insert(e, 0, 5) # back into the first printer: its old copy still says 800 g
|
||
poll(e, 30)
|
||
assert ("adjust", "s0001", 700.0) in e["cc2s"]["p1"].calls
|
||
assert e["cc2s"]["p1"].library[0]["remaining"] == 700.0
|
||
assert [c for c in e["sm"].calls if c[0] in ("use", "used")] == [("use", 5, 100.0)] # nothing booked back
|
||
poll(e, 30)
|
||
assert len([c for c in e["cc2s"]["p1"].calls if c[0] == "adjust"]) == 1
|
||
|
||
|
||
def test_weighed_in_spoolman_goes_to_cc2(env):
|
||
env["cc2"].clock = env["clock"]
|
||
env["sm"].add(5, remaining=800)
|
||
insert(env, 0, 5)
|
||
poll(env, 2)
|
||
env["sm"].db[5]["remaining_weight"] = 612.0 # weighed and entered in Spoolman
|
||
poll(env, 30)
|
||
assert ("adjust", "s0001", 612.0) in env["cc2"].calls
|
||
assert not [c for c in env["sm"].calls if c[0] in ("use", "used")]
|
||
assert booked(env)["s0001"]["remaining"] == 612.0 and "adjusted" not in booked(env)["s0001"]
|
||
|
||
|
||
def test_rest_not_adjusted_before_use_is_booked(env):
|
||
env["sm"].add(5, remaining=800)
|
||
insert(env, 0, 5)
|
||
poll(env, 2)
|
||
env["cc2"].job = {"active": True}
|
||
env["cc2"].consume("s0001", 50.0)
|
||
poll(env, 20) # printing: neither booked nor adjusted
|
||
assert not [c for c in env["cc2"].calls if c[0] == "adjust"]
|
||
env["cc2"].job = {"active": False}
|
||
poll(env, 30)
|
||
assert uses(env) == [("use", 5, 50.0)]
|
||
assert not [c for c in env["cc2"].calls if c[0] == "adjust"]
|
||
|
||
|
||
def test_master_data_follows_spoolman(env):
|
||
env["sm"].add(5, color="4A4A4A")
|
||
insert(env, 0, 5, color="0x898989")
|
||
poll(env, 2)
|
||
assert env["cc2"].library[0]["color"] == "#898989" # created with the tag's colour
|
||
env["sm"].db[5]["filament"]["density"] = 1.3
|
||
env["sm"].db[5]["lot_nr"] = "SUNLU005B"
|
||
env["bridge"].spool_cache.clear()
|
||
poll(env, 14)
|
||
assert ("save", {"id": "s0001", "name": "SUNLU005B", "density": 1.3}) in env["cc2"].calls
|
||
poll(env, 14)
|
||
assert len([c for c in env["cc2"].calls if c[0] == "save" and "id" in c[1]]) == 1
|
||
|
||
|
||
def test_copy_colour_follows_the_printers_report(env):
|
||
"""A copy made with Spoolman's exact colour is flagged by CC2 Control; it gets the tray's colour."""
|
||
env["sm"].add(4, color="FF0000")
|
||
env["cc2"].save_spool({"name": "SUNLU004", "brand": "SUNLU", "material": "PLA+", "color": "#FF0000",
|
||
"note": "spoolman:4", "net": 1000, "tare": 150, "density": 1.24, "diameter": 1.75})
|
||
insert(env, 0, 4, color="0xF72221")
|
||
env["cc2"].slots[0]["printer"] = {"status": 1, "color": "#F72221", "type": "PLA"}
|
||
poll(env, 14)
|
||
assert env["cc2"].library[0]["color"] == "#F72221"
|
||
take_out_of_tray(env, 0)
|
||
env["sm"].db[4]["filament"]["color_hex"] = "00FF00"
|
||
env["bridge"].spool_cache.clear()
|
||
poll(env, 14)
|
||
assert env["cc2"].library[0]["color"] == "#F72221" # not in a tray: colour left alone
|
||
|
||
|
||
def test_copies_of_archived_or_deleted_spools_are_removed_when_not_loaded(env):
|
||
env["sm"].add(5)
|
||
env["sm"].add(6)
|
||
env["sm"].add(7)
|
||
insert(env, 0, 5)
|
||
insert(env, 1, 6)
|
||
insert(env, 2, 7)
|
||
poll(env, 2)
|
||
take_out_of_tray(env, 0)
|
||
take_out_of_tray(env, 1)
|
||
poll(env, 6)
|
||
env["sm"].db[5]["archived"] = True
|
||
del env["sm"].db[6]
|
||
env["sm"].db[7]["archived"] = True # still in a tray: kept
|
||
env["bridge"].spool_cache.clear()
|
||
poll(env, 14)
|
||
deleted = [c[1] for c in env["cc2"].calls if c[0] == "delete"]
|
||
assert deleted == ["s0001", "s0002"]
|
||
assert [s["id"] for s in env["cc2"].library] == ["s0003"]
|
||
|
||
|
||
def test_scan_of_archived_spool_is_not_mapped(env):
|
||
env["sm"].add(5)["archived"] = True
|
||
insert(env, 0, 5)
|
||
poll(env, 2)
|
||
assert slots(env) == {} and not env["cc2"].calls
|
||
assert any("ist archiviert" in m for m in msgs(env))
|
||
|
||
|
||
def test_runout_does_not_archive_when_switched_off(env):
|
||
env["settings"].save({"archive_empty": False})
|
||
env["sm"].add(5, remaining=40)
|
||
insert(env, 0, 5)
|
||
poll(env)
|
||
cc2 = env["cc2"]
|
||
cc2.job = {"active": True}
|
||
cc2.consume("s0001", 40.0)
|
||
poll(env)
|
||
cc2.slots[0]["printer"] = {"status": 0}
|
||
cc2.slots[0]["spool"] = ""
|
||
cc2.add_log(env["clock"](), "runout", "s0001", 0, 0.0, 0, "vase.gcode", "runout")
|
||
cc2.job = {"active": False}
|
||
poll(env, 15)
|
||
assert env["sm"].db[5]["remaining_weight"] == 0 and not env["sm"].db[5]["archived"]
|
||
|
||
|
||
def test_roll_back_in_its_old_slot_after_the_other_printer(tmp_path):
|
||
"""Scanned again with identical tag data: the record does not change, CC2 Control asks; the roll is recognised."""
|
||
e = make_env(tmp_path, printers=2)
|
||
e["sm"].add(5, remaining=800)
|
||
insert(e, 3, 5)
|
||
poll(e, 2)
|
||
take_out_of_tray(e, 3)
|
||
insert(e, 0, 5, pid="p2") # scanned and loaded in the other printer
|
||
poll(e, 6)
|
||
assert slots(e, "p2")["0"]["spool"] == 5 and slots(e) == {}
|
||
take_out_of_tray(e, 0, pid="p2")
|
||
e["readers"]["p2"].record["channels0"] = channel("")
|
||
poll(e, 6)
|
||
assert slots(e, "p2") == {}
|
||
put_back(e, 3) # scanned at the side again (same data) and back into its old tray
|
||
poll(e, 2)
|
||
assert slots(e)["3"]["spool"] == 5 and slots(e)["3"].get("reinserted")
|
||
assert e["cc2s"]["p1"].slots[3]["spool"] == "s0001"
|
||
|
||
|
||
def test_archived_spool_with_roll_in_tray_is_not_deleted(env):
|
||
env["sm"].add(5)
|
||
insert(env, 0, 5)
|
||
poll(env, 2)
|
||
env["bridge"].release(env["bridge"].workers["p1"], 0, "Test")
|
||
env["cc2"].slots[0]["spool"] = "" # unbound, roll still in the tray with its tag
|
||
env["sm"].db[5]["archived"] = True
|
||
env["bridge"].spool_cache.clear()
|
||
poll(env, 14)
|
||
assert not [c for c in env["cc2"].calls if c[0] == "delete"]
|
||
|
||
|
||
def test_spoolman_outage_does_not_delete_copies(env):
|
||
env["sm"].add(5)
|
||
insert(env, 0, 5)
|
||
poll(env, 2)
|
||
take_out_of_tray(env, 0)
|
||
poll(env, 6)
|
||
env["bridge"].spool_cache.clear()
|
||
env["sm"].fail = True
|
||
poll(env, 14)
|
||
assert not [c for c in env["cc2"].calls if c[0] == "delete"]
|
||
assert len(env["cc2"].library) == 1
|