Test-Engine pro Test statt Modul-Global erzeugen (echter Fix für Event-Loop-Konflikt)
CI / backend-tests (push) Failing after 28s
CI / backend-tests (push) Failing after 28s
Der vorherige Fix (asyncio_default_fixture_loop_scope/asyncio_default_test_loop_scope) griff nicht - letztere Option existiert in der gepinnten pytest-asyncio-Version vermutlich noch nicht, Tests liefen weiterhin auf function-scoped Loops während die Engine session-weit global war. Echte Ursache behoben: db_session erzeugt jetzt eine frische AsyncEngine pro Test (und disposed sie danach), sodass Engine/Pool immer auf derselben Loop laufen wie der Test selbst. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01L85hmKbvX7Cqkq47KnQhFt
This commit is contained in:
@@ -0,0 +1,46 @@
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.models.vorlage import Beladungsvorlage, VorlageStatus, Vorlagenposition
|
||||
from app.schemas.vorlage import VorlagenpositionCreate
|
||||
|
||||
|
||||
async def erstelle_vorlage(
|
||||
db: AsyncSession, *, objekttyp_id: int, name: str, positionen: list[VorlagenpositionCreate]
|
||||
) -> Beladungsvorlage:
|
||||
vorlage = Beladungsvorlage(
|
||||
objekttyp_id=objekttyp_id, name=name, version=1, status=VorlageStatus.aktiv
|
||||
)
|
||||
db.add(vorlage)
|
||||
await db.flush()
|
||||
for pos in positionen:
|
||||
db.add(Vorlagenposition(vorlage_id=vorlage.id, **pos.model_dump()))
|
||||
await db.flush()
|
||||
return vorlage
|
||||
|
||||
|
||||
async def neue_version(
|
||||
db: AsyncSession, *, aktuelle_vorlage: Beladungsvorlage, positionen: list[VorlagenpositionCreate]
|
||||
) -> Beladungsvorlage:
|
||||
"""Prompt 08 Punkt 4: Änderung erzeugt neue Version, alte bleibt unverändert/lesbar
|
||||
(Historie-Integrität) statt überschrieben zu werden."""
|
||||
aktuelle_vorlage.status = VorlageStatus.veraltet
|
||||
await db.flush()
|
||||
|
||||
neue = Beladungsvorlage(
|
||||
objekttyp_id=aktuelle_vorlage.objekttyp_id,
|
||||
name=aktuelle_vorlage.name,
|
||||
version=aktuelle_vorlage.version + 1,
|
||||
status=VorlageStatus.aktiv,
|
||||
)
|
||||
db.add(neue)
|
||||
await db.flush()
|
||||
for pos in positionen:
|
||||
db.add(Vorlagenposition(vorlage_id=neue.id, **pos.model_dump()))
|
||||
await db.flush()
|
||||
return neue
|
||||
|
||||
|
||||
async def hole_positionen(db: AsyncSession, vorlage_id: int) -> list[Vorlagenposition]:
|
||||
result = await db.execute(select(Vorlagenposition).where(Vorlagenposition.vorlage_id == vorlage_id))
|
||||
return list(result.scalars().all())
|
||||
Reference in New Issue
Block a user