import importlib
-import types # Add import for types
+import types # Add import for types
from decimal import Decimal
-from unittest.mock import MagicMock # Keep MagicMock for type hint, though not strictly necessary for runtime
import pytest
from sqlmodel import create_engine
-from ...conftest import needs_py310, PrintMock # Import PrintMock for type hint
+from ...conftest import PrintMock, needs_py310 # Import PrintMock for type hint
expected_calls = [
[
return importlib.import_module(f"docs_src.advanced.decimal.{module_name}")
-def test_tutorial(print_mock: PrintMock, module: types.ModuleType): # Use PrintMock for type hint and types.ModuleType
+def test_tutorial(
+ print_mock: PrintMock, module: types.ModuleType
+): # Use PrintMock for type hint and types.ModuleType
module.sqlite_url = "sqlite://"
module.engine = create_engine(module.sqlite_url)
module.main()
- assert print_mock.calls == expected_calls # Use .calls instead of .mock_calls
+ assert print_mock.calls == expected_calls # Use .calls instead of .mock_calls
)
def get_module(request: pytest.FixtureRequest) -> ModuleType:
module_name = request.param
- mod = importlib.import_module(
- f"docs_src.tutorial.connect.delete.{module_name}"
- )
+ mod = importlib.import_module(f"docs_src.tutorial.connect.delete.{module_name}")
mod.sqlite_url = "sqlite://"
mod.engine = create_engine(mod.sqlite_url)
return mod
)
def get_module(request: pytest.FixtureRequest) -> ModuleType:
module_name = request.param
- mod = importlib.import_module(
- f"docs_src.tutorial.connect.insert.{module_name}"
- )
+ mod = importlib.import_module(f"docs_src.tutorial.connect.insert.{module_name}")
mod.sqlite_url = "sqlite://"
mod.engine = create_engine(mod.sqlite_url)
return mod
)
def get_module(request: pytest.FixtureRequest) -> ModuleType:
module_name = request.param
- mod = importlib.import_module(
- f"docs_src.tutorial.connect.select.{module_name}"
- )
+ mod = importlib.import_module(f"docs_src.tutorial.connect.select.{module_name}")
mod.sqlite_url = "sqlite://"
mod.engine = create_engine(mod.sqlite_url)
return mod
)
def get_module(request: pytest.FixtureRequest) -> ModuleType:
module_name = request.param
- mod = importlib.import_module(
- f"docs_src.tutorial.connect.select.{module_name}"
- )
+ mod = importlib.import_module(f"docs_src.tutorial.connect.select.{module_name}")
mod.sqlite_url = "sqlite://"
mod.engine = create_engine(mod.sqlite_url)
return mod
)
def get_module(request: pytest.FixtureRequest) -> ModuleType:
module_name = request.param
- mod = importlib.import_module(
- f"docs_src.tutorial.connect.select.{module_name}"
- )
+ mod = importlib.import_module(f"docs_src.tutorial.connect.select.{module_name}")
mod.sqlite_url = "sqlite://"
mod.engine = create_engine(mod.sqlite_url)
return mod
import importlib
from types import ModuleType
-from typing import Any # For clear_sqlmodel type hint
+from typing import Any # For clear_sqlmodel type hint
import pytest
from sqlmodel import create_engine
)
def get_module(request: pytest.FixtureRequest) -> ModuleType:
module_name = request.param
- mod = importlib.import_module(
- f"docs_src.tutorial.connect.update.{module_name}"
- )
+ mod = importlib.import_module(f"docs_src.tutorial.connect.update.{module_name}")
mod.sqlite_url = "sqlite://"
mod.engine = create_engine(mod.sqlite_url)
return mod
-def test_tutorial(clear_sqlmodel: Any, print_mock: PrintMock, module: ModuleType) -> None:
+def test_tutorial(
+ clear_sqlmodel: Any, print_mock: PrintMock, module: ModuleType
+) -> None:
module.main()
assert print_mock.calls == expected_calls
import importlib
from types import ModuleType
-from typing import Any # For clear_sqlmodel type hint
+from typing import Any # For clear_sqlmodel type hint
import pytest
from sqlalchemy import inspect
import importlib
from types import ModuleType
-from typing import Any # For clear_sqlmodel type hint
+from typing import Any # For clear_sqlmodel type hint
import pytest
from sqlalchemy import inspect
import importlib
-import sys # Add sys import
+import sys # Add sys import
from types import ModuleType
from typing import Any, Generator
import pytest
from fastapi.testclient import TestClient
-from sqlmodel import Session, SQLModel, create_engine # Keep this for session_fixture
-from sqlmodel.pool import StaticPool # Keep this for session_fixture
+from sqlmodel import Session, SQLModel, create_engine # Keep this for session_fixture
+from sqlmodel.pool import StaticPool # Keep this for session_fixture
from ....conftest import needs_py39, needs_py310
+
# This will be our parametrized fixture providing the versioned 'main' module
@pytest.fixture(
name="module",
pytest.param("tutorial001_py310", marks=needs_py310),
],
)
-def get_module(request: pytest.FixtureRequest, clear_sqlmodel: Any) -> ModuleType: # clear_sqlmodel is autouse
+def get_module(
+ request: pytest.FixtureRequest, clear_sqlmodel: Any
+) -> ModuleType: # clear_sqlmodel is autouse
module_name = f"docs_src.tutorial.fastapi.app_testing.{request.param}.main"
# Forcing reload to try to get a fresh state for models
module = importlib.import_module(module_name)
return module
+
@pytest.fixture(name="session", scope="function")
def session_fixture(module: ModuleType) -> Generator[Session, None, None]:
# Store original engine-related attributes from the module
# Force module to use a fresh in-memory SQLite DB for this test run
module.sqlite_url = "sqlite://"
- module.connect_args = {"check_same_thread": False} # Crucial for FastAPI + SQLite
+ module.connect_args = {"check_same_thread": False} # Crucial for FastAPI + SQLite
# Re-create the engine in the module to use these new settings
test_engine = create_engine(
module.sqlite_url,
connect_args=module.connect_args,
- poolclass=StaticPool # Recommended for tests
+ poolclass=StaticPool, # Recommended for tests
)
module.engine = test_engine
# Fallback if the function isn't named create_db_and_tables
SQLModel.metadata.create_all(module.engine)
- with Session(module.engine) as session: # Use the module's (now test-configured) engine
+ with Session(
+ module.engine
+ ) as session: # Use the module's (now test-configured) engine
yield session
# Teardown: drop tables from the module's engine
module.connect_args = original_connect_args
if original_engine is not None:
module.engine = original_engine
- else: # If engine didn't exist, remove the one we created
+ else: # If engine didn't exist, remove the one we created
if hasattr(module, "engine"):
del module.engine
@pytest.fixture(name="client", scope="function")
-def client_fixture(session: Session, module: ModuleType) -> Generator[TestClient, None, None]:
- def get_session_override() -> Generator[Session, None, None]: # Must be a generator
+def client_fixture(
+ session: Session, module: ModuleType
+) -> Generator[TestClient, None, None]:
+ def get_session_override() -> Generator[Session, None, None]: # Must be a generator
yield session
module.app.dependency_overrides[module.get_session] = get_session_override
def test_read_hero(session: Session, client: TestClient, module: ModuleType):
- hero_1 = module.Hero(name="Deadpond", secret_name="Dive Wilson") # Use module.Hero
+ hero_1 = module.Hero(name="Deadpond", secret_name="Dive Wilson") # Use module.Hero
session.add(hero_1)
session.commit()
def test_update_hero(session: Session, client: TestClient, module: ModuleType):
- hero_1 = module.Hero(name="Deadpond", secret_name="Dive Wilson") # Use module.Hero
+ hero_1 = module.Hero(name="Deadpond", secret_name="Dive Wilson") # Use module.Hero
session.add(hero_1)
session.commit()
def test_delete_hero(session: Session, client: TestClient, module: ModuleType):
- hero_1 = module.Hero(name="Deadpond", secret_name="Dive Wilson") # Use module.Hero
+ hero_1 = module.Hero(name="Deadpond", secret_name="Dive Wilson") # Use module.Hero
session.add(hero_1)
session.commit()
response = client.delete(f"/heroes/{hero_1.id}")
- hero_in_db = session.get(module.Hero, hero_1.id) # Use module.Hero
+ hero_in_db = session.get(module.Hero, hero_1.id) # Use module.Hero
assert response.status_code == 200
assert hero_in_db is None
import importlib
import sys
from types import ModuleType
-from typing import Any # For clear_sqlmodel type hint
+from typing import Any # For clear_sqlmodel type hint
import pytest
from dirty_equals import IsDict
from fastapi.testclient import TestClient
-from sqlmodel import SQLModel, create_engine # Import SQLModel for metadata operations
+from sqlmodel import SQLModel, create_engine # Import SQLModel for metadata operations
from sqlmodel.pool import StaticPool
from ....conftest import needs_py39, needs_py310
],
)
def get_module(request: pytest.FixtureRequest, clear_sqlmodel: Any) -> ModuleType:
- module_name = f"docs_src.tutorial.fastapi.delete.{request.param}" # No .main here
+ module_name = f"docs_src.tutorial.fastapi.delete.{request.param}" # No .main here
if module_name in sys.modules:
module = importlib.reload(sys.modules[module_name])
else:
module.sqlite_url = "sqlite://"
module.engine = create_engine(
module.sqlite_url,
- connect_args={"check_same_thread": False}, # connect_args from original main.py
- poolclass=StaticPool
+ connect_args={"check_same_thread": False}, # connect_args from original main.py
+ poolclass=StaticPool,
)
# Assuming the module has a create_db_and_tables or similar, or uses SQLModel.metadata directly
if hasattr(module, "create_db_and_tables"):
module.create_db_and_tables()
else:
- SQLModel.metadata.create_all(module.engine) # Fallback, ensure tables are created
+ SQLModel.metadata.create_all(
+ module.engine
+ ) # Fallback, ensure tables are created
return module
-def test_tutorial(clear_sqlmodel: Any, module: ModuleType): # clear_sqlmodel is autouse but explicit for safety
+def test_tutorial(
+ clear_sqlmodel: Any, module: ModuleType
+): # clear_sqlmodel is autouse but explicit for safety
# The engine and tables are now set up by the 'module' fixture
# The app's dependency overrides for get_session will use module.engine
hero2_data = {
"name": "Spider-Boy",
"secret_name": "Pedro Parqueador",
- "id": 9000, # Note: ID is part of creation data here
+ "id": 9000, # Note: ID is part of creation data here
}
hero3_data = {
"name": "Rusty-Man",
}
response = client.post("/heroes/", json=hero1_data)
assert response.status_code == 200, response.text
- hero1 = response.json() # Get actual ID of hero1
+ hero1 = response.json() # Get actual ID of hero1
hero1_id = hero1["id"]
response = client.post("/heroes/", json=hero2_data)
assert response.status_code == 200, response.text
hero2 = response.json()
- hero2_id = hero2["id"] # This will be the ID assigned by DB, not 9000 if 9000 is not allowed on POST
+ hero2_id = hero2[
+ "id"
+ ] # This will be the ID assigned by DB, not 9000 if 9000 is not allowed on POST
response = client.post("/heroes/", json=hero3_data)
assert response.status_code == 200, response.text
# For robustness, let's check for a non-existent ID based on actual data.
# If hero2_id is 1, check for 9000. If it's 9000, check for 1 (assuming hero1_id is 1).
non_existent_id_check = 9000
- if hero2_id == non_existent_id_check: # if DB somehow used 9000
- non_existent_id_check = hero1_id + hero2_id + 100 # just some other ID
+ if hero2_id == non_existent_id_check: # if DB somehow used 9000
+ non_existent_id_check = hero1_id + hero2_id + 100 # just some other ID
response = client.get(f"/heroes/{non_existent_id_check}")
assert response.status_code == 404, response.text
)
assert response.status_code == 200, response.text
- response = client.patch(f"/heroes/{non_existent_id_check}", json={"name": "Dragon Cube X"})
+ response = client.patch(
+ f"/heroes/{non_existent_id_check}", json={"name": "Dragon Cube X"}
+ )
assert response.status_code == 404, response.text
response = client.delete(f"/heroes/{hero2_id}")
response = client.get("/heroes/")
assert response.status_code == 200, response.text
data = response.json()
- assert len(data) == 2 # After deleting one hero
+ assert len(data) == 2 # After deleting one hero
response = client.delete(f"/heroes/{non_existent_id_check}")
assert response.status_code == 404, response.text
import importlib
import sys
from types import ModuleType
-from typing import Any # For clear_sqlmodel type hint
+from typing import Any # For clear_sqlmodel type hint
import pytest
from dirty_equals import IsDict
from fastapi.testclient import TestClient
-from sqlmodel import SQLModel, create_engine # Import SQLModel for metadata operations
+from sqlmodel import SQLModel, create_engine # Import SQLModel for metadata operations
from sqlmodel.pool import StaticPool
from ....conftest import needs_py39, needs_py310
],
)
def get_module(request: pytest.FixtureRequest, clear_sqlmodel: Any) -> ModuleType:
- module_name = f"docs_src.tutorial.fastapi.limit_and_offset.{request.param}" # No .main
+ module_name = (
+ f"docs_src.tutorial.fastapi.limit_and_offset.{request.param}" # No .main
+ )
if module_name in sys.modules:
module = importlib.reload(sys.modules[module_name])
else:
module.sqlite_url = "sqlite://"
module.engine = create_engine(
module.sqlite_url,
- connect_args={"check_same_thread": False}, # Assuming connect_args was in original mod or default
- poolclass=StaticPool
+ connect_args={
+ "check_same_thread": False
+ }, # Assuming connect_args was in original mod or default
+ poolclass=StaticPool,
)
if hasattr(module, "create_db_and_tables"):
module.create_db_and_tables()
response = client.post("/heroes/", json=hero2_data)
assert response.status_code == 200, response.text
hero2 = response.json()
- hero2_id = hero2["id"] # Use the actual ID from response
+ hero2_id = hero2["id"] # Use the actual ID from response
# Create hero 3
response = client.post("/heroes/", json=hero3_data)
assert response.status_code == 200, response.text
data_limit2 = response.json()
assert len(data_limit2) == 2
- assert data_limit2[0]["name"] == hero1["name"] # Compare with actual created hero data
+ assert (
+ data_limit2[0]["name"] == hero1["name"]
+ ) # Compare with actual created hero data
assert data_limit2[1]["name"] == hero2["name"]
response = client.get("/heroes/", params={"offset": 1})
import importlib
import sys
from types import ModuleType
-from typing import Any # For clear_sqlmodel type hint
+from typing import Any # For clear_sqlmodel type hint
import pytest
from dirty_equals import IsDict
from fastapi.testclient import TestClient
from sqlalchemy import inspect
from sqlalchemy.engine.reflection import Inspector
-from sqlmodel import SQLModel, create_engine # Import SQLModel
+from sqlmodel import SQLModel, create_engine # Import SQLModel
from sqlmodel.pool import StaticPool
from ....conftest import needs_py39, needs_py310
],
)
def get_module(request: pytest.FixtureRequest, clear_sqlmodel: Any) -> ModuleType:
- module_name = f"docs_src.tutorial.fastapi.multiple_models.{request.param}" # No .main
+ module_name = (
+ f"docs_src.tutorial.fastapi.multiple_models.{request.param}" # No .main
+ )
if module_name in sys.modules:
module = importlib.reload(sys.modules[module_name])
else:
# Ensure connect_args is available in module, default if not.
# Some tutorial files might not define it if they don't use on_event("startup") for engine creation.
connect_args = getattr(module, "connect_args", {"check_same_thread": False})
- if "check_same_thread" not in connect_args: # Ensure this specific arg for SQLite
+ if "check_same_thread" not in connect_args: # Ensure this specific arg for SQLite
connect_args["check_same_thread"] = False
module.engine = create_engine(
- module.sqlite_url,
- connect_args=connect_args,
- poolclass=StaticPool
+ module.sqlite_url, connect_args=connect_args, poolclass=StaticPool
)
if hasattr(module, "create_db_and_tables"):
module.create_db_and_tables()
assert data["secret_name"] == hero1_data["secret_name"]
assert data["id"] is not None
assert data["age"] is None
- hero1_id = data["id"] # Store actual ID
+ hero1_id = data["id"] # Store actual ID
response = client.post("/heroes/", json=hero2_data)
data = response.json()
# This is true if ID is auto-generated and not 9000.
assert data["id"] is not None
assert data["age"] is None
- hero2_id = data["id"] # Store actual ID
-
+ hero2_id = data["id"] # Store actual ID
response = client.get("/heroes/")
data = response.json()
assert data[1]["name"] == hero2_data["name"]
assert data[1]["secret_name"] == hero2_data["secret_name"]
-
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
# OpenAPI schema check - kept as is from original test
}
# Test inherited indexes
- insp: Inspector = inspect(module.engine) # Use module.engine
- indexes = insp.get_indexes(str(module.Hero.__tablename__)) # Use module.Hero
+ insp: Inspector = inspect(module.engine) # Use module.engine
+ indexes = insp.get_indexes(str(module.Hero.__tablename__)) # Use module.Hero
expected_indexes = [
{
"name": "ix_hero_name",
]
# Convert list of dicts to list of tuples of sorted items for order-agnostic comparison
indexes_for_comparison = [tuple(sorted(d.items())) for d in indexes]
- expected_indexes_for_comparison = [tuple(sorted(d.items())) for d in expected_indexes]
+ expected_indexes_for_comparison = [
+ tuple(sorted(d.items())) for d in expected_indexes
+ ]
for index_data_tuple in expected_indexes_for_comparison:
- assert index_data_tuple in indexes_for_comparison, f"Expected index {index_data_tuple} not found in DB indexes {indexes_for_comparison}"
+ assert index_data_tuple in indexes_for_comparison, (
+ f"Expected index {index_data_tuple} not found in DB indexes {indexes_for_comparison}"
+ )
indexes_for_comparison.remove(index_data_tuple)
- assert len(indexes_for_comparison) == 0, f"Unexpected extra indexes found in DB: {indexes_for_comparison}"
+ assert len(indexes_for_comparison) == 0, (
+ f"Unexpected extra indexes found in DB: {indexes_for_comparison}"
+ )
import importlib
import sys
from types import ModuleType
-from typing import Any # For clear_sqlmodel type hint
+from typing import Any # For clear_sqlmodel type hint
import pytest
from dirty_equals import IsDict
from fastapi.testclient import TestClient
from sqlalchemy import inspect
from sqlalchemy.engine.reflection import Inspector
-from sqlmodel import SQLModel, create_engine # Import SQLModel
+from sqlmodel import SQLModel, create_engine # Import SQLModel
from sqlmodel.pool import StaticPool
from ....conftest import needs_py39, needs_py310
name="module",
scope="function",
params=[
- "tutorial002", # Changed to tutorial002
- pytest.param("tutorial002_py39", marks=needs_py39), # Changed to tutorial002_py39
- pytest.param("tutorial002_py310", marks=needs_py310), # Changed to tutorial002_py310
+ "tutorial002", # Changed to tutorial002
+ pytest.param(
+ "tutorial002_py39", marks=needs_py39
+ ), # Changed to tutorial002_py39
+ pytest.param(
+ "tutorial002_py310", marks=needs_py310
+ ), # Changed to tutorial002_py310
],
)
def get_module(request: pytest.FixtureRequest, clear_sqlmodel: Any) -> ModuleType:
connect_args["check_same_thread"] = False
module.engine = create_engine(
- module.sqlite_url,
- connect_args=connect_args,
- poolclass=StaticPool
+ module.sqlite_url, connect_args=connect_args, poolclass=StaticPool
)
if hasattr(module, "create_db_and_tables"):
module.create_db_and_tables()
assert data["age"] is None
hero2_id = data["id"]
-
response = client.get("/heroes/")
data = response.json()
assert data[1]["name"] == hero2_data["name"]
assert data[1]["secret_name"] == hero2_data["secret_name"]
-
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
assert response.json() == {
indexes = insp.get_indexes(str(module.Hero.__tablename__))
expected_indexes = [
{
- "name": "ix_hero_age", # For tutorial002, order of expected indexes is different
+ "name": "ix_hero_age", # For tutorial002, order of expected indexes is different
"dialect_options": {},
"column_names": ["age"],
"unique": 0,
},
]
indexes_for_comparison = [tuple(sorted(d.items())) for d in indexes]
- expected_indexes_for_comparison = [tuple(sorted(d.items())) for d in expected_indexes]
+ expected_indexes_for_comparison = [
+ tuple(sorted(d.items())) for d in expected_indexes
+ ]
for index_data_tuple in expected_indexes_for_comparison:
- assert index_data_tuple in indexes_for_comparison, f"Expected index {index_data_tuple} not found in DB indexes {indexes_for_comparison}"
+ assert index_data_tuple in indexes_for_comparison, (
+ f"Expected index {index_data_tuple} not found in DB indexes {indexes_for_comparison}"
+ )
indexes_for_comparison.remove(index_data_tuple)
- assert len(indexes_for_comparison) == 0, f"Unexpected extra indexes found in DB: {indexes_for_comparison}"
+ assert len(indexes_for_comparison) == 0, (
+ f"Unexpected extra indexes found in DB: {indexes_for_comparison}"
+ )
],
)
def get_module(request: pytest.FixtureRequest, clear_sqlmodel: Any) -> ModuleType:
- module_name = f"docs_src.tutorial.fastapi.read_one.{request.param}" # No .main
+ module_name = f"docs_src.tutorial.fastapi.read_one.{request.param}" # No .main
if module_name in sys.modules:
module = importlib.reload(sys.modules[module_name])
else:
connect_args["check_same_thread"] = False
module.engine = create_engine(
- module.sqlite_url,
- connect_args=connect_args,
- poolclass=StaticPool
+ module.sqlite_url, connect_args=connect_args, poolclass=StaticPool
)
if hasattr(module, "create_db_and_tables"):
module.create_db_and_tables()
}
response = client.post("/heroes/", json=hero1_data)
assert response.status_code == 200, response.text
- hero1 = response.json() # Store created hero1 data
+ hero1 = response.json() # Store created hero1 data
response = client.post("/heroes/", json=hero2_data)
assert response.status_code == 200, response.text
- hero2 = response.json() # Store created hero2 data
+ hero2 = response.json() # Store created hero2 data
response_get_all = client.get("/heroes/")
assert response_get_all.status_code == 200, response_get_all.text
data_all = response_get_all.json()
assert len(data_all) == 2
- hero_id_to_get = hero2["id"] # Use actual ID from created hero2
+ hero_id_to_get = hero2["id"] # Use actual ID from created hero2
response_get_one = client.get(f"/heroes/{hero_id_to_get}")
assert response_get_one.status_code == 200, response_get_one.text
data_one = response_get_one.json()
assert data_one["id"] == hero2["id"]
# Check for a non-existent ID
- non_existent_id = hero1["id"] + hero2["id"] + 100 # A likely non-existent ID
+ non_existent_id = hero1["id"] + hero2["id"] + 100 # A likely non-existent ID
response_get_non_existent = client.get(f"/heroes/{non_existent_id}")
- assert response_get_non_existent.status_code == 404, response_get_non_existent.text
+ assert response_get_non_existent.status_code == 404, (
+ response_get_non_existent.text
+ )
response_openapi = client.get("/openapi.json")
assert response_openapi.status_code == 200, response_openapi.text
from typing import Any
import pytest
-from dirty_equals import IsDict
from fastapi.testclient import TestClient
-from sqlmodel import create_engine, SQLModel
+from sqlmodel import SQLModel, create_engine
from sqlmodel.pool import StaticPool
from ....conftest import needs_py39, needs_py310
hero2_data = {
"name": "Spider-Boy",
"secret_name": "Pedro Parqueador",
- "id": 9000, # This ID might be problematic if the DB auto-increments differently or if this ID is expected to be user-settable and unique
+ "id": 9000, # This ID might be problematic if the DB auto-increments differently or if this ID is expected to be user-settable and unique
}
hero3_data = {
"name": "Rusty-Man",
hero2_id = hero2["id"]
response = client.post("/heroes/", json=hero3_data)
assert response.status_code == 200, response.text
- response = client.get("/heroes/9000") # This might fail if hero2_id is not 9000
- assert response.status_code == 404, response.text # Original test expects 404, this implies ID 9000 is not found after creation. This needs to align with how IDs are handled.
+ response = client.get("/heroes/9000") # This might fail if hero2_id is not 9000
+ assert response.status_code == 404, (
+ response.text
+ ) # Original test expects 404, this implies ID 9000 is not found after creation. This needs to align with how IDs are handled.
response = client.get("/heroes/")
assert response.status_code == 200, response.text
data = response.json()
assert data["name"] == hero1_data["name"]
# Ensure team is loaded and correct
- if "team" in data and data["team"] is not None: # Team might not be present if not correctly loaded by the endpoint
+ if (
+ "team" in data and data["team"] is not None
+ ): # Team might not be present if not correctly loaded by the endpoint
assert data["team"]["name"] == team_z_force["name"]
- elif short_module_name != "tutorial001_py310": # tutorial001_py310.py doesn't include team in HeroPublic
- # If team is expected, this is a failure. For tutorial001 and tutorial001_py39, team should be present.
- assert "team" in data and data["team"] is not None, "Team data missing in hero response"
-
+ elif (
+ short_module_name != "tutorial001_py310"
+ ): # tutorial001_py310.py doesn't include team in HeroPublic
+ # If team is expected, this is a failure. For tutorial001 and tutorial001_py39, team should be present.
+ assert "team" in data and data["team"] is not None, (
+ "Team data missing in hero response"
+ )
response = client.patch(
f"/heroes/{hero2_id}", json={"secret_name": "Spider-Youngster"}
)
assert response.status_code == 200, response.text
- response = client.patch("/heroes/9001", json={"name": "Dragon Cube X"}) # Test patching non-existent hero
+ response = client.patch(
+ "/heroes/9001", json={"name": "Dragon Cube X"}
+ ) # Test patching non-existent hero
assert response.status_code == 404, response.text
response = client.delete(f"/heroes/{hero2_id}")
assert response.status_code == 200, response.text
data = response.json()
assert len(data) == 2
- response = client.delete("/heroes/9000") # Test deleting non-existent hero
+ response = client.delete("/heroes/9000") # Test deleting non-existent hero
assert response.status_code == 404, response.text
response = client.get(f"/teams/{team_preventers_id}")
data = response.json()
assert response.status_code == 200, response.text
assert data["name"] == team_preventers_data["name"]
- assert len(data["heroes"]) > 0 # Ensure heroes are loaded
+ assert len(data["heroes"]) > 0 # Ensure heroes are loaded
assert data["heroes"][0]["name"] == hero3_data["name"]
response = client.delete(f"/teams/{team_preventers_id}")
assert response.status_code == 200, response.text
- response = client.delete("/teams/9000") # Test deleting non-existent team
+ response = client.delete("/teams/9000") # Test deleting non-existent team
assert response.status_code == 404, response.text
response = client.get("/teams/")
assert response.status_code == 200, response.text
data = response.json()
- assert len(data) == 1 # Only Z-Force should remain
+ assert len(data) == 1 # Only Z-Force should remain
# OpenAPI schema check - this is a long part, keeping it as is from the original.
# Small modification to handle potential differences in Pydantic v1 vs v2 for optional fields in schema
# short_module_name is already defined at the start of the 'with TestClient' block
# All versions (base, py39, py310) use HeroPublicWithTeam for this endpoint based on previous test run.
- assert get_hero_path["responses"]["200"]["content"]["application/json"]["schema"]["$ref"] == "#/components/schemas/HeroPublicWithTeam"
+ assert (
+ get_hero_path["responses"]["200"]["content"]["application/json"]["schema"][
+ "$ref"
+ ]
+ == "#/components/schemas/HeroPublicWithTeam"
+ )
# Check HeroCreate schema for age and team_id nullability based on IsDict usage in original
- hero_create_props = openapi_schema["components"]["schemas"]["HeroCreate"]["properties"]
+ hero_create_props = openapi_schema["components"]["schemas"]["HeroCreate"][
+ "properties"
+ ]
# For Pydantic v2 style (anyOf with type and null) vs Pydantic v1 (just type, optionality by not being in required)
# This test was written with IsDict which complicates exact schema matching without knowing SQLModel version's Pydantic interaction
# For simplicity, we check if 'age' and 'team_id' are present. Detailed check would need to adapt to SQLModel's Pydantic version.
# It's better to check for key components and structures.
# Check if TeamPublicWithHeroes has heroes list
- team_public_with_heroes_props = openapi_schema["components"]["schemas"]["TeamPublicWithHeroes"]["properties"]
+ team_public_with_heroes_props = openapi_schema["components"]["schemas"][
+ "TeamPublicWithHeroes"
+ ]["properties"]
assert "heroes" in team_public_with_heroes_props
assert team_public_with_heroes_props["heroes"]["type"] == "array"
# short_module_name is already defined
if short_module_name == "tutorial001_py310":
- assert team_public_with_heroes_props["heroes"]["items"]["$ref"] == "#/components/schemas/HeroPublic" # tutorial001_py310 uses HeroPublic for heroes list
+ assert (
+ team_public_with_heroes_props["heroes"]["items"]["$ref"]
+ == "#/components/schemas/HeroPublic"
+ ) # tutorial001_py310 uses HeroPublic for heroes list
else:
- assert team_public_with_heroes_props["heroes"]["items"]["$ref"] == "#/components/schemas/HeroPublic" # Original tutorial001.py seems to imply HeroPublic as well.
+ assert (
+ team_public_with_heroes_props["heroes"]["items"]["$ref"]
+ == "#/components/schemas/HeroPublic"
+ ) # Original tutorial001.py seems to imply HeroPublic as well.
import pytest
from dirty_equals import IsDict
from fastapi.testclient import TestClient
-from sqlmodel import create_engine, SQLModel
+from sqlmodel import SQLModel, create_engine
from sqlmodel.pool import StaticPool
from ....conftest import needs_py39, needs_py310
assert data[0]["secret_name"] == hero_data["secret_name"]
# Ensure other fields are present as per the model Hero (which is used as response_model)
assert "id" in data[0]
- assert "age" in data[0] # Even if None, it should be in the response
+ assert "age" in data[0] # Even if None, it should be in the response
response = client.get("/openapi.json")
assert response.status_code == 200, response.text
import pytest
from dirty_equals import IsDict
from fastapi.testclient import TestClient
-from sqlmodel import create_engine, SQLModel
+from sqlmodel import create_engine
from sqlmodel.pool import StaticPool
from ....conftest import needs_py39, needs_py310
# Let's rely on the app's startup event as per the tutorial's design.
# If `create_db_and_tables` exists as a global function in the module (outside app event), then call it.
if hasattr(mod, "create_db_and_tables") and callable(mod.create_db_and_tables):
- # Check if it's the function that FastAPI would call, or a standalone one.
- # This tutorial series usually has `create_db_and_tables` called by `app.on_event("startup")`.
- # If the tests run TestClient(mod.app), startup events will run.
- pass # Assuming startup event handles it.
+ # Check if it's the function that FastAPI would call, or a standalone one.
+ # This tutorial series usually has `create_db_and_tables` called by `app.on_event("startup")`.
+ # If the tests run TestClient(mod.app), startup events will run.
+ pass # Assuming startup event handles it.
return mod
hero2_data = {
"name": "Spider-Boy",
"secret_name": "Pedro Parqueador",
- "id": 9000, # This ID might be ignored by DB if it's auto-incrementing primary key
+ "id": 9000, # This ID might be ignored by DB if it's auto-incrementing primary key
}
hero3_data = {
"name": "Rusty-Man",
response = client.post("/heroes/", json=hero2_data)
assert response.status_code == 200, response.text
- hero2_created = response.json() # Use the ID from the created hero
+ hero2_created = response.json() # Use the ID from the created hero
hero2_id = hero2_created["id"]
response = client.post("/heroes/", json=hero3_data)
assert response.status_code == 200, response.text
- response = client.get(f"/heroes/{hero2_id}") # Use the actual ID from DB
+ response = client.get(f"/heroes/{hero2_id}") # Use the actual ID from DB
assert response.status_code == 200, response.text
# If hero ID 9000 was intended to be a specific test case for a non-existent ID
# Otherwise, if hero2 was expected to have ID 9000, this needs adjustment.
# Given typical auto-increment, ID 9000 for hero2 is unlikely unless DB is reset and hero2 is first entry.
# The original test implies hero2_data's ID is not necessarily the created ID.
- response = client.get("/heroes/9000") # Check for a potentially non-existent ID
- assert response.status_code == 404, response.text # Expect 404 if 9000 is not hero2_id and not another hero's ID
+ response = client.get("/heroes/9000") # Check for a potentially non-existent ID
+ assert response.status_code == 404, (
+ response.text
+ ) # Expect 404 if 9000 is not hero2_id and not another hero's ID
response = client.get("/heroes/")
assert response.status_code == 200, response.text
)
assert response.status_code == 200, response.text
- response = client.patch("/heroes/9001", json={"name": "Dragon Cube X"}) # Non-existent ID
+ response = client.patch(
+ "/heroes/9001", json={"name": "Dragon Cube X"}
+ ) # Non-existent ID
assert response.status_code == 404, response.text
response = client.delete(f"/heroes/{hero2_id}")
data = response.json()
assert len(data) == 2
- response = client.delete("/heroes/9000") # Non-existent ID (same as the GET check)
+ response = client.delete(
+ "/heroes/9000"
+ ) # Non-existent ID (same as the GET check)
assert response.status_code == 404, response.text
response = client.get("/openapi.json")
import pytest
from dirty_equals import IsDict
from fastapi.testclient import TestClient
-from sqlmodel import create_engine, SQLModel
+from sqlmodel import create_engine
from sqlmodel.pool import StaticPool
# Adjust the import path based on the file's new location or structure
# Assuming conftest.py is located at tests/conftest.py
-from ....conftest import needs_py310 # This needs to be relative to this file's location
+from ....conftest import (
+ needs_py310, # This needs to be relative to this file's location
+)
@pytest.fixture(
)
def get_module(request: pytest.FixtureRequest, clear_sqlmodel: Any):
module_name = request.param
- full_module_name = (
- f"docs_src.tutorial.fastapi.simple_hero_api.{module_name}"
- )
+ full_module_name = f"docs_src.tutorial.fastapi.simple_hero_api.{module_name}"
if full_module_name in sys.modules:
mod = importlib.reload(sys.modules[full_module_name])
return mod
-def test_tutorial(module: types.ModuleType): # clear_sqlmodel is implicitly used by get_module
+def test_tutorial(
+ module: types.ModuleType,
+): # clear_sqlmodel is implicitly used by get_module
with TestClient(module.app) as client:
hero1_data = {"name": "Deadpond", "secret_name": "Dive Wilson"}
hero2_data = {
"name": "Spider-Boy",
"secret_name": "Pedro Parqueador",
- "id": 9000, # This ID is part of the test logic for this tutorial specifically
+ "id": 9000, # This ID is part of the test logic for this tutorial specifically
}
response = client.post("/heroes/", json=hero1_data)
data = response.json()
import pytest
from dirty_equals import IsDict
from fastapi.testclient import TestClient
-from sqlmodel import create_engine, SQLModel
+from sqlmodel import create_engine
from sqlmodel.pool import StaticPool
from ....conftest import needs_py39, needs_py310
return mod
-def test_tutorial(module: types.ModuleType): # clear_sqlmodel is implicitly used by get_module
+def test_tutorial(
+ module: types.ModuleType,
+): # clear_sqlmodel is implicitly used by get_module
with TestClient(module.app) as client:
# Hero Operations
hero1_data = {"name": "Deadpond", "secret_name": "Dive Wilson"}
- hero2_data = { # This hero's ID might be overridden by DB if not specified or if ID is auto-incrementing
+ hero2_data = { # This hero's ID might be overridden by DB if not specified or if ID is auto-incrementing
"name": "Spider-Boy",
"secret_name": "Pedro Parqueador",
"id": 9000,
response = client.post("/heroes/", json=hero2_data)
assert response.status_code == 200, response.text
hero2_created = response.json()
- hero2_id = hero2_created["id"] # Use the actual ID returned by the DB
+ hero2_id = hero2_created["id"] # Use the actual ID returned by the DB
response = client.post("/heroes/", json=hero3_data)
assert response.status_code == 200, response.text
- response = client.get(f"/heroes/{hero2_id}") # Use DB generated ID
+ response = client.get(f"/heroes/{hero2_id}") # Use DB generated ID
assert response.status_code == 200, response.text
- response = client.get("/heroes/9000") # Check for ID 9000 specifically (could be hero2_id or not)
- if hero2_id == 9000 : # If hero2 got ID 9000
- assert response.status_code == 200, response.text
- else: # If hero2 got a different ID, then 9000 should not exist
- assert response.status_code == 404, response.text
+ response = client.get(
+ "/heroes/9000"
+ ) # Check for ID 9000 specifically (could be hero2_id or not)
+ if hero2_id == 9000: # If hero2 got ID 9000
+ assert response.status_code == 200, response.text
+ else: # If hero2 got a different ID, then 9000 should not exist
+ assert response.status_code == 404, response.text
response = client.get("/heroes/")
assert response.status_code == 200, response.text
data = response.json()
assert len(data) == 3
- response = client.patch(f"/heroes/{hero2_id}", json={"secret_name": "Spider-Youngster"})
+ response = client.patch(
+ f"/heroes/{hero2_id}", json={"secret_name": "Spider-Youngster"}
+ )
assert response.status_code == 200, response.text
- response = client.patch("/heroes/9001", json={"name": "Dragon Cube X"}) # Non-existent ID
+ response = client.patch(
+ "/heroes/9001", json={"name": "Dragon Cube X"}
+ ) # Non-existent ID
assert response.status_code == 404, response.text
response = client.delete(f"/heroes/{hero2_id}")
data = response.json()
assert len(data) == 2
- response = client.delete("/heroes/9000") # Try deleting ID 9000
- if hero2_id == 9000 and hero2_id not in [h["id"] for h in data]: # If it was hero2's ID and hero2 was deleted
- assert response.status_code == 404 # Already deleted
- elif hero2_id != 9000 and 9000 not in [h["id"] for h in data]: # If 9000 was never a valid ID among current heroes
+ response = client.delete("/heroes/9000") # Try deleting ID 9000
+ if hero2_id == 9000 and hero2_id not in [
+ h["id"] for h in data
+ ]: # If it was hero2's ID and hero2 was deleted
+ assert response.status_code == 404 # Already deleted
+ elif hero2_id != 9000 and 9000 not in [
+ h["id"] for h in data
+ ]: # If 9000 was never a valid ID among current heroes
assert response.status_code == 404
- else: # If 9000 was a valid ID of another hero still present (should not happen with current data)
- assert response.status_code == 200 # This case is unlikely with current test data
+ else: # If 9000 was a valid ID of another hero still present (should not happen with current data)
+ assert (
+ response.status_code == 200
+ ) # This case is unlikely with current test data
# Team Operations
team_preventers_data = {"name": "Preventers", "headquarters": "Sharp Tower"}
assert data["headquarters"] == team_preventers_created["headquarters"]
assert data["id"] == team_preventers_created["id"]
- response = client.get("/teams/9000") # Non-existent team ID
+ response = client.get("/teams/9000") # Non-existent team ID
assert response.status_code == 404, response.text
response = client.patch(
)
data = response.json()
assert response.status_code == 200, response.text
- assert data["name"] == team_preventers_data["name"] # Name should be unchanged
+ assert data["name"] == team_preventers_data["name"] # Name should be unchanged
assert data["headquarters"] == "Preventers Tower"
- response = client.patch("/teams/9000", json={"name": "Freedom League"}) # Non-existent
+ response = client.patch(
+ "/teams/9000", json={"name": "Freedom League"}
+ ) # Non-existent
assert response.status_code == 404, response.text
response = client.delete(f"/teams/{team_preventers_id}")
assert response.status_code == 200, response.text
- response = client.delete("/teams/9000") # Non-existent
+ response = client.delete("/teams/9000") # Non-existent
assert response.status_code == 404, response.text
response = client.get("/teams/")
projects / thirdparty / fastapi / sqlmodel.git / commitdiff
