|
|
@ -1,5 +1,4 @@ |
|
|
import importlib |
|
|
import importlib |
|
|
import warnings |
|
|
|
|
|
|
|
|
|
|
|
import pytest |
|
|
import pytest |
|
|
from dirty_equals import IsInt |
|
|
from dirty_equals import IsInt |
|
|
@ -30,12 +29,9 @@ def clear_sqlmodel(): |
|
|
) |
|
|
) |
|
|
def get_client(request: pytest.FixtureRequest): |
|
|
def get_client(request: pytest.FixtureRequest): |
|
|
clear_sqlmodel() |
|
|
clear_sqlmodel() |
|
|
# TODO: remove when updating SQL tutorial to use new lifespan API |
|
|
mod = importlib.import_module(f"docs_src.sql_databases.{request.param}") |
|
|
with warnings.catch_warnings(record=True): |
|
|
clear_sqlmodel() |
|
|
warnings.simplefilter("always") |
|
|
importlib.reload(mod) |
|
|
mod = importlib.import_module(f"docs_src.sql_databases.{request.param}") |
|
|
|
|
|
clear_sqlmodel() |
|
|
|
|
|
importlib.reload(mod) |
|
|
|
|
|
mod.sqlite_url = "sqlite://" |
|
|
mod.sqlite_url = "sqlite://" |
|
|
mod.engine = create_engine( |
|
|
mod.engine = create_engine( |
|
|
mod.sqlite_url, connect_args={"check_same_thread": False}, poolclass=StaticPool |
|
|
mod.sqlite_url, connect_args={"check_same_thread": False}, poolclass=StaticPool |
|
|
@ -48,103 +44,93 @@ def get_client(request: pytest.FixtureRequest): |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_crud_app(client: TestClient): |
|
|
def test_crud_app(client: TestClient): |
|
|
# TODO: this warns that SQLModel.from_orm is deprecated in Pydantic v1, refactor |
|
|
# No heroes before creating |
|
|
# this if using obj.model_validate becomes independent of Pydantic v2 |
|
|
response = client.get("heroes/") |
|
|
with warnings.catch_warnings(record=True): |
|
|
assert response.status_code == 200, response.text |
|
|
warnings.simplefilter("always") |
|
|
assert response.json() == [] |
|
|
# No heroes before creating |
|
|
|
|
|
response = client.get("heroes/") |
|
|
|
|
|
assert response.status_code == 200, response.text |
|
|
|
|
|
assert response.json() == [] |
|
|
|
|
|
|
|
|
|
|
|
# Create a hero |
|
|
# Create a hero |
|
|
response = client.post( |
|
|
response = client.post( |
|
|
"/heroes/", |
|
|
"/heroes/", |
|
|
json={ |
|
|
json={ |
|
|
"id": 9000, |
|
|
"id": 9000, |
|
|
"name": "Dead Pond", |
|
|
"name": "Dead Pond", |
|
|
"age": 30, |
|
|
"age": 30, |
|
|
"secret_name": "Dive Wilson", |
|
|
"secret_name": "Dive Wilson", |
|
|
}, |
|
|
}, |
|
|
) |
|
|
) |
|
|
assert response.status_code == 200, response.text |
|
|
assert response.status_code == 200, response.text |
|
|
assert response.json() == snapshot( |
|
|
assert response.json() == snapshot({"age": 30, "id": IsInt(), "name": "Dead Pond"}) |
|
|
{"age": 30, "id": IsInt(), "name": "Dead Pond"} |
|
|
assert response.json()["id"] != 9000, "The ID should be generated by the database" |
|
|
) |
|
|
|
|
|
assert response.json()["id"] != 9000, ( |
|
|
|
|
|
"The ID should be generated by the database" |
|
|
|
|
|
) |
|
|
|
|
|
|
|
|
|
|
|
# Read a hero |
|
|
# Read a hero |
|
|
hero_id = response.json()["id"] |
|
|
hero_id = response.json()["id"] |
|
|
response = client.get(f"/heroes/{hero_id}") |
|
|
response = client.get(f"/heroes/{hero_id}") |
|
|
assert response.status_code == 200, response.text |
|
|
assert response.status_code == 200, response.text |
|
|
assert response.json() == snapshot( |
|
|
assert response.json() == snapshot({"name": "Dead Pond", "age": 30, "id": IsInt()}) |
|
|
{"name": "Dead Pond", "age": 30, "id": IsInt()} |
|
|
|
|
|
) |
|
|
|
|
|
|
|
|
|
|
|
# Read all heroes |
|
|
# Read all heroes |
|
|
# Create more heroes first |
|
|
# Create more heroes first |
|
|
response = client.post( |
|
|
response = client.post( |
|
|
"/heroes/", |
|
|
"/heroes/", |
|
|
json={"name": "Spider-Boy", "age": 18, "secret_name": "Pedro Parqueador"}, |
|
|
json={"name": "Spider-Boy", "age": 18, "secret_name": "Pedro Parqueador"}, |
|
|
) |
|
|
) |
|
|
assert response.status_code == 200, response.text |
|
|
assert response.status_code == 200, response.text |
|
|
response = client.post( |
|
|
response = client.post( |
|
|
"/heroes/", json={"name": "Rusty-Man", "secret_name": "Tommy Sharp"} |
|
|
"/heroes/", json={"name": "Rusty-Man", "secret_name": "Tommy Sharp"} |
|
|
) |
|
|
) |
|
|
assert response.status_code == 200, response.text |
|
|
assert response.status_code == 200, response.text |
|
|
|
|
|
|
|
|
response = client.get("/heroes/") |
|
|
response = client.get("/heroes/") |
|
|
assert response.status_code == 200, response.text |
|
|
assert response.status_code == 200, response.text |
|
|
assert response.json() == snapshot( |
|
|
assert response.json() == snapshot( |
|
|
[ |
|
|
[ |
|
|
{"name": "Dead Pond", "age": 30, "id": IsInt()}, |
|
|
{"name": "Dead Pond", "age": 30, "id": IsInt()}, |
|
|
{"name": "Spider-Boy", "age": 18, "id": IsInt()}, |
|
|
{"name": "Spider-Boy", "age": 18, "id": IsInt()}, |
|
|
{"name": "Rusty-Man", "age": None, "id": IsInt()}, |
|
|
{"name": "Rusty-Man", "age": None, "id": IsInt()}, |
|
|
] |
|
|
] |
|
|
) |
|
|
) |
|
|
|
|
|
|
|
|
response = client.get("/heroes/?offset=1&limit=1") |
|
|
response = client.get("/heroes/?offset=1&limit=1") |
|
|
assert response.status_code == 200, response.text |
|
|
assert response.status_code == 200, response.text |
|
|
assert response.json() == snapshot( |
|
|
assert response.json() == snapshot( |
|
|
[{"name": "Spider-Boy", "age": 18, "id": IsInt()}] |
|
|
[{"name": "Spider-Boy", "age": 18, "id": IsInt()}] |
|
|
) |
|
|
) |
|
|
|
|
|
|
|
|
# Update a hero |
|
|
# Update a hero |
|
|
response = client.patch( |
|
|
response = client.patch( |
|
|
f"/heroes/{hero_id}", json={"name": "Dog Pond", "age": None} |
|
|
f"/heroes/{hero_id}", json={"name": "Dog Pond", "age": None} |
|
|
) |
|
|
) |
|
|
assert response.status_code == 200, response.text |
|
|
assert response.status_code == 200, response.text |
|
|
assert response.json() == snapshot( |
|
|
assert response.json() == snapshot( |
|
|
{"name": "Dog Pond", "age": None, "id": Is(hero_id)} |
|
|
{"name": "Dog Pond", "age": None, "id": Is(hero_id)} |
|
|
) |
|
|
) |
|
|
|
|
|
|
|
|
# Get updated hero |
|
|
# Get updated hero |
|
|
response = client.get(f"/heroes/{hero_id}") |
|
|
response = client.get(f"/heroes/{hero_id}") |
|
|
assert response.status_code == 200, response.text |
|
|
assert response.status_code == 200, response.text |
|
|
assert response.json() == snapshot( |
|
|
assert response.json() == snapshot( |
|
|
{"name": "Dog Pond", "age": None, "id": Is(hero_id)} |
|
|
{"name": "Dog Pond", "age": None, "id": Is(hero_id)} |
|
|
) |
|
|
) |
|
|
|
|
|
|
|
|
# Delete a hero |
|
|
# Delete a hero |
|
|
response = client.delete(f"/heroes/{hero_id}") |
|
|
response = client.delete(f"/heroes/{hero_id}") |
|
|
assert response.status_code == 200, response.text |
|
|
assert response.status_code == 200, response.text |
|
|
assert response.json() == snapshot({"ok": True}) |
|
|
assert response.json() == snapshot({"ok": True}) |
|
|
|
|
|
|
|
|
# The hero is no longer found |
|
|
# The hero is no longer found |
|
|
response = client.get(f"/heroes/{hero_id}") |
|
|
response = client.get(f"/heroes/{hero_id}") |
|
|
assert response.status_code == 404, response.text |
|
|
assert response.status_code == 404, response.text |
|
|
|
|
|
|
|
|
# Delete a hero that does not exist |
|
|
# Delete a hero that does not exist |
|
|
response = client.delete(f"/heroes/{hero_id}") |
|
|
response = client.delete(f"/heroes/{hero_id}") |
|
|
assert response.status_code == 404, response.text |
|
|
assert response.status_code == 404, response.text |
|
|
assert response.json() == snapshot({"detail": "Hero not found"}) |
|
|
assert response.json() == snapshot({"detail": "Hero not found"}) |
|
|
|
|
|
|
|
|
# Update a hero that does not exist |
|
|
# Update a hero that does not exist |
|
|
response = client.patch(f"/heroes/{hero_id}", json={"name": "Dog Pond"}) |
|
|
response = client.patch(f"/heroes/{hero_id}", json={"name": "Dog Pond"}) |
|
|
assert response.status_code == 404, response.text |
|
|
assert response.status_code == 404, response.text |
|
|
assert response.json() == snapshot({"detail": "Hero not found"}) |
|
|
assert response.json() == snapshot({"detail": "Hero not found"}) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def test_openapi_schema(client: TestClient): |
|
|
def test_openapi_schema(client: TestClient): |
|
|
|