108 lines
4.0 KiB
Python
108 lines
4.0 KiB
Python
"""Leere History-Hülle, Resume-Beweis und Chargen-Tag."""
|
|
|
|
import unittest
|
|
from datetime import date, datetime, timedelta, timezone
|
|
from unittest.mock import MagicMock
|
|
|
|
from grotti.jobs import (
|
|
history_in_order,
|
|
history_shell_is_empty,
|
|
is_filament_resume,
|
|
)
|
|
from grotti.store import chargen_day, history_job_retired
|
|
|
|
|
|
class HistoryShellTests(unittest.TestCase):
|
|
def test_pause_shell_is_empty(self):
|
|
self.assertTrue(history_shell_is_empty({"duration_s": 0, "ended_at": None}))
|
|
|
|
def test_recorded_cancel_is_not_a_shell(self):
|
|
self.assertFalse(history_shell_is_empty({
|
|
"duration_s": 653,
|
|
"ended_at": datetime(2026, 10, 18, 13, 18, tzinfo=timezone.utc),
|
|
}))
|
|
|
|
def test_missing_row_is_not_a_shell(self):
|
|
self.assertFalse(history_shell_is_empty(None))
|
|
|
|
|
|
class FilamentResumeProofTests(unittest.TestCase):
|
|
def test_duration_continued_with_tiny_gap(self):
|
|
# 2026-10-25 Kobra 3 A1: 000040 → 000041, 0,04 s Lücke, gleiche print_duration
|
|
end = datetime(2026, 10, 25, 5, 48, 27, 569483, tzinfo=timezone.utc)
|
|
start = end + timedelta(milliseconds=40)
|
|
self.assertTrue(is_filament_resume(
|
|
end, start, prev_duration_s=11711, new_duration_s=11711
|
|
))
|
|
|
|
def test_fresh_restart_after_cancel_is_not_resume(self):
|
|
# Lauf 27941: 8 s abgebrochen, 45 s später Neustart mit Dauer ~0
|
|
end = datetime(2026, 10, 24, 11, 46, 8, tzinfo=timezone.utc)
|
|
start = end + timedelta(seconds=45)
|
|
self.assertFalse(is_filament_resume(
|
|
end, start, prev_duration_s=8, new_duration_s=5
|
|
))
|
|
|
|
def test_two_arg_call_without_proof_is_false(self):
|
|
end = datetime(2026, 10, 21, 19, 32, 20, tzinfo=timezone.utc)
|
|
start = end + timedelta(milliseconds=40)
|
|
self.assertFalse(is_filament_resume(end, start))
|
|
|
|
def test_watcher_signal_allows_longer_gap(self):
|
|
end = datetime(2026, 10, 25, 1, 0, 0, tzinfo=timezone.utc)
|
|
start = end + timedelta(hours=4)
|
|
self.assertTrue(is_filament_resume(
|
|
end, start, prev_duration_s=1000, new_duration_s=1000, watcher_resume=True
|
|
))
|
|
|
|
def test_duration_reset_is_not_resume(self):
|
|
end = datetime(2026, 10, 25, 5, 48, 27, tzinfo=timezone.utc)
|
|
start = end + timedelta(milliseconds=40)
|
|
self.assertFalse(is_filament_resume(
|
|
end, start, prev_duration_s=11711, new_duration_s=12
|
|
))
|
|
|
|
|
|
class HistoryOrderTests(unittest.TestCase):
|
|
def test_cancelled_part_before_resume(self):
|
|
items = [
|
|
{"job_id": "000041", "start_time": 1792907307.607},
|
|
{"job_id": "000040", "start_time": 1792879062.392},
|
|
]
|
|
self.assertEqual([j["job_id"] for j in history_in_order(items)], ["000040", "000041"])
|
|
|
|
def test_missing_start_first(self):
|
|
items = [{"job_id": "b", "start_time": 5}, {"job_id": "a"}]
|
|
self.assertEqual([j["job_id"] for j in history_in_order(items)], ["a", "b"])
|
|
|
|
|
|
class HistoryJobRetiredTests(unittest.TestCase):
|
|
def _conn(self, row):
|
|
conn = MagicMock()
|
|
conn.execute.return_value.fetchone.return_value = row
|
|
return conn
|
|
|
|
def test_folded_predecessor_is_retired(self):
|
|
conn = self._conn({"moonraker_job_id": "000041"})
|
|
self.assertTrue(history_job_retired(conn, "h", "000040"))
|
|
|
|
def test_own_job_is_not_retired(self):
|
|
conn = self._conn({"moonraker_job_id": "000041"})
|
|
self.assertFalse(history_job_retired(conn, "h", "000041"))
|
|
|
|
def test_unknown_job_is_not_retired(self):
|
|
self.assertFalse(history_job_retired(self._conn(None), "h", "000042"))
|
|
|
|
|
|
class ChargenDayTests(unittest.TestCase):
|
|
def test_old_print_keeps_its_day(self):
|
|
started = datetime(2026, 10, 18, 13, 4, 16, tzinfo=timezone.utc)
|
|
self.assertEqual(chargen_day(started, today=date(2026, 10, 24)), date(2026, 10, 18))
|
|
|
|
def test_without_start_uses_today(self):
|
|
self.assertEqual(chargen_day(None, today=date(2026, 10, 24)), date(2026, 10, 24))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|