194 lines
8.5 KiB
Python
194 lines
8.5 KiB
Python
import os
|
|
import sys
|
|
import unittest
|
|
from datetime import datetime, time
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, call, patch
|
|
|
|
PROJECT_DIR = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(PROJECT_DIR / "stempelbot"))
|
|
os.environ.setdefault("USERNAME", "test-user")
|
|
os.environ.setdefault("PASSWORD", "test-password")
|
|
os.environ.setdefault("SMART_TIME_PASSWORD", "smarttime-password")
|
|
os.environ.setdefault("AUTOSTAMP_FILE", "autostamps.json")
|
|
os.environ.setdefault("STAMPHISTORY_FILE", "timestamp_history.json")
|
|
os.environ.setdefault("LOG_LEVEL", "CRITICAL")
|
|
|
|
import smarttime_client as smarttime # noqa: E402
|
|
import client as coeo # noqa: E402
|
|
|
|
|
|
class SmartTimeVerificationTests(unittest.TestCase):
|
|
def test_verification_checks_status_once(self) -> None:
|
|
client = smarttime.SmartTimeClient()
|
|
snapshot = smarttime.SmartTimeStatus(time(9, 0), "Anwesend", datetime.now())
|
|
calls = 0
|
|
|
|
def get_status():
|
|
nonlocal calls
|
|
calls += 1
|
|
return snapshot
|
|
|
|
client.get_status = get_status
|
|
ok, message, returned_snapshot = client.verify_stamp(
|
|
datetime(2026, 8, 25, 10, 0), initial_delay=0
|
|
)
|
|
|
|
self.assertFalse(ok)
|
|
self.assertEqual(calls, 1)
|
|
self.assertIs(returned_snapshot, snapshot)
|
|
self.assertIn("Nicht verifiziert", message)
|
|
|
|
def test_verification_does_not_retry_without_existing_stamp(self) -> None:
|
|
client = smarttime.SmartTimeClient()
|
|
snapshot = smarttime.SmartTimeStatus(None, "Anwesend", datetime.now())
|
|
retry_calls = 0
|
|
|
|
def retry_action() -> bool:
|
|
nonlocal retry_calls
|
|
retry_calls += 1
|
|
return True
|
|
|
|
client.get_status = lambda: snapshot
|
|
ok, _, _ = client.verify_stamp(
|
|
datetime(2026, 8, 25, 10, 0),
|
|
initial_delay=0,
|
|
retry_action=retry_action,
|
|
)
|
|
|
|
self.assertFalse(ok)
|
|
self.assertEqual(retry_calls, 0)
|
|
|
|
def test_failed_read_does_not_poison_next_check(self) -> None:
|
|
client = smarttime.SmartTimeClient()
|
|
payload = {"last_stamp": "09:29", "status_text": "Anwesend",
|
|
"query_time": "2026-09-25T10:00:00"}
|
|
with patch.object(smarttime, "run_browser_job", side_effect=[RuntimeError("hung"), payload]):
|
|
with self.assertRaisesRegex(RuntimeError, "hung"):
|
|
client.get_status()
|
|
result = client.get_status()
|
|
self.assertEqual(result.last_stamp, time(9, 29))
|
|
self.assertTrue(result.is_present)
|
|
|
|
|
|
class StampRetryTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.now = datetime(2026, 9, 25, 10, 0)
|
|
self.missing = smarttime.SmartTimeStatus(time(9, 0), "Anwesend", self.now)
|
|
self.applied = smarttime.SmartTimeStatus(time(10, 0), "Abwesend", self.now)
|
|
self.smarttime = smarttime.SmartTimeClient()
|
|
self.coeo = coeo.CoeoClient("test-user", "test-password")
|
|
self.addCleanup(self.coeo.session.close)
|
|
self.coeo._login = Mock(return_value=(True, 200))
|
|
self.coeo._send_stamp_request = Mock(return_value=(True, 200))
|
|
self.progress = Mock()
|
|
for module in (smarttime, coeo):
|
|
clock = self.enterContext(patch.object(module, "datetime", wraps=datetime))
|
|
clock.now.return_value = self.now
|
|
if module is smarttime:
|
|
self.clock = clock
|
|
self.sleep = self.enterContext(patch.object(smarttime._time, "sleep"))
|
|
self.enterContext(patch.object(smarttime.settings, "SMART_TIME_VERIFY_INITIAL_DELAY", 0))
|
|
self.enterContext(patch.object(coeo, "get_smarttime_client", return_value=self.smarttime))
|
|
|
|
def run_stamp(self, snapshots):
|
|
self.smarttime.get_status = Mock(side_effect=snapshots)
|
|
return self.coeo.execute_stamp(on_progress=self.progress)
|
|
|
|
def test_stops_after_five_total_attempts_with_five_second_waits(self) -> None:
|
|
events = Mock()
|
|
events.attach_mock(self.sleep, "sleep")
|
|
events.attach_mock(self.coeo._send_stamp_request, "send")
|
|
ok, message, verified, snap = self.run_stamp([self.missing] * 9)
|
|
|
|
self.assertTrue(ok)
|
|
self.assertFalse(verified)
|
|
self.assertIs(snap, self.missing)
|
|
self.assertIn("Versuch 5/5", message)
|
|
self.assertEqual(self.coeo._send_stamp_request.call_count, 5)
|
|
self.assertEqual(events.mock_calls, [call.send()] + [call.sleep(5), call.send()] * 4)
|
|
messages = [args.args[0] for args in self.progress.call_args_list]
|
|
for attempt in range(1, 6):
|
|
self.assertTrue(any(f"Versuch {attempt}/5" in message for message in messages))
|
|
self.assertEqual(sum("in 5s" in message for message in messages), 4)
|
|
|
|
def test_stops_as_soon_as_retry_is_verified(self) -> None:
|
|
_, message, verified, _ = self.run_stamp([self.missing, self.missing, self.applied])
|
|
self.assertTrue(verified)
|
|
self.assertIn("Versuch 2/5", message)
|
|
self.assertEqual(self.coeo._send_stamp_request.call_count, 2)
|
|
self.sleep.assert_called_once_with(5)
|
|
|
|
def test_success_on_fifth_attempt(self) -> None:
|
|
_, _, verified, _ = self.run_stamp([self.missing] * 8 + [self.applied])
|
|
self.assertTrue(verified)
|
|
self.assertEqual(self.coeo._send_stamp_request.call_count, 5)
|
|
|
|
def test_delayed_original_stamp_prevents_resend(self) -> None:
|
|
_, message, verified, _ = self.run_stamp([self.missing, self.applied])
|
|
self.assertTrue(verified)
|
|
self.assertIn("Versuch 1/5", message)
|
|
self.coeo._send_stamp_request.assert_called_once()
|
|
self.sleep.assert_called_once_with(5)
|
|
|
|
def test_initial_success_does_not_wait_or_retry(self) -> None:
|
|
_, _, verified, _ = self.run_stamp([self.applied])
|
|
self.assertTrue(verified)
|
|
self.coeo._send_stamp_request.assert_called_once()
|
|
self.sleep.assert_not_called()
|
|
|
|
def test_verification_errors_stop_at_every_check(self) -> None:
|
|
for preceding, sends in (([], 1), ([self.missing], 1), ([self.missing] * 2, 2)):
|
|
with self.subTest(preceding=len(preceding)):
|
|
self.coeo._send_stamp_request.reset_mock()
|
|
_, _, verified, snap = self.run_stamp(preceding + [RuntimeError("offline")])
|
|
self.assertFalse(verified)
|
|
self.assertIsNone(snap)
|
|
self.assertEqual(self.coeo._send_stamp_request.call_count, sends)
|
|
|
|
def test_ambiguous_snapshots_do_not_retry(self) -> None:
|
|
for snapshot in (
|
|
smarttime.SmartTimeStatus(None, "Anwesend", self.now),
|
|
smarttime.SmartTimeStatus(time(9, 0), None, self.now),
|
|
smarttime.SmartTimeStatus(time(10, 5), "Anwesend", self.now),
|
|
smarttime.SmartTimeStatus(time(9, 0), "Anwesend", datetime(2026, 9, 26, 10)),
|
|
smarttime.SmartTimeStatus(time(9, 0), "Anwesend", datetime(2026, 9, 25, 9)),
|
|
):
|
|
with self.subTest(snapshot=snapshot):
|
|
self.coeo._send_stamp_request.reset_mock()
|
|
_, _, verified, _ = self.run_stamp([snapshot])
|
|
self.assertFalse(verified)
|
|
self.coeo._send_stamp_request.assert_called_once()
|
|
self.sleep.assert_not_called()
|
|
|
|
def test_retry_failure_does_not_trigger_another_send(self) -> None:
|
|
for failure in ((False, "timeout"), RuntimeError("timeout")):
|
|
with self.subTest(failure=failure):
|
|
self.coeo._send_stamp_request.reset_mock()
|
|
self.coeo._send_stamp_request.side_effect = [(True, 200), failure]
|
|
_, message, verified, _ = self.run_stamp([self.missing, self.missing])
|
|
self.assertFalse(verified)
|
|
self.assertIn("Versuch 2/5", message)
|
|
self.assertEqual(self.coeo._send_stamp_request.call_count, 2)
|
|
|
|
def test_retry_uses_its_own_send_time_after_slow_verification(self) -> None:
|
|
self.clock.now.return_value = datetime(2026, 9, 25, 10, 5)
|
|
later_stamp = smarttime.SmartTimeStatus(time(10, 5), "Abwesend", datetime(2026, 9, 25, 10, 5))
|
|
_, _, verified, _ = self.run_stamp([self.missing, self.missing, later_stamp])
|
|
self.assertTrue(verified)
|
|
self.assertEqual(self.coeo._send_stamp_request.call_count, 2)
|
|
|
|
def test_no_verification_means_no_retries(self) -> None:
|
|
self.smarttime.get_status = Mock()
|
|
_, _, verified, _ = self.coeo.execute_stamp(verify=False)
|
|
self.assertIsNone(verified)
|
|
self.coeo._send_stamp_request.assert_called_once()
|
|
self.smarttime.get_status.assert_not_called()
|
|
self.sleep.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|
|
|