""" Portal Users, Sign-in rules and sessions — `web/users.py` (spec 2026-10-22 Stage 2, EEWA-32), against an in-memory store (no SQL Server). python tests\\test_portal_users.py * Passwords: hashed with their own salt, never kept in plain; 10+ characters, not the username. * Sign-in: a wrong password and an unknown username read the same; 5 wrong in a row lock the user for 15 minutes; a disabled user cannot sign in; every outcome is recorded. * Admin changes: a new user must pick their own password at first Sign-in, a reset does the same, the last active Admin cannot be demoted or disabled, nobody disables themselves. * Sessions: 30 minutes idle ends one; ending a user's sessions works. """ import sys import pathlib from datetime import datetime, timedelta ROOT = pathlib.Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT)) sys.path.insert(0, str(ROOT / "src")) from web import users # noqa: E402 users.ITERATIONS = 1_000 # fast tests; the real count is checked below CASES = [] def case(fn): CASES.append(fn) return fn class FakeUserStore: """The SqlUserStore methods, in memory.""" def __init__(self): self.users = {} self.events = [] def get(self, username): u = self.users.get(username) return dict(u) if u else None def list(self): return [dict(u) for _, u in sorted(self.users.items())] def create(self, user): self.users[user["username"]] = dict(user) def update(self, username, **fields): self.users[username].update(fields) def record(self, username, event, by=""): self.events.append((username, event, by or "")) def history(self, limit=300): return [{"at": datetime(2026, 10, 22, 12, 0), "username": u, "event": e, "by_user": b or None} for u, e, b in reversed(self.events)][:limit] class Clock: def __init__(self): self.t = datetime(2026, 10, 22, 9, 0, 0) def __call__(self): return self.t def _accounts(): store, clock = FakeUserStore(), Clock() acc = users.Accounts(store, now=clock) acc.create_user("Asha", "Asha R", "Super Admin", "first-password-1") return acc, store, clock @case def the_real_iteration_count_is_at_least_600k(): import importlib fresh = importlib.reload(users) try: assert fresh.ITERATIONS >= 600_000, fresh.ITERATIONS finally: fresh.ITERATIONS = 1_000 @case def a_password_is_kept_only_as_a_salted_hash(): acc, store, _ = _accounts() u = store.users["asha"] assert "first-password-1" not in str(u), u assert len(u["pw_hash"]) == 64 and len(u["pw_salt"]) == 32, u a, s1, _ = users.hash_password("same-password!") b, s2, _ = users.hash_password("same-password!") assert a != b and s1 != s2, "each hash has its own salt" @case def password_rules(): assert users.password_problem("short", "x") assert users.password_problem("asha.r.123", "asha.r.123") assert users.password_problem("long-enough-1", "asha") is None @case def a_user_made_by_the_tool_signs_in_and_names_are_case_blind(): acc, store, _ = _accounts() assert store.users["asha"]["must_change"] is False, "tool user" user, why = acc.sign_in(" ASHA ", "first-password-1") assert why == "ok" and user["username"] == "asha", why assert store.events[-1] == ("asha", "sign_in", "") @case def wrong_password_and_unknown_user_read_the_same(): acc, store, _ = _accounts() assert acc.sign_in("asha", "wrong-password-x") == (None, "bad") assert acc.sign_in("nobody", "wrong-password-x") == (None, "bad") assert [e[1] for e in store.events[-2:]] == ["wrong_password"] * 2 @case def five_wrong_passwords_lock_for_fifteen_minutes(): acc, store, clock = _accounts() for _ in range(4): assert acc.sign_in("asha", "nope-nope-nope")[1] == "bad" assert acc.sign_in("asha", "nope-nope-nope")[1] == "locked" assert acc.sign_in("asha", "first-password-1")[1] == "locked", \ "even the right password waits" clock.t += timedelta(minutes=15, seconds=1) assert acc.sign_in("asha", "first-password-1")[1] == "ok" assert store.users["asha"]["failed_count"] == 0 @case def a_good_sign_in_clears_the_wrong_count(): acc, store, _ = _accounts() for _ in range(4): acc.sign_in("asha", "nope-nope-nope") acc.sign_in("asha", "first-password-1") for _ in range(4): assert acc.sign_in("asha", "nope-nope-nope")[1] == "bad" @case def an_admin_made_user_must_change_and_a_reset_does_the_same(): acc, store, _ = _accounts() acc.create_user("ravi", "Ravi", "Viewer", "temp-password-1", by="asha") assert store.users["ravi"]["must_change"] is True assert store.users["ravi"]["created_by"] == "asha" acc.change_password("ravi", "temp-password-1", "ravis-own-pass") assert store.users["ravi"]["must_change"] is False acc.reset_password("ravi", "another-temp-1", by="asha") assert store.users["ravi"]["must_change"] is True assert ("ravi", "password_reset", "asha") in store.events @case def change_password_needs_the_current_one_and_a_new_one(): acc, _, _ = _accounts() for current, new in (("wrong-current", "brand-new-pass"), ("first-password-1", "first-password-1"), ("first-password-1", "short")): try: acc.change_password("asha", current, new) raise AssertionError(f"accepted {current!r} -> {new!r}") except users.UserError: pass @case def bad_new_users_are_refused(): acc, _, _ = _accounts() for args in (("a", "A", "Viewer", "long-enough-1"), ("bad name", "B", "Viewer", "long-enough-1"), ("bob", "B", "Boss", "long-enough-1"), ("bob", "B", "Viewer", "short"), ("asha", "A", "Viewer", "long-enough-1")): try: acc.create_user(*args, by="asha") raise AssertionError(f"accepted {args}") except users.UserError: pass @case def a_disabled_user_cannot_sign_in_and_is_never_deleted(): acc, store, _ = _accounts() acc.create_user("ravi", "Ravi", "Viewer", "temp-password-1", by="asha") acc.set_disabled("ravi", True, by="asha") assert acc.sign_in("ravi", "temp-password-1") == (None, "disabled") assert "ravi" in store.users acc.set_disabled("ravi", False, by="asha") assert acc.sign_in("ravi", "temp-password-1")[1] == "ok" @case def an_admin_adds_and_changes_viewers_only(): """Operator, 2026-10-27: an Admin creates users, and users means Viewers.""" acc, store, _ = _accounts() acc.create_user("ravi", "Ravi", "Admin", "temp-password-1", by="asha") acc.create_user("vee", "Vee", "Viewer", "temp-password-1", by="ravi") acc.reset_password("vee", "temp-password-2", by="ravi") acc.set_disabled("vee", True, by="ravi") acc.set_disabled("vee", False, by="ravi") acc.create_user("mani", "Mani", "Admin", "temp-password-1", by="asha") for fn in (lambda: acc.create_user("kim", "K", "Admin", "temp-password-1", by="ravi"), lambda: acc.set_role("vee", "Admin", by="ravi"), lambda: acc.set_disabled("mani", True, by="ravi"), lambda: acc.reset_password("mani", "temp-password-3", by="ravi"), lambda: acc.set_role("asha", "Viewer", by="ravi"), lambda: acc.set_disabled("asha", True, by="ravi")): try: fn() raise AssertionError("an Admin went past Viewers") except users.UserError: pass assert "kim" not in store.users and store.users["vee"]["role"] == "Viewer" @case def the_super_admin_runs_admins_and_nobody_touches_a_super_admin(): acc, store, _ = _accounts() assert store.users["asha"]["role"] == "Super Admin" acc.create_user("ravi", "Ravi", "Admin", "temp-password-1", by="asha") acc.set_role("ravi", "Viewer", by="asha") acc.set_role("ravi", "Admin", by="asha") acc.set_disabled("ravi", True, by="asha") assert ("ravi", "role_admin", "asha") in store.events for fn in (lambda: acc.create_user("boss", "B", "Super Admin", "temp-password-1", by="asha"), lambda: acc.set_role("asha", "Viewer", by="asha"), lambda: acc.set_disabled("asha", True, by="asha"), lambda: acc.set_role("ravi", "Super Admin", by="asha"), lambda: acc.create_user("bob", "B", "Viewer", "temp-password-1", by="nobody")): try: fn() raise AssertionError("allowed") except users.UserError: pass assert store.users["asha"]["role"] == "Super Admin" acc.create_user("boss", "B", "Super Admin", "temp-password-1") assert store.users["boss"]["role"] == "Super Admin", "the PC tool may" @case def the_old_operator_role_is_gone(): assert users.ROLES == ("Super Admin", "Admin", "Viewer"), users.ROLES assert users.CAN_RUN == ("Super Admin", "Admin") acc, _, _ = _accounts() try: acc.create_user("olu", "Olu", "Operator", "temp-password-1") raise AssertionError("Operator accepted") except users.UserError: pass @case def the_sql_table_allows_the_same_roles_and_moves_old_ones(): sql = (pathlib.Path(__file__).resolve().parent.parent / "tools" / "sql" / "create_tables.sql").read_text(encoding="utf-8") check = "CHECK (role IN ('Super Admin', 'Admin', 'Viewer'))" assert sql.count(check) == 2, "the new table and the old one moved" first = sql.index("SET role = 'Super Admin' WHERE role = 'Admin'") assert first < sql.index("SET role = 'Admin' WHERE role = 'Operator'"), "Admins move up before Operators take their place" # ── Ecare office sign-in (2026-10-27) ──────────────────────────────────────── class FakeEcare: """EcareSignIn.check, canned: "ok", "bad" or "down".""" def __init__(self, verdict="ok"): self.verdict = verdict self.calls = [] def check(self, username, password): self.calls.append(username) return self.verdict def _ecare_accounts(verdict="ok"): store, clock, ecare = FakeUserStore(), Clock(), FakeEcare(verdict) acc = users.Accounts(store, now=clock, ecare=ecare) acc.create_user("Asha", "Asha R", "Super Admin", "first-password-1") acc.create_user("ravi", "Ravi", "Viewer", None, by="asha", ecare=True) return acc, store, clock, ecare class _Reply: def __init__(self, body: bytes, status=200): self._body, self.status = body, status def read(self): return self._body def __enter__(self): return self def __exit__(self, *a): return False def _service(answer): """(EcareSignIn, the requests it sent) with urlopen giving `answer`: bytes for a 200 body, an int for an HTTP error, or an exception.""" import urllib.error sent = [] def urlopen(request, timeout=None): sent.append((request, timeout)) if isinstance(answer, BaseException): raise answer if isinstance(answer, int): raise urllib.error.HTTPError(request.full_url, answer, "x", {}, None) return _Reply(answer) return users.EcareSignIn("https://auth.example/api/login/token", urlopen=urlopen), sent @case def the_ecare_service_is_asked_from_the_backend_with_a_15_second_limit(): import json svc, sent = _service(b'{"success": true, "token": "abc"}') assert svc.check(" Ravi ", "pw-123") == "ok" request, timeout = sent[0] assert timeout == 15 and request.get_method() == "POST" assert request.full_url == "https://auth.example/api/login/token" assert request.get_header("Content-type") == "application/json" assert json.loads(request.data) == {"username": "Ravi", "password": "pw-123"} @case def the_ecare_answer_reads_as_ok_bad_or_down(): import socket import urllib.error for answer, want in ((b'{"token": "abc"}', "ok"), (b'{"token": ""}', "bad"), (b'{"success": false, "token": "abc"}', "bad"), (b'{"success": true}', "bad"), (b"not json", "down"), (401, "bad"), (403, "bad"), (400, "bad"), # A wrong AUTH_TOKEN_API_URL, not a wrong password: # else every user would lock themselves out. (404, "down"), (405, "down"), (500, "down"), (503, "down"), (socket.timeout("slow"), "down"), (urllib.error.URLError("no route"), "down"), (ConnectionResetError(), "down")): svc, _ = _service(answer) got = svc.check("ravi", "pw-123") assert got == want, (answer, got, want) @case def an_ecare_user_signs_in_with_the_office_password_only(): acc, store, _, ecare = _ecare_accounts("ok") u = store.users["ravi"] assert u["pw_iterations"] == 0 and u["must_change"] is False, u user, why = acc.sign_in(" RAVI ", "office-pass") assert why == "ok" and user["username"] == "ravi", why assert ecare.calls == ["RAVI"], "the name as typed, trimmed" assert ("ravi", "sign_in", "") in store.events assert users.is_ecare(user) and not users.is_ecare(store.get("asha")) @case def a_local_user_never_reaches_the_ecare_service(): acc, _, _, ecare = _ecare_accounts("ok") assert acc.sign_in("asha", "first-password-1")[1] == "ok" assert acc.sign_in("asha", "office-pass")[1] == "bad" assert ecare.calls == [] @case def a_good_ecare_login_without_a_portal_user_is_no_access(): acc, store, _, _ = _ecare_accounts("ok") assert acc.sign_in("stranger", "office-pass") == (None, "no_access") assert ("stranger", "no_access", "") in store.events assert "stranger" not in store.users, "never added by itself" @case def a_wrong_ecare_password_counts_toward_the_lock(): acc, store, clock, ecare = _ecare_accounts("bad") for _ in range(4): assert acc.sign_in("ravi", "nope")[1] == "bad" assert acc.sign_in("ravi", "nope")[1] == "locked" ecare.verdict = "ok" n = len(ecare.calls) assert acc.sign_in("ravi", "office-pass")[1] == "locked" assert len(ecare.calls) == n, "a locked user is not sent to Ecare" clock.t += timedelta(minutes=15, seconds=1) assert acc.sign_in("ravi", "office-pass")[1] == "ok" @case def ecare_down_is_never_a_wrong_password(): acc, store, _, _ = _ecare_accounts("down") assert acc.sign_in("ravi", "office-pass") == (None, "down") assert store.users["ravi"]["failed_count"] == 0 assert ("ravi", "service_down", "") in store.events acc.ecare = None # AUTH_TOKEN_API_URL not set assert acc.sign_in("ravi", "office-pass") == (None, "down") @case def a_disabled_ecare_user_stays_out_and_empty_input_never_leaves(): acc, _, _, ecare = _ecare_accounts("ok") acc.set_disabled("ravi", True, by="asha") assert acc.sign_in("ravi", "office-pass") == (None, "disabled") n = len(ecare.calls) assert acc.sign_in("ravi", "")[1] == "bad" assert acc.sign_in("", "pw")[1] == "bad" assert len(ecare.calls) == n, "nothing empty is sent to Ecare" @case def an_ecare_users_password_is_changed_in_ecare_not_here(): acc, _, _, _ = _ecare_accounts() for fn in (lambda: acc.reset_password("ravi", "temp-password-9", by="asha"), lambda: acc.change_password("ravi", "a", "new-password-1")): try: fn() raise AssertionError("allowed") except users.UserError as exc: assert "Ecare" in str(exc), exc @case def sessions_time_out_after_thirty_idle_minutes(): clock = Clock() s = users.Sessions(now=clock) tok = s.new({"username": "asha", "role": "Admin", "pw_hash": "x"}) user, _ = s.get(tok) assert user["username"] == "asha" and "pw_hash" not in user, user clock.t += timedelta(minutes=29) assert s.get(tok)[0], "29 idle minutes is fine, and resets the clock" clock.t += timedelta(minutes=30, seconds=1) assert s.get(tok) == (None, "asha"), "timed out, says whose" assert s.get(tok) == (None, None), "gone" @case def a_users_sessions_can_be_ended_and_refreshed(): s = users.Sessions() t1 = s.new({"username": "ravi", "role": "Admin"}) t2 = s.new({"username": "asha", "role": "Admin"}) s.refresh_user({"username": "ravi", "role": "Viewer"}) assert s.get(t1)[0]["role"] == "Viewer" s.drop_user("ravi") assert s.get(t1) == (None, None) and s.get(t2)[0] def main() -> int: failures = 0 for fn in CASES: name = fn.__name__.replace("_", " ") try: fn() print(f" ok {name}") except AssertionError as exc: print(f" FAIL {name}: {exc}") failures += 1 except Exception as exc: print(f" ERROR {name}: {type(exc).__name__}: {exc}") failures += 1 print(f"\n{len(CASES) - failures}/{len(CASES)} passed") return 1 if failures else 0 if __name__ == "__main__": sys.exit(main())