Grotti 3.2.0: Jetzt-Liste, Filament-Pause, Strom-Torte, UI-Polish.

Jetzt/Topbar und Material-Routing, Pause bei Filament-Ende, Druckbericht mit Stromanteilen, Design- und Kamera-Stream-Polish; Bedarf-Alert-Spam und Platte-belegt-Bugs behoben.

tests/test_auth.py

@@ -48,6 +48,7 @@ class CameraFieldsTest(unittest.TestCase):
         moonraker.CAMERA_CACHE.clear()
         fields = moonraker._camera_fields("10.0.0.1", "10.0.0.1")
         self.assertTrue(fields["camera"])
+        self.assertIn("/api/v1/kamera/stream?", fields["camera_stream"])
         self.assertIsNone(fields["has_camera"])
 
     def test_known_dead_hides_url(self):
@@ -56,6 +57,7 @@ class CameraFieldsTest(unittest.TestCase):
         moonraker.CAMERA_CACHE["10.0.0.2"] = (time.time(), 30, None)
         fields = moonraker._camera_fields("10.0.0.2", "10.0.0.2")
         self.assertEqual(fields["camera"], "")
+        self.assertEqual(fields["camera_stream"], "")
         self.assertFalse(fields["has_camera"])
         moonraker.CAMERA_CACHE.pop("10.0.0.2", None)
 

tests/test_cost_ed.py

@@ -46,6 +46,22 @@ class EnergyEdTests(unittest.TestCase):
         self.assertAlmostEqual(est["print_kwh"], 0.2, places=3)
         self.assertAlmostEqual(est["cost"], 0.4, places=3)
 
+    def test_ed_parts_split_base_nozzle_bed(self):
+        samples = [sample_at(i * 30, np=1.0, bp=0.5) for i in range(0, 121)]
+        est = energy_estimate(3600, {
+            "watts": 40,
+            "nozzle_watts": 60,
+            "bed_watts": 220,
+            "euro_per_kwh": 0.35,
+        }, samples)
+        self.assertEqual(est["source"], "ed")
+        self.assertAlmostEqual(est["base_kwh"], 0.04, places=3)
+        self.assertAlmostEqual(est["nozzle_kwh"], 0.06, places=3)
+        self.assertAlmostEqual(est["bed_kwh"], 0.11, places=3)
+        keys = {p["key"] for p in est["parts"]}
+        self.assertEqual(keys, {"base", "nozzle", "bed"})
+        self.assertAlmostEqual(sum(p["kwh"] for p in est["parts"]), est["kwh"], places=6)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_daily_ops_alerts.py

@@ -0,0 +1,48 @@
+"""Ops-Alerts: Dedup-Art muss zum gespeicherten Event-kind passen."""
+
+import unittest
+from unittest.mock import MagicMock, patch
+
+from grotti.daily_report import push_ops_alerts
+
+
+class OpsAlertDedupeTests(unittest.TestCase):
+    def test_bedarf_records_matching_kind(self):
+        conn = MagicMock()
+        cfg = {"alerts": {"queue": True}}
+        shortfall = [{
+            "material_type": "PETG-CF",
+            "color": "#161616",
+            "shortfall_g": 15,
+        }]
+        with patch("grotti.daily_report.jetzt_mod.overdue_orders", return_value=[]), \
+             patch("grotti.daily_report.jetzt_mod.overdue_maintenance", return_value=[]), \
+             patch("grotti.daily_report.bedarf_mod.compute", return_value={"shortfall": shortfall}), \
+             patch("grotti.daily_report.store.list_printers", return_value=[]), \
+             patch("grotti.daily_report.store.recent_event_kind", return_value=False) as recent, \
+             patch("grotti.daily_report.notify.record") as record:
+            push_ops_alerts(conn, cfg, printers=[])
+            recent.assert_called_with(
+                conn, "bedarf:PETG-CF:#161616", "bedarf", within_s=21600,
+            )
+            record.assert_called_once()
+            args = record.call_args
+            self.assertEqual(args.args[4], "bedarf")
+            self.assertIn("15 g fehlen", args.args[5])
+
+    def test_bedarf_skips_when_recent(self):
+        conn = MagicMock()
+        cfg = {"alerts": {"queue": True}}
+        shortfall = [{"material_type": "PLA", "color": "", "shortfall_g": 10}]
+        with patch("grotti.daily_report.jetzt_mod.overdue_orders", return_value=[]), \
+             patch("grotti.daily_report.jetzt_mod.overdue_maintenance", return_value=[]), \
+             patch("grotti.daily_report.bedarf_mod.compute", return_value={"shortfall": shortfall}), \
+             patch("grotti.daily_report.store.list_printers", return_value=[]), \
+             patch("grotti.daily_report.store.recent_event_kind", return_value=True), \
+             patch("grotti.daily_report.notify.record") as record:
+            push_ops_alerts(conn, cfg, printers=[])
+            record.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_filament_pause.py

@@ -0,0 +1,43 @@
+"""Filament-Pause-Heuristik aus ACE/mmu und Meldungen."""
+
+import unittest
+
+from grotti.moonraker import filament_out_hint
+
+
+class FilamentOutHintTests(unittest.TestCase):
+    def test_message_10107(self):
+        code, text = filament_out_hint({}, "Report code 10107")
+        self.assertEqual(code, "message")
+        self.assertIn("10107", text)
+
+    def test_active_filament_empty(self):
+        code, text = filament_out_hint({"active_filament": {"empty": True}})
+        self.assertEqual(code, "active_empty")
+        self.assertIn("leer", text.lower())
+
+    def test_gate_status_zero_for_tool(self):
+        code, text = filament_out_hint({
+            "tool": 2,
+            "gate_status": [1, 1, 0, 1],
+            "active_filament": {"type": "PLA"},
+        })
+        self.assertEqual(code, "gate_empty")
+        self.assertIn("3", text)
+
+    def test_mmu_paused_without_active_type(self):
+        code, text = filament_out_hint({
+            "is_paused": True,
+            "active_filament": {},
+        })
+        self.assertEqual(code, "mmu_paused")
+        self.assertTrue(text)
+
+    def test_no_hint_for_plain_pause(self):
+        code, text = filament_out_hint({"tool": 0, "gate_status": [1, 1]}, "paused")
+        self.assertIsNone(code)
+        self.assertIsNone(text)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_print_queue.py

@@ -38,12 +38,16 @@ class DispatcherHelpers(unittest.TestCase):
 class ResolveSpoolTests(unittest.TestCase):
     def test_material_match(self):
         from grotti.jobs import assert_material_match
-        from grotti.errors import GrottiError
 
         file_row = {"filaments": [{"type": "PLA"}]}
         assert_material_match(file_row, [{"material_type": "PLA"}])
-        with self.assertRaises(GrottiError):
-            assert_material_match(file_row, [{"material_type": "PETG"}])
+        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__":