full_encoach_platform/test_reports_flows.py

#!/usr/bin/env python3
"""API smoke tests for the admin Reports section.

Covers the three pages in `AdminLmsLayout.tsx` under "Reports":

  * /admin/student-performance  -> /api/reports/student-performance
  * /admin/stats-corporate      -> /api/reports/stats-corporate
  * /admin/record               -> /api/reports/record
  * (shared filter picker)      -> /api/reports/filters

Usage:
    python3 test_reports_flows.py
"""
import json
import sys
from urllib import request as urlreq
from urllib.error import HTTPError

BASE = "http://localhost:8069/api"
TOKEN = None

GREEN = "\033[32m"; RED = "\033[31m"; RESET = "\033[0m"; BOLD = "\033[1m"

results = []


def log(page, action, ok_, note=""):
    results.append((page, action, ok_, note))
    tag = f"{GREEN}PASS{RESET}" if ok_ else f"{RED}FAIL{RESET}"
    print(f"  [{tag}] {page:22s} {action:42s} {note}")


def _req(method, path, body=None, params=None):
    url = BASE + path
    if params:
        qs = "&".join(f"{k}={v}" for k, v in params.items() if v is not None)
        url += ("?" + qs) if qs else ""
    data = None
    headers = {"Accept": "application/json"}
    if TOKEN:
        headers["Authorization"] = f"Bearer {TOKEN}"
    if body is not None:
        data = json.dumps(body).encode()
        headers["Content-Type"] = "application/json"
    req = urlreq.Request(url, data=data, method=method, headers=headers)
    try:
        with urlreq.urlopen(req, timeout=20) as resp:
            raw = resp.read().decode()
            return resp.status, (json.loads(raw) if raw else {})
    except HTTPError as e:
        raw = e.read().decode() if e.fp else ""
        try:
            payload = json.loads(raw)
        except Exception:
            payload = {"error": raw[:200]}
        return e.code, payload
    except Exception as e:
        return 0, {"error": str(e)}


def GET(path, **params): return _req("GET", path, params=params)
def POST(path, body=None): return _req("POST", path, body=body or {})


def ok(s): return 200 <= s < 300


def login():
    global TOKEN
    s, b = POST("/login", {"email": "admin", "password": "admin"})
    TOKEN = (b or {}).get("token")
    return bool(TOKEN)


# ── Filters picker ──────────────────────────────────────────────────────
def test_filters():
    print(f"{BOLD}== /api/reports/filters =={RESET}")
    s, b = GET("/reports/filters")
    entities = (b or {}).get("entities") or []
    students = (b or {}).get("students") or []
    log("reports-filters", "GET", ok(s) and isinstance(entities, list) and isinstance(students, list),
        f"status={s} entities={len(entities)} students={len(students)}")
    return entities, students


# ── 1. Student Performance ─────────────────────────────────────────────
def test_student_performance(entities):
    print(f"{BOLD}== /admin/student-performance =={RESET}")

    s, b = GET("/reports/student-performance")
    rows = (b or {}).get("items") or []
    log("student-performance", "LIST returns rows",
        ok(s) and isinstance(rows, list) and len(rows) > 0,
        f"status={s} rows={len(rows)}")

    if rows:
        r0 = rows[0]
        expected_keys = {"student_id", "student_name", "entity_name",
                         "reading", "listening", "writing", "speaking",
                         "overall", "level", "attempts_count"}
        log("student-performance", "row shape complete",
            expected_keys.issubset(r0.keys()),
            f"missing={sorted(expected_keys - set(r0.keys()))}")

        log("student-performance", "overall is numeric/None",
            r0["overall"] is None or isinstance(r0["overall"], (int, float)),
            f"overall={r0['overall']}")

        log("student-performance", "CEFR level assigned",
            r0["level"] in {"Pre-A1","A1","A2","B1","B2","C1","C2", None},
            f"level={r0['level']}")

        s2, b2 = GET("/reports/student-performance",
                     search=(r0["student_name"] or "").split()[0])
        rows2 = (b2 or {}).get("items") or []
        log("student-performance", "search filter",
            ok(s2) and any(r["student_id"] == r0["student_id"] for r in rows2),
            f"status={s2} matched={len(rows2)}")

        if r0["level"]:
            s3, b3 = GET("/reports/student-performance", level=r0["level"])
            rows3 = (b3 or {}).get("items") or []
            log("student-performance", "level filter",
                ok(s3) and all(r["level"] == r0["level"] for r in rows3),
                f"status={s3} filtered={len(rows3)}")

    if entities:
        eid = entities[0]["id"]
        s4, b4 = GET("/reports/student-performance", entity_id=eid)
        rows4 = (b4 or {}).get("items") or []
        log("student-performance", "entity filter",
            ok(s4) and all(r["entity_id"] in (None, eid) or r["entity_id"] == eid for r in rows4),
            f"status={s4} entity_id={eid} rows={len(rows4)}")


