88 lines
2.4 KiB
Python
88 lines
2.4 KiB
Python
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from sqlalchemy.dialects import postgresql
|
|
|
|
from app.infra.db.metadb.repositories import metadata_repository
|
|
from app.infra.db.metadb.repositories.metadata_repository import MetadataRepository
|
|
|
|
|
|
@pytest.fixture
|
|
def anyio_backend():
|
|
return "asyncio"
|
|
|
|
|
|
class _ProjectListResult:
|
|
def __init__(self, rows):
|
|
self._rows = rows
|
|
|
|
def all(self):
|
|
return self._rows
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_list_projects_for_user_only_queries_active_projects():
|
|
user_id = uuid4()
|
|
project = SimpleNamespace(
|
|
id=uuid4(),
|
|
name="Active Project",
|
|
code="active-project",
|
|
description=None,
|
|
gs_workspace="active-project",
|
|
map_extent=None,
|
|
status="active",
|
|
)
|
|
session = SimpleNamespace(
|
|
execute=AsyncMock(return_value=_ProjectListResult([(project, "member")]))
|
|
)
|
|
|
|
projects = await MetadataRepository(session).list_projects_for_user(user_id)
|
|
|
|
statement = session.execute.await_args.args[0]
|
|
sql = str(
|
|
statement.compile(
|
|
dialect=postgresql.dialect(),
|
|
compile_kwargs={"literal_binds": True},
|
|
)
|
|
)
|
|
assert "user_project_membership.user_id" in sql
|
|
assert "projects.status = 'active'" in sql
|
|
assert [item.code for item in projects] == ["active-project"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_provisioned_project_commits_project_routes_and_membership_once(
|
|
monkeypatch,
|
|
):
|
|
added = []
|
|
session = SimpleNamespace(
|
|
add=added.append,
|
|
commit=AsyncMock(),
|
|
refresh=AsyncMock(),
|
|
)
|
|
monkeypatch.setattr(
|
|
metadata_repository,
|
|
"_encrypt_database_secret",
|
|
lambda value: f"encrypted::{value}",
|
|
)
|
|
user_id = uuid4()
|
|
|
|
project = await MetadataRepository(session).create_provisioned_project(
|
|
name="Demo",
|
|
code="demo",
|
|
description=None,
|
|
gs_workspace="demo",
|
|
map_extent={"bbox": [1, 2, 3, 4], "zoom": 14},
|
|
creator_user_id=user_id,
|
|
business_dsn="postgresql://business/demo",
|
|
timescale_dsn="postgresql://timescale/demo",
|
|
)
|
|
|
|
assert project.status == "active"
|
|
assert [record.db_role for record in added[1:3]] == ["biz_data", "iot_data"]
|
|
assert added[3].user_id == user_id
|
|
session.commit.assert_awaited_once()
|
|
session.refresh.assert_awaited_once_with(project)
|