automation-tools/ecw-charge-validation
8
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d500f750da37cd7e9884393a4d41f31b6c2f9233
ecw-charge-validation/tests/test_portal_users.py
T
ManikandanandClaude Opus 5.5 64efb68ea1 Portal sign-in with the Ecare office login 
Ecare staff sign in with their office username and password, checked by
the Ecare sign-in service from the backend (AUTH_TOKEN_API_URL); a good
Ecare login still needs a Portal User. Service down = "try again
shortly", never a wrong password. /api/auth/login, me, logout; sign-in
tries limited per address. Users page: Ecare office login or Portal
password per user. Context.md: 27/10 runs, the login retry bug.
71 test files, 1239 tests.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-27 17:55:50 +05:30

485 lines
17 KiB
Python
Raw Blame History

"""
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())
Reference in new issue View Git Blame Copy Permalink