File size: 2,519 Bytes
b2c1c67
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
from httpx import AsyncClient

from tests.conftest import register_and_login


async def test_register_login_me(client: AsyncClient) -> None:
    headers = await register_and_login(client)

    me = await client.get("/api/auth/me", headers=headers)
    assert me.status_code == 200
    body = me.json()
    assert body["email"] == "adi@example.com"
    assert body["username"] == "adi"
    assert "password" not in body and "password_hash" not in body


async def test_duplicate_email_rejected(client: AsyncClient) -> None:
    await register_and_login(client)
    response = await client.post(
        "/api/auth/register",
        json={"email": "ADI@example.com", "username": "xx", "password": "supersecret123"},
    )
    assert response.status_code == 409


async def test_wrong_password_rejected(client: AsyncClient) -> None:
    await register_and_login(client)
    response = await client.post(
        "/api/auth/login",
        json={"email": "adi@example.com", "password": "wrong-password"},
    )
    assert response.status_code == 401


async def test_short_password_rejected(client: AsyncClient) -> None:
    response = await client.post(
        "/api/auth/register",
        json={"email": "a@b.com", "username": "a", "password": "short"},
    )
    assert response.status_code == 422


async def test_refresh_flow(client: AsyncClient) -> None:
    register = await client.post(
        "/api/auth/register",
        json={"email": "r@example.com", "username": "rr", "password": "supersecret123"},
    )
    refresh_token = register.json()["refresh_token"]

    refreshed = await client.post(
        "/api/auth/refresh", json={"refresh_token": refresh_token}
    )
    assert refreshed.status_code == 200
    new_access = refreshed.json()["access_token"]
    me = await client.get(
        "/api/auth/me", headers={"Authorization": f"Bearer {new_access}"}
    )
    assert me.status_code == 200


async def test_access_token_cannot_be_used_as_refresh(client: AsyncClient) -> None:
    register = await client.post(
        "/api/auth/register",
        json={"email": "t@example.com", "username": "tt", "password": "supersecret123"},
    )
    access_token = register.json()["access_token"]
    response = await client.post(
        "/api/auth/refresh", json={"refresh_token": access_token}
    )
    assert response.status_code == 401


async def test_protected_route_requires_token(client: AsyncClient) -> None:
    response = await client.get("/api/sessions")
    assert response.status_code == 401