80 lines
2.2 KiB
Python
80 lines
2.2 KiB
Python
import io
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.api.import_data import router as import_router
|
|
from app.database.base import engine, SessionLocal
|
|
from app.models.base import BaseModel
|
|
from app.models.user import User
|
|
from app.models.flexible import FlexibleImport
|
|
from app.auth.security import get_current_user
|
|
|
|
|
|
def test_batch_upload_unknown_csv_saved_as_flexible_rows():
|
|
# Fresh DB
|
|
BaseModel.metadata.drop_all(bind=engine)
|
|
BaseModel.metadata.create_all(bind=engine)
|
|
|
|
# Seed an admin user
|
|
db = SessionLocal()
|
|
try:
|
|
user = User(
|
|
username="tester",
|
|
email="tester@example.com",
|
|
hashed_password="x",
|
|
is_active=True,
|
|
is_admin=True,
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
finally:
|
|
db.close()
|
|
|
|
# Minimal app with import router and auth override
|
|
app = FastAPI()
|
|
app.include_router(import_router, prefix="/api/import")
|
|
|
|
def _override_current_user():
|
|
return user # type: ignore[return-value]
|
|
|
|
app.dependency_overrides[get_current_user] = _override_current_user
|
|
|
|
client = TestClient(app)
|
|
|
|
# Unknown CSV that should fall back to flexible import
|
|
csv_bytes = b"alpha,beta\n1,2\n3,4\n"
|
|
files = [("files", ("UNKNOWN.csv", io.BytesIO(csv_bytes), "text/csv"))]
|
|
|
|
resp = client.post("/api/import/batch-upload", files=files)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
|
|
# Assert API result shows success and correct row count
|
|
results = body.get("batch_results", [])
|
|
assert any(
|
|
r.get("file_type") == "UNKNOWN.csv"
|
|
and r.get("status") == "success"
|
|
and r.get("imported_count") == 2
|
|
for r in results
|
|
)
|
|
|
|
# Assert rows persisted in flexible storage
|
|
db = SessionLocal()
|
|
try:
|
|
rows = (
|
|
db.query(FlexibleImport)
|
|
.filter(FlexibleImport.file_type == "UNKNOWN.csv")
|
|
.order_by(FlexibleImport.id.asc())
|
|
.all()
|
|
)
|
|
assert len(rows) == 2
|
|
assert rows[0].target_table is None
|
|
assert set(rows[0].extra_data.keys()) == {"alpha", "beta"}
|
|
assert set(rows[1].extra_data.keys()) == {"alpha", "beta"}
|
|
finally:
|
|
db.close()
|
|
|
|
|