Add a retry when smart time succeeds but gives an outdated timestamp

stempelbot/client.py

@@ -64,7 +64,10 @@ class CoeoClient:
             return True, "Successfully stamped (unverified)!", None, None
         try:
             st = get_smarttime_client()
-            ok, msg, snap = st.verify_stamp(sent_at)
+            ok, msg, snap = st.verify_stamp(
+                sent_at,
+                retry_action=lambda: self._send_stamp_request()[0],
+            )
             combined = f"Stempel gesendet – {msg}"
             return True, combined, ok, snap
         except Exception as e:  # verification must never break the primary flow

stempelbot/smarttime_client.py

@@ -319,12 +319,16 @@ class SmartTimeClient:
         *,
         initial_delay: Optional[float] = None,
         tolerance_sec: Optional[float] = None,
+        retry_action: Optional[Callable[[], bool]] = None,
     ) -> tuple[bool, str, Optional[SmartTimeStatus]]:
         """Check SmartTime Plus once and compare its `Letzte Buchung` to
         ``expected_at`` within ``tolerance_sec`` seconds.
 
-        No retry loop – if the check misses, the caller (UI button) can
-        simply trigger another check.
+        If a dashboard snapshot was fetched successfully and contains a
+        different existing stamp, ``retry_action`` is called once and the
+        stamp is checked again.  Fetch errors and snapshots without an
+        existing stamp are intentionally not treated as proof that the
+        original stamp was not applied, so they never trigger a retry.
         """
         initial_delay = (
             initial_delay if initial_delay is not None
@@ -345,21 +349,55 @@ class SmartTimeClient:
             log.exception("SmartTime verify: error fetching status: %s", e)
             return False, f"SmartTime nicht erreichbar: {e}", None
 
-        if snap.last_stamp:
-            observed_dt = datetime.combine(expected_dt.date(), snap.last_stamp)
-            diff_sec = abs((observed_dt - expected_dt).total_seconds())
-            if diff_sec <= tolerance_sec:
-                return True, (
-                    f"Verifiziert: SmartTime zeigt Buchung um "
-                    f"{snap.last_stamp.strftime('%H:%M')} "
-                    f"(Δ {diff_sec:.0f}s)"
-                ), snap
+        def evaluate(current: SmartTimeStatus) -> tuple[bool, str]:
+            if current.last_stamp:
+                observed_dt = datetime.combine(expected_dt.date(), current.last_stamp)
+                diff_sec = abs((observed_dt - expected_dt).total_seconds())
+                if diff_sec <= tolerance_sec:
+                    return True, (
+                        f"Verifiziert: SmartTime zeigt Buchung um "
+                        f"{current.last_stamp.strftime('%H:%M')} "
+                        f"(Δ {diff_sec:.0f}s)"
+                    )
 
-        obs = snap.last_stamp.strftime("%H:%M") if snap.last_stamp else "—"
-        return False, (
-            f"Nicht verifiziert: erwartet {expected_dt.strftime('%H:%M')}, "
-            f"SmartTime letzte Buchung: {obs}"
-        ), snap
+            obs = current.last_stamp.strftime("%H:%M") if current.last_stamp else "—"
+            return False, (
+                f"Nicht verifiziert: erwartet {expected_dt.strftime('%H:%M')}, "
+                f"SmartTime letzte Buchung: {obs}"
+            )
+
+        verified, message = evaluate(snap)
+        if verified:
+            return True, message, snap
+
+        # A parsed, existing but mismatching stamp is the only state strong
+        # enough to justify sending a duplicate request.  In particular, an
+        # empty last-stamp field may be a scraper/backend issue, not proof of
+        # a missing stamp.
+        if retry_action is None or snap.last_stamp is None:
+            return False, message, snap
+
+        log.warning("SmartTime stamp mismatch; sending one retry")
+        try:
+            retry_sent = retry_action()
+        except Exception as e:
+            log.exception("SmartTime stamp retry failed: %s", e)
+            return False, f"{message}; Retry fehlgeschlagen: {e}", snap
+        if not retry_sent:
+            return False, f"{message}; Retry konnte nicht gesendet werden", snap
+
+        if initial_delay > 0:
+            _time.sleep(initial_delay)
+        try:
+            retry_snap = self.get_status()
+        except Exception as e:
+            log.exception("SmartTime verify after retry: error fetching status: %s", e)
+            return False, f"{message}; Retry gesendet, Verifikation fehlgeschlagen: {e}", snap
+
+        retry_verified, retry_message = evaluate(retry_snap)
+        if retry_verified:
+            return True, f"{retry_message} (nach einmaligem Retry)", retry_snap
+        return False, f"{retry_message} (auch nach einmaligem Retry)", retry_snap
 
     def close(self) -> None:
         with self._worker_lock:

tests/test_smarttime_client.py

@@ -126,6 +126,54 @@ class SmartTimeVerificationTests(unittest.TestCase):
         self.assertIs(returned_snapshot, snapshot)
         self.assertIn("Nicht verifiziert", message)
 
+    def test_verification_retries_once_on_confident_mismatch(self) -> None:
+        client = smarttime.SmartTimeClient()
+        snapshots = [
+            smarttime.SmartTimeStatus(time(9, 0), "Anwesend", datetime.now()),
+            smarttime.SmartTimeStatus(time(10, 0), "Anwesend", datetime.now()),
+        ]
+        retry_calls = 0
+
+        def get_status():
+            return snapshots.pop(0)
+
+        def retry_action() -> bool:
+            nonlocal retry_calls
+            retry_calls += 1
+            return True
+
+        client.get_status = get_status
+        ok, message, returned_snapshot = client.verify_stamp(
+            datetime(2026, 8, 25, 10, 0),
+            initial_delay=0,
+            retry_action=retry_action,
+        )
+
+        self.assertTrue(ok)
+        self.assertEqual(retry_calls, 1)
+        self.assertEqual(returned_snapshot.last_stamp, time(10, 0))
+        self.assertIn("Retry", 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_stopped_worker_is_replaced_for_next_explicit_check(self) -> None:
         expected = smarttime.SmartTimeStatus(time(11, 30), "Anwesend", datetime.now())
 

tests/test_weekly_report.py

@@ -0,0 +1,239 @@
+import os
+import sys
+import tempfile
+import unittest
+from datetime import date
+from io import BytesIO
+from pathlib import Path
+from types import SimpleNamespace
+
+from openpyxl import load_workbook
+
+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")
+
+from excel_export import (  # noqa: E402
+    ExcelExportError,
+    export_weekly_workbook,
+    load_template_options,
+    save_workbook,
+)
+from jira_client import JiraActivity, JiraClient, JiraError, _body_preview  # noqa: E402
+from weekly_report import calculate_week, iso_week_bounds  # noqa: E402
+
+TEMPLATE = PROJECT_DIR / "TimeTracking Activation_Firstname_Lastname_CW_Version 15.xlsx"
+
+
+class FakeResponse:
+    def __init__(self, payload, status_code=200):
+        self.payload = payload
+        self.status_code = status_code
+        self.headers = {}
+        self.request = None
+
+    def raise_for_status(self):
+        if self.status_code >= 400:
+            import requests
+            raise requests.HTTPError(response=self)
+
+    def json(self):
+        return self.payload
+
+    @property
+    def text(self):
+        return str(self.payload)
+
+    @property
+    def reason(self):
+        return "Unauthorized" if self.status_code == 401 else "OK"
+
+
+class FakeSession:
+    def __init__(self, responses):
+        self.responses = iter(responses)
+        self.headers = {}
+        self.auth = None
+        self.calls = []
+
+    def get(self, url, **kwargs):
+        self.calls.append((url, kwargs))
+        response = next(self.responses)
+        return response if isinstance(response, FakeResponse) else FakeResponse(response)
+
+
+class WeeklyReportTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.options = load_template_options(TEMPLATE)
+
+    def test_iso_week_can_cross_year_boundary(self):
+        self.assertEqual(
+            iso_week_bounds(2026, 1),
+            (date(2025, 12, 29), date(2026, 1, 4)),
+        )
+
+    def test_massive_html_error_body_is_abbreviated(self):
+        body = "<html>" + ("route-data" * 2000) + "</html>"
+        preview = _body_preview(body)
+        self.assertLess(len(preview), len(body))
+        self.assertIn("characters omitted", preview)
+        self.assertIn(f"total response size {len(body)}", preview)
+        self.assertTrue(preview.startswith("<html>"))
+        self.assertTrue(preview.endswith("</html>"))
+
+    def test_daily_net_time_applies_legal_break_and_accounts_for_minutes(self):
+        day = date(2026, 8, 24)
+        rows = calculate_week(
+            2026,
+            35,
+            {day: ["08:00", "12:00", "12:15", "16:15"]},
+            [JiraActivity(day, "ABC-1", "Implement export", "Other", "change", "code updated", "url")],
+            self.options,
+        )
+        self.assertEqual(len(rows), 1)
+        self.assertEqual(rows[0].minutes, 465)
+        self.assertEqual(rows[0].issue_keys, "ABC-1")
+        self.assertEqual(rows[0].project, "Other")
+
+    def test_incomplete_historical_stamp_is_not_guessed(self):
+        day = date(2026, 8, 25)
+        rows = calculate_week(2026, 35, {day: ["08:00"]}, [], self.options)
+        self.assertEqual(rows[0].minutes, 0)
+        self.assertIn("Incomplete", rows[0].warning)
+
+    def test_jira_pagination_and_authored_activity_filtering(self):
+        issue = {
+            "key": "ABC-7",
+            "fields": {
+                "summary": "Build report",
+                "project": {"name": "Other"},
+                "assignee": {"accountId": "me"},
+                "creator": {"accountId": "someone"},
+                "updated": "2026-08-24T15:00:00+00:00",
+                "comment": {
+                    "comments": [
+                        {"author": {"accountId": "me"}, "created": "2026-08-24T12:00:00+00:00"},
+                        {"author": {"accountId": "other"}, "created": "2026-08-24T13:00:00+00:00"},
+                    ]
+                },
+                "worklog": {"worklogs": []},
+            },
+            "changelog": {"histories": []},
+        }
+        session = FakeSession([
+            {"accountId": "me", "displayName": "Tester"},
+            {"issues": [issue], "nextPageToken": "next"},
+            {"issues": [], "isLast": True},
+        ])
+        client = JiraClient("https://jira.example", "mail", "token", session=session)
+        activities = client.get_week_activity(date(2026, 8, 24), date(2026, 8, 30))
+        self.assertEqual([(item.issue_key, item.activity_type) for item in activities], [("ABC-7", "comment")])
+        self.assertEqual(session.calls[2][1]["params"]["nextPageToken"], "next")
+
+    def test_jira_http_failure_logs_response_without_token(self):
+        response = FakeResponse(
+            {"errorMessages": ["Client must be authenticated"]}, status_code=401
+        )
+        response.request = SimpleNamespace(
+            method="GET",
+            url="https://jira.example/rest/api/3/myself",
+            headers={"Accept": "application/json", "Authorization": "Basic super-secret-token"},
+        )
+        response.headers = {
+            "WWW-Authenticate": 'Bearer realm="Atlassian"',
+            "Set-Cookie": "secret-session-cookie",
+        }
+        session = FakeSession([
+            response
+        ])
+        client = JiraClient(
+            "https://jira.example", "mail@example.com", "super-secret-token", session=session
+        )
+        with self.assertLogs("jira_client", level="ERROR") as captured:
+            with self.assertRaisesRegex(JiraError, "HTTP 401"):
+                client._get("/rest/api/3/myself")
+        output = "\n".join(captured.output)
+        self.assertIn("Client must be authenticated", output)
+        self.assertIn("401 Unauthorized", output)
+        self.assertIn("GET https://jira.example/rest/api/3/myself", output)
+        self.assertIn("WWW-Authenticate", output)
+        self.assertIn("<redacted>", output)
+        self.assertNotIn("super-secret-token", output)
+        self.assertNotIn("secret-session-cookie", output)
+
+    def test_scoped_token_discovers_cloud_id_and_retries_gateway(self):
+        session = FakeSession([
+            FakeResponse("<html>Jira application shell</html>", status_code=401),
+            {"cloudId": "cloud-123"},
+            {"accountId": "me", "displayName": "Tester"},
+        ])
+        client = JiraClient(
+            "https://company.atlassian.net/jira/your-work",
+            "mail@example.com",
+            "scoped-token",
+            session=session,
+        )
+        with self.assertNoLogs("jira_client", level="INFO"):
+            payload = client._get("/rest/api/3/myself")
+        self.assertEqual(payload["accountId"], "me")
+        self.assertEqual(
+            [call[0] for call in session.calls],
+            [
+                "https://company.atlassian.net/rest/api/3/myself",
+                "https://company.atlassian.net/_edge/tenant_info",
+                "https://api.atlassian.com/ex/jira/cloud-123/rest/api/3/myself",
+            ],
+        )
+        self.assertEqual(client.site_url, "https://company.atlassian.net")
+
+    def test_export_preserves_template_and_exact_total(self):
+        first = date(2026, 8, 24)
+        rows = calculate_week(
+            2026,
+            35,
+            {
+                first: ["08:00", "12:00"],
+                date(2026, 8, 25): ["09:00", "10:01"],
+            },
+            [],
+            self.options,
+        )
+        data = export_weekly_workbook(TEMPLATE, "Test Person", 2026, 35, rows)
+        workbook = load_workbook(BytesIO(data), data_only=False)
+        sheet = workbook["weekly time tracking"]
+        self.assertEqual(workbook.sheetnames, ["weekly time tracking", "Dropdowns"])
+        self.assertEqual(sheet["B3"].value, "Test Person")
+        self.assertEqual(sheet["B6"].value, "KW 35")
+        self.assertEqual(sheet["C6"].value, "24.08. bis 30.08.")
+        self.assertAlmostEqual(sheet["H6"].value + sheet["H7"].value, 301 / 60)
+        self.assertEqual(sheet["H6"].number_format, "0.00")
+        self.assertEqual(len(sheet.data_validations.dataValidation), 5)
+        self.assertTrue(workbook.calculation.fullCalcOnLoad)
+        self.assertGreater(len(sheet.merged_cells.ranges), 0)
+        workbook.close()
+
+    def test_save_workbook_creates_export_directory(self):
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            export_directory = Path(temporary_directory) / "nested" / "exports"
+            destination = save_workbook(b"workbook-data", export_directory, "week.xlsx")
+            self.assertEqual(destination, (export_directory / "week.xlsx").resolve())
+            self.assertEqual(destination.read_bytes(), b"workbook-data")
+
+    def test_save_workbook_rejects_filename_with_directories(self):
+        with tempfile.TemporaryDirectory() as temporary_directory:
+            with self.assertRaisesRegex(ExcelExportError, "filename is invalid"):
+                save_workbook(b"workbook-data", temporary_directory, "../week.xlsx")
+
+
+if __name__ == "__main__":
+    unittest.main()
+
+
+
+