| """ |
| Database integration tests for workflow analytics and debugger. |
| |
| Tests cover database-heavy code paths that unit tests with mocks cannot reach: |
| - AgentExecution database operations (create, update, query, aggregate) |
| - WorkflowExecution database operations and lifecycle |
| - WorkflowDebugSession database operations |
| - Database persistence and retrieval |
| - Database aggregation and analytics queries |
| |
| Uses transaction rollback pattern for test isolation. |
| """ |
|
|
| import pytest |
| from datetime import datetime, timedelta |
| from pathlib import Path |
| from sqlalchemy.orm import Session |
| from sqlalchemy import func, and_ |
| from unittest.mock import MagicMock, patch |
|
|
| import sys |
| from pathlib import Path as SysPath |
| sys.path.insert(0, str(SysPath(__file__).parent.parent.parent)) |
|
|
| from core.models import ( |
| AgentRegistry, |
| AgentExecution, |
| WorkflowExecution, |
| WorkflowDebugSession, |
| ExecutionTrace, |
| WorkflowExecutionLog, |
| User, |
| UserRole, |
| WorkflowExecutionStatus |
| ) |
| from core.workflow_debugger import WorkflowDebugger |
| from tests.factories.agent_factory import AgentFactory |
| from tests.factories.user_factory import UserFactory |
| from tests.factories.execution_factory import AgentExecutionFactory |
|
|
|
|
| class TestWorkflowDebuggerIntegration: |
| """Test WorkflowDebugger database operations.""" |
|
|
| def test_create_debug_session(self, db_session: Session): |
| """Test creating debug session in database.""" |
| debugger = WorkflowDebugger(db=db_session) |
|
|
| |
| user = UserFactory(email="debugger@test.com", _session=db_session) |
| db_session.commit() |
|
|
| |
| session = debugger.create_debug_session( |
| workflow_id="workflow-debug-1", |
| user_id=user.id, |
| session_name="Test Debug Session" |
| ) |
|
|
| assert session.id is not None, "Session should have ID" |
| assert session.workflow_id == "workflow-debug-1" |
| assert session.user_id == user.id |
| assert session.status == "active" |
| assert session.session_name == "Test Debug Session" |
|
|
| |
| retrieved = db_session.query(WorkflowDebugSession).filter( |
| WorkflowDebugSession.id == session.id |
| ).first() |
| assert retrieved is not None, "Session should be persisted" |
|
|
| def test_query_debug_history(self, db_session: Session): |
| """Test querying debug session history from database.""" |
| debugger = WorkflowDebugger(db=db_session) |
|
|
| user = UserFactory(email="history@test.com", _session=db_session) |
| db_session.commit() |
|
|
| |
| session1 = debugger.create_debug_session( |
| workflow_id="workflow-history", |
| user_id=user.id, |
| session_name="Session 1" |
| ) |
| db_session.commit() |
|
|
| session2 = debugger.create_debug_session( |
| workflow_id="workflow-history", |
| user_id=user.id, |
| session_name="Session 2" |
| ) |
| db_session.commit() |
|
|
| |
| sessions = debugger.get_active_debug_sessions( |
| workflow_id="workflow-history", |
| user_id=user.id |
| ) |
|
|
| assert len(sessions) >= 2, "Should find at least 2 sessions" |
| assert any(s.session_name == "Session 1" for s in sessions) |
| assert any(s.session_name == "Session 2" for s in sessions) |
|
|
| def test_pause_and_resume_session(self, db_session: Session): |
| """Test pausing and resuming debug session.""" |
| debugger = WorkflowDebugger(db=db_session) |
|
|
| user = UserFactory(email="pause-resume@test.com", _session=db_session) |
| db_session.commit() |
|
|
| session = debugger.create_debug_session( |
| workflow_id="workflow-pause", |
| user_id=user.id |
| ) |
| db_session.commit() |
|
|
| |
| result = debugger.pause_debug_session(session.id) |
| assert result is True, "Should pause successfully" |
|
|
| retrieved = db_session.query(WorkflowDebugSession).filter( |
| WorkflowDebugSession.id == session.id |
| ).first() |
| assert retrieved.status == "paused" |
|
|
| |
| result = debugger.resume_debug_session(session.id) |
| assert result is True, "Should resume successfully" |
|
|
| retrieved = db_session.query(WorkflowDebugSession).filter( |
| WorkflowDebugSession.id == session.id |
| ).first() |
| assert retrieved.status == "active" |
|
|
| def test_add_breakpoint(self, db_session: Session): |
| """Test adding breakpoint to debug session.""" |
| debugger = WorkflowDebugger(db=db_session) |
|
|
| user = UserFactory(email="breakpoint@test.com", _session=db_session) |
| db_session.commit() |
|
|
| session = debugger.create_debug_session( |
| workflow_id="workflow-breakpoint", |
| user_id=user.id |
| ) |
| db_session.commit() |
|
|
| |
| breakpoint = debugger.add_breakpoint( |
| workflow_id="workflow-breakpoint", |
| node_id="node-123", |
| user_id=user.id, |
| debug_session_id=session.id |
| ) |
|
|
| assert breakpoint is not None, "Breakpoint should be created" |
| assert breakpoint.node_id == "node-123" |
|
|
| |
| retrieved = db_session.query(WorkflowDebugSession).filter( |
| WorkflowDebugSession.id == session.id |
| ).first() |
|
|
| assert retrieved is not None |
| |
| |
|
|
| def test_create_execution_trace(self, db_session: Session): |
| """Test recording execution traces to database.""" |
| debugger = WorkflowDebugger(db=db_session) |
|
|
| user = UserFactory(email="trace@test.com", _session=db_session) |
| agent = AgentFactory(name="TraceAgent", _session=db_session) |
| execution = AgentExecutionFactory(agent_id=agent.id, _session=db_session) |
| db_session.commit() |
|
|
| session = debugger.create_debug_session( |
| workflow_id="workflow-trace", |
| user_id=user.id, |
| execution_id=execution.id |
| ) |
| db_session.commit() |
|
|
| |
| trace = debugger.create_trace( |
| workflow_id="workflow-trace", |
| execution_id=execution.id, |
| step_number=1, |
| node_id="test-node", |
| node_type="test", |
| debug_session_id=session.id |
| ) |
|
|
| assert trace is not None, "Trace should be created" |
| |
|
|
| |
| traces = db_session.query(ExecutionTrace).filter( |
| ExecutionTrace.debug_session_id == session.id |
| ).all() |
|
|
| assert len(traces) > 0, "Should have trace record" |
|
|
| def test_get_execution_traces(self, db_session: Session): |
| """Test retrieving execution traces from database.""" |
| debugger = WorkflowDebugger(db=db_session) |
|
|
| user = UserFactory(email="get-trace@test.com", _session=db_session) |
| agent = AgentFactory(name="GetTraceAgent", _session=db_session) |
| execution = AgentExecutionFactory(agent_id=agent.id, _session=db_session) |
| db_session.commit() |
|
|
| session = debugger.create_debug_session( |
| workflow_id="workflow-get-trace", |
| user_id=user.id, |
| execution_id=execution.id |
| ) |
| db_session.commit() |
|
|
| |
| trace = debugger.create_trace( |
| workflow_id="workflow-get-trace", |
| execution_id=execution.id, |
| step_number=1, |
| node_id="test-node", |
| node_type="test", |
| debug_session_id=session.id |
| ) |
| db_session.commit() |
|
|
| |
| traces = debugger.get_execution_traces(execution_id=execution.id, debug_session_id=session.id) |
|
|
| assert len(traces) > 0, "Should find traces" |
| assert traces[0].debug_session_id == session.id |
|
|
|
|
| class TestAgentExecutionDatabaseCoverage: |
| """Test AgentExecution database-heavy operations.""" |
|
|
| def test_agent_execution_lifecycle(self, db_session: Session): |
| """Test complete lifecycle of AgentExecution record.""" |
| |
| agent = AgentFactory(name="LifecycleAgent", _session=db_session) |
| db_session.commit() |
|
|
| |
| execution = AgentExecutionFactory( |
| agent_id=agent.id, |
| status="running", |
| _session=db_session |
| ) |
| db_session.commit() |
| execution_id = execution.id |
|
|
| |
| execution.status = "completed" |
| execution.completed_at = datetime.utcnow() |
| execution.duration_seconds = 5.5 |
| execution.result_summary = "Test completed successfully" |
| db_session.commit() |
|
|
| |
| retrieved = db_session.query(AgentExecution).filter( |
| AgentExecution.id == execution_id |
| ).first() |
|
|
| assert retrieved.status == "completed" |
| assert retrieved.duration_seconds == 5.5 |
| assert retrieved.result_summary == "Test completed successfully" |
|
|
| def test_execution_with_error(self, db_session: Session): |
| """Test execution with error recording.""" |
| agent = AgentFactory(name="ErrorAgent", _session=db_session) |
| db_session.commit() |
|
|
| execution = AgentExecutionFactory( |
| agent_id=agent.id, |
| status="failed", |
| error_message="Test error: validation failed", |
| _session=db_session |
| ) |
| db_session.commit() |
|
|
| |
| failed_executions = db_session.query(AgentExecution).filter( |
| AgentExecution.status == "failed" |
| ).all() |
|
|
| assert len(failed_executions) > 0 |
| assert "validation failed" in failed_executions[0].error_message |
|
|
| def test_execution_query_by_agent(self, db_session: Session): |
| """Test querying executions by agent.""" |
| agent = AgentFactory(name="QueryAgent", _session=db_session) |
| db_session.commit() |
|
|
| |
| for i in range(5): |
| AgentExecutionFactory( |
| agent_id=agent.id, |
| status="completed", |
| _session=db_session |
| ) |
| db_session.commit() |
|
|
| |
| executions = db_session.query(AgentExecution).filter( |
| AgentExecution.agent_id == agent.id |
| ).all() |
|
|
| assert len(executions) >= 5, "Should find all executions" |
|
|
| def test_execution_duration_statistics(self, db_session: Session): |
| """Test calculating execution duration statistics.""" |
| agent = AgentFactory(name="StatsAgent", _session=db_session) |
| db_session.commit() |
|
|
| |
| durations = [1.0, 2.5, 3.0, 4.5, 5.0] |
| for duration in durations: |
| execution = AgentExecutionFactory( |
| agent_id=agent.id, |
| status="completed", |
| duration_seconds=duration, |
| _session=db_session |
| ) |
| db_session.commit() |
|
|
| |
| result = db_session.query( |
| func.avg(AgentExecution.duration_seconds) |
| ).filter( |
| AgentExecution.agent_id == agent.id, |
| AgentExecution.status == "completed" |
| ).scalar() |
|
|
| expected_avg = sum(durations) / len(durations) |
| assert abs(result - expected_avg) < 0.01, "Average should match" |
|
|
| def test_execution_with_input_output(self, db_session: Session): |
| """Test execution with input and output summaries.""" |
| agent = AgentFactory(name="IOAgent", _session=db_session) |
| db_session.commit() |
|
|
| execution = AgentExecutionFactory( |
| agent_id=agent.id, |
| input_summary="Process workflow: test-workflow", |
| output_summary="Completed 3 steps successfully", |
| _session=db_session |
| ) |
| db_session.commit() |
|
|
| retrieved = db_session.query(AgentExecution).filter( |
| AgentExecution.id == execution.id |
| ).first() |
|
|
| assert "Process workflow" in retrieved.input_summary |
| assert "Completed 3 steps" in retrieved.output_summary |
|
|
| def test_execution_triggered_by(self, db_session: Session): |
| """Test execution trigger source tracking.""" |
| agent = AgentFactory(name="TriggerAgent", _session=db_session) |
| db_session.commit() |
|
|
| |
| for trigger in ["manual", "schedule", "websocket", "event"]: |
| execution = AgentExecutionFactory( |
| agent_id=agent.id, |
| triggered_by=trigger, |
| _session=db_session |
| ) |
| db_session.commit() |
|
|
| |
| scheduled = db_session.query(AgentExecution).filter( |
| and_( |
| AgentExecution.agent_id == agent.id, |
| AgentExecution.triggered_by == "schedule" |
| ) |
| ).count() |
|
|
| assert scheduled >= 1, "Should find scheduled execution" |
|
|
| def test_execution_time_range_query(self, db_session: Session): |
| """Test querying executions by time range.""" |
| agent = AgentFactory(name="TimeRangeAgent", _session=db_session) |
| db_session.commit() |
|
|
| now = datetime.utcnow() |
|
|
| |
| execution = AgentExecutionFactory( |
| agent_id=agent.id, |
| started_at=now, |
| _session=db_session |
| ) |
| db_session.commit() |
|
|
| |
| recent = db_session.query(AgentExecution).filter( |
| and_( |
| AgentExecution.agent_id == agent.id, |
| AgentExecution.started_at >= now - timedelta(minutes=5) |
| ) |
| ).all() |
|
|
| assert len(recent) >= 1, "Should find recent execution" |
|
|
|
|
| class TestDatabaseAggregationQueries: |
| """Test database aggregation queries for analytics.""" |
|
|
| def test_count_executions_by_status(self, db_session: Session): |
| """Test counting executions grouped by status.""" |
| agent = AgentFactory(name="AggregationAgent", _session=db_session) |
| db_session.commit() |
|
|
| |
| statuses = ["running", "completed", "completed", "failed"] |
| for status in statuses: |
| AgentExecutionFactory( |
| agent_id=agent.id, |
| status=status, |
| _session=db_session |
| ) |
| db_session.commit() |
|
|
| |
| result = db_session.query( |
| AgentExecution.status, |
| func.count() |
| ).filter( |
| AgentExecution.agent_id == agent.id |
| ).group_by(AgentExecution.status).all() |
|
|
| status_counts = {row[0]: row[1] for row in result} |
| assert status_counts.get("completed", 0) == 2 |
| assert status_counts.get("running", 0) == 1 |
| assert status_counts.get("failed", 0) == 1 |
|
|
| def test_recent_executions_limit(self, db_session: Session): |
| """Test querying recent executions with limit.""" |
| agent = AgentFactory(name="RecentAgent", _session=db_session) |
| db_session.commit() |
|
|
| |
| for i in range(10): |
| AgentExecutionFactory( |
| agent_id=agent.id, |
| status="completed", |
| _session=db_session |
| ) |
| db_session.commit() |
|
|
| |
| recent = db_session.query(AgentExecution).filter( |
| AgentExecution.agent_id == agent.id |
| ).order_by( |
| AgentExecution.started_at.desc() |
| ).limit(5).all() |
|
|
| assert len(recent) == 5 |
|
|
| def test_execution_exists_query(self, db_session: Session): |
| """Test checking if execution exists.""" |
| agent = AgentFactory(name="ExistsAgent", _session=db_session) |
| db_session.commit() |
|
|
| execution = AgentExecutionFactory( |
| agent_id=agent.id, |
| _session=db_session |
| ) |
| db_session.commit() |
|
|
| |
| exists = db_session.query(AgentExecution).filter( |
| AgentExecution.id == execution.id |
| ).first() is not None |
|
|
| assert exists is True |
|
|
| |
| not_exists = db_session.query(AgentExecution).filter( |
| AgentExecution.id == "non-existent-id" |
| ).first() is not None |
|
|
| assert not_exists is False |
|
|