Files
TJWaterServerBinary/tests/api/test_model_import_endpoints.py
T

69 lines
2.1 KiB
Python

from types import SimpleNamespace
from unittest.mock import AsyncMock
from uuid import uuid4
from fastapi import HTTPException
from fastapi.testclient import TestClient
from app.api.v1.endpoints import model_import
from app.auth.metadata_dependencies import (
get_current_metadata_admin,
get_metadata_repository,
)
from tests.conftest import build_test_app
VALID_INP = b"[TITLE]\nDesktop model\n[JUNCTIONS]\n;ID Elev Demand\n"
def _client(*, admin=None, repo=None) -> TestClient:
app = build_test_app(model_import.router, "/api/v1")
if admin is not None:
app.dependency_overrides[get_current_metadata_admin] = lambda: admin
if repo is not None:
app.dependency_overrides[get_metadata_repository] = lambda: repo
return TestClient(app)
def test_system_admin_can_import_model_without_project_membership(monkeypatch):
project_id = uuid4()
repo = SimpleNamespace(
session=object(),
get_project_by_id=AsyncMock(
return_value=SimpleNamespace(id=project_id, code="demo", status="active")
),
)
monkeypatch.setattr(
model_import, "_run_uploaded_inp", AsyncMock(return_value="imported")
)
monkeypatch.setattr(model_import, "log_audit_event", AsyncMock())
client = _client(
admin=SimpleNamespace(id=uuid4(), role="admin", is_superuser=False),
repo=repo,
)
response = client.post(
f"/api/v1/admin/projects/{project_id}/model/import",
files={"file": ("desktop-model.inp", VALID_INP)},
)
assert response.status_code == 200
assert response.json()["result"] == "imported"
repo.get_project_by_id.assert_awaited_once_with(project_id)
def test_non_admin_is_denied_model_import():
def deny_admin():
raise HTTPException(status_code=403, detail="Admin access required")
app = build_test_app(model_import.router, "/api/v1")
app.dependency_overrides[get_current_metadata_admin] = deny_admin
client = TestClient(app)
response = client.post(
f"/api/v1/admin/projects/{uuid4()}/model/import",
files={"file": ("desktop-model.inp", VALID_INP)},
)
assert response.status_code == 403