tests.py - main - mkiiis/task_tracker

tests.py
165 строк6 KB
mkiiis
lab4
14 дек 2025, 05:32
14 дек 2025, 05:3288d2183
import os, json, uuid, time, sys
import pika

RABBIT_URL = os.getenv("RABBIT_URL", "amqp://lab4u:lab4p@localhost:5672/%2F")
INTERNAL_API_KEY = os.getenv("INTERNAL_API_KEY", "int123")

def rpc_call(action, data, version="v1", token=""):
    conn = pika.BlockingConnection(pika.URLParameters(RABBIT_URL))
    ch = conn.channel()
    corr_id = str(uuid.uuid4())

    ch.basic_publish(
        exchange="api",
        routing_key="requests",
        properties=pika.BasicProperties(correlation_id=corr_id),
        body=json.dumps({
            "id": corr_id,
            "version": version,
            "action": action,
            "data": data,
            "auth": token or "",
        }).encode("utf-8"),
    )

    deadline = time.time() + 30
    got = None
    while time.time() < deadline:
        m, p, b = ch.basic_get(queue="api.responses", auto_ack=True)
        if b:
            resp = json.loads(b.decode("utf-8"))
            if resp.get("correlation_id") in (None, corr_id):
                got = resp
                break
        time.sleep(0.2)

    conn.close()
    if not got:
        raise TimeoutError(f"No response for {action} within 15s")
    if got.get("status") != "ok":
        raise RuntimeError(f"{action} error: {got.get('error')}")
    return got["data"]

def must(cond: bool, msg: str):
    if not cond:
        raise AssertionError(msg)

def main():
    print(f"RABBIT_URL={RABBIT_URL}")
    # ---------- register ----------
    email = f"user_{uuid.uuid4().hex[:6]}@example.com"
    name = "User1"
    password = "pass123"
    print("register:", email)
    reg = rpc_call("register", {"email": email, "name": name, "password": password})
    must("access_token" in reg and "refresh_token" in reg, "register must return tokens")

    # ---------- login ----------
    print("login")
    login = rpc_call("login", {"email": email, "password": password})
    token = login["access_token"]
    must(token and isinstance(token, str), "login must return access_token")

    # ---------- projects ----------
    print("create_project")
    proj = rpc_call("create_project", {"name": "Demo", "description": "First"}, token=token)
    proj_id = proj["id"]
    must(proj_id > 0, "project id must be positive")

    print("list_projects")
    projs = rpc_call("list_projects", {"limit": 10, "offset": 0}, token=token)
    must(any(p["id"] == proj_id for p in projs), "created project should be in list")

    print("get_project")
    gp = rpc_call("get_project", {"project_id": proj_id}, token=token)
    must(gp["id"] == proj_id, "get_project id mismatch")

    print("update_project")
    up = rpc_call("update_project", {"project_id": proj_id, "name": "DemoRenamed"}, token=token)
    must(up["name"] == "DemoRenamed", "update_project didn't change name")

    # ---------- tasks v1 ----------
    print("[TASK v1] create_task_v1")
    t1 = rpc_call("create_task_v1", {
        "project_id": proj_id,
        "title": "T1",
        "description": "desc"
    }, token=token)
    task_id = t1["id"]
    must(task_id > 0, "task id must be positive")

    print("[TASK v1] list_tasks_v1")
    lt1 = rpc_call("list_tasks_v1", {"project_id": proj_id, "limit": 10, "offset": 0}, token=token)
    must(any(t["id"] == task_id for t in lt1), "created task (v1) should be in list")
    print("[TASK v1] get_task_v1 with include")
    gt1 = rpc_call("get_task_v1", {"task_id": task_id, "include": "project,comments"}, token=token)
    must(gt1["id"] == task_id, "get_task_v1 id mismatch")
    must("project" in gt1 and gt1["project"]["id"] == proj_id, "get_task_v1 include project missing")

    print("[TASK v1] update_task_v1")
    ut1 = rpc_call("update_task_v1", {"task_id": task_id, "status": "done"}, token=token)
    must(ut1["status"] == "done", "update_task_v1 didn't update status")

    # ---------- comments v1 ----------
    print("create_comment_v1")
    c1 = rpc_call("create_comment_v1", {"task_id": task_id, "body": "hello"}, token=token)
    cid = c1["id"]
    must(cid > 0, "comment id must be positive")

    print("list_comments_v1")
    lc = rpc_call("list_comments_v1", {"task_id": task_id, "limit": 10, "offset": 0}, token=token)
    must(any(c["id"] == cid for c in lc), "created comment must be in list")

    print("update_comment_v1")
    uc = rpc_call("update_comment_v1", {"task_id": task_id, "comment_id": cid, "body": "edited"}, token=token)
    must(uc["body"] == "edited", "update_comment_v1 didn't change body")

    # ---------- tasks v2 ----------
    print("[TASK v2] create_task_v2")
    t2 = rpc_call("create_task_v2", {
        "project_id": proj_id,
        "title": "T2",
        "description": "d2",
        "estimated_time_minutes": 30
    }, version="v1", token=token)  
    task2_id = t2["id"]
    must(task2_id > 0, "task v2 id must be positive")

    print("[TASK v2] get_task_v2")
    gt2 = rpc_call("get_task_v2", {"task_id": task2_id, "include": "project,comments"}, version="v2", token=token)
    must(gt2["id"] == task2_id, "get_task_v2 id mismatch")
    must("project" in gt2 and gt2["project"]["id"] == proj_id, "get_task_v2 include project missing")

    print("[TASK v2] update_task_v2")
    ut2 = rpc_call("update_task_v2", {"task_id": task2_id, "priority": 3, "estimated_time_minutes": 45}, version="v2", token=token)
    must(ut2.get("estimated_time_minutes") == 45, "update_task_v2 didn't change estimated_time_minutes")

    # ---------- internal stats ----------
    print("[INTERNAL] internal_get_stats")
    stats = rpc_call("internal_get_stats", {}, token=INTERNAL_API_KEY)
    for k in ("total_users", "total_projects", "total_tasks", "total_comments"):
        must(k in stats, f"stats must contain {k}")

    # ---------- cleanup ----------
    print("delete_comment_v1")
    rpc_call("delete_comment_v1", {"task_id": task_id, "comment_id": cid}, token=token)

    print("delete_task_v1")
    rpc_call("delete_task_v1", {"task_id": task_id}, token=token)

    print("delete_task_v2")
    rpc_call("delete_task_v2", {"task_id": task2_id}, version="v2", token=token)

    print("delete_project_v1")
    rpc_call("delete_project", {"project_id": proj_id}, token=token)


    print("\nPASSED")
    return 0

if __name__ == "__main__":
    try:
        sys.exit(main())
    except Exception as e:
        print("\nFAILED:", e)
        sys.exit(2)