101 lines
3.2 KiB
Python
101 lines
3.2 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()
|
|
project = SimpleNamespace(id=project_id, code="demo", status="active")
|
|
repo = SimpleNamespace(
|
|
session=object(),
|
|
get_project_by_id=AsyncMock(return_value=project),
|
|
)
|
|
admin = SimpleNamespace(id=uuid4(), role="admin", is_superuser=False)
|
|
monkeypatch.setattr(
|
|
model_import,
|
|
"_run_uploaded_inp",
|
|
AsyncMock(return_value="imported"),
|
|
)
|
|
monkeypatch.setattr(model_import, "log_audit_event", AsyncMock())
|
|
client = _client(admin=admin, repo=repo)
|
|
|
|
response = client.post(
|
|
f"/api/v1/admin/projects/{project_id}/model-imports",
|
|
files={"file": ("desktop-model.inp", VALID_INP)},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["project_id"] == str(project_id)
|
|
assert response.json()["result"] == "imported"
|
|
repo.get_project_by_id.assert_awaited_once_with(project_id)
|
|
model_import.log_audit_event.assert_awaited_once()
|
|
|
|
|
|
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-imports",
|
|
files={"file": ("desktop-model.inp", VALID_INP)},
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
assert response.json()["detail"] == "Admin access required"
|
|
|
|
|
|
def test_model_import_rejects_non_inp_file(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, "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-imports",
|
|
files={"file": ("desktop-model.txt", VALID_INP)},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"] == "Only .inp model files are accepted"
|
|
model_import.log_audit_event.assert_not_awaited()
|