Snapshot of the working project as of 2026-10-23 (local 2d4d4fc), one commit with no earlier history: the tracked files were de-identified first (synthetic claim ids and patient data only; see Context.md section 7). Credentials, run logs, screenshots, claim reports and client workbooks are gitignored and not included. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
114 lines
4.2 KiB
Python
114 lines
4.2 KiB
Python
"""
|
|
Attempt-guard tests.
|
|
|
|
python tests\\test_guard.py
|
|
|
|
Plain asserts, no pytest — same reasoning as the hand-rolled `.env` parser:
|
|
these machines are locked down and every added dependency is one more thing to
|
|
get approved. `python tests\\test_guard.py` needs nothing that is not already
|
|
installed.
|
|
|
|
These are the only tests that can exist for the guard, and they matter more
|
|
than usual: the alternative is a real run, and a real run costs one of a
|
|
handful of hourly login attempts plus a mailed 2FA link. Verifying the cooldown
|
|
arithmetic against a live portal would be absurd.
|
|
|
|
Both clocks are faked from **one** source. That is the point of the harness,
|
|
not an accident: an earlier version of `wait_if_cooling` measured with
|
|
`time.time()` while `check()` measured with `datetime.now()`, and two clocks
|
|
for one deadline is how a run waits out its cooldown and is then refused for
|
|
being a fraction of a second early. A single fake clock catches that.
|
|
"""
|
|
|
|
import io
|
|
import sys
|
|
import json
|
|
import pathlib
|
|
import tempfile
|
|
import contextlib
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent / "src"))
|
|
|
|
from ecw_bot import settings as cfg # noqa: E402
|
|
from ecw_bot.core import guard # noqa: E402
|
|
|
|
BASE = datetime.now(timezone.utc)
|
|
_ago = lambda s: (BASE - timedelta(seconds=s)).isoformat() # noqa: E731
|
|
|
|
|
|
class Clock:
|
|
"""One fake clock driving both `_now()` and `time.sleep()`."""
|
|
|
|
def __init__(self):
|
|
self.off = 0.0
|
|
self.slept = 0.0
|
|
|
|
def now(self):
|
|
return BASE + timedelta(seconds=self.off)
|
|
|
|
def sleep(self, seconds):
|
|
self.off += seconds
|
|
self.slept += seconds
|
|
|
|
|
|
def run(state: dict):
|
|
"""Apply `state`, run wait_if_cooling() then check(). → (slept, verdict)."""
|
|
tmp = pathlib.Path(tempfile.mkdtemp()) / "attempts.json"
|
|
tmp.write_text(json.dumps(state), encoding="utf-8")
|
|
|
|
clock = Clock()
|
|
saved = (guard.STATE_PATH, guard._now, guard.time)
|
|
guard.STATE_PATH = tmp
|
|
guard._now = clock.now
|
|
guard.time = type("_t", (), {"sleep": staticmethod(clock.sleep)})
|
|
try:
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
guard.wait_if_cooling()
|
|
verdict = guard.check()
|
|
finally:
|
|
guard.STATE_PATH, guard._now, guard.time = saved
|
|
return clock.slept, verdict
|
|
|
|
|
|
CASES = [
|
|
# name, state, slept, clear
|
|
("first ever run", {}, 0, True),
|
|
("cooldown long expired", {"last_attempt": _ago(4000)}, 0, True),
|
|
# The one that used to exit 3 with three seconds to go.
|
|
("3s left of the cooldown", {"last_attempt": _ago(297)}, 4, True),
|
|
("280s left of the cooldown", {"last_attempt": _ago(20)}, 281, True),
|
|
# Neither of these may be slept through — they mean stop, not "not yet".
|
|
("burst limit reached", {"last_attempt": _ago(4000),
|
|
"recent": [_ago(60 * m) for m in (1, 2, 3, 4)]},
|
|
0, False),
|
|
("restriction lockout", {"last_attempt": _ago(10),
|
|
"locked_until":
|
|
(BASE + timedelta(minutes=4)).isoformat(),
|
|
"restricted_at": BASE.isoformat()}, 0, False),
|
|
]
|
|
|
|
|
|
def main() -> int:
|
|
assert cfg.LOGIN_COOLDOWN_SECONDS == 300, (
|
|
f"these expectations assume a 300s cooldown, .env says "
|
|
f"{cfg.LOGIN_COOLDOWN_SECONDS}")
|
|
|
|
failures = 0
|
|
for name, state, want_slept, want_clear in CASES:
|
|
slept, verdict = run(state)
|
|
clear = not verdict
|
|
ok = (round(slept) == want_slept) and (clear == want_clear)
|
|
failures += not ok
|
|
print(f" {'ok ' if ok else 'FAIL'} {name:26s} "
|
|
f"slept {slept:6.0f}s (want {want_slept:4d}) "
|
|
f"{'clear' if clear else 'refused'}"
|
|
f"{'' if clear == want_clear else f' (want {want_clear})'}")
|
|
|
|
print(f"\n{len(CASES) - failures}/{len(CASES)} passed")
|
|
return 1 if failures else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|