Grotti 3.0 Phase 1: Migrationen, Pool, Queue- und Abzug-Fixes.
Nummerierte SQL-Migrationen, psycopg-Pool, waitress, hängende Queue-Jobs, sichere History-Abzüge, Chargen-Locks, Upload-/Host-Validierung, ProxyFix.
This commit is contained in:
1 parent
9e22cebb81
commit
0236278929
19 files changed
+986
-81
No files matched your search
@@ -0,0 +1,65 @@
|
||||
"""Unit-Tests für Spulenwahl und Steuer-Parameter-Aliase."""
|
||||
|
||||
import unittest
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from grotti import jobs
|
||||
from grotti import moonraker
|
||||
|
||||
|
||||
class ChooseSpoolTests(unittest.TestCase):
|
||||
def test_rejects_spool_created_after_print_end(self):
|
||||
end = datetime(2026, 10, 1, 12, 0, tzinfo=timezone.utc)
|
||||
old = {
|
||||
"id": 1,
|
||||
"material_type": "PLA",
|
||||
"created_at": end - timedelta(days=2),
|
||||
}
|
||||
new = {
|
||||
"id": 2,
|
||||
"material_type": "PLA",
|
||||
"created_at": end + timedelta(hours=1),
|
||||
}
|
||||
conn = MagicMock()
|
||||
with patch("grotti.jobs.store.spools_for_printer", return_value=[new]):
|
||||
with patch("grotti.jobs.store.get_spool_row", return_value=new):
|
||||
self.assertIsNone(jobs.choose_spool(conn, "h", "PLA", 2, ended_at=end))
|
||||
with patch("grotti.jobs.store.spools_for_printer", return_value=[old]):
|
||||
self.assertEqual(jobs.choose_spool(conn, "h", "PLA", None, ended_at=end)["id"], 1)
|
||||
|
||||
def test_spool_valid_for_end_naive(self):
|
||||
end = datetime(2026, 10, 1, 12, 0, tzinfo=timezone.utc)
|
||||
spool = {"created_at": datetime(2026, 9, 1, 12, 0)}
|
||||
self.assertTrue(jobs._spool_valid_for_end(spool, end))
|
||||
|
||||
|
||||
class SteuernAliasTests(unittest.TestCase):
|
||||
@patch("grotti.moonraker.gcode_script", return_value={"ok": True})
|
||||
def test_value_alias_for_temp(self, gcode):
|
||||
moonraker.steuern("10.0.0.1", "nozzle", {"value": 210})
|
||||
script = gcode.call_args[0][1]
|
||||
self.assertIn("TARGET=210.0", script)
|
||||
|
||||
@patch("grotti.moonraker.gcode_script", return_value={"ok": True})
|
||||
def test_percent_alias_for_speed(self, gcode):
|
||||
moonraker.steuern("10.0.0.1", "speed", {"value": 120})
|
||||
script = gcode.call_args[0][1]
|
||||
self.assertIn("M220 S120", script)
|
||||
|
||||
|
||||
class HostValidationTests(unittest.TestCase):
|
||||
def test_validate_printer_host(self):
|
||||
from web.app import validate_printer_host
|
||||
from grotti.errors import GrottiError
|
||||
|
||||
self.assertEqual(validate_printer_host("10.10.1.60"), "10.10.1.60")
|
||||
self.assertEqual(validate_printer_host("kobra.local:80"), "kobra.local:80")
|
||||
with self.assertRaises(GrottiError):
|
||||
validate_printer_host("http://evil")
|
||||
with self.assertRaises(GrottiError):
|
||||
validate_printer_host("../etc/passwd")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user