"""Kontaktverlust beendet einen Druck nicht.""" import unittest from unittest import mock from grotti.watcher import StateWatcher, contact_gap_keeps_run HOST = "10.10.1.88" FILE = "boden_PETG_0.2_2h15m.gcode" class ContactGapTests(unittest.TestCase): def test_same_file_stays(self): self.assertTrue(contact_gap_keeps_run(FILE, FILE, "printing")) self.assertTrue(contact_gap_keeps_run(FILE, FILE, "paused")) def test_other_file_does_not_stay(self): self.assertFalse(contact_gap_keeps_run(FILE, "andere.gcode", "printing")) def test_terminal_state_does_not_stay(self): self.assertFalse(contact_gap_keeps_run(FILE, FILE, "standby")) self.assertFalse(contact_gap_keeps_run(FILE, FILE, "complete")) self.assertFalse(contact_gap_keeps_run(FILE, FILE, "error")) def test_missing_name_is_not_a_change(self): self.assertTrue(contact_gap_keeps_run("", FILE, "printing")) self.assertTrue(contact_gap_keeps_run(FILE, "", "paused")) class WatcherOfflineTests(unittest.TestCase): def setUp(self): self.watcher = StateWatcher("postgresql://x", {}) self.watcher._prev[HOST] = { "state": "printing", "offline": False, "filename": FILE, "run_id": 42, } def test_two_misses_keep_the_run(self): live = { "host": HOST, "name": "Kobra S1", "state": "offline", "ok": False, "filename": "", } with mock.patch.object(self.watcher, "_emit") as emit, \ mock.patch("grotti.watcher.store.complete_open_run") as close, \ mock.patch("grotti.watcher.jobs.apply_deduction") as deduct: self.watcher._one(mock.Mock(), live) self.watcher._one(mock.Mock(), live) close.assert_not_called() deduct.assert_not_called() remembered = self.watcher._prev[HOST] self.assertEqual(remembered["state"], "printing") self.assertTrue(remembered["offline"]) self.assertEqual(remembered["run_id"], 42) self.assertEqual(remembered["filename"], FILE) self.assertEqual([call.args[3] for call in emit.call_args_list], ["offline"]) def test_same_print_after_gap_is_the_same_run(self): self.watcher._prev[HOST]["offline"] = True live = { "host": HOST, "name": "Kobra S1", "state": "printing", "ok": True, "filename": FILE, "weight": 0, } run = {"id": 42, "filename": FILE} with mock.patch.object(self.watcher, "_emit") as emit, \ mock.patch.object(self.watcher, "_track_print", return_value=run) as track, \ mock.patch.object(self.watcher, "_check_spool", return_value=False), \ mock.patch("grotti.watcher.store.find_live_run", return_value=run), \ mock.patch("grotti.watcher.store.abandon_live_runs") as abandon, \ mock.patch("grotti.watcher.store.complete_open_run") as close, \ mock.patch("grotti.watcher.jobs.sync_history") as sync, \ mock.patch("grotti.watcher.jobs.apply_deduction") as deduct: self.watcher._one(mock.Mock(), live) close.assert_not_called() deduct.assert_not_called() abandon.assert_not_called() sync.assert_not_called() track.assert_called_once() self.assertEqual([call.args[3] for call in emit.call_args_list], ["offline"]) self.assertIn("wieder erreichbar", emit.call_args_list[0].args[4]) self.assertEqual(self.watcher._prev[HOST]["state"], "printing") self.assertFalse(self.watcher._prev[HOST]["offline"]) def test_standby_after_gap_closes(self): self.watcher._prev[HOST]["offline"] = True live = { "host": HOST, "name": "Kobra S1", "state": "standby", "ok": True, "filename": "", } run = {"id": 42, "filename": FILE} with mock.patch.object(self.watcher, "_emit") as emit, \ mock.patch.object(self.watcher, "_track_print", return_value=None), \ mock.patch.object(self.watcher, "_check_spool", return_value=False), \ mock.patch("grotti.watcher.store.find_live_run", return_value=run), \ mock.patch("grotti.watcher.store.running_print_on_host", return_value=None), \ mock.patch("grotti.watcher.store.abandon_live_runs") as abandon, \ mock.patch("grotti.watcher.store.set_bed_needs_clear"), \ mock.patch("grotti.watcher.jobs.sync_history") as sync: self.watcher._one(mock.Mock(), live) abandon.assert_called_once() sync.assert_called_once() self.assertIn("abbruch", [call.args[3] for call in emit.call_args_list]) def test_other_file_after_gap_releases_the_old_run(self): self.watcher._prev[HOST]["offline"] = True live = { "host": HOST, "name": "Kobra S1", "state": "printing", "ok": True, "filename": "andere.gcode", "weight": 0, } old = {"id": 42, "filename": FILE} new = {"id": 99, "filename": "andere.gcode"} with mock.patch.object(self.watcher, "_emit") as emit, \ mock.patch.object(self.watcher, "_track_print", return_value=new), \ mock.patch.object(self.watcher, "_ensure_start_segment"), \ mock.patch.object(self.watcher, "_check_spool", return_value=False), \ mock.patch("grotti.watcher.store.find_live_run", side_effect=[old, old, None]), \ mock.patch("grotti.watcher.store.abandon_live_runs") as abandon, \ mock.patch("grotti.watcher.jobs.sync_history") as sync: self.watcher._one(mock.Mock(), live) sync.assert_called_once() self.assertGreaterEqual(abandon.call_count, 1) kinds = [call.args[3] for call in emit.call_args_list] self.assertEqual(kinds, ["abbruch", "offline", "start"]) def test_other_file_without_offline_releases_the_old_run(self): live = { "host": HOST, "name": "Kobra S1", "state": "printing", "ok": True, "filename": "andere.gcode", "weight": 0, } old = {"id": 42, "filename": FILE} new = {"id": 99, "filename": "andere.gcode"} seen = {"n": 0} def live_run(_conn, _host, filename=""): seen["n"] += 1 if seen["n"] <= 3: return old return None with mock.patch.object(self.watcher, "_emit") as emit, \ mock.patch.object(self.watcher, "_track_print", return_value=new), \ mock.patch.object(self.watcher, "_ensure_start_segment"), \ mock.patch.object(self.watcher, "_check_spool", return_value=False), \ mock.patch("grotti.watcher.store.find_live_run", side_effect=live_run), \ mock.patch("grotti.watcher.store.abandon_live_runs") as abandon, \ mock.patch("grotti.watcher.jobs.sync_history") as sync: self.watcher._one(mock.Mock(), live) sync.assert_called_once() abandon.assert_called() self.assertIn("abbruch", [call.args[3] for call in emit.call_args_list]) self.assertIn("start", [call.args[3] for call in emit.call_args_list]) def test_cancel_event_survives_deleted_live_run(self): self.watcher._cancel_hits[HOST] = 2 live = { "host": HOST, "name": "Kobra S1", "state": "cancelled", "ok": True, "filename": FILE, } dangling = {"id": 42, "filename": FILE, "state": "in_progress"} def row(_conn, run_id): if run_id == 77: return {"id": 77, "filename": FILE} return None with mock.patch("grotti.watcher.store.find_live_run", side_effect=[dangling, None]), \ mock.patch("grotti.watcher.store.find_open_history_run", return_value=None), \ mock.patch("grotti.watcher.store.get_run_row", side_effect=row), \ mock.patch("grotti.watcher.store.latest_run_id", return_value=77), \ mock.patch("grotti.watcher.store.abandon_live_runs"), \ mock.patch("grotti.watcher.store.set_bed_needs_clear"), \ mock.patch("grotti.watcher.store.set_printer_active_run"), \ mock.patch("grotti.watcher.jobs.sync_history") as sync, \ mock.patch("grotti.watcher.notify.record") as record: self.watcher._one(mock.Mock(), live) sync.assert_called_once() record.assert_called() self.assertEqual(record.call_args.kwargs["run_id"], 77) self.assertEqual(record.call_args.args[4], "abbruch") if __name__ == "__main__": unittest.main()