Marmelator/stempelbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update for new smart time ui

Browse Source 
Up to 5 retries for stamps
This commit is contained in:
Kilian Schuettler
2026-09-25 18:10:05 +02:00
parent faad4b0849
commit b8f8830064
8 changed files with 633 additions and 487 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -3,3 +3,4 @@ __pycache__/
*.py[cod]
autostamps.json
timestamp_history.json
/.local/
README.md
+56 -8
View File
@@ -33,6 +33,9 @@ stempelbot/
Stempelbot.py # Streamlit UI
cli.py # `stempelbot` entry point → runs `streamlit run Stempelbot.py`
client.py # CoeoClient: login + stamp HTTP calls
smarttime_client.py # Stamp verification and guarded resends
smarttime_browser.py # Isolated browser jobs with timeout/recovery
smarttime_ui.py # SmartTime 9 login, dashboard and monthly DOM reader
autostamp.py # AutoStampService: background scheduler
stamp_history.py # Local JSON stamp history
time_calculator.py # Work/break time calculations
@@ -41,6 +44,9 @@ stempelbot/
excel_export.py # Template-preserving XLSX export
pages/ # Weekly Jira & Excel Streamlit page
settings.py # Pydantic settings loaded from .env
scripts/
smarttime_cli.py # Read-only status / complete month JSON export
smarttime_probe.py # Interactive browser exploration and snapshots
```
## Installation
@@ -251,15 +257,57 @@ then uses a headless Chromium (Playwright) to log in to SmartTime Plus
`SMART_TIME_VERIFY_TOLERANCE_SEC` after `SMART_TIME_VERIFY_INITIAL_DELAY` seconds, the
UI flags the stamp as **unverified**.
Each verification performs one SmartTime check. That check waits up to 30 seconds
for the asynchronously rendered login form or complete dashboard values; it does
not retry a failed verification.
SmartTime **9.0** uses Vaadin Flow. The reader logs in using `username` / `password`,
opens the root dashboard through its sidebar entry, and reads the personal
**Letzte Buchung** card. Presence comes from that card's building icon, matched to
the dashboard's status-color legend. The old GWT hash route and global status-text
matching are no longer used.
SmartTime Plus is a GWT app (all traffic is `text/x-gwt-rpc` with per-build
strong-name hashes), so we drive the DOM instead of trying to speak the RPC
protocol. A persistent Chromium profile in `~/.stempelbot/smarttime-profile`
keeps session cookies between runs so subsequent verifications skip the login
step.
Each read runs in a fresh headless browser subprocess. A stuck job is terminated
with its browser children and retried, up to **three read attempts**, with two
seconds between attempts. The hard limit per attempt is 75 seconds for dashboard
reads and 180 seconds for a monthly export. A permanent outage returns an error;
it cannot leave the shared checker stuck on an old worker. Recognized login
rejections stop immediately. Reads no longer use the old persistent browser profile.
Stamping allows **five total stamp attempts**, with a five-second wait before each
resend. A fresh SmartTime check after the wait can catch a delayed booking and
cancel the resend. Only a complete, same-day snapshot with an older booking allows
a resend; unreadable, missing, future or ambiguous results stop the workflow.
Each resend is verified against its own send time. The manual-stamp spinner shows
attempts and waits. Browser recovery attempts only read data; they never send stamps.
### Standalone SmartTime CLI
Run from the repository root with the same `.env` used by the app:
```powershell
# Check login and retrieve the current last booking and presence.
poetry run python scripts/smarttime_cli.py
# Export every day of a selected month to a new JSON file.
poetry run python scripts/smarttime_cli.py --month 2026-09 --output .local/smarttime/2026-09.json
# Also capture screenshots and sanitized HTML, including Vaadin shadow DOM.
poetry run python scripts/smarttime_cli.py --month 2026-09 --artifacts .local/smarttime/inspection
```
Alternatively use `.venv\Scripts\python.exe` in place of `poetry run python`.
The CLI is read-only: it never calls the coeo stamping endpoint or modifies local
stamp history. Existing output files are refused. Diagnostic files under `.local/`
are git-ignored; they contain personal time data, so keep them local.
The monthly reader navigates to **Monatsübersicht** (`/MonthlyOV`), selects the
month/year, and scrolls the virtualized table to collect all calendar dates. It
rejects missing dates, duplicate dates, changed columns or inconsistent repeated
rows instead of exporting a partial month. JSON contains the ordered booking times,
Soll/Ist, daily/cumulative balances, breaks, absence labels and original cell text.
Unpaired bookings are marked `incomplete_pair`; no missing departure is invented.
SmartTime's displayed totals are preserved as-is, including provisional current-day
and future-day values. This export is not yet wired into weekly reports.
Implementation notes and the observed SmartTime 9 selectors are in
[`docs/smarttime-9.md`](docs/smarttime-9.md).
### SmartTime Abgleich panel
Below the daily timeline there's a **"SmartTime Abgleich"** panel with a
stempelbot/Stempelbot.py
+11 -2
View File
@@ -1,4 +1,5 @@
from datetime import datetime
from contextlib import ExitStack
from time import sleep
import streamlit as st
@@ -131,9 +132,17 @@ with st.container(border=True):
# Full width button for primary action
if st.button("Jetzt Stempeln", type="primary", use_container_width=True):
with st.spinner("Stempel wird übermittelt und mit SmartTime abgeglichen..."):
spinner_slot = st.empty()
with ExitStack() as spinner_stack:
def update_stamp_progress(message: str) -> None:
spinner_stack.close()
with spinner_slot.container():
spinner_stack.enter_context(st.spinner(message))
client = get_coeo_client()
ok, msg, verified, snap = client.execute_stamp()
ok, msg, verified, snap = client.execute_stamp(on_progress=update_stamp_progress)
spinner_stack.close()
spinner_slot.empty()
if ok:
if verified is True:
# Only record locally once SmartTime confirmed the stamp.
stempelbot/client.py
+8 -2
View File
@@ -1,10 +1,11 @@
from functools import lru_cache
from datetime import datetime
from typing import Callable, Optional
import requests
from settings import settings
from smarttime_client import get_smarttime_client, SmartTimeStatus
from smarttime_client import get_smarttime_client, STAMP_MAX_ATTEMPTS
class CoeoClient:
@@ -39,7 +40,9 @@ class CoeoClient:
except requests.RequestException as e:
return False, str(e)
def execute_stamp(self, verify: bool = True):
def execute_stamp(
self, verify: bool = True, *, on_progress: Optional[Callable[[str], None]] = None
):
"""Public method to run the full workflow.
Returns a 4-tuple ``(ok, message, verified, smarttime_status)``:
@@ -49,6 +52,8 @@ class CoeoClient:
smarttime_status – SmartTimeStatus snapshot or None
"""
# 1. Login
if on_progress is not None:
on_progress(f"Versuch 1/{STAMP_MAX_ATTEMPTS}: Anmeldung und Stempelübermittlung …")
login_success, login_msg = self._login()
if not login_success:
return False, f"Login failed: HTTP {login_msg}", None, None
@@ -67,6 +72,7 @@ class CoeoClient:
ok, msg, snap = st.verify_stamp(
sent_at,
retry_action=lambda: self._send_stamp_request()[0],
on_progress=on_progress,
)
combined = f"Stempel gesendet – {msg}"
return True, combined, ok, snap
stempelbot/smarttime_browser.py
+120
View File
@@ -0,0 +1,120 @@
"""Bounded, isolated Playwright reads; a hung browser cannot poison later jobs."""
from __future__ import annotations
import json
import logging
import os
import signal
import subprocess
import sys
import time
from pathlib import Path
log = logging.getLogger(__name__)
BROWSER_ATTEMPTS = 3
STATUS_TIMEOUT_SEC = 75
MONTH_TIMEOUT_SEC = 180
def stop_process_tree(process: subprocess.Popen) -> None:
"""Only terminate this job and the Chromium/Node children it launched."""
if process.poll() is not None:
return
if sys.platform == "win32":
subprocess.run(
["taskkill", "/PID", str(process.pid), "/T", "/F"],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10,
creationflags=subprocess.CREATE_NO_WINDOW,
)
else:
os.killpg(process.pid, signal.SIGKILL)
if process.poll() is None:
process.kill()
process.wait(timeout=10)
def run_browser_job(action: str = "status", *, year: int | None = None,
month: int | None = None, artifacts: Path | None = None) -> dict:
if action not in {"status", "month"}:
raise ValueError("Unknown read-only SmartTime action")
request = {"action": action, "year": year, "month": month}
timeout = MONTH_TIMEOUT_SEC if action == "month" else STATUS_TIMEOUT_SEC
error = "Unknown browser error"
for attempt in range(1, BROWSER_ATTEMPTS + 1):
if artifacts:
request["artifacts"] = str(artifacts.resolve() / f"attempt-{attempt}")
log.info("SmartTime browser read %s/%s", attempt, BROWSER_ATTEMPTS)
options = ({"creationflags": subprocess.CREATE_NO_WINDOW} if sys.platform == "win32"
else {"start_new_session": True})
process = subprocess.Popen(
[sys.executable, "-X", "utf8", str(Path(__file__).resolve()), "--worker"],
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
text=True, encoding="utf-8", **options,
)
try:
output, _ = process.communicate(json.dumps(request), timeout=timeout)
response = json.loads(output)
if process.returncode == 0 and "result" in response:
return response["result"]
error = response.get("error", "SmartTime browser failed")
if response.get("login_error"):
raise PermissionError(error)
except subprocess.TimeoutExpired:
error = f"SmartTime browser did not respond within {timeout}s"
except (json.JSONDecodeError, UnicodeError):
error = "SmartTime browser exited without a valid response"
finally:
stop_process_tree(process)
for stream in (process.stdin, process.stdout, process.stderr):
if stream:
stream.close()
log.warning("SmartTime read %s failed: %s", attempt, error)
if attempt < BROWSER_ATTEMPTS:
time.sleep(2)
raise RuntimeError(f"SmartTime unavailable after {BROWSER_ATTEMPTS} browser reads: {error}")
def worker_main() -> int:
# Imported only inside a fresh process: Windows' main-thread event loop
# supports Playwright subprocesses even when the caller is Streamlit.
from playwright.sync_api import sync_playwright
from settings import settings
from smarttime_ui import SmartTimeLoginError, capture, open_dashboard, read_dashboard, read_month
request = json.load(sys.stdin)
artifacts = Path(request["artifacts"]) if request.get("artifacts") else None
try:
with sync_playwright() as pw:
browser = pw.chromium.launch(headless=True)
context = browser.new_context(ignore_https_errors=True, locale="de-DE",
viewport={"width": 1440, "height": 1100})
page = context.new_page()
try:
open_dashboard(page, settings.SMART_TIME_URL, settings.USERNAME, settings.SMART_TIME_PASSWORD)
result = read_dashboard(page) if request["action"] == "status" else None
if artifacts:
capture(page, artifacts, "dashboard")
if request["action"] == "month":
result = read_month(page, request["year"], request["month"], artifacts)
except Exception:
if artifacts:
try:
capture(page, artifacts, "error")
except Exception:
pass
raise
finally:
browser.close()
print(json.dumps({"result": result}, ensure_ascii=True))
return 0
except Exception as exc:
message = str(exc)
for secret in (settings.SMART_TIME_PASSWORD, settings.PASSWORD):
if secret:
message = message.replace(secret, "***")
print(json.dumps({"error": message, "login_error": isinstance(exc, SmartTimeLoginError)}))
return 1
if __name__ == "__main__":
raise SystemExit(worker_main())
stempelbot/smarttime_client.py
+77 -333
View File
@@ -1,317 +1,52 @@
"""
SmartTime Plus verification client.
"""SmartTime 9 verification with bounded, isolated browser reads.
The internal SmartTime Plus web app (https://zeiterfassung/) is a GWT
application whose only transport is a serialized RPC protocol tied to a
per-build strong-name hash. Reverse-engineering that protocol is brittle,
so this client instead drives a headless Chromium via Playwright, logs in
with the SmartTime credentials, opens the dashboard, and scrapes the
"Letzte Buchung" (last booking) and status fields.
This is used to confirm that a stamp submitted through the coeo portal
actually arrived at the underlying time-tracking system.
Threading notes
---------------
Playwright's sync API spawns a Node.js subprocess via asyncio, which on
Windows requires a ``ProactorEventLoop``. Streamlit executes user scripts
on worker threads whose default loop is a ``SelectorEventLoop`` (no
subprocess support), which causes ``NotImplementedError`` when starting
Playwright directly.
To dodge this, all Playwright interaction runs inside a **single dedicated
worker thread** (`_PlaywrightWorker`) that installs a proactor loop before
touching Playwright. Callers from any thread submit closures to that
worker and block on the result.
The Vaadin DOM reader lives in smarttime_ui.py. Each read runs in a fresh
subprocess so Streamlit worker threads need no asyncio policy changes and a
hung Chromium/Node process can be terminated without poisoning later checks.
"""
from __future__ import annotations
import asyncio
import logging
import queue
import re
import sys
import threading
import time as _time
from dataclasses import dataclass
from datetime import datetime, time
from pathlib import Path
from typing import Any, Callable, Optional
from playwright.sync_api import (
BrowserContext,
Page,
sync_playwright,
)
from typing import Callable, Optional
from settings import settings
from smarttime_browser import run_browser_job
log = logging.getLogger(__name__)
PROFILE_DIR = Path.home() / ".stempelbot" / "smarttime-profile"
DASHBOARD_LOAD_TIMEOUT_SEC = 300.0
DASHBOARD_POLL_INTERVAL_SEC = 0.5
# Regexes to pick "Letzte Buchung 11:43" / "Aktueller Status …" out of the
# dashboard text. Kept intentionally lenient because the SmartTime Plus GWT
# UI reshuffles labels between builds (extra whitespace, colons, dashes,
# line breaks, adjacent icons, …).
LAST_STAMP_RE = re.compile(
r"Letzte\s*Buchung[^\d]{0,40}?(\d{1,2}):(\d{2})",
re.IGNORECASE | re.DOTALL,
)
# Fallback: any "Buchung" label followed shortly by HH:MM.
LAST_STAMP_FALLBACK_RE = re.compile(
r"Buchung[^\d]{0,60}?(\d{1,2}):(\d{2})",
re.IGNORECASE | re.DOTALL,
)
STATUS_RE = re.compile(
r"Aktueller\s*Status[^A-Za-zÄÖÜäöü]{0,40}?([A-Za-zÄÖÜäöü][^\n\r]{0,80})",
re.IGNORECASE,
)
# Fallback: just look for the two words we care about.
STATUS_FALLBACK_RE = re.compile(
r"\b(Anwesend|Abwesend[^\n\r]{0,60})",
re.IGNORECASE,
)
STAMP_MAX_ATTEMPTS = 5 # Includes the original request sent by CoeoClient.
STAMP_RETRY_DELAY_SEC = 5
@dataclass
class SmartTimeStatus:
"""Snapshot of the SmartTime Plus dashboard for today."""
"""Snapshot of the logged-in user's SmartTime dashboard."""
last_stamp: Optional[time]
status_text: Optional[str] # e.g. "Anwesend" / "Abwesend (Haus verlassen)"
status_text: Optional[str]
query_time: datetime
@property
def is_present(self) -> bool:
return bool(self.status_text and "anwesend" in self.status_text.lower()
and "abwesend" not in self.status_text.lower())
return self.status_text == "Anwesend"
# ---------------------------------------------------------------------------
# Dedicated Playwright worker thread
# ---------------------------------------------------------------------------
_STOP = object()
class _PlaywrightWorker(threading.Thread):
"""Owns the Playwright instance + browser. Runs jobs from a queue."""
def __init__(self) -> None:
super().__init__(name="SmartTimeWorker", daemon=True)
self._jobs: "queue.Queue[tuple[Callable[[Page], Any], queue.Queue]]" = queue.Queue()
self._ready = threading.Event()
self._startup_error: Optional[BaseException] = None
self._start_lock = threading.Lock()
self._pw = None
self._context: Optional[BrowserContext] = None
self._page: Optional[Page] = None
# ---------- called from worker thread only ----------
def _init_playwright(self) -> None:
# On Windows we NEED a proactor loop for subprocess support.
if sys.platform == "win32":
try:
asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
except Exception:
pass
asyncio.set_event_loop(asyncio.ProactorEventLoop())
PROFILE_DIR.mkdir(parents=True, exist_ok=True)
log.info("🕵️ SmartTime worker: launching Chromium (profile=%s)", PROFILE_DIR)
self._pw = sync_playwright().start()
self._context = self._pw.chromium.launch_persistent_context(
user_data_dir=str(PROFILE_DIR),
headless=True,
ignore_https_errors=True,
locale="de-DE",
args=["--ignore-certificate-errors"],
)
self._page = self._context.new_page()
def _ensure_page(self) -> Page:
if self._page is None or self._page.is_closed():
self._page = self._context.new_page() # type: ignore[union-attr]
return self._page
def _shutdown(self) -> None:
try:
if self._context:
self._context.close()
finally:
self._context = None
self._page = None
if self._pw:
try:
self._pw.stop()
except Exception:
pass
self._pw = None
def run(self) -> None: # noqa: D401
try:
self._init_playwright()
except BaseException as e: # capture and surface to caller
self._startup_error = e
self._ready.set()
log.exception("SmartTime worker failed to start: %s", e)
return
self._ready.set()
while True:
item = self._jobs.get()
if item is _STOP:
break
fn, result_q = item
try:
page = self._ensure_page()
result_q.put(("ok", fn(page)))
except BaseException as e: # send back to caller
log.exception("SmartTime worker job failed: %s", e)
result_q.put(("err", e))
self._shutdown()
# ---------- called from other threads ----------
def submit(self, fn: Callable[[Page], Any], timeout: float = 120.0) -> Any:
with self._start_lock:
if not self.is_alive():
if self.ident is not None:
raise RuntimeError("SmartTime worker has stopped")
self.start()
if not self._ready.wait(timeout=30):
raise RuntimeError("SmartTime worker did not start in time")
if self._startup_error is not None:
raise RuntimeError(
f"SmartTime worker failed to start: {self._startup_error!r}"
) from self._startup_error
result_q: "queue.Queue[tuple[str, Any]]" = queue.Queue()
self._jobs.put((fn, result_q))
kind, val = result_q.get(timeout=timeout)
if kind == "err":
raise val
return val
def stop(self) -> None:
self._jobs.put(_STOP) # type: ignore[arg-type]
# ---------------------------------------------------------------------------
# Public client
# ---------------------------------------------------------------------------
class SmartTimeClient:
def __init__(self) -> None:
self._worker = _PlaywrightWorker()
self._worker_lock = threading.Lock()
self._read_lock = threading.Lock()
# ---------------------------------------------------------------- helpers
@staticmethod
def _open_dashboard(page: Page) -> None:
try:
page.evaluate("location.hash = 'dashboard_tab'")
except Exception:
# The GWT page may be replacing its execution context just now. The
# polling loop below will either observe the dashboard or fail with
# a useful timeout instead of accepting a half-loaded page.
pass
@staticmethod
def _parse_dashboard(body_text: str) -> tuple[Optional[time], Optional[str]]:
last_stamp: Optional[time] = None
match = LAST_STAMP_RE.search(body_text) or LAST_STAMP_FALLBACK_RE.search(body_text)
if match:
try:
last_stamp = time(int(match.group(1)), int(match.group(2)))
except ValueError:
pass
status_text: Optional[str] = None
status_match = STATUS_RE.search(body_text) or STATUS_FALLBACK_RE.search(body_text)
if status_match:
status_text = status_match.group(1).strip()
return last_stamp, status_text
@classmethod
def _fetch_status(cls, page: Page) -> SmartTimeStatus:
t0 = _time.monotonic()
log.info("🕵️ SmartTime: fetching dashboard")
page.goto(
settings.SMART_TIME_URL,
wait_until="domcontentloaded",
timeout=15_000,
)
cls._open_dashboard(page)
# GWT renders both the login form and dashboard asynchronously. Poll
# until the values (not merely their labels) can actually be parsed.
# No networkidle wait – GWT keeps long-poll XHRs open, so that
# would always block until timeout.
deadline = _time.monotonic() + DASHBOARD_LOAD_TIMEOUT_SEC
body_text = ""
login_submitted = False
reopen_dashboard = False
while True:
try:
body_text = page.locator("body").inner_text(timeout=1500)
except Exception:
body_text = ""
last_stamp, status_text = cls._parse_dashboard(body_text)
if last_stamp is not None and status_text is not None:
snap = SmartTimeStatus(
last_stamp=last_stamp,
status_text=status_text,
query_time=datetime.now(),
)
log.info(
"🕵️ SmartTime status: %s (took %.1fs)",
snap,
_time.monotonic() - t0,
)
return snap
try:
user_input = page.query_selector("input[name='useraccount']")
pw_input = page.query_selector("input[name='password']")
except Exception:
user_input = pw_input = None
if user_input and pw_input and not login_submitted:
log.info("🕵️ SmartTime: logging in as %s", settings.USERNAME)
user_input.fill(settings.USERNAME)
pw_input.fill(settings.SMART_TIME_PASSWORD)
pw_input.press("Enter")
login_submitted = True
reopen_dashboard = True
elif login_submitted and not user_input and not pw_input and reopen_dashboard:
cls._open_dashboard(page)
reopen_dashboard = False
if _time.monotonic() >= deadline:
preview = re.sub(r"\s+", " ", body_text)[:400]
reason = "login failed" if login_submitted and user_input and pw_input else "dashboard did not finish loading"
raise RuntimeError(
f"SmartTime {reason} within {DASHBOARD_LOAD_TIMEOUT_SEC:g}s "
f"(body: {preview or 'empty'})"
)
_time.sleep(DASHBOARD_POLL_INTERVAL_SEC)
# ---------------------------------------------------------------- public
def get_status(self) -> SmartTimeStatus:
with self._worker_lock:
if self._worker.ident is not None and not self._worker.is_alive():
self._worker = _PlaywrightWorker()
worker = self._worker
try:
return worker.submit(self._fetch_status)
except Exception:
# A Thread object is one-shot. Discard a worker that failed during
# startup so a later explicit check gets a fresh browser worker.
if worker.ident is not None and not worker.is_alive():
with self._worker_lock:
if self._worker is worker:
self._worker = _PlaywrightWorker()
raise
with self._read_lock:
result = run_browser_job()
return SmartTimeStatus(
last_stamp=time.fromisoformat(result["last_stamp"]),
status_text=result["status_text"],
query_time=datetime.fromisoformat(result["query_time"]),
)
def verify_stamp(
self,
@@ -320,15 +55,13 @@ class SmartTimeClient:
initial_delay: Optional[float] = None,
tolerance_sec: Optional[float] = None,
retry_action: Optional[Callable[[], bool]] = None,
on_progress: Optional[Callable[[str], None]] = None,
) -> tuple[bool, str, Optional[SmartTimeStatus]]:
"""Check SmartTime Plus once and compare its `Letzte Buchung` to
``expected_at`` within ``tolerance_sec`` seconds.
"""Verify the initial stamp, allowing at most four resends.
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.
Before each resend, wait five seconds and confirm the dashboard still
shows an older stamp. Unknown results stop the workflow. Progress is
reported synchronously on the caller's thread for UI updates.
"""
initial_delay = (
initial_delay if initial_delay is not None
@@ -339,15 +72,7 @@ class SmartTimeClient:
else settings.SMART_TIME_VERIFY_TOLERANCE_SEC
)
if initial_delay > 0:
_time.sleep(initial_delay)
expected_dt = expected_at.replace(second=0, microsecond=0)
try:
snap = self.get_status()
except Exception as e:
log.exception("SmartTime verify: error fetching status: %s", e)
return False, f"SmartTime nicht erreichbar: {e}", None
def evaluate(current: SmartTimeStatus) -> tuple[bool, str]:
if current.last_stamp:
@@ -366,48 +91,67 @@ class SmartTimeClient:
f"SmartTime letzte Buchung: {obs}"
)
verified, message = evaluate(snap)
if verified:
return True, message, snap
def report(message: str) -> None:
log.info(message)
if on_progress is not None:
on_progress(message)
# 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
def can_retry(current: SmartTimeStatus) -> bool:
# A later stamp, incomplete data or a date rollover is ambiguous.
return (
retry_action is not None
and current.last_stamp is not None
and bool(current.status_text)
and current.query_time.date() == expected_dt.date()
and current.query_time >= expected_at
and datetime.combine(expected_dt.date(), current.last_stamp) < expected_dt
)
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
for attempt in range(1, STAMP_MAX_ATTEMPTS + 1):
label = f"Versuch {attempt}/{STAMP_MAX_ATTEMPTS}"
report(f"{label}: SmartTime-Buchung wird geprüft …")
if initial_delay > 0:
_time.sleep(initial_delay)
try:
snap = self.get_status()
except Exception as e:
log.exception("SmartTime verify: error fetching status: %s", e)
return False, f"{label}: SmartTime nicht erreichbar: {e}", None
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
verified, message = evaluate(snap)
if verified or attempt == STAMP_MAX_ATTEMPTS or not can_retry(snap):
return verified, f"{message} ({label})", 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
next_label = f"Versuch {attempt + 1}/{STAMP_MAX_ATTEMPTS}"
report(f"{label}: Buchung fehlt. {next_label} in {STAMP_RETRY_DELAY_SEC}s …")
_time.sleep(STAMP_RETRY_DELAY_SEC)
# The previous request may have arrived during the wait.
report(f"{label}: SmartTime vor erneutem Stempeln prüfen …")
try:
snap = self.get_status()
except Exception as e:
log.exception("SmartTime check before retry failed: %s", e)
return False, f"{label}: SmartTime nicht erreichbar: {e}", None
verified, message = evaluate(snap)
if verified or not can_retry(snap):
return verified, f"{message} ({label})", snap
report(f"{next_label}: Stempel wird übermittelt …")
expected_at = datetime.now()
expected_dt = expected_at.replace(second=0, microsecond=0)
try:
retry_sent = retry_action()
except Exception as e:
log.exception("SmartTime stamp retry failed: %s", e)
return False, f"{next_label}: Retry fehlgeschlagen: {e}", snap
if not retry_sent:
return False, f"{next_label}: Retry konnte nicht bestätigt werden", snap
def close(self) -> None:
with self._worker_lock:
worker = self._worker
self._worker = _PlaywrightWorker()
if worker.is_alive():
worker.stop()
"""Compatibility hook; browser resources are closed after every read."""
# --- singleton -------------------------------------------------------------
_singleton: Optional[SmartTimeClient] = None
_singleton_lock = threading.Lock()
stempelbot/smarttime_ui.py
+235
View File
@@ -0,0 +1,235 @@
"""Read-only DOM navigation for SmartTime 9 (Vaadin Flow).
No booking, correction, approval or account-setting controls are used here.
"""
from __future__ import annotations
import calendar
import re
import time
from datetime import date, datetime
from pathlib import Path
from playwright.sync_api import Page
MONTHS = ("Januar", "Februar", "März", "April", "Mai", "Juni", "Juli", "August",
"September", "Oktober", "November", "Dezember")
class SmartTimeReadError(RuntimeError):
pass
class SmartTimeLoginError(SmartTimeReadError):
pass
# Use the logged-in user's card, never the employee-status legend text: all
# statuses occur in that legend regardless of whether the user is present.
DASHBOARD_JS = """() => {
const labels = [...document.querySelectorAll('label')];
const label = labels.find(e => e.textContent.trim() === 'Letzte Buchung');
const card = label?.closest('.shadow-xs');
if (!card) return null;
const stamp = label.nextElementSibling?.textContent.trim();
const icon = card.querySelector('vaadin-icon[icon="far:building"]');
if (!icon) return null;
const color = getComputedStyle(icon).fill;
const statuses = ['Anwesend', 'Außer Haus', 'Abwesend', 'Entschuldigt'];
const matches = statuses.filter(status => {
const legend = labels.find(e => e.textContent.trim() === status);
const swatch = legend?.parentElement.querySelector('div[style*="background-color"]');
return swatch && getComputedStyle(swatch).backgroundColor === color;
});
return {last_stamp: stamp, status_text: matches.length === 1 ? matches[0] : null};
}"""
GRID_JS = """() => {
const grid = document.querySelector('vaadin-grid.Customers-grid');
const root = grid?.shadowRoot;
if (!root || grid.loading || !root.querySelector('#header tr') || !root.querySelector('#table')) return null;
const cells = row => [...row.querySelectorAll('slot')].map(slot =>
slot.assignedElements().map(e => e.innerText).join('').trim());
return {
size: grid.size,
headers: cells(root.querySelector('#header tr')),
rows: [...root.querySelectorAll('#items tr')].filter(r => !r.hidden).map(cells),
scroll: root.querySelector('#table').scrollTop
};
}"""
def wait_for(page: Page, predicate, *, timeout: float = 40, description: str):
deadline = time.monotonic() + timeout
while True:
value = predicate()
if value:
return value
if time.monotonic() >= deadline:
raise SmartTimeReadError(f"Timed out waiting for {description}")
page.wait_for_timeout(250)
def wait_idle(page: Page) -> None:
# Vaadin's request queue must be empty; a menu label can update before
# the server has returned the new rows. networkidle never works reliably.
wait_for(page, lambda: page.evaluate("""() => {
const clients = Object.values(window.Vaadin?.Flow?.clients || {});
return clients.length > 0 && clients.every(c => !c.isActive());
}"""), description="Vaadin response")
def open_dashboard(page: Page, url: str, username: str, password: str) -> None:
page.set_default_timeout(8000)
page.goto(url, wait_until="domcontentloaded", timeout=20000)
submitted = False
def ready():
nonlocal submitted
user = page.locator('input[name="username"]')
secret = page.locator('input[name="password"]')
if user.is_visible() and secret.is_visible():
if page.locator('vaadin-login-form[error], vaadin-login-overlay[error]').count():
raise SmartTimeLoginError("SmartTime rejected the login")
if not submitted:
user.fill(username)
secret.fill(password)
secret.press("Enter")
submitted = True
return False
return page.locator('vaadin-side-nav-item[path=""]').is_visible()
wait_for(page, ready, description="SmartTime login/application")
dashboard = page.locator('vaadin-side-nav-item[path=""]')
if dashboard.get_attribute("current") is None:
dashboard.click()
wait_for(page, lambda: page.locator('vaadin-side-nav-item[path=""][current]').count(),
description="dashboard navigation")
wait_idle(page)
def read_dashboard(page: Page, *, timeout: float = 40) -> dict:
def ready():
value = page.evaluate(DASHBOARD_JS)
if not value or not value["status_text"]:
return False
stamp = value["last_stamp"] or ""
if not re.fullmatch(r"(?:[01]\d|2[0-3]):[0-5]\d", stamp):
return False
return value
result = wait_for(page, ready, timeout=timeout, description="complete personal dashboard card")
result["query_time"] = datetime.now().isoformat()
return result
def capture(page: Page, output: Path, name: str) -> None:
"""Local diagnostics, including shadow DOM omitted by ordinary page.content()."""
output.mkdir(parents=True, exist_ok=True)
page.screenshot(path=str(output / f"{name}.png"), full_page=True, timeout=8000)
html = page.evaluate("""() => {
function clone(node) {
const copy = node.cloneNode(false);
if (copy.nodeType === 1 && copy.tagName === 'INPUT') copy.removeAttribute('value');
for (const child of node.childNodes) copy.appendChild(clone(child));
if (node.shadowRoot) {
const template = document.createElement('template');
template.setAttribute('shadowrootmode', 'open');
for (const child of node.shadowRoot.childNodes) template.content.appendChild(clone(child));
copy.appendChild(template);
}
return copy;
}
return '<!DOCTYPE html>' + clone(document.documentElement).outerHTML;
}""")
(output / f"{name}.html").write_text(html, encoding="utf-8")
def period(page: Page) -> tuple[int, int] | None:
# Only top-level menu buttons: closed menus retain their option elements.
texts = page.locator('vaadin-menu-bar-button[role="menuitem"]:visible').all_text_contents()
months = [MONTHS.index(t.strip()) + 1 for t in texts if t.strip() in MONTHS]
years = [int(t.strip()) for t in texts if re.fullmatch(r"\d{4}", t.strip())]
return (years[0], months[0]) if len(months) == len(years) == 1 else None
def select_period(page: Page, year: int, month: int) -> None:
current = wait_for(page, lambda: period(page), description="month/year controls")
for index, target in ((0, year), (1, month)):
if current[index] == target:
continue
old_label = str(current[0]) if index == 0 else MONTHS[current[1] - 1]
label = str(year) if index == 0 else MONTHS[month - 1]
page.locator('vaadin-menu-bar-button[role="menuitem"]:visible').filter(has_text=re.compile(r"^\s*" + re.escape(old_label) + r"\s*$")).click()
# Menu options have role=menuitem in Vaadin's overlay as well.
page.get_by_role("menuitem", name=label, exact=True).click()
wanted = (year, current[1]) if index == 0 else (year, month)
wait_for(page, lambda: period(page) == wanted, description=f"period {wanted}")
wait_idle(page)
current = wanted
def normalize_month(year: int, month: int, headers: list[str], rows: list[list[str]]) -> dict:
expected_headers = ["Datum", "", "Kommen / Gehen", "Soll", "Ist", "TSaldo", "Saldo", "Pause", "Abwesenheit"]
if headers != expected_headers:
raise SmartTimeReadError(f"Unexpected monthly columns: {headers!r}")
days = []
for cells in rows:
if len(cells) != len(headers) or not re.fullmatch(r"\d{2}", cells[0]):
raise SmartTimeReadError("Incomplete or unknown monthly row")
day = date(year, month, int(cells[0]))
stamps = re.findall(r"(?<!\d)(?:[01]\d|2[0-3]):[0-5]\d(?!\d)", cells[2])
days.append({
"date": day.isoformat(), "stamps": stamps,
"incomplete_pair": bool(len(stamps) % 2),
"scheduled": cells[3], "actual": cells[4], "daily_balance": cells[5],
"balance": cells[6], "break": cells[7], "absence": cells[8],
"raw_cells": cells,
})
count = calendar.monthrange(year, month)[1]
expected_dates = {date(year, month, d).isoformat() for d in range(1, count + 1)}
if len(days) != count or {d["date"] for d in days} != expected_dates:
raise SmartTimeReadError("Incomplete month: refusing to export partial or duplicate dates")
return {"schema_version": 1, "source": "SmartTime 9 Monatsübersicht",
"month": f"{year:04d}-{month:02d}", "exported_at": datetime.now().isoformat(),
"columns": headers, "days": sorted(days, key=lambda d: d["date"])}
def read_month(page: Page, year: int, month: int, artifacts: Path | None = None) -> dict:
count = calendar.monthrange(year, month)[1]
page.locator('vaadin-side-nav-item[path="MonthlyOV"]').click()
wait_for(page, lambda: page.locator('vaadin-side-nav-item[path="MonthlyOV"][current]').count(),
description="Monatsübersicht navigation")
wait_idle(page)
select_period(page, year, month)
grid = page.locator("vaadin-grid.Customers-grid")
wait_for(page, lambda: page.evaluate(GRID_JS), description="monthly grid")
rows: dict[str, list[str]] = {}
headers = None
for position in range(0, count, 8):
# Public Vaadin method scrolls virtual rows into the rendered DOM.
grid.evaluate("(g, row) => g.scrollToIndex(row)", position)
target = f"{position + 1:02d}"
def batch_ready():
batch = page.evaluate(GRID_JS)
if not batch or batch["size"] != count:
return False
return batch if any(r and r[0] == target for r in batch["rows"]) else False
batch = wait_for(page, batch_ready, description=f"monthly row {target}")
wait_idle(page)
batch = page.evaluate(GRID_JS)
if headers is not None and headers != batch["headers"]:
raise SmartTimeReadError("Monthly columns changed during export")
headers = batch["headers"]
for row in batch["rows"]:
if row and re.fullmatch(r"\d{2}", row[0]):
if row[0] in rows and rows[row[0]] != row:
raise SmartTimeReadError("Monthly values changed during export; retry required")
rows[row[0]] = row
if artifacts:
capture(page, artifacts, f"month-{year}-{month:02d}-{position:02d}")
if period(page) != (year, month):
raise SmartTimeReadError("Month changed during export")
return normalize_month(year, month, headers, list(rows.values()))
tests/test_smarttime_client.py
+125 -142
View File
@@ -3,7 +3,7 @@ import sys
import unittest
from datetime import datetime, time
from pathlib import Path
from unittest.mock import patch
from unittest.mock import Mock, call, patch
PROJECT_DIR = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(PROJECT_DIR / "stempelbot"))
@@ -15,94 +15,7 @@ os.environ.setdefault("STAMPHISTORY_FILE", "timestamp_history.json")
os.environ.setdefault("LOG_LEVEL", "CRITICAL")
import smarttime_client as smarttime # noqa: E402
class FakeInput:
def __init__(self) -> None:
self.filled_with = None
self.pressed = None
def fill(self, value: str) -> None:
self.filled_with = value
def press(self, key: str) -> None:
self.pressed = key
class FakeLocator:
def __init__(self, page: "FakePage") -> None:
self.page = page
def inner_text(self, timeout: int) -> str:
index = min(self.page.body_reads, len(self.page.body_texts) - 1)
self.page.body_reads += 1
return self.page.body_texts[index]
class FakePage:
def __init__(self, body_texts: list[str], login_on_read: int | None = None) -> None:
self.body_texts = body_texts
self.login_on_read = login_on_read
self.body_reads = 0
self.user_input = FakeInput()
self.password_input = FakeInput()
self.goto_calls = 0
self.dashboard_opens = 0
def goto(self, *args, **kwargs) -> None:
self.goto_calls += 1
def evaluate(self, expression: str) -> None:
self.dashboard_opens += 1
def locator(self, selector: str) -> FakeLocator:
return FakeLocator(self)
def query_selector(self, selector: str):
if self.login_on_read == self.body_reads:
if "useraccount" in selector:
return self.user_input
if "password" in selector:
return self.password_input
return None
class SmartTimeLoadingTests(unittest.TestCase):
def test_waits_for_parseable_values_and_handles_delayed_login(self) -> None:
page = FakePage(
[
"SmartTime wird geladen",
"Anmeldung",
"Dashboard wird geladen",
"Letzte Buchung 08:42\nAktueller Status: Anwesend",
],
login_on_read=2,
)
with (
patch.object(smarttime, "DASHBOARD_LOAD_TIMEOUT_SEC", 1.0),
patch.object(smarttime, "DASHBOARD_POLL_INTERVAL_SEC", 0.0),
):
result = smarttime.SmartTimeClient._fetch_status(page)
self.assertEqual(result.last_stamp, time(8, 42))
self.assertEqual(result.status_text, "Anwesend")
self.assertEqual(page.user_input.filled_with, smarttime.settings.USERNAME)
self.assertEqual(
page.password_input.filled_with, smarttime.settings.SMART_TIME_PASSWORD
)
self.assertEqual(page.password_input.pressed, "Enter")
self.assertEqual(page.dashboard_opens, 2)
def test_incomplete_dashboard_raises_instead_of_returning_empty_snapshot(self) -> None:
page = FakePage(["Letzte Buchung\nAktueller Status"])
with (
patch.object(smarttime, "DASHBOARD_LOAD_TIMEOUT_SEC", 0.0),
patch.object(smarttime, "DASHBOARD_POLL_INTERVAL_SEC", 0.0),
):
with self.assertRaisesRegex(RuntimeError, "did not finish loading"):
smarttime.SmartTimeClient._fetch_status(page)
import client as coeo # noqa: E402
class SmartTimeVerificationTests(unittest.TestCase):
@@ -126,34 +39,6 @@ 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())
@@ -174,34 +59,132 @@ class SmartTimeVerificationTests(unittest.TestCase):
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())
class DeadWorker:
ident = 1
@staticmethod
def is_alive() -> bool:
return False
class ReplacementWorker:
ident = None
@staticmethod
def is_alive() -> bool:
return False
@staticmethod
def submit(fn):
return expected
def test_failed_read_does_not_poison_next_check(self) -> None:
client = smarttime.SmartTimeClient()
client._worker = DeadWorker()
with patch.object(smarttime, "_PlaywrightWorker", ReplacementWorker):
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)
self.assertIs(result, expected)
self.assertIsInstance(client._worker, ReplacementWorker)
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__":