from datetime import timedelta from pathlib import Path import pytest from fastapi import HTTPException from sqlalchemy import UniqueConstraint, create_engine, func, select, text from sqlalchemy.orm import Session, sessionmaker from app.application.events import EventDispatchService from app.application.scheduling import create_scheduler from app.core.config import get_settings from app.core.database import Base from app.core.utils.time import utc_now from app.main import app from app.modules.ai_agent.constants import AIContextKey from app.modules.ai_agent.service import AIService from app.modules.audit.constants import AuditSource from app.modules.events.constants import EventStatus from app.modules.events.models import DomainEvent from app.modules.events.services import EventService from app.modules.legacy_mysql.services import LegacyMySQLService from app.modules.observability.constants import ( HeartbeatComponent, ObservabilityKey, ObservabilityStatus, ) from app.modules.observability.models import SystemHeartbeat from app.modules.observability.service import ObservabilityService from app.tasks import celery_app def _session_factory(): engine = create_engine("sqlite://") Base.metadata.create_all(engine) return engine, sessionmaker(bind=engine, expire_on_commit=False) def test_readiness_and_metrics_degrade_when_schema_is_missing() -> None: engine = create_engine("sqlite://") with Session(engine) as db: readiness = ObservabilityService(db).ready() metrics = ObservabilityService(db).metrics() assert readiness[ObservabilityKey.STATUS] == ObservabilityStatus.DEGRADED assert readiness[ObservabilityKey.CHECKS][ObservabilityKey.DATABASE][ ObservabilityKey.STATUS ] == ObservabilityStatus.OK assert readiness[ObservabilityKey.CHECKS][ObservabilityKey.EVENTS][ ObservabilityKey.STATUS ] == ObservabilityStatus.ERROR assert metrics[ObservabilityKey.METRICS][ObservabilityKey.EVENTS][ ObservabilityKey.STATUS ] == ObservabilityStatus.ERROR @pytest.mark.parametrize( "sql", [ "SELECT 1; DELETE FROM projects", "SELECT * FROM projects /* hidden operation */", "SELECT * FROM projects -- hidden operation", "SELECT * FROM projects FOR UPDATE", "SELECT * INTO OUTFILE '/tmp/data' FROM projects", ], ) def test_legacy_query_rejects_unsafe_allowlist_sql(sql: str) -> None: with pytest.raises(HTTPException) as exc_info: LegacyMySQLService(None)._ensure_readonly(sql) assert exc_info.value.status_code == 400 def test_legacy_query_accepts_parameterized_select() -> None: LegacyMySQLService(None)._ensure_readonly( "SELECT code, name FROM projects WHERE owner = :owner LIMIT :limit" ) def test_celery_and_scheduler_have_safe_runtime_defaults() -> None: assert celery_app.conf.task_serializer == "json" assert celery_app.conf.result_serializer == "json" assert tuple(celery_app.conf.accept_content) == ("json",) assert celery_app.conf.task_acks_late is True assert celery_app.conf.worker_prefetch_multiplier == 1 scheduler = create_scheduler() assert scheduler._job_defaults["coalesce"] is True assert scheduler._job_defaults["max_instances"] == 1 def test_unknown_event_type_fails_instead_of_being_discarded() -> None: engine, factory = _session_factory() try: with factory() as db: event = EventService(db).emit( event_type="unknown.event", source="pytest", aggregate_type="test", aggregate_id="unknown", idempotency_key="unknown-event", ) event.max_attempts = 1 db.commit() result = EventDispatchService(db).dispatch_event( event.event_id, worker_id="pytest", ) assert result.status == EventStatus.FAILED assert "Unsupported domain event type" in (result.last_error or "") finally: engine.dispose() def test_event_failure_rolls_back_handler_transaction( monkeypatch: pytest.MonkeyPatch, ) -> None: engine, factory = _session_factory() try: with factory() as db: event = EventService(db).emit( event_type="test.database_failure", source="pytest", aggregate_type="test", aggregate_id="database-failure", idempotency_key="database-failure", ) event.max_attempts = 1 db.commit() dispatcher = EventDispatchService(db) def fail_with_database_error(record: DomainEvent) -> None: _ = record db.execute(text("SELECT * FROM missing_handler_table")).all() monkeypatch.setattr(dispatcher, "_handle_event", fail_with_database_error) result = dispatcher.dispatch_event(event.event_id, worker_id="pytest") assert result.status == EventStatus.FAILED assert result.last_error finally: engine.dispose() def test_retry_rejects_an_active_event_lease() -> None: engine, factory = _session_factory() try: with factory() as db: event = EventService(db).emit( event_type="test.locked", source="pytest", aggregate_type="test", aggregate_id="locked", idempotency_key="locked-event", ) event.locked_by = "active-worker" event.locked_until = utc_now() + timedelta(minutes=5) db.commit() with pytest.raises(HTTPException) as exc_info: EventDispatchService(db).retry_event(event.event_id) assert exc_info.value.status_code == 409 finally: engine.dispose() def test_heartbeat_identity_is_unique_and_expired_instances_are_ignored( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setenv("HEARTBEAT_RETENTION_SECONDS", "3600") get_settings.cache_clear() engine, factory = _session_factory() try: constraints = SystemHeartbeat.__table__.constraints assert any( isinstance(item, UniqueConstraint) and {column.name for column in item.columns} == {"component", "instance_id"} for item in constraints ) with factory() as db: service = ObservabilityService(db) service.record_heartbeat( component=HeartbeatComponent.WORKER, instance_id="worker-current", ) service.record_heartbeat( component=HeartbeatComponent.WORKER, instance_id="worker-current", ) db.add( SystemHeartbeat( component=HeartbeatComponent.WORKER, instance_id="worker-expired", status="ok", last_seen_at=utc_now() - timedelta(hours=2), ) ) db.commit() count = db.scalar(select(func.count()).select_from(SystemHeartbeat)) summary = service.heartbeat_summary() assert count == 2 assert summary["total"] == 1 assert summary["stale"] == 0 finally: get_settings.cache_clear() engine.dispose() @pytest.mark.parametrize( ("field", "value"), [ (AIContextKey.OPENCLAW_TOOL, "sessions_list"), (AIContextKey.OPENCLAW_ACTION, "json"), (AIContextKey.OPENCLAW_ARGS, {"limit": 1}), (AIContextKey.OPENCLAW_SESSION_KEY, "main"), ], ) def test_external_ai_cannot_invoke_openclaw_tools( field: AIContextKey, value: object, ) -> None: with pytest.raises(HTTPException) as exc_info: AIService(Session()).ask( "run a tool", context={field: value}, source=AuditSource.API, ) assert exc_info.value.status_code == 403 assert "/api/v1/ai/openclaw/tools/invoke" not in app.openapi()["paths"] def test_sample_requests_are_read_only() -> None: sample = Path("scripts/sample_requests.http").read_text(encoding="utf-8") assert "POST http://127.0.0.1:8010/api/v1/business/" not in sample assert "/sync" not in sample assert "/assign" not in sample