from datetime import date, timedelta import pytest from fastapi import HTTPException from sqlalchemy import create_engine, func, select from sqlalchemy.orm import sessionmaker from app.application.events import EventDispatchService from app.core.database import Base from app.core.utils.time import utc_now from app.modules.ai_memory.constants import ( AIMemoryPayloadKey, AIMemoryStatus, ) from app.modules.ai_memory.models import AIMemoryEntry from app.modules.ai_memory.service import AIMemoryService from app.modules.business.constants import StatusValue from app.modules.business.models import FundAccount, PerformanceMetric, Project from app.modules.events.constants import ( EventAggregateType, EventSource, EventStatus, EventType, ) from app.modules.events.models import DomainEvent from app.modules.events.services import EventService from app.modules.reports.constants import EnterpriseAnalyticsKey, MetricKey from app.modules.reports.services import ReportService from app.modules.workflows.constants import WorkflowStatus, WorkflowType from app.modules.workflows.models import WorkflowAction, WorkflowInstance def _session_factory(): engine = create_engine("sqlite://") Base.metadata.create_all(engine) return engine, sessionmaker(bind=engine, expire_on_commit=False) def test_workflow_actions_are_idempotent_per_source_event() -> None: engine, factory = _session_factory() try: with factory() as db: event = EventService(db).emit( event_type=EventType.ENTERPRISE_ANALYTICS_GENERATED, source=EventSource.ANALYTICS, aggregate_type=EventAggregateType.ENTERPRISE_ANALYTICS, aggregate_id="snapshot-1", idempotency_key="analytics:snapshot-1", ) first = EventDispatchService(db).dispatch_event( event.event_id, worker_id="pytest-first", ) assert first.status == EventStatus.PROCESSED first.status = EventStatus.PENDING first.processed_at = None first.next_attempt_at = utc_now() db.commit() second = EventDispatchService(db).dispatch_event( event.event_id, worker_id="pytest-second", ) assert second.status == EventStatus.PROCESSED assert db.scalar(select(func.count()).select_from(WorkflowInstance)) == 1 assert db.scalar(select(func.count()).select_from(WorkflowAction)) == 1 action = db.execute(select(WorkflowAction)).scalar_one() assert action.source_event_id == event.event_id finally: engine.dispose() def test_ai_memory_has_no_unrelated_fallback_and_archives_expired_entries() -> None: engine, factory = _session_factory() try: with factory() as db: service = AIMemoryService(db) first = service.auto_write( prompt="Remember concise operational summaries", context={ AIMemoryPayloadKey.SCOPE: "project", AIMemoryPayloadKey.SUBJECT: "P-MEM-IDEMPOTENT", }, answer="Use concise bullet summaries for project updates.", ) duplicate = service.auto_write( prompt="Remember concise operational summaries", context={ AIMemoryPayloadKey.SCOPE: "project", AIMemoryPayloadKey.SUBJECT: "P-MEM-IDEMPOTENT", }, answer="Use concise bullet summaries for project updates.", ) assert first is not None assert duplicate is not None assert duplicate.code == first.code assert db.scalar(select(func.count()).select_from(AIMemoryEntry)) == 1 assert ( service.recall( query="no-such-memory-token", scope="project", subject="P-MEM-IDEMPOTENT", ) == [] ) first.expires_at = utc_now() - timedelta(seconds=1) db.commit() assert ( service.list_entries( scope="project", subject="P-MEM-IDEMPOTENT", status_filter=AIMemoryStatus.ACTIVE, ) == [] ) db.refresh(first) assert first.status == AIMemoryStatus.ARCHIVED finally: engine.dispose() def test_enterprise_analytics_validates_scope_and_reuses_identical_snapshot() -> None: engine, factory = _session_factory() try: with factory() as db: db.add_all( [ Project( code="P-SCOPED", name="Scoped project", owner="owner-a", status=StatusValue.RUNNING_CN, ), PerformanceMetric( code="PERF-GLOBAL", name="Global metric", weight=10, auto_score=80, confirmed_score=75, status=StatusValue.REVIEWED, ), FundAccount( code="FUND-GLOBAL", name="Global account", current_balance=1000, ), ] ) db.commit() service = ReportService(db) with pytest.raises(HTTPException) as exc_info: service.enterprise_analytics( period_start=date(2030, 2, 2), period_end=date(2030, 2, 1), ) assert exc_info.value.status_code == 422 first = service.enterprise_analytics(project_code="P-SCOPED") second = service.enterprise_analytics(project_code="P-SCOPED") assert first[EnterpriseAnalyticsKey.CODE] == second[EnterpriseAnalyticsKey.CODE] assert first[EnterpriseAnalyticsKey.PERFORMANCE][MetricKey.TOTAL] == 0 assert ( first[EnterpriseAnalyticsKey.FINANCE][ MetricKey.CURRENT_BALANCE_TOTAL ] == 0 ) events = list( db.execute( select(DomainEvent).where( DomainEvent.event_type == EventType.ENTERPRISE_ANALYTICS_GENERATED ) ).scalars() ) assert len(events) == 1 assert events[0].aggregate_id == first[EnterpriseAnalyticsKey.CODE] dispatched = EventDispatchService(db).dispatch_event( events[0].event_id, worker_id="pytest-enterprise", ) assert dispatched.status == EventStatus.PROCESSED workflow = db.execute( select(WorkflowInstance).where( WorkflowInstance.workflow_type == WorkflowType.ENTERPRISE_ANALYTICS ) ).scalar_one() assert workflow.status == WorkflowStatus.COMPLETED assert workflow.aggregate_id == first[EnterpriseAnalyticsKey.CODE] finally: engine.dispose()