patrickandClaude Sonnet 5
76a984ad56
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
89 lines
3.4 KiB
Python
89 lines
3.4 KiB
Python
import pytest
|
|
|
|
from tests.conftest import login
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_objekte_export_import_roundtrip(client, admin_user, standort_factory, objekt_factory):
|
|
standort = await standort_factory("Wache-Export")
|
|
await objekt_factory(name="Rucksack A", code="RS-EXP-1", standort=standort)
|
|
token = await login(client, "admin1")
|
|
header = {"Authorization": f"Bearer {token}"}
|
|
|
|
export = await client.get("/api/v1/objekte/export", headers=header)
|
|
assert export.status_code == 200
|
|
assert "RS-EXP-1" in export.text
|
|
|
|
upload = {"datei": ("objekte.csv", export.text, "text/csv")}
|
|
dry = await client.post("/api/v1/objekte/import?commit=false", files=upload, headers=header)
|
|
assert dry.status_code == 200
|
|
daten = dry.json()
|
|
assert daten["gesamt"] == 1
|
|
assert daten["uebernommen"] == 1
|
|
assert daten["fehler"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_objekte_import_neue_zeile_mit_fehler(client, admin_user, standort_factory):
|
|
standort = await standort_factory("Wache-Import")
|
|
token = await login(client, "admin1")
|
|
header = {"Authorization": f"Bearer {token}"}
|
|
|
|
csv_inhalt = (
|
|
"code,name,objekttyp,standort,status,fahrzeug_code\n"
|
|
f"RS-IMP-1,Rucksack Import,Typ-Fehlt,{standort.name},aktiv,\n"
|
|
)
|
|
upload = {"datei": ("objekte.csv", csv_inhalt, "text/csv")}
|
|
result = await client.post("/api/v1/objekte/import?commit=true", files=upload, headers=header)
|
|
assert result.status_code == 200
|
|
daten = result.json()
|
|
assert daten["uebernommen"] == 0
|
|
assert len(daten["fehler"]) == 1
|
|
assert "Objekttyp" in daten["fehler"][0]["fehler"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_geraete_export_import_roundtrip(
|
|
client, admin_user, standort_factory, objekt_factory, material_factory, db_session
|
|
):
|
|
from app.models.objektposition import Objektposition
|
|
from decimal import Decimal
|
|
|
|
standort = await standort_factory("Wache-Geraete")
|
|
objekt = await objekt_factory(name="Rucksack B", code="RS-GER-1", standort=standort)
|
|
material = await material_factory("Pulsoxymeter")
|
|
position = Objektposition(
|
|
objekt_id=objekt.id, material_id=material.id, sollmenge_override=Decimal(1), istmenge=Decimal(0)
|
|
)
|
|
db_session.add(position)
|
|
await db_session.flush()
|
|
|
|
from app.services.geraet_instanz import erstelle
|
|
|
|
await erstelle(db_session, objektposition_id=position.id, seriennummer="SN-0001")
|
|
await db_session.commit()
|
|
|
|
token = await login(client, "admin1")
|
|
header = {"Authorization": f"Bearer {token}"}
|
|
|
|
export = await client.get("/api/v1/geraete/export", headers=header)
|
|
assert export.status_code == 200
|
|
assert "SN-0001" in export.text
|
|
assert "RS-GER-1" in export.text
|
|
|
|
csv_neu = (
|
|
"objekt_code,material_name,seriennummer,pruefdatum,naechste_pruefung,status,bemerkung\n"
|
|
f"RS-GER-1,Pulsoxymeter,SN-0002,,,einsatzbereit,\n"
|
|
)
|
|
upload = {"datei": ("geraete.csv", csv_neu, "text/csv")}
|
|
result = await client.post("/api/v1/geraete/import?commit=true", files=upload, headers=header)
|
|
assert result.status_code == 200
|
|
daten = result.json()
|
|
assert daten["uebernommen"] == 1
|
|
assert daten["fehler"] == []
|
|
|
|
doppelt = await client.post("/api/v1/geraete/import?commit=true", files=upload, headers=header)
|
|
daten_doppelt = doppelt.json()
|
|
assert daten_doppelt["uebernommen"] == 0
|
|
assert "existiert bereits" in daten_doppelt["fehler"][0]["fehler"]
|