95 lines
3.3 KiB
Python
95 lines
3.3 KiB
Python
"""Zustandsmaschine der Druckwarteschlange mit Fake-Moonraker."""
|
|
|
|
import unittest
|
|
from pathlib import Path
|
|
import sys
|
|
from unittest import mock
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from grotti.jobs import is_filament_resume
|
|
from grotti.print_queue import (
|
|
PRINT_START_TIMEOUT_S,
|
|
TERMINAL_CONFIRM_HITS,
|
|
PrintQueueDispatcher,
|
|
_cool_wait_text,
|
|
)
|
|
|
|
|
|
class CoolTextTests(unittest.TestCase):
|
|
def test_wait_text(self):
|
|
cool = {"bed_max": 30, "nozzle_max": 45}
|
|
text = _cool_wait_text({"bed": 40, "nozzle": 50}, cool)
|
|
self.assertIn("Düse", text)
|
|
self.assertIn("Bett", text)
|
|
self.assertNotIn(" · ", text)
|
|
|
|
|
|
class FilamentResumeTests(unittest.TestCase):
|
|
def test_same_moment_with_duration_is_resume(self):
|
|
end = datetime(2026, 10, 21, 19, 32, 20, tzinfo=timezone.utc)
|
|
start = end + timedelta(milliseconds=40)
|
|
self.assertTrue(is_filament_resume(
|
|
end, start, prev_duration_s=500, new_duration_s=520
|
|
))
|
|
|
|
def test_later_reprint_is_not_resume(self):
|
|
end = datetime(2026, 10, 21, 19, 32, 20, tzinfo=timezone.utc)
|
|
start = end + timedelta(minutes=10)
|
|
self.assertFalse(is_filament_resume(
|
|
end, start, prev_duration_s=500, new_duration_s=10
|
|
))
|
|
|
|
|
|
class DispatcherHelpers(unittest.TestCase):
|
|
def test_timeout_constant(self):
|
|
self.assertGreaterEqual(PRINT_START_TIMEOUT_S, 300)
|
|
self.assertGreaterEqual(TERMINAL_CONFIRM_HITS, 2)
|
|
|
|
def test_worker_busy_empty(self):
|
|
d = PrintQueueDispatcher("postgresql://x", {}, watcher=None)
|
|
self.assertFalse(d._worker_busy(1))
|
|
|
|
def test_auto_enabled_default(self):
|
|
d = PrintQueueDispatcher("postgresql://x", {}, watcher=None)
|
|
self.assertTrue(d._auto_enabled())
|
|
d.cfg = {"auto_queue": False}
|
|
self.assertFalse(d._auto_enabled())
|
|
|
|
def test_no_retry_after_abort(self):
|
|
d = PrintQueueDispatcher("postgresql://x", {}, watcher=None)
|
|
self.assertFalse(d._retryable_message("Druck abgebrochen."))
|
|
self.assertTrue(d._retryable_message("Start-Timeout"))
|
|
|
|
def test_advance_simple_ignores_single_cancelled(self):
|
|
d = PrintQueueDispatcher("postgresql://x", {}, watcher=None)
|
|
item = {"id": 1, "phase": "print_active", "filename": "a.gcode"}
|
|
printer = {"host": "h", "name": "P"}
|
|
live = {"state": "cancelled", "print_active": False}
|
|
with mock.patch("grotti.print_queue.store.update_print_queue") as upd:
|
|
d._advance_simple(mock.Mock(), item, printer, live)
|
|
upd.assert_not_called()
|
|
self.assertEqual(d._term_hits.get("1:end"), 1)
|
|
|
|
|
|
class ResolveSpoolTests(unittest.TestCase):
|
|
def test_material_match(self):
|
|
from grotti.jobs import assert_material_match
|
|
|
|
file_row = {"filaments": [{"type": "PLA"}]}
|
|
assert_material_match(file_row, [{"material_type": "PLA"}])
|
|
assert_material_match(file_row, [{"material_type": "PETG"}])
|
|
|
|
def test_ace_type_for_pa12(self):
|
|
from grotti.catalog import ace_material_type
|
|
|
|
self.assertEqual(ace_material_type("PA12-CF"), "PA6-CF")
|
|
self.assertEqual(ace_material_type("PETG"), "PETG")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|