fastapi-sqlalchemy-asyncpg/tests/api/test_stuff.py at e9ea2c627a36c207596c269ac1be383201da5549 · grillazz/fastapi-sqlalchemy-asyncpg · 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
import pytest
from dirty_equals import IsUUID
from fastapi import status
from httpx import AsyncClient
from inline_snapshot import snapshot
from polyfactory.factories.pydantic_factory import ModelFactory
from sqlalchemy.ext.asyncio import AsyncSession

from app.schemas.stuff import StuffSchema
from app.models import Stuff

pytestmark = pytest.mark.anyio


class StuffFactory(ModelFactory[StuffSchema]):
    __model__ = StuffSchema


async def test_add_stuff(client: AsyncClient, db_session: AsyncSession):
    stuff = StuffFactory.build(factory_use_constructors=True).model_dump(mode="json")
    response = await client.post("/stuff", json=stuff)
    assert response.status_code == status.HTTP_201_CREATED
    assert response.json() == snapshot(
        {
            "id": IsUUID(4),
            "name": stuff["name"],
            "description": stuff["description"],
        }
    )
    response = await client.post("/stuff", json=stuff)
    assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
    assert response.json() == snapshot(
        {"message": "A database error occurred. Please try again later."}
    )


async def test_get_stuff(client: AsyncClient, db_session: AsyncSession):
    response = await client.get("/stuff/nonexistent")
    assert response.status_code == status.HTTP_404_NOT_FOUND
    assert response.json() == snapshot(
        {"no_response": "The requested resource was not found"}
    )
    stuff = StuffFactory.build(factory_use_constructors=True).model_dump(mode="json")
    # await client.post("/stuff", json=stuff)
    # name = stuff["name"]
    stuff = Stuff(**stuff)
    name = stuff.name
    db_session.add(stuff)
    await db_session.commit()

    response = await client.get(f"/stuff/{name}")
    assert response.status_code == status.HTTP_200_OK
    assert response.json() == snapshot(
        {
            "id": IsUUID(4),
            "name": stuff.name,
            "description": stuff.description,
        }
    )


async def test_delete_stuff(client: AsyncClient):
    stuff = StuffFactory.build(factory_use_constructors=True).model_dump(mode="json")
    await client.post("/stuff", json=stuff)
    name = stuff["name"]
    response = await client.delete(f"/stuff/{name}")
    assert response.status_code == status.HTTP_200_OK
    assert response.json() == snapshot(True)


async def test_update_stuff(client: AsyncClient):
    stuff = StuffFactory.build(factory_use_constructors=True).model_dump(mode="json")
    response = await client.post("/stuff", json=stuff)
    assert response.json() == snapshot(
        {
            "id": IsUUID(4),
            "name": stuff["name"],
            "description": stuff["description"],
        }
    )
    name = stuff["name"]
    response = await client.patch(
        f"/stuff/{name}",
        json={"name": stuff["name"], "description": "we play rock and roll"},
    )
    assert response.json() == snapshot(
        {
            "id": IsUUID(4),
            "name": stuff["name"],
            "description": "we play rock and roll",
        }
    )
    assert response.status_code == status.HTTP_200_OK