128 lines
5.2 KiB
Python
128 lines
5.2 KiB
Python
"""ACE-Startabschnitte und Spulenwechsel 0→jetzt."""
|
|
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
from grotti import store
|
|
from grotti.watcher import StateWatcher
|
|
|
|
|
|
class PrevSpoolForSlotTests(unittest.TestCase):
|
|
def test_prefers_old_at_slot(self):
|
|
conn = mock.Mock()
|
|
live = {"id": 1, "filename": "a.gcode", "spool_id": 99}
|
|
got = store.prev_spool_id_for_slot(
|
|
conn, "h", 4, live, 21, old_rows=[{"id": 18}],
|
|
)
|
|
self.assertEqual(got, 18)
|
|
|
|
def test_ace_map_send_when_old_empty(self):
|
|
conn = mock.Mock()
|
|
live = {"id": 33805, "filename": "x.gcode", "spool_id": None}
|
|
with mock.patch(
|
|
"grotti.store.latest_send",
|
|
return_value={
|
|
"ace_map": [{"paint_index": 0, "ams_index": 3}],
|
|
"spool_id": 18,
|
|
},
|
|
):
|
|
got = store.prev_spool_id_for_slot(conn, "h", 4, live, 21, old_rows=[])
|
|
self.assertEqual(got, 18)
|
|
|
|
def test_last_segment_on_slot(self):
|
|
conn = mock.Mock()
|
|
live = {"id": 7, "filename": "x.gcode", "spool_id": None}
|
|
with mock.patch("grotti.store.latest_send", return_value={"ace_map": [], "spool_id": None}), \
|
|
mock.patch(
|
|
"grotti.store.list_run_segments",
|
|
return_value=[
|
|
{"spool_id": 10, "ace_slot": 2, "mm_start": 0, "mm_end": 100},
|
|
{"spool_id": 18, "ace_slot": 4, "mm_start": 0, "mm_end": 500},
|
|
],
|
|
):
|
|
got = store.prev_spool_id_for_slot(conn, "h", 4, live, 21, old_rows=[])
|
|
self.assertEqual(got, 18)
|
|
|
|
|
|
class EnsureStartSegmentAceTests(unittest.TestCase):
|
|
def test_opens_mapped_slots_with_spool(self):
|
|
watcher = StateWatcher("postgresql://x", {})
|
|
conn = mock.Mock()
|
|
live = {"filament_used": 0}
|
|
with mock.patch("grotti.watcher.store.list_run_segments", return_value=[]), \
|
|
mock.patch(
|
|
"grotti.watcher.store.latest_send",
|
|
return_value={"ace_map": [{"paint_index": 0, "ams_index": 3}]},
|
|
), \
|
|
mock.patch(
|
|
"grotti.watcher.store.spool_at_slot",
|
|
return_value={"id": 18},
|
|
) as at_slot, \
|
|
mock.patch("grotti.watcher.store.open_segment") as open_seg:
|
|
watcher._ensure_start_segment(conn, "h", "a.gcode", live, 33805)
|
|
at_slot.assert_called_once_with(conn, "h", 4)
|
|
open_seg.assert_called_once_with(conn, 33805, 18, 4, 0.0)
|
|
|
|
def test_ace_without_spool_opens_nothing(self):
|
|
watcher = StateWatcher("postgresql://x", {})
|
|
conn = mock.Mock()
|
|
with mock.patch("grotti.watcher.store.list_run_segments", return_value=[]), \
|
|
mock.patch(
|
|
"grotti.watcher.store.latest_send",
|
|
return_value={"ace_map": [{"paint_index": 0, "ams_index": 3}]},
|
|
), \
|
|
mock.patch("grotti.watcher.store.spool_at_slot", return_value=None), \
|
|
mock.patch("grotti.watcher.store.open_segment") as open_seg:
|
|
watcher._ensure_start_segment(conn, "h", "a.gcode", {"filament_used": 0}, 1)
|
|
open_seg.assert_not_called()
|
|
|
|
|
|
class SpoolLoadFallbackSegmentsTests(unittest.TestCase):
|
|
"""ACE ohne Segment + Einlegen ohne old → 0→jetzt + neuer offener Abschnitt."""
|
|
|
|
def test_creates_closed_and_open_segment(self):
|
|
conn = mock.Mock()
|
|
live = {"id": 33805, "filename": "x.gcode", "spool_id": None, "state": "in_progress"}
|
|
opened = []
|
|
closed = []
|
|
|
|
def open_segment(_c, run_id, spool_id, ace_slot, mm_start):
|
|
opened.append((run_id, spool_id, ace_slot, mm_start))
|
|
return True
|
|
|
|
def close_open_segments(_c, run_id, ace_slot, mm_end):
|
|
closed.append((run_id, ace_slot, mm_end))
|
|
# Erster Aufruf: nichts offen; zweiter: nach open_segment
|
|
return 1 if opened else 0
|
|
|
|
with mock.patch("grotti.store.close_open_segments", side_effect=close_open_segments), \
|
|
mock.patch("grotti.store.open_segment", side_effect=open_segment), \
|
|
mock.patch(
|
|
"grotti.store.prev_spool_id_for_slot",
|
|
return_value=18,
|
|
):
|
|
# Logik wie in spool_load (Auszug)
|
|
slot = 4
|
|
spool_id = 21
|
|
mm_now = 5291.0
|
|
old = []
|
|
closed_n = store.close_open_segments(conn, live["id"], slot, mm_now)
|
|
if not closed_n and old:
|
|
pass
|
|
elif not closed_n and mm_now > 0:
|
|
prev_id = store.prev_spool_id_for_slot(
|
|
conn, "h", slot, live, spool_id, old_rows=old,
|
|
)
|
|
if prev_id:
|
|
store.open_segment(conn, live["id"], prev_id, slot, 0)
|
|
store.close_open_segments(conn, live["id"], slot, mm_now)
|
|
store.open_segment(conn, live["id"], spool_id, slot, mm_now)
|
|
|
|
self.assertEqual(opened[0], (33805, 18, 4, 0))
|
|
self.assertEqual(closed[-1], (33805, 4, 5291.0))
|
|
self.assertEqual(opened[-1], (33805, 21, 4, 5291.0))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|