# ── 2. Stats Corporate ─────────────────────────────────────────────────
def test_stats_corporate(entities):
    print(f"{BOLD}== /admin/stats-corporate =={RESET}")

    s, b = GET("/reports/stats-corporate")
    by_module = (b or {}).get("by_module") or []
    trend = (b or {}).get("trend") or []
    dist = (b or {}).get("distribution") or []
    comp = (b or {}).get("comparison") or []
    meta = (b or {}).get("meta") or {}

    log("stats-corporate", "LIST returns envelope",
        ok(s) and all(k in (b or {}) for k in
                       ("by_module", "trend", "distribution",
                        "comparison", "meta")),
        f"status={s} keys={sorted((b or {}).keys()) if b else 'n/a'}")
    log("stats-corporate", "by_module has 4 modules",
        len(by_module) == 4 and {m["module"] for m in by_module} ==
        {"Reading","Listening","Writing","Speaking"},
        f"modules={[m['module'] for m in by_module]}")
    log("stats-corporate", "trend length == 6 (default months)",
        len(trend) == 6,
        f"trend={len(trend)}")
    log("stats-corporate", "distribution has CEFR levels",
        isinstance(dist, list) and len(dist) > 0 and
        all("name" in d and "value" in d and "color" in d for d in dist),
        f"bins={len(dist)}")
    log("stats-corporate", "comparison is entity-level",
        isinstance(comp, list) and all("entity_name" in c for c in comp),
        f"entities={len(comp)}")
    log("stats-corporate", "meta.attempts_considered > 0",
        meta.get("attempts_considered", 0) > 0,
        f"meta={meta}")

    # Threshold filter -> should narrow attempts considered.
    s2, b2 = GET("/reports/stats-corporate", threshold=70)
    log("stats-corporate", "threshold=70 filter",
        ok(s2) and (b2 or {}).get("meta", {}).get("threshold") == 70.0,
        f"status={s2} meta={(b2 or {}).get('meta')}")

    # Entity filter
    if entities:
        eid = entities[0]["id"]
        s3, b3 = GET("/reports/stats-corporate", entity_id=eid)
        cmp3 = (b3 or {}).get("comparison") or []
        log("stats-corporate", "entity filter narrows comparison",
            ok(s3) and all(c["entity_id"] in (None, eid) or c["entity_id"] == eid
                            for c in cmp3),
            f"status={s3} entities={len(cmp3)}")

    # Months horizon
    s4, b4 = GET("/reports/stats-corporate", months=3)
    log("stats-corporate", "months=3 trend len",
        ok(s4) and len((b4 or {}).get("trend") or []) == 3,
        f"status={s4} trend={len((b4 or {}).get('trend') or [])}")


# ── 3. Record ──────────────────────────────────────────────────────────
def test_record(entities, students):
    print(f"{BOLD}== /admin/record =={RESET}")

    s, b = GET("/reports/record", size=10)
    items = (b or {}).get("items") or []
    log("record", "LIST (paginated)", ok(s) and isinstance(items, list),
        f"status={s} got={len(items)} total={(b or {}).get('total')}")

    if items:
        r0 = items[0]
        keys = {"id", "student_name", "assignment", "exam", "exam_code",
                "date", "score", "duration_min", "status", "status_label"}
        log("record", "row shape complete",
            keys.issubset(r0.keys()),
            f"missing={sorted(keys - set(r0.keys()))}")
        log("record", "exam_code is EX-###",
            r0["exam_code"].startswith("EX-") if r0["exam_code"] else True,
            f"exam_code={r0['exam_code']}")
        log("record", "status_label is titlecase",
            isinstance(r0["status_label"], str) and r0["status_label"][:1].isupper(),
            f"status_label={r0['status_label']}")

    if students:
        sid = students[0]["id"]
        s2, b2 = GET("/reports/record", user_id=sid, size=50)
        items2 = (b2 or {}).get("items") or []
        log("record", "user_id filter",
            ok(s2) and all(i["student_id"] == sid for i in items2),
            f"status={s2} user_id={sid} rows={len(items2)}")

    if entities:
        eid = entities[0]["id"]
        s3, b3 = GET("/reports/record", entity_id=eid, size=50)
        items3 = (b3 or {}).get("items") or []
        log("record", "entity_id filter",
            ok(s3) and all(i["entity_id"] in (None, eid) or i["entity_id"] == eid
                            for i in items3),
            f"status={s3} entity_id={eid} rows={len(items3)}")

    # Period filter (month should return most data)
    s4, b4 = GET("/reports/record", period="month", size=50)
    log("record", "period=month filter",
        ok(s4) and isinstance((b4 or {}).get("items"), list),
        f"status={s4} rows={len((b4 or {}).get('items') or [])}")

    # Pagination
    s5, b5 = GET("/reports/record", size=5, page=1)
    s6, b6 = GET("/reports/record", size=5, page=2)
    log("record", "pagination size=5",
        ok(s5) and ok(s6)
        and len((b5 or {}).get("items") or []) <= 5
        and len((b6 or {}).get("items") or []) <= 5,
        f"p1={len((b5 or {}).get('items') or [])} p2={len((b6 or {}).get('items') or [])}")


# ── Main ────────────────────────────────────────────────────────────────
if __name__ == "__main__":
    print(f"{BOLD}Reports Section API Smoke Tests{RESET}")
    if not login():
        print(f"{RED}Login failed. Is Odoo running on :8069 with admin/admin?{RESET}")
        sys.exit(1)
    print(f"  logged in, token={TOKEN[:12] if TOKEN else 'none'}...\n")

    entities, students = test_filters()
    print()
    test_student_performance(entities)
    print()
    test_stats_corporate(entities)
    print()
    test_record(entities, students)

    total = len(results)
    passed = sum(1 for r in results if r[2])
    failed = total - passed
    print("\n" + "=" * 70)
    print(f"{BOLD}Summary: {GREEN}{passed} passed{RESET}, "
          f"{RED}{failed} failed{RESET}, {total} total")
    if failed:
        print(f"\n{RED}Failing:{RESET}")
        for page, action, ok_, note in results:
            if not ok_:
                print(f"  - {page:22s} {action:42s} {note}")
        sys.exit(2)