Files
TJWaterServerBinary/tests/api/test_project_endpoints.py
T

101 lines
3.5 KiB
Python

from types import SimpleNamespace
from unittest.mock import AsyncMock
from uuid import uuid4
import pytest
from fastapi import HTTPException
from tests.conftest import install_stub, load_module_from_path
class DummyChangeSet:
def __init__(self, operations=None):
if operations is None:
self.operations = []
elif isinstance(operations, dict):
self.operations = [operations]
else:
self.operations = operations
def _load_project_module(monkeypatch):
install_stub(monkeypatch, "app.services", package=True)
install_stub(monkeypatch, "app.services.project_info", {})
install_stub(
monkeypatch,
"app.services.tjnetwork",
{
"ChangeSet": DummyChangeSet,
"list_project": lambda: ["demo"],
"have_project": lambda network: network == "demo",
"create_project": lambda network: None,
"delete_project": lambda network: None,
"is_project_open": lambda network: False,
"open_project": lambda network: None,
"close_project": lambda network: None,
"copy_project": lambda source, target: None,
"import_inp": lambda network, cs: {"ok": True},
"export_inp": lambda network, version: DummyChangeSet({"kind": "export"}),
"read_inp": lambda network, inp: True,
"dump_inp": lambda network, inp: True,
"get_all_vertices": lambda network: [],
"get_all_scada_elements": lambda network: [],
"get_all_district_metering_areas": lambda network: [],
"get_all_service_areas": lambda network: [],
"get_all_virtual_districts": lambda network: [],
"get_extension_data": lambda network, key: None,
"convert_inp_v3_to_v2": lambda inp: DummyChangeSet({"inp": inp}),
},
)
install_stub(
monkeypatch,
"app.auth.project_dependencies",
{"get_metadata_repository": lambda: None},
)
install_stub(
monkeypatch,
"app.infra.db.postgresql.database",
{"get_database_instance": lambda network: None},
)
install_stub(
monkeypatch,
"app.infra.db.timescaledb.database",
{"get_database_instance": lambda network: None},
)
return load_module_from_path(
"tests_project_endpoints_module",
"app/api/v1/endpoints/project.py",
)
@pytest.mark.anyio
async def test_project_info_returns_404_when_missing(monkeypatch):
module = _load_project_module(monkeypatch)
repo = SimpleNamespace(get_project_detail_by_code=AsyncMock(return_value=None))
with pytest.raises(HTTPException) as exc_info:
await module.get_project_info_endpoint(network="missing", metadata_repo=repo)
assert exc_info.value.status_code == 404
assert exc_info.value.detail == "Project missing not found"
@pytest.mark.anyio
async def test_project_info_returns_project_metadata(monkeypatch):
module = _load_project_module(monkeypatch)
detail = SimpleNamespace(
project_id=uuid4(),
name="Demo Project",
code="demo",
description="desc",
gs_workspace="ws",
map_extent={"xmin": 1, "ymin": 2, "xmax": 3, "ymax": 4},
status="active",
)
repo = SimpleNamespace(get_project_detail_by_code=AsyncMock(return_value=detail))
payload = await module.get_project_info_endpoint(network="demo", metadata_repo=repo)
assert payload.code == "demo"
assert payload.map_extent == {"xmin": 1, "ymin": 2, "xmax": 3, "ymax": 4}