fastapi-sqlalchemy-template/tests/test_cards.py at b01acc0f5d9e46fd7c5a9d096f3e4ade681f1c66 · modern-python/fastapi-sqlalchemy-template · 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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
from typing import TYPE_CHECKING

import pytest
from fastapi import status

from tests import factories


if TYPE_CHECKING:
    from httpx import AsyncClient


pytestmark = [pytest.mark.usefixtures("set_async_session_in_base_sqlalchemy_factory")]


async def test_get_cards_empty(client: AsyncClient) -> None:
    deck = await factories.DeckModelFactory.create_async()

    response = await client.get(f"/api/decks/{deck.id}/cards/")
    assert response.status_code == status.HTTP_200_OK
    assert len(response.json()["items"]) == 0

    response = await client.get("/api/cards/0/")
    assert response.status_code == status.HTTP_404_NOT_FOUND


async def test_get_cards(client: AsyncClient) -> None:
    deck = await factories.DeckModelFactory.create_async()
    card = await factories.CardModelFactory.create_async(deck_id=deck.id)

    response = await client.get(f"/api/decks/{card.deck_id}/cards/")
    assert response.status_code == status.HTTP_200_OK
    data = response.json()
    assert len(data["items"]) == 1
    for k, v in data["items"][0].items():
        assert v == getattr(card, k)


async def test_get_card(client: AsyncClient) -> None:
    deck = await factories.DeckModelFactory.create_async()
    card = await factories.CardModelFactory.create_async(deck_id=deck.id)

    response = await client.get(f"/api/cards/{card.id}/")
    assert response.status_code == status.HTTP_200_OK
    for k, v in response.json().items():
        assert v == getattr(card, k)


async def test_get_card_not_exist(client: AsyncClient) -> None:
    response = await client.get("/api/cards/999/")
    assert response.status_code == status.HTTP_404_NOT_FOUND


async def test_create_cards(client: AsyncClient) -> None:
    deck = await factories.DeckModelFactory.create_async()

    cards_to_create = [factories.CardCreateSchemaFactory.build(), factories.CardCreateSchemaFactory.build()]
    response = await client.post(
        f"/api/decks/{deck.id}/cards/",
        json=[x.model_dump() for x in cards_to_create],
    )
    assert response.status_code == status.HTTP_200_OK
    created_data = response.json()

    # check creation
    response = await client.get(f"/api/decks/{deck.id}/cards/")
    assert response.status_code == status.HTTP_200_OK
    data = response.json()
    assert created_data == data
    assert len(data["items"]) == len(cards_to_create)
    for k, v in cards_to_create[0].model_dump().items():
        assert data["items"][0][k] == v
    for k, v in cards_to_create[1].model_dump().items():
        assert data["items"][1][k] == v

    # unique constraint error
    response = await client.post(
        f"/api/decks/{deck.id}/cards/",
        json=[cards_to_create[0].model_dump(), cards_to_create[1].model_dump()],
    )
    data = response.json()
    assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT
    assert data["detail"] == "A record matching the supplied data already exists."


async def test_update_cards(client: AsyncClient) -> None:
    deck = await factories.DeckModelFactory.create_async()
    card1, card2 = await factories.CardModelFactory.create_batch_async(size=2, deck_id=deck.id)

    updated_data = [
        {
            "id": card1.id,
            "front": "card front updated",
            "back": "card back updated",
            "hint": "card hint updated",
        },
        {
            "id": card2.id,
            "front": "card front2 updated",
            "back": "card back2 updated",
            "hint": "card hint2 updated",
        },
    ]
    response = await client.put(
        f"/api/decks/{deck.id}/cards/",
        json=updated_data,
    )
    assert response.status_code == status.HTTP_200_OK
    cards = response.json()["items"]
    for x in cards:
        assert x.pop("deck_id") == deck.id
    assert cards == updated_data

    # check creation
    response = await client.get(f"/api/decks/{deck.id}/cards/")
    assert response.status_code == status.HTTP_200_OK
    cards = response.json()["items"]
    assert len(cards) == len(updated_data)
    for x in cards:
        assert x.pop("deck_id") == deck.id
    assert cards == updated_data