88 lines
2.6 KiB
Python
88 lines
2.6 KiB
Python
from types import SimpleNamespace
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
from tests.conftest import install_stub, load_module_from_path
|
|
|
|
|
|
def _load_meta_module(monkeypatch):
|
|
install_stub(monkeypatch, "app.auth", package=True)
|
|
install_stub(
|
|
monkeypatch,
|
|
"app.auth.project_dependencies",
|
|
{
|
|
"ProjectContext": object,
|
|
"get_project_context": lambda: None,
|
|
"get_project_pg_session": lambda: None,
|
|
"get_project_timescale_connection": lambda: None,
|
|
"get_metadata_repository": lambda: None,
|
|
},
|
|
)
|
|
install_stub(
|
|
monkeypatch,
|
|
"app.auth.metadata_dependencies",
|
|
{"get_current_metadata_user": lambda: None},
|
|
)
|
|
return load_module_from_path(
|
|
"tests_meta_endpoints_module",
|
|
"app/api/v1/endpoints/meta.py",
|
|
)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_meta_project_returns_map_extent(monkeypatch):
|
|
module = _load_meta_module(monkeypatch)
|
|
project_id = uuid4()
|
|
repo = SimpleNamespace(
|
|
get_project_by_id=lambda _project_id: None,
|
|
)
|
|
|
|
async def get_project_by_id(_project_id):
|
|
return SimpleNamespace(
|
|
id=project_id,
|
|
name="Demo Project",
|
|
code="demo",
|
|
description="desc",
|
|
gs_workspace="workspace",
|
|
map_extent={"xmin": 1, "ymin": 2, "xmax": 3, "ymax": 4},
|
|
status="active",
|
|
)
|
|
|
|
repo.get_project_by_id = get_project_by_id
|
|
|
|
response = await module.get_project_metadata(
|
|
ctx=SimpleNamespace(
|
|
project_id=project_id,
|
|
project_role="editor",
|
|
),
|
|
metadata_repo=repo,
|
|
)
|
|
|
|
assert response.map_extent == {"xmin": 1, "ymin": 2, "xmax": 3, "ymax": 4}
|
|
assert response.project_id == project_id
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_meta_db_health_returns_503_for_postgres_errors(monkeypatch):
|
|
module = _load_meta_module(monkeypatch)
|
|
|
|
class BrokenSession:
|
|
async def execute(self, _query):
|
|
raise SQLAlchemyError("pg unavailable")
|
|
|
|
class DummyTimescaleConnection:
|
|
def cursor(self):
|
|
raise AssertionError("timescale should not be queried after postgres failure")
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await module.project_db_health(
|
|
pg_session=BrokenSession(),
|
|
ts_conn=DummyTimescaleConnection(),
|
|
)
|
|
|
|
assert exc_info.value.status_code == 503
|
|
assert exc_info.value.detail == "Project PostgreSQL health check failed: pg unavailable"
|