fastapi-tutorial/test_main.py at main · rafi021/fastapi-tutorial · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
import os

os.environ["DATABASE_URL"] = "sqlite:///./test.db"
if os.path.exists("test.db"):
    os.remove("test.db")

from fastapi.testclient import TestClient

from app.db.base import Base
from app.db.init_db import init_db
from app.db.session import SessionLocal, engine
from main import app

Base.metadata.create_all(bind=engine)
db = SessionLocal()
try:
    init_db(db)
finally:
    db.close()

client = TestClient(app)


def get_auth_header() -> dict[str, str]:
    response = client.post(
        "/api/v1/auth/token",
        data={"username": "admin", "password": "admin123"},
    )
    assert response.status_code == 200
    token = response.json()["access_token"]

    refresh_response = client.post(
        "/api/v1/auth/refresh",
        json={"refresh_token": response.json()["refresh_token"]},
    )
    assert refresh_response.status_code == 200

    return {"Authorization": f"Bearer {token}"}


def test_health() -> None:
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "ok"


def test_request_id_header_is_set() -> None:
    response = client.get("/health")
    assert response.status_code == 200
    assert "X-Request-ID" in response.headers
    assert "X-Trace-ID" in response.headers


def test_health_dependencies() -> None:
    response = client.get("/health/dependencies")
    assert response.status_code in (200, 503)
    body = response.json()
    assert "dependencies" in body
    assert "redis" in body["dependencies"]
    assert "rabbitmq" in body["dependencies"]
    assert isinstance(body["dependencies"]["redis"]["ready"], bool)
    assert isinstance(body["dependencies"]["rabbitmq"]["ready"], bool)


def test_metrics_endpoint() -> None:
    response = client.get("/metrics")
    assert response.status_code == 200
    assert "http_requests_total" in response.text


def test_category_crud_flow() -> None:
    headers = get_auth_header()

    create_response = client.post(
        "/api/v1/categories",
        headers=headers,
        json={"name": "Electronics", "description": "Devices and gadgets"},
    )
    assert create_response.status_code == 201
    category_id = create_response.json()["id"]

    list_response = client.get(
        "/api/v1/categories?search=Elect&page=1&size=10",
        headers=headers,
    )
    assert list_response.status_code == 200
    assert list_response.json()["total"] >= 1

    update_response = client.put(
        f"/api/v1/categories/{category_id}",
        headers=headers,
        json={"name": "Electronics and Gadgets"},
    )
    assert update_response.status_code == 200
    assert update_response.json()["name"] == "Electronics and Gadgets"

    delete_response = client.delete(f"/api/v1/categories/{category_id}", headers=headers)
    assert delete_response.status_code == 204