full_encoach_platform/smoke_entity_isolation.py

"""End-to-end smoke test for LMS entity isolation.

Verifies that an admin scoped to entity A cannot:
  • see entity B's courses / classrooms / batches / students / teachers
  • assign an entity-B student into an entity-A classroom
  • assign an entity-B teacher into an entity-A classroom
  • create a batch that mixes entities (course/classroom/section/teachers)
  • assign an entity-B course (or its section) to an entity-A classroom

We do this entirely through the REST API the admin UI uses. We bootstrap
entity B with a *superadmin* (`admin / admin`), then run the cross-entity
attempts as the entity-A admin (`admin@encoach.test`).
"""
from __future__ import annotations
import json
import sys
import urllib.request
import urllib.error

API = "http://127.0.0.1:8069"
SUPER_LOGIN = "admin"
SUPER_PASSWORD = "admin"
A_LOGIN = "admin@encoach.test"
A_PASSWORD = "admin123"


def _req(method, path, token=None, body=None):
    url = f"{API}{path}"
    data = None
    headers = {"Content-Type": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    if body is not None:
        data = json.dumps(body).encode()
    req = urllib.request.Request(url, data=data, method=method, headers=headers)
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            payload = r.read().decode() or "{}"
            return r.status, json.loads(payload) if payload.strip() else {}
    except urllib.error.HTTPError as e:
        body = e.read().decode() if e.fp else ""
        try:
            parsed = json.loads(body)
        except Exception:
            parsed = {"raw": body}
        return e.code, parsed


def login(login, password):
    code, data = _req("POST", "/api/login", body={"login": login, "password": password})
    assert code == 200, f"login {login}: {code} {data}"
    return data["access_token"]


def expect_ok(label, code, data, ok_codes=(200,)):
    if code not in ok_codes:
        raise AssertionError(f"{label}: expected {ok_codes}, got {code} -> {data}")
    print(f"  OK   {label}  ({code})")


def expect_blocked(label, code, data):
    if code in (200, 201):
        raise AssertionError(
            f"{label}: expected blocked but got {code} -> {data}"
        )
    print(f"  OK   {label}  → blocked ({code})")


def main():
    super_tok = login(SUPER_LOGIN, SUPER_PASSWORD)
    print(f"[1] superadmin token acquired")

    code, ents = _req("GET", "/api/entities?size=200", token=super_tok)
    items = ents.get("items", ents.get("data", []))
    by_code = {e.get("code"): e for e in items}
    entity_a = next((e for e in items if e.get("id") == 1), items[0] if items else None)
    if not entity_a:
        print("FATAL: no entities present")
        sys.exit(1)
    print(f"[2] entity A = {entity_a['name']} (id={entity_a['id']})")

    entity_b = next(
        (e for e in items if e.get("id") != entity_a["id"] and e.get("code", "").startswith("ISO_TEST")),
        None,
    )
    if not entity_b:
        code, created = _req(
            "POST", "/api/entities", token=super_tok,
            body={"name": "Iso Test Entity", "code": "ISO_TEST", "type": "school"},
        )
        if code in (200, 201):
            entity_b = created.get("data") or created
        else:
            print(f"FATAL: cannot create entity B: {code} {created}")
            sys.exit(1)
    print(f"[3] entity B = {entity_b['name']} (id={entity_b['id']})")

    code, ulist = _req(
        "GET", f"/api/entities/{entity_b['id']}/users", token=super_tok,
    )
    cur = (ulist.get("items") or ulist.get("data") or []) if code == 200 else []
    cur_ids = [u["id"] for u in cur]
    if 2 not in cur_ids:
        cur_ids.append(2)
        code, _ = _req(
            "PATCH", f"/api/entities/{entity_b['id']}/users", token=super_tok,
            body={"user_ids": cur_ids},
        )
        if code in (200, 201):
            print(f"  (super admin uid=2 attached to entity B; users now={cur_ids})")
        else:
            print(f"  WARN: cannot attach uid=2 to entity B: {code}")

    code, listing = _req(
        "GET", f"/api/courses?entity_id={entity_b['id']}&size=200", token=super_tok,
    )
    course_items = (listing.get("items") or listing.get("data") or [])
    existing = next((c for c in course_items if c.get("code", "").startswith("EB-ISO")), None)
    if existing:
        course_b_id = existing["id"]
        print(f"  (reusing existing entity-B course id={course_b_id})")
    else:
        import time
        course_code = f"EB-ISO{int(time.time())%100000}"
        code, course_b = _req(
            "POST", "/api/courses", token=super_tok,
            body={"name": "Entity-B Course", "code": course_code, "entity_id": entity_b["id"]},
        )
        expect_ok("create course in entity B (as super)", code, course_b, (200, 201))
        course_b_id = (course_b.get("data") or course_b)["id"]

    code, sect_b = _req(
        "POST", f"/api/courses/{course_b_id}/sections/generate-defaults",
        token=super_tok, body={"codes": ["A"]},
    )
    expect_ok("generate section A in entity B (as super)", code, sect_b, (200, 201))
    sect_b_id = (sect_b.get("items") or sect_b.get("data"))[0]["id"]

    code, slist = _req(
        "GET", f"/api/students?entity_id={entity_b['id']}&search=EntityB", token=super_tok,
    )
    s_items = slist.get("items") or slist.get("data") or []
    existing_s = next((s for s in s_items if 'EntityB' in (s.get('name') or '')), None)
    if existing_s:
        stud_b_id = existing_s["id"]
        print(f"  (reusing existing entity-B student id={stud_b_id})")
    else:
        code, stud_b = _req(
            "POST", "/api/students", token=super_tok,
            body={
                "first_name": "EntityB", "last_name": "Student",
                "email": f"entityb.student.{entity_b['id']}@iso.test",
                "create_portal_user": False,
                "gender": "m",
                "entity_id": entity_b["id"],
            },
        )
        expect_ok("create student in entity B (as super)", code, stud_b, (200, 201))
        stud_b_id = (stud_b.get("data") or stud_b)["id"]

    code, tlist = _req(
        "GET", f"/api/teachers?entity_id={entity_b['id']}&search=EntityB", token=super_tok,
    )
    t_items = tlist.get("items") or tlist.get("data") or []
    existing_t = next((t for t in t_items if 'EntityB' in (t.get('name') or '')), None)
    if existing_t:
        teach_b_id = existing_t["id"]
        print(f"  (reusing existing entity-B teacher id={teach_b_id})")
    else:
        code, teach_b = _req(
            "POST", "/api/teachers", token=super_tok,
            body={
                "first_name": "EntityB", "last_name": "Teacher",
                "email": f"entityb.teacher.{entity_b['id']}@iso.test",
                "gender": "female",
                "entity_id": entity_b["id"],
            },
        )
        expect_ok("create teacher in entity B (as super)", code, teach_b, (200, 201))
        teach_b_id = (teach_b.get("data") or teach_b)["id"]

    code, rlist = _req(
        "GET", f"/api/classrooms?entity_id={entity_b['id']}&search=EntityB", token=super_tok,
    )
    r_items = rlist.get("items") or rlist.get("data") or []
    existing_r = next((r for r in r_items if 'EntityB' in (r.get('name') or '')), None)
    if existing_r:
        room_b_id = existing_r["id"]
        print(f"  (reusing existing entity-B classroom id={room_b_id})")
    else:
        import time as _t
        code, room_b = _req(
            "POST", "/api/classrooms", token=super_tok,
            body={"name": "EntityB Room", "code": f"EBR{int(_t.time())%100000}", "entity_id": entity_b["id"]},
        )
        expect_ok("create classroom in entity B (as super)", code, room_b, (200, 201))
        room_b_id = (room_b.get("data") or room_b)["id"]

    a_tok = login(A_LOGIN, A_PASSWORD)
    print(f"\n[4] entity-A admin token acquired (entity A only)")

    print(f"\n[5] visibility checks — entity-A admin must NOT see entity B's records")
    for path in ("/api/courses", "/api/classrooms", "/api/students", "/api/teachers", "/api/batches"):
        code, data = _req("GET", f"{path}?size=500", token=a_tok)
        items = data.get("items") or data.get("data") or []
        leaks = [
            it for it in items
            if (it.get("entity_id") or (it.get("entity") or {}).get("id")) == entity_b["id"]
        ]
        if leaks:
            raise AssertionError(f"{path} leaked entity-B records: {leaks[:3]}")
        print(f"  OK   {path} hides entity-B  ({len(items)} entity-A items)")

    print(f"\n[6] write-path checks — entity-A admin must be BLOCKED from cross-entity writes")
    code, room_a_list = _req(
        "GET", f"/api/classrooms?entity_id={entity_a['id']}&size=10", token=a_tok,
    )
    rooms_a = room_a_list.get("items") or room_a_list.get("data") or []
    if not rooms_a:
        code, room_a = _req(
            "POST", "/api/classrooms", token=a_tok,
            body={"name": "Iso Test Class A", "code": "ITA1"},
        )
        expect_ok("create classroom in entity A", code, room_a, (200, 201))
        room_a_id = (room_a.get("data") or room_a)["id"]
    else:
        room_a_id = rooms_a[0]["id"]
    print(f"  using classroom A id={room_a_id}")

    code, data = _req(
        "POST", f"/api/classrooms/{room_a_id}/students", token=a_tok,
        body={"student_ids": [stud_b_id], "mode": "add"},
    )
    expect_blocked("attach entity-B student to entity-A classroom", code, data)

    code, data = _req(
        "POST", f"/api/classrooms/{room_a_id}/teachers", token=a_tok,
        body={"teacher_ids": [teach_b_id], "mode": "add"},
    )
    expect_blocked("attach entity-B teacher to entity-A classroom", code, data)

    code, data = _req(
        "POST", f"/api/classrooms/{room_a_id}/assign-course", token=a_tok,
        body={"course_id": course_b_id, "term_key": "2026-T1"},
    )
    expect_blocked("assign entity-B course to entity-A classroom", code, data)

    code, data = _req(
        "POST", f"/api/classrooms/{room_a_id}/assign-section", token=a_tok,
        body={"course_id": course_b_id, "section_id": sect_b_id, "term_key": "2026-T1"},
    )
    expect_blocked("assign entity-B section to entity-A classroom", code, data)

    code, data = _req(
        "POST", "/api/batches", token=a_tok,
        body={"name": "Mixed Batch", "course_id": course_b_id, "term_key": "2026-T1"},
    )
    expect_blocked("create batch with entity-B course as entity-A admin", code, data)

    code, data = _req(
        "POST", "/api/batches", token=a_tok,
        body={
            "name": "Mixed Batch 2", "course_id": course_b_id,
            "course_section_id": sect_b_id, "term_key": "2026-T1",
        },
    )
    expect_blocked("create batch with entity-B section as entity-A admin", code, data)

    code, data = _req(
        "POST", "/api/batches", token=a_tok,
        body={"name": "Mixed Batch 3", "teacher_ids": [teach_b_id]},
    )
    expect_blocked("create batch with entity-B teacher as entity-A admin", code, data)

    print(f"\n[7] sanity — entity-A admin can still operate on its own classroom")
    code, room_a_full = _req("GET", f"/api/classrooms/{room_a_id}", token=a_tok)
    expect_ok("read own classroom A", code, room_a_full)

    print("\nALL ENTITY-ISOLATION CHECKS PASSED")


if __name__ == "__main__":
    main()