encoach_backend_new_v2/test_training_flows.py

#!/usr/bin/env python3
"""
API-level write-flow sweep for the admin Training section
(`/admin/training/vocabulary`, `/admin/training/grammar`).

Exercises CREATE / EDIT / DELETE / FILTER / SEARCH / PROGRESS on both models.

Usage:
    python3 test_training_flows.py
"""
import json
import sys
import time
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:26s} {action:38s} {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 PATCH(path, body=None): return _req("PATCH", path, body=body or {})
def DELETE(path): return _req("DELETE", path)


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


def items(payload):
    return payload.get("items") or payload.get("data") or []


def extract_id(payload):
    if isinstance(payload, dict):
        if payload.get("id"):
            return payload["id"]
        d = payload.get("data")
        if isinstance(d, dict) and d.get("id"):
            return d["id"]
    return None


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


# ── Vocabulary ───────────────────────────────────────────────────────────
def test_vocab():
    print(f"{BOLD}== /admin/training/vocabulary =={RESET}")

    s, b = GET("/training/vocabulary")
    summary = (b or {}).get("summary") or {}
    log("vocabulary", "LIST seeded",
        ok(s) and len(items(b)) >= 10,
        f"status={s} total={summary.get('total')} completed={summary.get('completed')} rate={summary.get('completion_rate')}%")

    # Filter by level
    s, b = GET("/training/vocabulary", level="B2")
    all_b2 = all(v["level"] == "B2" for v in items(b))
    log("vocabulary", "LIST filter level=B2",
        ok(s) and all_b2 and len(items(b)) >= 1,
        f"status={s} count={len(items(b))} all_B2={all_b2}")

    # Search
    s, b = GET("/training/vocabulary", search="eloqu")
    has_eloquent = any("eloquent" in v["word"].lower() for v in items(b))
    log("vocabulary", "LIST search='eloqu'",
        ok(s) and has_eloquent,
        f"status={s} count={len(items(b))}")

    # CREATE
    word = f"Testword{int(time.time()*1000) % 100000}"
    s, b = POST("/training/vocabulary", {
        "word": word,
        "meaning": "A word created by the API smoke test.",
        "example_sentence": f"This is a {word} in a sentence.",
        "level": "C1",
        "part_of_speech": "noun",
        "category": "test",
    })
    vid = extract_id(b)
    log("vocabulary", "CREATE", ok(s) and vid,
        f"status={s} id={vid} word={word}")
    if not vid:
        return

    # Uniqueness — creating the same word should fail
    s, _ = POST("/training/vocabulary", {
        "word": word, "meaning": "duplicate attempt", "level": "C1",
    })
    log("vocabulary", "CREATE duplicate word rejected",
        s >= 400, f"status={s}")

    # Required fields — missing meaning
    s, _ = POST("/training/vocabulary", {"word": f"{word}-no-meaning"})
    log("vocabulary", "CREATE missing meaning rejected",
        s == 400, f"status={s}")

    # GET detail
    s, b = GET(f"/training/vocabulary/{vid}")
    log("vocabulary", "GET detail",
        ok(s) and b.get("word") == word,
        f"status={s} word={b.get('word')!r}")

    # EDIT level + meaning
    s, b = PATCH(f"/training/vocabulary/{vid}", {
        "level": "B2",
        "meaning": "An updated definition by the test.",
    })
    log("vocabulary", "EDIT level+meaning",
        ok(s) and b.get("level") == "B2" and "updated definition" in (b.get("meaning") or ""),
        f"status={s} level={b.get('level')}")

    # PROGRESS — mark completed + mastered
    s, b = POST(f"/training/vocabulary/{vid}/progress",
                {"completed": True, "mastery": "mastered"})
    log("vocabulary", "PROGRESS mark completed",
        ok(s) and b.get("completed") is True and b.get("mastery") == "mastered",
        f"status={s} completed={b.get('completed')}")

    # Verify summary reflects the completion after the new row was added
    s, b = GET("/training/vocabulary")
    summary2 = (b or {}).get("summary") or {}
    log("vocabulary", "SUMMARY reflects new completion",
        ok(s) and summary2.get("completed", 0) >= (summary.get("completed", 0) + 1),
        f"status={s} completed={summary2.get('completed')}")

    # PROGRESS — un-mark
    s, b = POST(f"/training/vocabulary/{vid}/progress", {"completed": False})
    log("vocabulary", "PROGRESS un-mark",
        ok(s) and b.get("completed") is False,
        f"status={s} completed={b.get('completed')}")

    # DELETE
    s, _ = DELETE(f"/training/vocabulary/{vid}")
    log("vocabulary", "DELETE", ok(s), f"status={s}")

    # GET after delete should 404
    s, _ = GET(f"/training/vocabulary/{vid}")
    log("vocabulary", "GET after DELETE returns 404",
        s == 404, f"status={s}")


# ── Grammar ──────────────────────────────────────────────────────────────
def test_grammar():
    print(f"{BOLD}== /admin/training/grammar =={RESET}")

    s, b = GET("/training/grammar")
    summary = (b or {}).get("summary") or {}
    log("grammar", "LIST seeded",
        ok(s) and len(items(b)) >= 8,
        f"status={s} total={summary.get('total')} completed={summary.get('completed')} rate={summary.get('completion_rate')}%")

    # Filter + search
    s, b = GET("/training/grammar", level="B1")
    all_b1 = all(r["level"] == "B1" for r in items(b))
    log("grammar", "LIST filter level=B1",
        ok(s) and all_b1 and len(items(b)) >= 1,
        f"status={s} count={len(items(b))} all_B1={all_b1}")

    s, b = GET("/training/grammar", search="passive")
    has_passive = any("passive" in r["name"].lower() for r in items(b))
    log("grammar", "LIST search='passive'",
        ok(s) and has_passive,
        f"status={s} count={len(items(b))}")

    # CREATE
    name = f"Test rule {int(time.time()*1000) % 100000}"
    s, b = POST("/training/grammar", {
        "name": name,
        "description": "A rule created by the API smoke test.",
        "example": "This is an example.",
        "level": "C1",
        "category": "test",
    })
    rid = extract_id(b)
    log("grammar", "CREATE",
        ok(s) and rid,
        f"status={s} id={rid}")
    if not rid:
        return

    # Required fields
    s, _ = POST("/training/grammar", {"name": f"{name}-no-desc"})
    log("grammar", "CREATE missing description rejected",
        s == 400, f"status={s}")

    # GET detail
    s, b = GET(f"/training/grammar/{rid}")
    log("grammar", "GET detail",
        ok(s) and b.get("name") == name,
        f"status={s} name={b.get('name')!r}")

    # EDIT
    s, b = PATCH(f"/training/grammar/{rid}", {
        "level": "B2",
        "category": "updated",
    })
    log("grammar", "EDIT level+category",
        ok(s) and b.get("level") == "B2" and b.get("category") == "updated",
        f"status={s} level={b.get('level')} category={b.get('category')}")

    # PROGRESS
    s, b = POST(f"/training/grammar/{rid}/progress", {"completed": True})
    log("grammar", "PROGRESS mark completed",
        ok(s) and b.get("completed") is True,
        f"status={s}")

    s, b = POST(f"/training/grammar/{rid}/progress", {"completed": False})
    log("grammar", "PROGRESS un-mark",
        ok(s) and b.get("completed") is False,
        f"status={s}")

    # DELETE
    s, _ = DELETE(f"/training/grammar/{rid}")
    log("grammar", "DELETE", ok(s), f"status={s}")

    s, _ = GET(f"/training/grammar/{rid}")
    log("grammar", "GET after DELETE returns 404",
        s == 404, f"status={s}")


# ── Cross-cutting: pagination sanity ─────────────────────────────────────
def test_pagination():
    print(f"{BOLD}== Pagination / envelope =={RESET}")
    s, b = GET("/training/vocabulary", size=3, page=1)
    log("pagination", "vocab size=3 page=1",
        ok(s) and len(items(b)) <= 3 and b.get("size") == 3 and b.get("page") == 1,
        f"status={s} got={len(items(b))} size={b.get('size')} page={b.get('page')}")
    s, b = GET("/training/grammar", size=3, page=2)
    log("pagination", "grammar size=3 page=2",
        ok(s) and b.get("page") == 2,
        f"status={s} got={len(items(b))} page={b.get('page')}")


# ── Main ─────────────────────────────────────────────────────────────────
def main():
    if not login():
        print("Login failed — is Odoo running at :8069?")
        sys.exit(1)

    test_vocab()
    test_grammar()
    test_pagination()

    print()
    total = len(results)
    passed = sum(1 for _, _, ok_, _ in results if ok_)
    failed = total - passed
    color = GREEN if failed == 0 else RED
    print("=" * 70)
    print(f"{BOLD}SUMMARY  {color}PASS={passed}{RESET}{BOLD}  "
          f"{RED if failed else GREEN}FAIL={failed}{RESET}{BOLD}  TOTAL={total}{RESET}")
    print("=" * 70)
    if failed:
        print(f"\n{BOLD}FAILURES{RESET}")
        for page, action, ok_, note in results:
            if not ok_:
                print(f"  {RED}×{RESET} {page:26s} {action:38s} {note}")
        sys.exit(1)


if __name__ == "__main__":
    main()