feat(admin): add project metadata config
This commit is contained in:
@@ -0,0 +1,646 @@
|
||||
from datetime import datetime, timezone
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
from fastapi import Response
|
||||
|
||||
from app.auth.metadata_dependencies import get_current_metadata_admin
|
||||
from app.api.v1.endpoints import admin_metadata
|
||||
from app.domain.schemas.admin_metadata import (
|
||||
AdminProjectCreateRequest,
|
||||
MetadataUsersBatchSyncRequest,
|
||||
MetadataUserSyncRequest,
|
||||
MetadataUserUpdateRequest,
|
||||
ProjectDatabaseUpsertRequest,
|
||||
ProjectGeoServerConfigUpsertRequest,
|
||||
ProjectMemberCreateRequest,
|
||||
ProjectMemberUpdateRequest,
|
||||
)
|
||||
from app.infra.db.metadb.repositories.metadata_repository import ProjectDbRouting
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anyio_backend():
|
||||
return "asyncio"
|
||||
|
||||
|
||||
def _user(**overrides):
|
||||
data = {
|
||||
"id": uuid4(),
|
||||
"keycloak_id": uuid4(),
|
||||
"username": "alice",
|
||||
"email": "alice@example.com",
|
||||
"role": "user",
|
||||
"is_active": True,
|
||||
"is_superuser": False,
|
||||
"created_at": datetime(2026, 1, 1, tzinfo=timezone.utc),
|
||||
"updated_at": datetime(2026, 1, 1, tzinfo=timezone.utc),
|
||||
"last_login_at": None,
|
||||
}
|
||||
data.update(overrides)
|
||||
return SimpleNamespace(**data)
|
||||
|
||||
|
||||
def _project(**overrides):
|
||||
data = {"id": uuid4(), "name": "Demo"}
|
||||
data.update(overrides)
|
||||
return SimpleNamespace(**data)
|
||||
|
||||
|
||||
def _membership(**overrides):
|
||||
data = {
|
||||
"id": uuid4(),
|
||||
"user_id": uuid4(),
|
||||
"project_id": uuid4(),
|
||||
"project_role": "viewer",
|
||||
}
|
||||
data.update(overrides)
|
||||
return SimpleNamespace(**data)
|
||||
|
||||
|
||||
def _database_config(**overrides):
|
||||
data = {
|
||||
"id": uuid4(),
|
||||
"project_id": uuid4(),
|
||||
"db_role": "biz_data",
|
||||
"db_type": "postgresql",
|
||||
"dsn_encrypted": "encrypted-dsn",
|
||||
"pool_min_size": 1,
|
||||
"pool_max_size": 5,
|
||||
}
|
||||
data.update(overrides)
|
||||
return SimpleNamespace(**data)
|
||||
|
||||
|
||||
def _geoserver_config(**overrides):
|
||||
data = {
|
||||
"id": uuid4(),
|
||||
"project_id": uuid4(),
|
||||
"gs_base_url": "http://geoserver",
|
||||
"gs_admin_user": "admin",
|
||||
"gs_admin_password_encrypted": "encrypted-password",
|
||||
"gs_datastore_name": "ds_postgis",
|
||||
"default_extent": {"bbox": [1, 2, 3, 4]},
|
||||
"srid": 4326,
|
||||
"updated_at": datetime(2026, 1, 1, tzinfo=timezone.utc),
|
||||
}
|
||||
data.update(overrides)
|
||||
return SimpleNamespace(**data)
|
||||
|
||||
|
||||
def test_to_async_sqlalchemy_url_preserves_password():
|
||||
url = admin_metadata._to_async_sqlalchemy_url(
|
||||
"postgresql://tjwater:secret@192.168.1.114:5433/tjwater"
|
||||
)
|
||||
|
||||
assert url == "postgresql+psycopg://tjwater:secret@192.168.1.114:5433/tjwater"
|
||||
assert "***" not in url
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_sync_metadata_user_upserts_without_password(monkeypatch):
|
||||
keycloak_id = uuid4()
|
||||
synced_user = _user(keycloak_id=keycloak_id, username="new-user")
|
||||
admin = _user(role="admin", is_superuser=True)
|
||||
repo = SimpleNamespace(
|
||||
session=object(),
|
||||
upsert_user_from_keycloak=AsyncMock(return_value=synced_user),
|
||||
)
|
||||
monkeypatch.setattr(admin_metadata, "log_audit_event", AsyncMock())
|
||||
|
||||
response = await admin_metadata.sync_metadata_user(
|
||||
MetadataUserSyncRequest(
|
||||
keycloak_id=keycloak_id,
|
||||
username="new-user",
|
||||
email="new-user@example.com",
|
||||
role="user",
|
||||
is_active=True,
|
||||
),
|
||||
current_user=admin,
|
||||
metadata_repo=repo,
|
||||
)
|
||||
|
||||
repo.upsert_user_from_keycloak.assert_awaited_once()
|
||||
kwargs = repo.upsert_user_from_keycloak.await_args.kwargs
|
||||
assert kwargs["keycloak_id"] == keycloak_id
|
||||
assert "password" not in kwargs
|
||||
assert response.username == "new-user"
|
||||
admin_metadata.log_audit_event.assert_awaited_once()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_batch_sync_metadata_users_returns_per_user_results(monkeypatch):
|
||||
users = [_user(username="alice"), _user(username="bob")]
|
||||
admin = _user(role="admin", is_superuser=True)
|
||||
repo = SimpleNamespace(
|
||||
session=SimpleNamespace(rollback=AsyncMock()),
|
||||
upsert_user_from_keycloak=AsyncMock(side_effect=users),
|
||||
)
|
||||
monkeypatch.setattr(admin_metadata, "log_audit_event", AsyncMock())
|
||||
|
||||
response = await admin_metadata.sync_metadata_users_batch(
|
||||
MetadataUsersBatchSyncRequest(
|
||||
users=[
|
||||
MetadataUserSyncRequest(
|
||||
keycloak_id=users[0].keycloak_id,
|
||||
username="alice",
|
||||
email="alice@example.com",
|
||||
role="user",
|
||||
is_active=True,
|
||||
),
|
||||
MetadataUserSyncRequest(
|
||||
keycloak_id=users[1].keycloak_id,
|
||||
username="bob",
|
||||
email="bob@example.com",
|
||||
role="viewer",
|
||||
is_active=True,
|
||||
),
|
||||
]
|
||||
),
|
||||
current_user=admin,
|
||||
metadata_repo=repo,
|
||||
)
|
||||
|
||||
assert [item.success for item in response] == [True, True]
|
||||
assert [item.user.username for item in response] == ["alice", "bob"]
|
||||
assert repo.upsert_user_from_keycloak.await_count == 2
|
||||
assert admin_metadata.log_audit_event.await_count == 2
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_update_metadata_user_updates_role_and_active_status(monkeypatch):
|
||||
user_id = uuid4()
|
||||
updated = _user(id=user_id, role="operator", is_active=False)
|
||||
repo = SimpleNamespace(
|
||||
session=object(),
|
||||
update_user_admin=AsyncMock(return_value=updated),
|
||||
)
|
||||
monkeypatch.setattr(admin_metadata, "log_audit_event", AsyncMock())
|
||||
|
||||
response = await admin_metadata.update_metadata_user(
|
||||
MetadataUserUpdateRequest(
|
||||
role="operator",
|
||||
is_active=False,
|
||||
),
|
||||
user_id=user_id,
|
||||
current_user=_user(role="admin", is_superuser=True),
|
||||
metadata_repo=repo,
|
||||
)
|
||||
|
||||
repo.update_user_admin.assert_awaited_once_with(
|
||||
user_id,
|
||||
updates={"role": "operator", "is_active": False},
|
||||
)
|
||||
assert response.role == "operator"
|
||||
admin_metadata.log_audit_event.assert_awaited_once()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_update_metadata_user_rejects_self_update(monkeypatch):
|
||||
current_user = _user(role="admin", is_superuser=True)
|
||||
repo = SimpleNamespace(
|
||||
session=object(),
|
||||
update_user_admin=AsyncMock(),
|
||||
)
|
||||
monkeypatch.setattr(admin_metadata, "log_audit_event", AsyncMock())
|
||||
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
await admin_metadata.update_metadata_user(
|
||||
MetadataUserUpdateRequest(role="viewer"),
|
||||
user_id=current_user.id,
|
||||
current_user=current_user,
|
||||
metadata_repo=repo,
|
||||
)
|
||||
|
||||
assert exc.value.status_code == 403
|
||||
repo.update_user_admin.assert_not_called()
|
||||
admin_metadata.log_audit_event.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_create_project_audits_metadata_admin_change(monkeypatch):
|
||||
project = SimpleNamespace(
|
||||
id=uuid4(),
|
||||
name="Demo Project",
|
||||
code="demo",
|
||||
description="desc",
|
||||
gs_workspace="demo_ws",
|
||||
map_extent={"bbox": [1, 2, 3, 4]},
|
||||
status="active",
|
||||
created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
|
||||
updated_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
|
||||
)
|
||||
repo = SimpleNamespace(
|
||||
session=object(),
|
||||
create_project=AsyncMock(return_value=project),
|
||||
)
|
||||
monkeypatch.setattr(admin_metadata, "log_audit_event", AsyncMock())
|
||||
|
||||
response = await admin_metadata.create_admin_project(
|
||||
AdminProjectCreateRequest(
|
||||
name="Demo Project",
|
||||
code="demo",
|
||||
description="desc",
|
||||
gs_workspace="demo_ws",
|
||||
map_extent={"bbox": [1, 2, 3, 4]},
|
||||
status="active",
|
||||
),
|
||||
current_user=_user(role="admin", is_superuser=True),
|
||||
metadata_repo=repo,
|
||||
)
|
||||
|
||||
assert response.project_id == project.id
|
||||
repo.create_project.assert_awaited_once()
|
||||
admin_metadata.log_audit_event.assert_awaited_once()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_upsert_project_database_hides_dsn_and_audits_without_plaintext(monkeypatch):
|
||||
project_id = uuid4()
|
||||
record = _database_config(project_id=project_id)
|
||||
repo = SimpleNamespace(
|
||||
session=object(),
|
||||
get_project_by_id=AsyncMock(return_value=_project(id=project_id)),
|
||||
upsert_project_database_config=AsyncMock(return_value=record),
|
||||
)
|
||||
monkeypatch.setattr(admin_metadata, "log_audit_event", AsyncMock())
|
||||
monkeypatch.setattr(admin_metadata, "_check_database_connection", AsyncMock())
|
||||
|
||||
response = await admin_metadata.upsert_project_database(
|
||||
ProjectDatabaseUpsertRequest(
|
||||
db_role="biz_data",
|
||||
dsn="postgresql://user:secret@localhost/db",
|
||||
pool_min_size=1,
|
||||
pool_max_size=5,
|
||||
),
|
||||
project_id=project_id,
|
||||
current_user=_user(role="admin", is_superuser=True),
|
||||
metadata_repo=repo,
|
||||
)
|
||||
|
||||
assert response.has_dsn is True
|
||||
assert "dsn" not in response.model_dump()
|
||||
admin_metadata._check_database_connection.assert_awaited_once()
|
||||
repo.upsert_project_database_config.assert_awaited_once()
|
||||
assert repo.upsert_project_database_config.await_args.kwargs["db_type"] == "postgresql"
|
||||
request_data = admin_metadata.log_audit_event.await_args.kwargs["request_data"]
|
||||
assert request_data["dsn_updated"] is True
|
||||
assert request_data["db_type"] == "postgresql"
|
||||
assert "dsn" not in request_data
|
||||
assert "postgresql://user:secret@localhost/db" not in str(request_data)
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_upsert_project_database_rejects_unhealthy_connection(monkeypatch):
|
||||
project_id = uuid4()
|
||||
repo = SimpleNamespace(
|
||||
session=object(),
|
||||
get_project_by_id=AsyncMock(return_value=_project(id=project_id)),
|
||||
upsert_project_database_config=AsyncMock(),
|
||||
)
|
||||
monkeypatch.setattr(admin_metadata, "log_audit_event", AsyncMock())
|
||||
monkeypatch.setattr(
|
||||
admin_metadata,
|
||||
"_check_database_connection",
|
||||
AsyncMock(
|
||||
side_effect=Exception(
|
||||
'FATAL: password authentication failed for user "tjwater"'
|
||||
)
|
||||
),
|
||||
)
|
||||
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
await admin_metadata.upsert_project_database(
|
||||
ProjectDatabaseUpsertRequest(
|
||||
db_role="iot_data",
|
||||
dsn="postgresql://tjwater:bad@192.168.1.114:5433/tjwater",
|
||||
pool_min_size=1,
|
||||
pool_max_size=5,
|
||||
),
|
||||
project_id=project_id,
|
||||
current_user=_user(role="admin", is_superuser=True),
|
||||
metadata_repo=repo,
|
||||
)
|
||||
|
||||
assert exc.value.status_code == 400
|
||||
assert exc.value.detail == "连通性测试失败：用户名或密码错误，请检查 DSN 中的账号密码。"
|
||||
repo.upsert_project_database_config.assert_not_called()
|
||||
admin_metadata.log_audit_event.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_project_database_health_returns_ok(monkeypatch):
|
||||
project_id = uuid4()
|
||||
repo = SimpleNamespace(
|
||||
get_project_db_routing=AsyncMock(
|
||||
return_value=ProjectDbRouting(
|
||||
project_id=project_id,
|
||||
db_role="biz_data",
|
||||
db_type="postgresql",
|
||||
dsn="postgresql://user:secret@localhost/db",
|
||||
pool_min_size=1,
|
||||
pool_max_size=5,
|
||||
)
|
||||
)
|
||||
)
|
||||
monkeypatch.setattr(admin_metadata, "_check_database_connection", AsyncMock())
|
||||
|
||||
response = await admin_metadata.check_project_database_health(
|
||||
project_id=project_id,
|
||||
db_role="biz_data",
|
||||
response=Response(),
|
||||
current_user=_user(role="admin", is_superuser=True),
|
||||
metadata_repo=repo,
|
||||
)
|
||||
|
||||
assert response.ok is True
|
||||
assert response.detail == "连通性测试通过"
|
||||
admin_metadata._check_database_connection.assert_awaited_once()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_project_database_health_can_test_unsaved_plaintext_dsn(monkeypatch):
|
||||
project_id = uuid4()
|
||||
repo = SimpleNamespace(
|
||||
get_project_db_routing=AsyncMock(),
|
||||
)
|
||||
monkeypatch.setattr(admin_metadata, "_check_database_connection", AsyncMock())
|
||||
|
||||
response = await admin_metadata.check_project_database_health(
|
||||
project_id=project_id,
|
||||
db_role="iot_data",
|
||||
payload=admin_metadata.ProjectDatabaseHealthRequest(
|
||||
dsn="postgresql://tjwater:secret@192.168.1.114:5433/tjwater"
|
||||
),
|
||||
response=Response(),
|
||||
current_user=_user(role="admin", is_superuser=True),
|
||||
metadata_repo=repo,
|
||||
)
|
||||
|
||||
assert response.ok is True
|
||||
assert response.db_type == "timescaledb"
|
||||
routing = admin_metadata._check_database_connection.await_args.args[0]
|
||||
assert routing.dsn == "postgresql://tjwater:secret@192.168.1.114:5433/tjwater"
|
||||
repo.get_project_db_routing.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_project_database_health_sanitizes_password_failures(monkeypatch):
|
||||
project_id = uuid4()
|
||||
repo = SimpleNamespace(
|
||||
get_project_db_routing=AsyncMock(
|
||||
return_value=ProjectDbRouting(
|
||||
project_id=project_id,
|
||||
db_role="iot_data",
|
||||
db_type="timescaledb",
|
||||
dsn="postgresql://tjwater:bad-password@192.168.1.114:5433/db",
|
||||
pool_min_size=1,
|
||||
pool_max_size=5,
|
||||
)
|
||||
)
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
admin_metadata,
|
||||
"_check_database_connection",
|
||||
AsyncMock(
|
||||
side_effect=Exception(
|
||||
'(psycopg.OperationalError) connection failed: FATAL: '
|
||||
'password authentication failed for user "tjwater"'
|
||||
)
|
||||
),
|
||||
)
|
||||
|
||||
fastapi_response = Response()
|
||||
response = await admin_metadata.check_project_database_health(
|
||||
project_id=project_id,
|
||||
db_role="iot_data",
|
||||
response=fastapi_response,
|
||||
current_user=_user(role="admin", is_superuser=True),
|
||||
metadata_repo=repo,
|
||||
)
|
||||
|
||||
assert fastapi_response.status_code == 503
|
||||
assert response.ok is False
|
||||
assert response.db_type == "timescaledb"
|
||||
assert response.detail == "连通性测试失败：用户名或密码错误，请检查 DSN 中的账号密码。"
|
||||
assert "psycopg" not in response.detail
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_upsert_geoserver_config_hides_password_and_audits_without_plaintext(monkeypatch):
|
||||
project_id = uuid4()
|
||||
record = _geoserver_config(project_id=project_id)
|
||||
repo = SimpleNamespace(
|
||||
session=object(),
|
||||
get_project_by_id=AsyncMock(return_value=_project(id=project_id)),
|
||||
upsert_geoserver_config=AsyncMock(return_value=record),
|
||||
)
|
||||
monkeypatch.setattr(admin_metadata, "log_audit_event", AsyncMock())
|
||||
|
||||
response = await admin_metadata.upsert_project_geoserver_config(
|
||||
ProjectGeoServerConfigUpsertRequest(
|
||||
gs_base_url="http://geoserver",
|
||||
gs_admin_user="admin",
|
||||
gs_admin_password="secret-password",
|
||||
gs_datastore_name="ds_postgis",
|
||||
default_extent={"bbox": [1, 2, 3, 4]},
|
||||
srid=4326,
|
||||
),
|
||||
project_id=project_id,
|
||||
current_user=_user(role="admin", is_superuser=True),
|
||||
metadata_repo=repo,
|
||||
)
|
||||
|
||||
assert response.has_password is True
|
||||
assert "password" not in response.model_dump()
|
||||
request_data = admin_metadata.log_audit_event.await_args.kwargs["request_data"]
|
||||
assert request_data["password_updated"] is True
|
||||
assert "secret-password" not in str(request_data)
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_get_geoserver_config_returns_empty_state_when_unconfigured():
|
||||
project_id = uuid4()
|
||||
repo = SimpleNamespace(
|
||||
get_project_by_id=AsyncMock(return_value=_project(id=project_id)),
|
||||
get_geoserver_config_record=AsyncMock(return_value=None),
|
||||
)
|
||||
|
||||
response = await admin_metadata.get_project_geoserver_config(
|
||||
project_id=project_id,
|
||||
current_user=_user(role="admin", is_superuser=True),
|
||||
metadata_repo=repo,
|
||||
)
|
||||
|
||||
assert response.project_id == project_id
|
||||
assert response.configured is False
|
||||
assert response.has_password is False
|
||||
assert response.gs_datastore_name == "ds_postgis"
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_metadata_admin_dependency_rejects_non_admin_user():
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
await get_current_metadata_admin(_user(role="user", is_superuser=False))
|
||||
|
||||
assert exc.value.status_code == 403
|
||||
assert exc.value.detail == "Admin access required"
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_add_project_member_rejects_duplicate(monkeypatch):
|
||||
project_id = uuid4()
|
||||
user_id = uuid4()
|
||||
repo = SimpleNamespace(
|
||||
session=object(),
|
||||
get_project_by_id=AsyncMock(return_value=_project(id=project_id)),
|
||||
get_user_by_id=AsyncMock(return_value=_user(id=user_id)),
|
||||
get_project_membership=AsyncMock(return_value=_membership()),
|
||||
add_project_member=AsyncMock(),
|
||||
)
|
||||
monkeypatch.setattr(admin_metadata, "log_audit_event", AsyncMock())
|
||||
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
await admin_metadata.add_project_member(
|
||||
ProjectMemberCreateRequest(user_id=user_id, project_role="viewer"),
|
||||
project_id=project_id,
|
||||
current_user=_user(role="admin", is_superuser=True),
|
||||
metadata_repo=repo,
|
||||
)
|
||||
|
||||
assert exc.value.status_code == 409
|
||||
repo.add_project_member.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_add_project_member_rejects_self_membership_change(monkeypatch):
|
||||
project_id = uuid4()
|
||||
current_user = _user(role="admin", is_superuser=True)
|
||||
repo = SimpleNamespace(
|
||||
session=object(),
|
||||
get_project_by_id=AsyncMock(),
|
||||
add_project_member=AsyncMock(),
|
||||
)
|
||||
monkeypatch.setattr(admin_metadata, "log_audit_event", AsyncMock())
|
||||
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
await admin_metadata.add_project_member(
|
||||
ProjectMemberCreateRequest(
|
||||
user_id=current_user.id,
|
||||
project_role="viewer",
|
||||
),
|
||||
project_id=project_id,
|
||||
current_user=current_user,
|
||||
metadata_repo=repo,
|
||||
)
|
||||
|
||||
assert exc.value.status_code == 403
|
||||
repo.get_project_by_id.assert_not_called()
|
||||
repo.add_project_member.assert_not_called()
|
||||
admin_metadata.log_audit_event.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_update_project_member_role_audits_change(monkeypatch):
|
||||
project_id = uuid4()
|
||||
user_id = uuid4()
|
||||
user = _user(id=user_id, username="bob", email="bob@example.com")
|
||||
membership = _membership(
|
||||
user_id=user_id,
|
||||
project_id=project_id,
|
||||
project_role="admin",
|
||||
)
|
||||
repo = SimpleNamespace(
|
||||
session=object(),
|
||||
get_user_by_id=AsyncMock(return_value=user),
|
||||
update_project_member_role=AsyncMock(return_value=membership),
|
||||
)
|
||||
monkeypatch.setattr(admin_metadata, "log_audit_event", AsyncMock())
|
||||
|
||||
response = await admin_metadata.update_project_member(
|
||||
ProjectMemberUpdateRequest(project_role="admin"),
|
||||
project_id=project_id,
|
||||
user_id=user_id,
|
||||
current_user=_user(role="admin", is_superuser=True),
|
||||
metadata_repo=repo,
|
||||
)
|
||||
|
||||
assert response.project_role == "admin"
|
||||
repo.update_project_member_role.assert_awaited_once_with(
|
||||
project_id, user_id, "admin"
|
||||
)
|
||||
admin_metadata.log_audit_event.assert_awaited_once()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_update_project_member_rejects_self_membership_change(monkeypatch):
|
||||
project_id = uuid4()
|
||||
current_user = _user(role="admin", is_superuser=True)
|
||||
repo = SimpleNamespace(
|
||||
session=object(),
|
||||
get_user_by_id=AsyncMock(),
|
||||
update_project_member_role=AsyncMock(),
|
||||
)
|
||||
monkeypatch.setattr(admin_metadata, "log_audit_event", AsyncMock())
|
||||
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
await admin_metadata.update_project_member(
|
||||
ProjectMemberUpdateRequest(project_role="admin"),
|
||||
project_id=project_id,
|
||||
user_id=current_user.id,
|
||||
current_user=current_user,
|
||||
metadata_repo=repo,
|
||||
)
|
||||
|
||||
assert exc.value.status_code == 403
|
||||
repo.get_user_by_id.assert_not_called()
|
||||
repo.update_project_member_role.assert_not_called()
|
||||
admin_metadata.log_audit_event.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_remove_project_member_audits_change(monkeypatch):
|
||||
project_id = uuid4()
|
||||
user_id = uuid4()
|
||||
repo = SimpleNamespace(
|
||||
session=object(),
|
||||
remove_project_member=AsyncMock(return_value=True),
|
||||
)
|
||||
monkeypatch.setattr(admin_metadata, "log_audit_event", AsyncMock())
|
||||
|
||||
response = await admin_metadata.remove_project_member(
|
||||
project_id=project_id,
|
||||
user_id=user_id,
|
||||
current_user=_user(role="admin", is_superuser=True),
|
||||
metadata_repo=repo,
|
||||
)
|
||||
|
||||
assert response is None
|
||||
repo.remove_project_member.assert_awaited_once_with(project_id, user_id)
|
||||
admin_metadata.log_audit_event.assert_awaited_once()
|
||||
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_remove_project_member_rejects_self_membership_change(monkeypatch):
|
||||
project_id = uuid4()
|
||||
current_user = _user(role="admin", is_superuser=True)
|
||||
repo = SimpleNamespace(
|
||||
session=object(),
|
||||
remove_project_member=AsyncMock(),
|
||||
)
|
||||
monkeypatch.setattr(admin_metadata, "log_audit_event", AsyncMock())
|
||||
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
await admin_metadata.remove_project_member(
|
||||
project_id=project_id,
|
||||
user_id=current_user.id,
|
||||
current_user=current_user,
|
||||
metadata_repo=repo,
|
||||
)
|
||||
|
||||
assert exc.value.status_code == 403
|
||||
repo.remove_project_member.assert_not_called()
|
||||
admin_metadata.log_audit_event.assert_not_called()
|
||||
Reference in New Issue
Block a user