import uuid from datetime import date, datetime, timedelta, timezone from types import SimpleNamespace from unittest.mock import AsyncMock, Mock import pytest from app.api.v1 import visits as visits_api from app.models.notification import Notification from app.models.desktop_notification import DesktopNotificationDelivery from app.models.collaboration import CollaborationEditRequest from app.services import collaboration_service, desktop_notification_service, notification_service, project_reminder_service def test_generic_notification_table_has_recipient_dedupe_and_state_indexes(): table = Notification.__table__ assert table.name == "notifications" assert { "study_id", "recipient_id", "category", "action_path", "source_type", "source_id", "requires_action", "due_at", "read_at", "resolved_at", } <= set(table.columns.keys()) assert any(constraint.name == "uq_notifications_recipient_dedupe" for constraint in table.constraints) assert {index.name for index in table.indexes} >= { "ix_notifications_recipient_study_state", "ix_notifications_source", } def test_desktop_delivery_targets_the_generic_notification_feed(): table = DesktopNotificationDelivery.__table__ assert table.columns.notification_id.nullable is True assert any( constraint.name == "uq_desktop_notification_user_notification" for constraint in table.constraints ) statement = str(desktop_notification_service._eligible_query( uuid.uuid4(), datetime.now(timezone.utc), datetime.now(timezone.utc), )) assert "notifications" in statement assert "notification_id" in statement assert "resolved_at" in statement @pytest.mark.asyncio async def test_recipient_notification_creation_is_deduplicated_per_recipient(): existing_id = uuid.uuid4() new_id = uuid.uuid4() result = SimpleNamespace(all=lambda: [existing_id]) db = SimpleNamespace(scalars=AsyncMock(return_value=result), add=Mock()) await notification_service.create_recipient_notifications( db, study_id=uuid.uuid4(), recipient_ids=[existing_id, new_id, new_id], category="COLLABORATION_EDIT_REQUEST", priority="NORMAL", title="新的编辑权限申请", message="申请编辑文件", action_path="/knowledge/collaboration?editRequestFile=file-id", source_type="COLLABORATION_EDIT_REQUEST", source_id="request-id", dedupe_key="collaboration-edit-request:request-id", ) db.add.assert_called_once() created = db.add.call_args.args[0] assert created.recipient_id == new_id assert created.source_id == "request-id" @pytest.mark.asyncio async def test_resolving_a_source_closes_every_recipient_notification(): db = SimpleNamespace(execute=AsyncMock()) await notification_service.resolve_source_notifications( db, source_type="COLLABORATION_EDIT_REQUEST", source_id="request-id", ) db.execute.assert_awaited_once() statement = str(db.execute.await_args.args[0]) assert "UPDATE notifications" in statement assert "source_type" in statement assert "source_id" in statement @pytest.mark.asyncio async def test_project_reminder_sync_runs_every_supported_rule_group(monkeypatch): study_id = uuid.uuid4() user = SimpleNamespace(id=uuid.uuid4(), is_admin=False) membership = SimpleNamespace(is_active=True, role_in_study="PM") db = SimpleNamespace(commit=AsyncMock()) risk_sync = AsyncMock() distribution_sync = AsyncMock() milestone_sync = AsyncMock() visit_sync = AsyncMock() collaboration_sync = AsyncMock() monkeypatch.setattr(project_reminder_service.member_crud, "get_member", AsyncMock(return_value=membership)) monkeypatch.setattr(project_reminder_service, "_sync_risk_reminders", risk_sync) monkeypatch.setattr(project_reminder_service, "_sync_document_distribution_reminders", distribution_sync) monkeypatch.setattr(project_reminder_service, "_sync_milestone_reminders", milestone_sync) monkeypatch.setattr(project_reminder_service, "_sync_visit_window_reminders", visit_sync) monkeypatch.setattr(project_reminder_service, "_sync_collaboration_edit_request_reminders", collaboration_sync) await project_reminder_service.sync_project_reminders(db, study_id, user) risk_sync.assert_awaited_once() distribution_sync.assert_awaited_once() milestone_sync.assert_awaited_once() visit_sync.assert_awaited_once() collaboration_sync.assert_awaited_once() assert collaboration_sync.await_args.kwargs["role"] == "PM" db.commit.assert_awaited_once() @pytest.mark.asyncio async def test_inactive_membership_resolves_stale_project_notifications(monkeypatch): db = SimpleNamespace(commit=AsyncMock()) resolve = AsyncMock() user = SimpleNamespace(id=uuid.uuid4(), is_admin=True) study_id = uuid.uuid4() monkeypatch.setattr(project_reminder_service.member_crud, "get_member", AsyncMock(return_value=None)) monkeypatch.setattr( project_reminder_service.notification_service, "resolve_recipient_study_notifications", resolve, ) await project_reminder_service.sync_project_reminders(db, study_id, user) resolve.assert_awaited_once_with(db, study_id=study_id, recipient_id=user.id) db.commit.assert_awaited_once() @pytest.mark.asyncio async def test_project_risks_include_due_soon_and_overdue_aggregates(monkeypatch): study_id = uuid.uuid4() user = SimpleNamespace(id=uuid.uuid4(), is_admin=False) now = datetime(2026, 7, 16, 2, 0, tzinfo=timezone.utc) db = SimpleNamespace() sync = AsyncMock() monkeypatch.setattr(project_reminder_service, "role_has_api_permission", AsyncMock(return_value=True)) monkeypatch.setattr(project_reminder_service, "get_cra_site_scope", AsyncMock(return_value=None)) monkeypatch.setattr( project_reminder_service, "_count_and_earliest", AsyncMock(side_effect=[ (2, date(2026, 7, 15)), (1, date(2026, 7, 18)), (3, now - timedelta(hours=2)), (4, now + timedelta(days=1)), ]), ) monkeypatch.setattr(project_reminder_service.notification_service, "sync_aggregate_notification", sync) await project_reminder_service._sync_risk_reminders( db, study_id=study_id, user=user, role="PM", current_time=now, ) assert [call.kwargs["category"] for call in sync.await_args_list] == [ "RISK_OVERDUE_AE", "RISK_AE_DUE_SOON", "RISK_OVERDUE_MONITORING", "RISK_MONITORING_DUE_SOON", ] assert [call.kwargs["count"] for call in sync.await_args_list] == [2, 1, 3, 4] @pytest.mark.asyncio async def test_visit_window_reminders_are_actionable_scoped_and_privacy_safe(monkeypatch): study_id = uuid.uuid4() user = SimpleNamespace(id=uuid.uuid4(), is_admin=False) site_id = uuid.uuid4() due_visit_id = uuid.uuid4() missed_visit_id = uuid.uuid4() due_subject_id = uuid.uuid4() missed_subject_id = uuid.uuid4() due_visit = SimpleNamespace( id=due_visit_id, planned_date=date(2026, 7, 19), window_start=date(2026, 7, 18), window_end=date(2026, 7, 20), ) missed_visit = SimpleNamespace( id=missed_visit_id, planned_date=date(2026, 7, 14), window_start=date(2026, 7, 13), window_end=date(2026, 7, 15), ) due_subject = SimpleNamespace(id=due_subject_id, subject_no="S-PRIVACY-001") missed_subject = SimpleNamespace(id=missed_subject_id, subject_no="S-PRIVACY-002") rows = SimpleNamespace(all=lambda: [ (due_visit, due_subject), (missed_visit, missed_subject), ]) db = SimpleNamespace(execute=AsyncMock(return_value=rows)) sync = AsyncMock() resolve = AsyncMock() monkeypatch.setattr(project_reminder_service, "role_has_api_permission", AsyncMock(return_value=True)) monkeypatch.setattr( project_reminder_service, "get_cra_site_scope", AsyncMock(return_value=({site_id}, {"研究中心"})), ) monkeypatch.setattr(project_reminder_service.notification_service, "sync_state_notification", sync) monkeypatch.setattr( project_reminder_service.notification_service, "resolve_missing_recipient_sources", resolve, ) await project_reminder_service._sync_visit_window_reminders( db, study_id=study_id, user=user, role="CRA", current_time=datetime(2026, 7, 16, 2, 0, tzinfo=timezone.utc), ) assert [call.kwargs["category"] for call in sync.await_args_list] == [ "VISIT_WINDOW_DUE_SOON", "VISIT_WINDOW_MISSED", ] assert [call.kwargs["priority"] for call in sync.await_args_list] == ["NORMAL", "HIGH"] assert sync.await_args_list[0].kwargs["action_path"] == f"/subjects/{due_subject_id}" assert sync.await_args_list[1].kwargs["action_path"] == f"/subjects/{missed_subject_id}" messages = " ".join(call.kwargs["message"] for call in sync.await_args_list) assert "S-PRIVACY-001" not in messages assert "S-PRIVACY-002" not in messages resolve.assert_awaited_once_with( db, study_id=study_id, recipient_id=user.id, source_type="SUBJECT_VISIT_WINDOW", active_source_ids={str(due_visit_id), str(missed_visit_id)}, ) @pytest.mark.asyncio async def test_visit_window_reminders_resolve_when_recipient_cannot_manage_visits(monkeypatch): study_id = uuid.uuid4() user = SimpleNamespace(id=uuid.uuid4(), is_admin=False) db = SimpleNamespace(execute=AsyncMock()) resolve = AsyncMock() monkeypatch.setattr(project_reminder_service, "role_has_api_permission", AsyncMock(return_value=False)) monkeypatch.setattr( project_reminder_service.notification_service, "resolve_missing_recipient_sources", resolve, ) await project_reminder_service._sync_visit_window_reminders( db, study_id=study_id, user=user, role="QA", current_time=datetime(2026, 7, 16, 2, 0, tzinfo=timezone.utc), ) db.execute.assert_not_awaited() resolve.assert_awaited_once_with( db, study_id=study_id, recipient_id=user.id, source_type="SUBJECT_VISIT_WINDOW", active_source_ids=set(), ) @pytest.mark.asyncio async def test_completed_visit_closes_every_recipient_reminder_immediately(monkeypatch): first_id = uuid.uuid4() second_id = uuid.uuid4() db = SimpleNamespace(commit=AsyncMock()) resolve = AsyncMock() monkeypatch.setattr(visits_api.notification_service, "resolve_source_notifications", resolve) await visits_api._resolve_visit_reminders(db, [first_id, second_id, first_id]) assert resolve.await_count == 2 assert {call.kwargs["source_id"] for call in resolve.await_args_list} == { str(first_id), str(second_id), } assert all( call.kwargs["source_type"] == "SUBJECT_VISIT_WINDOW" for call in resolve.await_args_list ) db.commit.assert_awaited_once() @pytest.mark.asyncio async def test_reading_an_informational_notification_archives_it(): item = SimpleNamespace( id=uuid.uuid4(), read_at=None, resolved_at=None, requires_action=False, ) db = SimpleNamespace( scalar=AsyncMock(return_value=item), commit=AsyncMock(), refresh=AsyncMock(), ) result = await notification_service.mark_read( db, study_id=uuid.uuid4(), recipient_id=uuid.uuid4(), notification_id=item.id, ) assert result.read_at is not None assert result.resolved_at == result.read_at db.commit.assert_awaited_once() @pytest.mark.asyncio async def test_edit_request_notifies_the_active_file_owner_and_managers(monkeypatch): study_id = uuid.uuid4() file_id = uuid.uuid4() owner_id = uuid.uuid4() manager_id = uuid.uuid4() request_id = uuid.uuid4() item = SimpleNamespace( id=file_id, study_id=study_id, owner_id=owner_id, title="PK_示例.xlsx", allow_edit_request=True, ) user = SimpleNamespace(id=uuid.uuid4(), full_name="周成成", email="member@example.com") added = [] def add(value): added.append(value) async def flush(): request = next(value for value in added if isinstance(value, CollaborationEditRequest)) request.id = request_id request.status = "PENDING" request.resolved_by = None request.resolved_at = None request.created_at = datetime.now(timezone.utc) db = SimpleNamespace( scalar=AsyncMock(return_value=None), scalars=AsyncMock(return_value=SimpleNamespace(all=lambda: [owner_id, manager_id])), add=Mock(side_effect=add), flush=AsyncMock(side_effect=flush), commit=AsyncMock(), refresh=AsyncMock(), ) notify = AsyncMock() monkeypatch.setattr(collaboration_service, "can_edit_file", AsyncMock(return_value=False)) monkeypatch.setattr( collaboration_service.member_crud, "get_member", AsyncMock(return_value=SimpleNamespace(is_active=True)), ) monkeypatch.setattr(collaboration_service.notification_service, "create_recipient_notifications", notify) result = await collaboration_service.create_edit_request(db, item, user) assert result.id == request_id assert set(notify.await_args.kwargs["recipient_ids"]) == {owner_id, manager_id} assert notify.await_args.kwargs["action_path"] == f"/knowledge/collaboration?editRequestFile={file_id}" assert notify.await_args.kwargs["source_id"] == str(request_id)