Wie funktionieren Unit-Tests mit (asynchronen) Scheinobjekten (Pytest, Unittest)? ⇐ Python

[Guest]

Ich bin ursprünglich Datenwissenschaftler, wurde aber mit der Durchführung von Komponententests für den von uns geschriebenen Code beauftragt. Aber um es kurz zu machen: Das ist nicht mein Fachgebiet und ich wurde damit beauftragt. Bisher habe ich Code und wollte eine ausführliche Erklärung, WIE er wirklich funktioniert, und eine Gewissheit darüber, ob das, was ich getan habe, richtig ist – und wenn nicht, wie ich es dann wirklich tun kann. Ich muss das meinen Kollegen erklären können, aber ich selbst bin etwas ratlos, wie diese Bibliothek funktioniert.
Konzentrieren wir uns zunächst auf den Komponententest eines API-Endpunkts. Hier ist ein Endpunkt, der bei Erreichen alle Fälle aus einer SQL-Datenbank für eine bestimmte Agenten-ID abruft:

Code: Select all

@router.get("/cases", response_model=List[CaseInDB])
async def get_all_cases(
agent_id: str = Query(..., description="The ID of the agent to retrieve the cases for"),
db: Session = Depends(get_db)
):
return await case_service.get_all_cases(db, agent_id)

Um nun einen Komponententest durchzuführen:

Code: Select all

import pytest
from fastapi.testclient import TestClient
from unittest.mock import AsyncMock, patch, MagicMock
from datetime import datetime

from app.main import app
from app.schemas.case import CaseCreate, CaseUpdate, CaseInDB
from app.models.case import Case
from app.repository.database import CaseRepository
from app.services.case_service import CaseService

client = TestClient(app)

# Mock Data
MOCK_AGENT_ID = "test_xhuma"
MOCK_CASE_ID = "66666"
MOCK_CASE_DATA = {
"case_title": "Test Case",
"case_description": "Test Description",
"case_status": "Open",
"client_id": "client123",
"client_name": "John Doe",
"client_phone": "+1234567890",
"client_email": "john@example.com",
"client_city": "Test City",
"client_coords": "12.345,67.890",
"client_address": "123 Test St",
"case_diagnosis": "Initial diagnosis"
}

MOCK_CASE_ID_2 = "66667"
MOCK_CASE_DATA_2 = {
"case_title": "Test Case 2",
"case_description": "Test Description",
"case_status": "Open",
"client_id": "client123",
"client_name": "Jane Doe",
"client_phone": "+1234567890",
"client_email": "john@example.com",
"client_city": "Test City",
"client_coords": "12.345,67.890",
"client_address": "123 Test St",
"case_diagnosis": "Initial diagnosis"
}

MOCK_DB_CASE = Case(
agent_id=MOCK_AGENT_ID,
case_id=MOCK_CASE_ID,
created_at=datetime.now(),
updated_at=datetime.now(),
**MOCK_CASE_DATA
)

MOCK_DB_CASE_2 = Case(
agent_id=MOCK_AGENT_ID,
case_id=MOCK_CASE_ID_2,
created_at=datetime.now(),
updated_at=datetime.now(),
**MOCK_CASE_DATA_2
)

@pytest.fixture
def mock_case_service():
with patch('app.api.cases.case_service') as mock:
yield mock

@pytest.fixture
def mock_db():
with patch('app.api.cases.get_db') as mock:
yield mock

@pytest.fixture
def mock_repository():
return MagicMock(spec=CaseRepository)

@pytest.fixture
def case_service(mock_repository):
return CaseService(mock_repository)

@pytest.fixture
def mock_redis_publisher():
with patch('app.services.case_service.publish_to_stream', new_callable=AsyncMock) as mock:
yield mock

class TestCaseAPI:

def test_get_all_cases(self, mock_case_service, mock_db):

mock_cases = [MOCK_DB_CASE, MOCK_DB_CASE_2]
mock_case_service.get_all_cases = AsyncMock(return_value=[CaseInDB.model_validate(case) for case in mock_cases])

response = client.get(f"/cases?agent_id={MOCK_AGENT_ID}")

assert response.status_code == 200
assert len(response.json()) == 2
assert response.json()[0]["case_id"] == MOCK_CASE_ID
assert response.json()[1]["case_id"] == MOCK_CASE_ID_2

Hier habe ich zwei Testfälle erstellt, die entsprechenden Fixtures oder Mocks, denke ich, für den Fastapi-Client und den case_service, der den Abruf durchführt (die Funktion get_all_cases). Ich verstehe, dass diese beiden Zeilen den Vorgang des Abrufens der Fälle aus der Datenbank „verspotten“:

Code: Select all

mock_cases = [MOCK_DB_CASE, MOCK_DB_CASE_2]
mock_case_service.get_all_cases = AsyncMock(return_value=[CaseInDB.model_validate(case) for case in mock_cases])

Aber wie startet der Testclient nun diesen Endpunkt und ruft die beiden Scheinfälle ab, anstatt auf die Datenbank zuzugreifen? Ich verstehe, dass es sich um ein TestClient-Objekt handelt, aber woher weiß es, dass es diese Scheinfälle erhält?
Hier ist nun der eigentliche Code, der in die Datenbank gelangt:

Code: Select all

class CaseRepository(BaseRepository[Case, CaseCreate, CaseUpdate]):
def __init__(self):
super().__init__(Case)

async def get_all(self, db: Session, agent_id: str) -> List[Case]:
return (
db.query(self.model)
.filter(self.model.agent_id == agent_id)
.order_by(desc(self.model.case_id))  # Order by case_id in descending order
.all()
)

Und dies ist ein Komponententest, um diesen Teil zu testen:

Code: Select all

class TestCaseService:

@pytest.mark.asyncio
async def test_get_all_cases(self, case_service, mock_repository, mock_db):

mock_cases = [MOCK_DB_CASE, MOCK_DB_CASE_2]
mock_repository.get_all = AsyncMock(return_value=mock_cases)

result = await case_service.get_all_cases(mock_db, MOCK_AGENT_ID)

assert len(result) == 2
assert isinstance(result[0], CaseInDB)
assert isinstance(result[1], CaseInDB)
assert result[0].case_id == MOCK_CASE_ID
assert result[1].case_id == MOCK_CASE_ID_2
mock_repository.get_all.assert_awaited_once_with(mock_db, MOCK_AGENT_ID)

Wieder erhalte ich, dass die ersten beiden Zeilen „simulieren“ und die beiden simulierten Fälle mit dieser Funktion get_all abrufen. Aber jetzt verwenden wir das Fixture case_service, um alle Fälle von einer mocked_db abzurufen. Wie werden diese beiden Scheinfälle zurückgegeben und nicht die tatsächlichen Fälle in der SQL-Datenbank?