196 lines
7.1 KiB
Python
196 lines
7.1 KiB
Python
"""Unit-Tests für Anycubic-MQTT-Codes, Task-Hygiene und 10101-Retry."""
|
|
|
|
import sys
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from grotti import anycubic_lan
|
|
from grotti.anycubic_lan import (
|
|
CODE_FILAMENT,
|
|
CODE_TASK_ACTIVE,
|
|
LanError,
|
|
ensure_task_free,
|
|
message_for_code,
|
|
start_print,
|
|
task_summary,
|
|
)
|
|
|
|
|
|
class MessageForCodeTests(unittest.TestCase):
|
|
def test_10101(self):
|
|
text = message_for_code(CODE_TASK_ACTIVE)
|
|
self.assertIn("10101", text)
|
|
self.assertIn("aktiv", text.lower())
|
|
|
|
def test_10107(self):
|
|
text = message_for_code(CODE_FILAMENT)
|
|
self.assertIn("10107", text)
|
|
self.assertIn("Filament", text)
|
|
|
|
def test_unknown_with_default(self):
|
|
text = message_for_code(99999, "Druckstart abgelehnt")
|
|
self.assertIn("99999", text)
|
|
self.assertIn("Druckstart abgelehnt", text)
|
|
|
|
def test_string_code(self):
|
|
self.assertEqual(
|
|
message_for_code("10107"),
|
|
message_for_code(10107),
|
|
)
|
|
|
|
|
|
class ReportErrorTests(unittest.TestCase):
|
|
def test_report_carries_code(self):
|
|
text, code = anycubic_lan._report_error(
|
|
{"code": 10101, "msg": "Print task already exists"},
|
|
"Druckstart abgelehnt",
|
|
)
|
|
self.assertEqual(code, 10101)
|
|
self.assertIn("10101", text)
|
|
|
|
def test_raw_msg_unknown_code(self):
|
|
text, code = anycubic_lan._report_error(
|
|
{"code": 42, "msg": "irgendwas"},
|
|
"Druckstart abgelehnt",
|
|
)
|
|
self.assertEqual(code, 42)
|
|
self.assertIn("irgendwas", text)
|
|
self.assertIn("42", text)
|
|
|
|
|
|
class TaskSummaryTests(unittest.TestCase):
|
|
def test_free(self):
|
|
summary = task_summary({"state": "free", "project": {}})
|
|
self.assertEqual(summary["state"], "free")
|
|
self.assertIsNone(summary["task_id"])
|
|
|
|
def test_busy_project(self):
|
|
summary = task_summary({
|
|
"state": "printing",
|
|
"project": {"task_id": 12345, "filename": "/tmp/foo.gcode"},
|
|
})
|
|
self.assertEqual(summary["task_id"], "12345")
|
|
self.assertEqual(summary["filename"], "foo.gcode")
|
|
|
|
def test_empty(self):
|
|
self.assertIsNone(task_summary({}))
|
|
self.assertIsNone(task_summary(None))
|
|
|
|
|
|
class TaskBusyTests(unittest.TestCase):
|
|
def test_free_idle(self):
|
|
self.assertFalse(anycubic_lan._task_busy({"state": "free"}))
|
|
self.assertFalse(anycubic_lan._task_busy({"state": "idle"}))
|
|
|
|
def test_s1_style_null_project(self):
|
|
# S1 liefert project: null + last_project; last_project zählt nicht.
|
|
self.assertFalse(anycubic_lan._task_busy({
|
|
"state": "free",
|
|
"project": None,
|
|
"last_project": {"task_id": "old"},
|
|
}))
|
|
|
|
def test_busy_state(self):
|
|
self.assertTrue(anycubic_lan._task_busy({"state": "printing"}))
|
|
self.assertTrue(anycubic_lan._task_busy({"state": "auto_leveling"}))
|
|
|
|
def test_busy_project(self):
|
|
self.assertTrue(anycubic_lan._task_busy({
|
|
"state": "free",
|
|
"project": {"task_id": "9"},
|
|
}))
|
|
|
|
|
|
class PrintInfoSoftFailTests(unittest.TestCase):
|
|
def tearDown(self):
|
|
anycubic_lan._FAIL_CACHE.clear()
|
|
anycubic_lan._INFO_CACHE.clear()
|
|
|
|
@mock.patch("grotti.anycubic_lan.credentials", side_effect=LanError("weg"))
|
|
def test_credentials_fail_returns_none(self, _cred):
|
|
self.assertIsNone(anycubic_lan.print_info("10.0.0.9", use_cache=False))
|
|
self.assertIn("10.0.0.9", anycubic_lan._FAIL_CACHE)
|
|
|
|
@mock.patch("grotti.anycubic_lan.credentials", side_effect=LanError("weg"))
|
|
def test_fail_cache_skips_retry(self, cred):
|
|
anycubic_lan._FAIL_CACHE["10.0.0.9"] = time.time()
|
|
self.assertIsNone(anycubic_lan.print_info("10.0.0.9", use_cache=True))
|
|
cred.assert_not_called()
|
|
|
|
@mock.patch("grotti.anycubic_lan.print_info", return_value=None)
|
|
def test_ensure_free_when_lan_down(self, _info):
|
|
self.assertIsNone(ensure_task_free("10.0.0.9"))
|
|
|
|
|
|
class EnsureTaskFreeTests(unittest.TestCase):
|
|
@mock.patch("grotti.anycubic_lan.time.sleep", return_value=None)
|
|
@mock.patch("grotti.anycubic_lan.cancel_print")
|
|
@mock.patch("grotti.anycubic_lan.print_info")
|
|
def test_already_free(self, print_info, cancel_print, _sleep):
|
|
print_info.return_value = {"state": "free", "project": {}}
|
|
info = ensure_task_free("10.0.0.1")
|
|
self.assertEqual(info["state"], "free")
|
|
cancel_print.assert_not_called()
|
|
|
|
@mock.patch("grotti.anycubic_lan.time.sleep", return_value=None)
|
|
@mock.patch("grotti.anycubic_lan.cancel_print")
|
|
@mock.patch("grotti.anycubic_lan.print_info")
|
|
def test_clears_busy(self, print_info, cancel_print, _sleep):
|
|
print_info.side_effect = [
|
|
{"state": "printing", "project": {"task_id": "1"}},
|
|
{"state": "free", "project": {}},
|
|
]
|
|
info = ensure_task_free("10.0.0.1", attempts=3, wait_s=0.01)
|
|
self.assertEqual(info["state"], "free")
|
|
cancel_print.assert_called_once()
|
|
|
|
@mock.patch("grotti.anycubic_lan.time.sleep", return_value=None)
|
|
@mock.patch("grotti.anycubic_lan.cancel_print")
|
|
@mock.patch("grotti.anycubic_lan.print_info")
|
|
def test_raises_when_stuck(self, print_info, cancel_print, _sleep):
|
|
print_info.return_value = {"state": "printing", "project": {"task_id": "1"}}
|
|
with self.assertRaises(LanError) as ctx:
|
|
ensure_task_free("10.0.0.1", attempts=2, wait_s=0.01)
|
|
self.assertEqual(cancel_print.call_count, 2)
|
|
self.assertIn("freimachen", ctx.exception.message)
|
|
|
|
|
|
class StartPrintRetryTests(unittest.TestCase):
|
|
@mock.patch("grotti.anycubic_lan.time.sleep", return_value=None)
|
|
@mock.patch("grotti.anycubic_lan.ensure_task_free")
|
|
@mock.patch("grotti.anycubic_lan.cancel_print")
|
|
@mock.patch("grotti.anycubic_lan._mqtt_print")
|
|
def test_retry_on_10101(self, mqtt_print, cancel_print, ensure_free, _sleep):
|
|
ensure_free.return_value = {"state": "free"}
|
|
mqtt_print.side_effect = [
|
|
LanError(message_for_code(CODE_TASK_ACTIVE), code=CODE_TASK_ACTIVE),
|
|
{"state": "printing"},
|
|
]
|
|
result = start_print("10.0.0.1", "demo.gcode", auto_leveling=False)
|
|
self.assertEqual(result["filename"], "demo.gcode")
|
|
self.assertEqual(mqtt_print.call_count, 2)
|
|
cancel_print.assert_called_once()
|
|
|
|
@mock.patch("grotti.anycubic_lan.time.sleep", return_value=None)
|
|
@mock.patch("grotti.anycubic_lan.ensure_task_free")
|
|
@mock.patch("grotti.anycubic_lan._mqtt_print")
|
|
def test_no_retry_on_filament(self, mqtt_print, ensure_free, _sleep):
|
|
ensure_free.return_value = {"state": "free"}
|
|
mqtt_print.side_effect = LanError(
|
|
message_for_code(CODE_FILAMENT),
|
|
code=CODE_FILAMENT,
|
|
)
|
|
with self.assertRaises(LanError) as ctx:
|
|
start_print("10.0.0.1", "demo.gcode")
|
|
self.assertEqual(ctx.exception.code, CODE_FILAMENT)
|
|
self.assertEqual(mqtt_print.call_count, 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|