patrickandClaude Sonnet 5
d61aabadf3
objektposition.seriennummer war ein Einzelfeld, konnte nur EIN Gerät pro Materialtyp/Objekt abbilden (z.B. 2 Pulsoxymeter im selben Rucksack nicht darstellbar). Neue Tabelle geraet_instanz erlaubt beliebig viele Exemplare pro Position (Seriennummer, Prüfdatum, nächste Prüfung, Status). Zusätzlich objektposition.pruefintervall_monate (individuell überschreibbar). Bewusst additiv: seriennummer-Spalte bleibt vorerst stehen (Service/ Endpunkte/Frontend hängen noch daran), bestehende Werte werden per INSERT...SELECT nach geraet_instanz kopiert. Drop + Umbau der abhängigen Schichten folgt in Umsetzungsschritt 2 (siehe arbeitskarten/14_...). Migration von postgres-expert gegengeprüft (ON CONFLICT DO NOTHING als Sicherheitsnetz, DROP TYPE IF EXISTS ergänzt), Tests decken Mehrfach- Instanzen, UNIQUE-Constraint je Position, gleiche SN an verschiedenen Positionen und pruefintervall_monate ab. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
100 lines
3.9 KiB
Python
100 lines
3.9 KiB
Python
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app.models.geraet_instanz import GeraetInstanz, GeraetStatus
|
|
from app.models.objektposition import Objektposition
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mehrere_geraete_instanzen_pro_position(db_session, objekt_mit_position):
|
|
"""Karte 14 Kernproblem: objektposition.seriennummer als Einzelfeld konnte nur
|
|
EIN Gerät pro Materialtyp/Objekt abbilden - geraet_instanz erlaubt beliebig
|
|
viele Exemplare (z.B. 2 Pulsoxymeter im selben Rucksack)."""
|
|
objekt, material = objekt_mit_position
|
|
result = await db_session.execute(
|
|
select(Objektposition).where(
|
|
Objektposition.objekt_id == objekt.id, Objektposition.material_id == material.id
|
|
)
|
|
)
|
|
position = result.scalar_one()
|
|
|
|
db_session.add(GeraetInstanz(objektposition_id=position.id, seriennummer="SN-001"))
|
|
db_session.add(GeraetInstanz(objektposition_id=position.id, seriennummer="SN-002"))
|
|
await db_session.flush()
|
|
|
|
instanzen = (
|
|
await db_session.execute(
|
|
select(GeraetInstanz).where(GeraetInstanz.objektposition_id == position.id)
|
|
)
|
|
).scalars().all()
|
|
assert {i.seriennummer for i in instanzen} == {"SN-001", "SN-002"}
|
|
assert all(i.status == GeraetStatus.einsatzbereit for i in instanzen)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_seriennummer_je_position_eindeutig(db_session, objekt_mit_position):
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
objekt, material = objekt_mit_position
|
|
result = await db_session.execute(
|
|
select(Objektposition).where(
|
|
Objektposition.objekt_id == objekt.id, Objektposition.material_id == material.id
|
|
)
|
|
)
|
|
position = result.scalar_one()
|
|
|
|
db_session.add(GeraetInstanz(objektposition_id=position.id, seriennummer="SN-001"))
|
|
await db_session.flush()
|
|
db_session.add(GeraetInstanz(objektposition_id=position.id, seriennummer="SN-001"))
|
|
with pytest.raises(IntegrityError):
|
|
await db_session.flush()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gleiche_seriennummer_an_verschiedenen_positionen_erlaubt(
|
|
db_session, objekt_mit_position, material_factory
|
|
):
|
|
"""UNIQUE(objektposition_id, seriennummer) greift nur je Position, nicht
|
|
global - zwei verschiedene Objektpositionen dürfen dieselbe SN führen
|
|
(z.B. baugleiche Geräte an unterschiedlichen Objekten)."""
|
|
objekt, material = objekt_mit_position
|
|
result = await db_session.execute(
|
|
select(Objektposition).where(
|
|
Objektposition.objekt_id == objekt.id, Objektposition.material_id == material.id
|
|
)
|
|
)
|
|
position_a = result.scalar_one()
|
|
|
|
material_b = await material_factory("Pulsoxymeter")
|
|
position_b = Objektposition(objekt_id=objekt.id, material_id=material_b.id, istmenge=0)
|
|
db_session.add(position_b)
|
|
await db_session.flush()
|
|
|
|
db_session.add(GeraetInstanz(objektposition_id=position_a.id, seriennummer="SN-GLEICH"))
|
|
db_session.add(GeraetInstanz(objektposition_id=position_b.id, seriennummer="SN-GLEICH"))
|
|
await db_session.flush() # darf nicht scheitern
|
|
|
|
anzahl = (
|
|
await db_session.execute(select(GeraetInstanz).where(GeraetInstanz.seriennummer == "SN-GLEICH"))
|
|
).scalars().all()
|
|
assert len(anzahl) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pruefintervall_monate_default_und_setzen(db_session, objekt_mit_position):
|
|
"""Karte 14: pruefintervall_monate individuell überschreibbar je
|
|
Objektposition (wie sollmenge_override), NULL = kein Prüfintervall."""
|
|
objekt, material = objekt_mit_position
|
|
result = await db_session.execute(
|
|
select(Objektposition).where(
|
|
Objektposition.objekt_id == objekt.id, Objektposition.material_id == material.id
|
|
)
|
|
)
|
|
position = result.scalar_one()
|
|
assert position.pruefintervall_monate is None
|
|
|
|
position.pruefintervall_monate = 24
|
|
await db_session.flush()
|
|
await db_session.refresh(position)
|
|
assert position.pruefintervall_monate == 24
|