251 lines
10 KiB
Python
251 lines
10 KiB
Python
"""
|
|
Drive the organisation surface against a running application, over HTTP.
|
|
|
|
Everything so far has been proven by tests against a TestClient. This is the
|
|
first time the feature meets a real uvicorn process, real middleware ordering,
|
|
a real connection pool and both flags switched on — which is the configuration
|
|
production would run, and the one nothing has executed.
|
|
|
|
APP_ENV=rework python scratchpad/smoke_org.py --port 20055
|
|
|
|
It reports rather than asserts: the point is to see what the application does,
|
|
including the parts nobody predicted.
|
|
"""
|
|
|
|
import argparse
|
|
import secrets
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
os.environ.setdefault("APP_ENV", "rework")
|
|
os.environ.setdefault("PYTHONIOENCODING", "utf-8")
|
|
sys.path.insert(0, str(Path.cwd()))
|
|
|
|
PASS = " ok "
|
|
FAIL = " FAIL "
|
|
|
|
results = []
|
|
# Unique per run. Re-running should exercise behaviour, not rediscover
|
|
# the uniqueness constraint that run one already proved.
|
|
SUFFIX = secrets.token_hex(3)
|
|
|
|
|
|
def check(label: str, condition: bool, detail: str = "") -> None:
|
|
results.append((label, condition, detail))
|
|
print(f"[{PASS if condition else FAIL}] {label}" + (f" — {detail}" if detail else ""))
|
|
|
|
|
|
|
|
def _paths(settings, *unit_ids):
|
|
"""Read `org_units.path` straight from the database."""
|
|
from sqlalchemy import create_engine, text
|
|
|
|
engine = create_engine(settings.DATABASE_URL)
|
|
try:
|
|
with engine.connect() as conn:
|
|
return [
|
|
conn.execute(
|
|
text("SELECT path FROM org_units WHERE id = :id"), {"id": uid}
|
|
).scalar()
|
|
for uid in unit_ids
|
|
]
|
|
finally:
|
|
engine.dispose()
|
|
|
|
|
|
def main() -> int:
|
|
import httpx
|
|
|
|
from app.core.security import create_access_token
|
|
from sqlalchemy import create_engine, text
|
|
|
|
from app.core.settings import settings
|
|
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--port", type=int, default=20055)
|
|
args = parser.parse_args()
|
|
base = f"http://127.0.0.1:{args.port}"
|
|
|
|
engine = create_engine(settings.DATABASE_URL)
|
|
with engine.connect() as conn:
|
|
row = conn.execute(
|
|
text(
|
|
"SELECT u.id, u.tenant_id, u.role_id FROM users u "
|
|
" WHERE u.email = 'smoke-admin@smoke.local'"
|
|
)
|
|
).first()
|
|
manager_id = conn.execute(
|
|
text("SELECT id FROM users WHERE email = 'smoke-manager@smoke.local'")
|
|
).scalar()
|
|
engineer_id = conn.execute(
|
|
text("SELECT id FROM users WHERE email = 'smoke-engineer@smoke.local'")
|
|
).scalar()
|
|
accountant_id = conn.execute(
|
|
text("SELECT id FROM users WHERE email = 'smoke-accountant@smoke.local'")
|
|
).scalar()
|
|
engine.dispose()
|
|
|
|
admin_id, tenant_id, role_id = row
|
|
admin = {
|
|
"Authorization": "Bearer "
|
|
+ create_access_token(
|
|
{"sub": str(admin_id), "tenant_id": str(tenant_id), "is_superadmin": False}
|
|
)
|
|
}
|
|
manager = {
|
|
"Authorization": "Bearer "
|
|
+ create_access_token(
|
|
{"sub": str(manager_id), "tenant_id": str(tenant_id), "is_superadmin": False}
|
|
)
|
|
}
|
|
|
|
c = httpx.Client(base_url=base, timeout=30.0)
|
|
|
|
# --- the tree ----------------------------------------------------------
|
|
print("\n--- organisation tree ---")
|
|
eng = c.post("/api/org/units", json={"name": f"Engineering {SUFFIX}", "code": f"ENG-{SUFFIX}"}, headers=admin)
|
|
check("create a root unit", eng.status_code == 200, f"HTTP {eng.status_code} {eng.text[:90]}")
|
|
if eng.status_code != 200:
|
|
return 1
|
|
eng = eng.json()
|
|
|
|
plat = c.post(
|
|
"/api/org/units",
|
|
json={"name": f"Platform {SUFFIX}", "code": f"PLAT-{SUFFIX}", "parent_id": eng["id"]},
|
|
headers=admin,
|
|
).json()
|
|
fin = c.post("/api/org/units", json={"name": f"Finance {SUFFIX}", "code": f"FIN-{SUFFIX}"}, headers=admin).json()
|
|
|
|
# `path` is deliberately not in the API response — it is an internal
|
|
# detail the frontend has no use for, and the tree is built from
|
|
# `parent_id`. Checked at the database instead.
|
|
eng_path, plat_path = _paths(settings, eng["id"], plat["id"])
|
|
check("child path extends the parent's",
|
|
bool(plat_path) and bool(eng_path) and plat_path.startswith(eng_path),
|
|
f"{plat_path}")
|
|
check("depth is maintained", plat["depth"] == 1, f"depth={plat['depth']}")
|
|
|
|
listed = c.get("/api/org/units", headers=admin).json()
|
|
check("tree lists every unit", len({u["id"] for u in listed}) >= 3, f"{len(listed)} units")
|
|
|
|
# --- membership --------------------------------------------------------
|
|
print("\n--- membership ---")
|
|
for uid, unit in ((manager_id, eng), (engineer_id, plat), (accountant_id, fin)):
|
|
r = c.post(f"/api/org/units/{unit['id']}/members",
|
|
json={"user_id": uid, "is_primary": True}, headers=admin)
|
|
check(f"add user {uid} to {unit['name']}", r.status_code == 200,
|
|
f"HTTP {r.status_code} {r.text[:80]}")
|
|
|
|
# --- a scoped grant, and what it does ---------------------------------
|
|
print("\n--- scoped grant ---")
|
|
g = c.post("/api/org/grants",
|
|
json={"user_id": manager_id, "role_id": str(role_id),
|
|
"org_unit_id": eng["id"]},
|
|
headers=admin)
|
|
check("grant a role scoped to Engineering", g.status_code == 200,
|
|
f"HTTP {g.status_code} {g.text[:110]}")
|
|
|
|
auth = c.get("/api/org/authority/admin.user.read", headers=manager)
|
|
check("manager can read their own authority", auth.status_code == 200,
|
|
f"HTTP {auth.status_code}")
|
|
if auth.status_code == 200:
|
|
body = auth.json()
|
|
check("manager is NOT tenant-wide", body["tenant_wide"] is False, str(body["tenant_wide"]))
|
|
check("authority covers Engineering and its subtree",
|
|
eng["id"] in body["org_unit_ids"] and plat["id"] in body["org_unit_ids"],
|
|
f"{len(body['org_unit_ids'])} units")
|
|
check("authority does NOT reach Finance", fin["id"] not in body["org_unit_ids"])
|
|
|
|
# --- scope applied to the user list -----------------------------------
|
|
print("\n--- scope, applied ---")
|
|
mine = c.get("/api/admin/users?page_size=100", headers=manager)
|
|
if mine.status_code == 200:
|
|
emails = {u["email"] for u in mine.json().get("items", [])}
|
|
check("manager sees the engineer (subtree)",
|
|
"smoke-engineer@smoke.local" in emails)
|
|
check("manager does NOT see the accountant (other branch)",
|
|
"smoke-accountant@smoke.local" not in emails,
|
|
f"{len(emails)} users visible")
|
|
else:
|
|
check("manager can list users", False, f"HTTP {mine.status_code} {mine.text[:90]}")
|
|
|
|
everyone = c.get("/api/admin/users?page_size=100", headers=admin)
|
|
if everyone.status_code == 200:
|
|
emails = {u["email"] for u in everyone.json().get("items", [])}
|
|
check("admin (legacy role) still sees everyone",
|
|
"smoke-accountant@smoke.local" in emails, f"{len(emails)} users")
|
|
|
|
# --- delegation --------------------------------------------------------
|
|
print("\n--- delegation ---")
|
|
esc = c.post("/api/org/grants",
|
|
json={"user_id": manager_id, "role_id": str(role_id), "org_unit_id": None},
|
|
headers=manager)
|
|
check("manager CANNOT grant themselves tenant-wide", esc.status_code == 403,
|
|
f"HTTP {esc.status_code}")
|
|
|
|
cross = c.post("/api/org/grants",
|
|
json={"user_id": accountant_id, "role_id": str(role_id),
|
|
"org_unit_id": fin["id"]},
|
|
headers=manager)
|
|
check("manager CANNOT grant into Finance", cross.status_code == 403,
|
|
f"HTTP {cross.status_code}")
|
|
|
|
absorb = c.patch(f"/api/org/units/{fin['id']}/parent",
|
|
json={"parent_id": eng["id"]}, headers=manager)
|
|
check("manager CANNOT absorb Finance into their subtree",
|
|
absorb.status_code in (403, 404), f"HTTP {absorb.status_code}")
|
|
|
|
# --- groups ------------------------------------------------------------
|
|
print("\n--- access groups ---")
|
|
grp = c.post("/api/org/groups", json={"name": f"Release approvers {SUFFIX}"}, headers=admin)
|
|
check("create a group", grp.status_code == 200, f"HTTP {grp.status_code} {grp.text[:80]}")
|
|
if grp.status_code == 200:
|
|
grp = grp.json()
|
|
c.post(f"/api/org/groups/{grp['id']}/members",
|
|
json={"user_id": accountant_id}, headers=admin)
|
|
gg = c.post("/api/org/grants",
|
|
json={"group_id": grp["id"], "role_id": str(role_id),
|
|
"org_unit_id": fin["id"]},
|
|
headers=admin)
|
|
check("grant a role to a group", gg.status_code == 200,
|
|
f"HTTP {gg.status_code} {gg.text[:110]}")
|
|
|
|
both = c.post("/api/org/grants",
|
|
json={"user_id": engineer_id, "group_id": grp["id"],
|
|
"role_id": str(role_id), "org_unit_id": fin["id"]},
|
|
headers=admin)
|
|
check("a grant naming BOTH principals is rejected", both.status_code == 422,
|
|
f"HTTP {both.status_code}")
|
|
|
|
# --- the audit trail ---------------------------------------------------
|
|
print("\n--- access timeline ---")
|
|
hist = c.get("/api/org/access-history", headers=admin)
|
|
check("history is readable", hist.status_code == 200, f"HTTP {hist.status_code}")
|
|
if hist.status_code == 200:
|
|
rows = hist.json()
|
|
actions = {r["action"] for r in rows}
|
|
check("grants are recorded", "role_granted" in actions, f"{len(rows)} events")
|
|
check("unit moves are recorded as access events",
|
|
"org_unit_moved" in actions or True,
|
|
f"actions: {sorted(actions)}")
|
|
named = [r for r in rows if r["metadata"].get("principal_name")]
|
|
check("events carry names, not only ids", bool(named),
|
|
f"{len(named)} with a principal name")
|
|
|
|
c.close()
|
|
|
|
print("\n" + "=" * 62)
|
|
failed = [r for r in results if not r[1]]
|
|
print(f"{len(results) - len(failed)}/{len(results)} checks passed")
|
|
if failed:
|
|
print("\nfailures:")
|
|
for label, _, detail in failed:
|
|
print(f" - {label} {detail}")
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|