File size: 10,433 Bytes
aef804e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 | """
Fixtures for agent execution property tests.
Provides test data and Hypothesis settings for testing agent execution
invariants (idempotence, termination, determinism).
"""
import pytest
from hypothesis import settings, HealthCheck
from hypothesis.strategies import (
text, integers, floats, lists, sampled_from,
booleans, dictionaries, tuples, datetimes, timedeltas, uuids
)
from datetime import datetime, timedelta
from unittest.mock import Mock, patch
import uuid
# Import fixtures from parent conftest to avoid duplication
from tests.property_tests.conftest import (
db_session, test_agent, test_agents,
DEFAULT_PROFILE
)
# Import models
from sqlalchemy.orm import Session
from core.models import (
AgentRegistry, AgentExecution, AgentStatus,
ExecutionStatus
)
# ============================================================================
# HYPOTHESIS SETTINGS FOR AGENT EXECUTION TESTS
# ============================================================================
#
# Property tests run with max_examples iterations to comprehensively validate
# invariants. Different invariants require different example counts:
#
# - CRITICAL: max_examples=200 (idempotence, determinism)
# - STANDARD: max_examples=100 (termination, state transitions)
# - IO_BOUND: max_examples=50 (database operations)
# ============================================================================
HYPOTHESIS_SETTINGS_CRITICAL = {
"suppress_health_check": [HealthCheck.function_scoped_fixture, HealthCheck.too_slow],
"max_examples": 200 # Critical invariants (idempotence, determinism)
}
HYPOTHESIS_SETTINGS_STANDARD = {
"suppress_health_check": [HealthCheck.function_scoped_fixture, HealthCheck.too_slow],
"max_examples": 100 # Standard invariants (termination, state transitions)
}
HYPOTHESIS_SETTINGS_IO = {
"suppress_health_check": [HealthCheck.function_scoped_fixture, HealthCheck.too_slow],
"max_examples": 50 # IO-bound operations (database queries)
}
# ============================================================================
# FIXTURES FOR AGENT EXECUTION TESTS
# ============================================================================
@pytest.fixture(scope="function")
def test_agent_execution(db_session: Session):
"""
Create a test agent execution with valid state.
Returns an AgentExecution with PENDING status for testing
state transitions and lifecycle invariants.
"""
# First create a test agent
agent = AgentRegistry(
name="ExecutionTestAgent",
tenant_id="default",
category="test",
module_path="test.module",
class_name="TestClass",
status=AgentStatus.INTERN.value,
confidence_score=0.6,
)
db_session.add(agent)
db_session.commit()
db_session.refresh(agent)
# Create execution
execution = AgentExecution(
agent_id=agent.id,
tenant_id="default",
status=ExecutionStatus.PENDING.value,
input_summary="Test input",
triggered_by="manual",
started_at=datetime.utcnow(),
duration_seconds=0.0,
result_summary=None,
error_message=None,
metadata_json={}
)
db_session.add(execution)
db_session.commit()
db_session.refresh(execution)
return execution
@pytest.fixture(scope="function")
def test_agent_executions(db_session: Session):
"""
Create multiple test agent executions with different statuses.
Returns a list of AgentExecution objects with various statuses
(PENDING, RUNNING, COMPLETED, FAILED, CANCELLED) for testing
state transition invariants.
"""
# First create a test agent
agent = AgentRegistry(
name="MultiExecutionTestAgent",
tenant_id="default",
category="test",
module_path="test.module",
class_name="TestClass",
status=AgentStatus.INTERN.value,
confidence_score=0.6,
)
db_session.add(agent)
db_session.commit()
db_session.refresh(agent)
executions = []
for status in ExecutionStatus:
execution = AgentExecution(
agent_id=agent.id,
tenant_id="default",
status=status.value,
input_summary=f"Test input for {status.value}",
triggered_by="manual",
started_at=datetime.utcnow(),
completed_at=datetime.utcnow() if status != ExecutionStatus.PENDING and status != ExecutionStatus.RUNNING else None,
duration_seconds=1.0 if status in [ExecutionStatus.COMPLETED, ExecutionStatus.FAILED, ExecutionStatus.CANCELLED] else 0.0,
result_summary="Test result" if status == ExecutionStatus.COMPLETED else None,
error_message="Test error" if status == ExecutionStatus.FAILED else None,
metadata_json={"test_key": "test_value"}
)
db_session.add(execution)
db_session.commit()
db_session.refresh(execution)
executions.append(execution)
return executions
@pytest.fixture(scope="function")
def mock_llm_response():
"""
Mock LLM responses for deterministic testing.
Returns a Mock object that simulates LLM responses with
consistent output for testing idempotence and determinism.
"""
mock_response = Mock()
mock_response.choices = [Mock()]
mock_response.choices[0].message.content = "Test response"
mock_response.choices[0].finish_reason = "stop"
mock_response.usage = Mock()
mock_response.usage.prompt_tokens = 10
mock_response.usage.completion_tokens = 20
mock_response.usage.total_tokens = 30
return mock_response
@pytest.fixture(scope="function")
def execution_params():
"""
Strategy for generating valid execution parameters.
Returns Hypothesis strategies for generating test data:
- agent_ids: UUID-based agent identifiers
- params: Dictionaries with string keys and values
- durations: Integer durations in seconds
"""
agent_ids = uuids()
params = dictionaries(
keys=text(min_size=1, max_size=50, alphabet='abcdefghijklmnopqrstuvwxyz_'),
values=text(min_size=0, max_size=100),
min_size=0,
max_size=10
)
durations = integers(min_value=0, max_value=3600)
return {
"agent_ids": agent_ids,
"params": params,
"durations": durations
}
# ============================================================================
# HYPOTHESIS STRATEGIES FOR AGENT EXECUTION TESTS
# ============================================================================
@pytest.fixture(scope="session")
def valid_agent_ids():
"""Strategy for generating valid agent IDs."""
return uuids()
@pytest.fixture(scope="session")
def execution_params_strategy():
"""Strategy for generating valid execution parameters."""
return dictionaries(
keys=text(min_size=1, max_size=50, alphabet='abcdefghijklmnopqrstuvwxyz_'),
values=text(min_size=0, max_size=100),
min_size=0,
max_size=10
)
@pytest.fixture(scope="session")
def execution_durations():
"""Strategy for generating execution durations."""
return integers(min_value=0, max_value=3600)
@pytest.fixture(scope="session")
def execution_statuses():
"""Strategy for generating execution status values."""
return sampled_from([
ExecutionStatus.PENDING.value,
ExecutionStatus.RUNNING.value,
ExecutionStatus.COMPLETED.value,
ExecutionStatus.FAILED.value,
ExecutionStatus.CANCELLED.value
])
# ============================================================================
# HELPER FUNCTIONS FOR AGENT EXECUTION TESTS
# ============================================================================
def create_execution_record(
db_session: Session,
agent_id: str,
status: str = ExecutionStatus.PENDING.value,
input_summary: str = None,
triggered_by: str = "manual",
started_at: datetime = None,
completed_at: datetime = None,
duration_seconds: float = 0.0,
result_summary: str = None,
error_message: str = None,
metadata_json: dict = None
) -> AgentExecution:
"""
Helper function to create an AgentExecution record.
Args:
db_session: Database session
agent_id: Agent ID
status: Execution status
input_summary: Input summary text
triggered_by: Trigger source (manual, schedule, websocket, event)
started_at: Start timestamp
completed_at: Completion timestamp
duration_seconds: Execution duration in seconds
result_summary: Result summary text
error_message: Error message text
metadata_json: Additional metadata
Returns:
AgentExecution object
"""
execution = AgentExecution(
agent_id=agent_id,
tenant_id="default",
status=status,
input_summary=input_summary,
triggered_by=triggered_by,
started_at=started_at or datetime.utcnow(),
completed_at=completed_at,
duration_seconds=duration_seconds,
result_summary=result_summary,
error_message=error_message,
metadata_json=metadata_json or {}
)
db_session.add(execution)
db_session.commit()
db_session.refresh(execution)
return execution
def simulate_execution(
db_session: Session,
execution_id: str,
result: str = None,
error: str = None,
duration: float = 1.0
) -> AgentExecution:
"""
Helper function to simulate agent execution completion.
Args:
db_session: Database session
execution_id: Execution ID to update
result: Result summary (if successful)
error: Error message (if failed)
duration: Execution duration in seconds
Returns:
Updated AgentExecution object
"""
execution = db_session.query(AgentExecution).filter(
AgentExecution.id == execution_id
).first()
if not execution:
raise ValueError(f"Execution {execution_id} not found")
execution.completed_at = datetime.utcnow()
execution.duration_seconds = duration
if error:
execution.status = ExecutionStatus.FAILED.value
execution.error_message = error
else:
execution.status = ExecutionStatus.COMPLETED.value
execution.result_summary = result
db_session.commit()
db_session.refresh(execution)
return execution
|