"""
Property-Based Tests for Agent Types - Agent Orchestration Platform
Comprehensive property-based testing for agent state management, resource tracking,
and lifecycle operations with security and consistency validation.
Author: Adder_3 | Created: 2025-06-26 | Last Modified: 2025-06-26
"""
import pytest
from hypothesis import given, strategies as st, assume, example, settings
from hypothesis.stateful import RuleBasedStateMachine, rule, invariant, initialize
from datetime import datetime, timedelta
from dataclasses import replace
from src.models.ids import create_agent_id, create_session_id, create_process_id, create_iterm_tab_id, create_message_id
from src.models.agent import (
AgentState, AgentStatus, AgentSpecialization, ClaudeConfig,
ResourceMetrics, Message
)
# Strategy for generating valid AgentStatus values
agent_status_strategy = st.sampled_from(AgentStatus)
# Strategy for generating valid AgentSpecialization values
agent_specialization_strategy = st.sampled_from(AgentSpecialization)
# Strategy for generating valid ClaudeConfig instances
claude_config_strategy = st.builds(
ClaudeConfig,
model=st.sampled_from(["sonnet-3.5", "opus-3", "haiku-3"]),
no_color=st.booleans(),
skip_permissions=st.booleans(),
verbose=st.booleans(),
output_format=st.sampled_from(["text", "json"]),
timeout_seconds=st.integers(min_value=30, max_value=3600),
max_memory_mb=st.integers(min_value=64, max_value=1024),
max_cpu_percent=st.floats(min_value=1.0, max_value=50.0),
custom_commands=st.tuples(*[st.text(alphabet=st.characters(whitelist_categories=["Ll", "Lu", "Nd"], whitelist_characters=" -_"), max_size=50) for _ in range(3)])
)
# Strategy for generating valid ResourceMetrics instances
resource_metrics_strategy = st.builds(
ResourceMetrics,
cpu_percent=st.floats(min_value=0.0, max_value=100.0),
memory_mb=st.floats(min_value=0.0, max_value=2048.0),
memory_percent=st.floats(min_value=0.0, max_value=100.0),
disk_io_read_mb=st.floats(min_value=0.0, max_value=1000.0),
disk_io_write_mb=st.floats(min_value=0.0, max_value=1000.0),
network_bytes_sent=st.integers(min_value=0, max_value=1000000),
network_bytes_recv=st.integers(min_value=0, max_value=1000000),
process_uptime_seconds=st.integers(min_value=0, max_value=86400),
thread_count=st.integers(min_value=0, max_value=200),
file_descriptor_count=st.integers(min_value=0, max_value=1000)
)
# Strategy for generating valid Message instances
message_strategy = st.builds(
Message,
message_id=st.text(min_size=1, max_size=50).map(lambda x: create_message_id()),
timestamp=st.datetimes(min_value=datetime(2020, 1, 1)),
role=st.sampled_from(["user", "assistant", "system"]),
content=st.text(min_size=1, max_size=1000).filter(lambda x: x.strip()),
token_count=st.one_of(st.none(), st.integers(min_value=1, max_value=10000)),
processing_time_ms=st.one_of(st.none(), st.integers(min_value=0, max_value=10000))
)
class TestClaudeConfigProperties:
"""Property-based tests for Claude configuration."""
@given(claude_config_strategy)
def test_claude_config_validation_properties(self, config):
"""Property: Valid Claude configs meet all constraints."""
# Property 1: Memory limits are enforced
assert 64 <= config.max_memory_mb <= 1024
# Property 2: CPU limits are enforced
assert 1.0 <= config.max_cpu_percent <= 50.0
# Property 3: Timeout is within bounds
assert 30 <= config.timeout_seconds <= 3600
# Property 4: Output format is valid
assert config.output_format in {"text", "json"}
@given(
st.one_of(st.integers(max_value=0), st.integers(min_value=4097)), # Invalid: <= 0 or > 4096
st.integers(min_value=1, max_value=4096), # Valid range
st.floats(min_value=1.0, max_value=50.0),
st.integers(min_value=30, max_value=3600)
)
def test_claude_config_memory_limit_enforcement(self, invalid_memory, valid_memory, cpu, timeout):
"""Property: Memory limits are strictly enforced."""
# Invalid memory should raise exception
if invalid_memory <= 0 or invalid_memory > 4096:
with pytest.raises(ValueError):
ClaudeConfig(
max_memory_mb=invalid_memory,
max_cpu_percent=cpu,
timeout_seconds=timeout
)
# Valid memory should work
config = ClaudeConfig(
max_memory_mb=valid_memory,
max_cpu_percent=cpu,
timeout_seconds=timeout
)
assert config.max_memory_mb == valid_memory
@given(st.text(min_size=1, max_size=500))
def test_claude_config_activation_command_safety(self, working_dir):
"""Property: Activation commands are safe from injection."""
assume('\\' not in working_dir) # Avoid Windows path issues
assume('"' not in working_dir) # Avoid quote issues
config = ClaudeConfig()
try:
from pathlib import Path
cmd = config.get_activation_command(Path(working_dir))
# Property: Command contains expected structure
assert "claude" in cmd
assert working_dir in cmd or working_dir.replace(' ', '\\ ') in cmd
# Property: No obvious injection patterns
assert ";" not in cmd
assert "&&" not in cmd.replace(" && claude", "")
except (TypeError, ValueError):
# Invalid working directory is handled gracefully
pass
class TestResourceMetricsProperties:
"""Property-based tests for resource metrics."""
@given(resource_metrics_strategy)
def test_resource_metrics_validation_properties(self, metrics):
"""Property: Valid resource metrics meet all constraints."""
# Property 1: CPU percentage is within bounds
assert 0.0 <= metrics.cpu_percent <= 100.0
# Property 2: Memory usage is non-negative
assert metrics.memory_mb >= 0
# Property 3: File descriptor count is non-negative
assert metrics.file_descriptor_count >= 0
# Property 4: Process uptime is non-negative
assert metrics.process_uptime_seconds >= 0
# Property 5: Thread count is non-negative
assert metrics.thread_count >= 0
# Property 6: Memory percentage is within bounds
assert 0.0 <= metrics.memory_percent <= 100.0
@given(
resource_metrics_strategy,
claude_config_strategy
)
def test_resource_limits_checking(self, metrics, config):
"""Property: Resource limit checking is consistent."""
# ResourceMetrics doesn't have is_within_limits method, so check manually
within_cpu_limit = metrics.cpu_percent <= config.max_cpu_percent
within_memory_limit = metrics.memory_mb <= config.max_memory_mb
within_thread_limit = metrics.thread_count <= 100
within_fd_limit = metrics.file_descriptor_count <= 500
# Property 1: Check resource health based on thresholds
is_healthy = metrics.is_healthy()
if is_healthy:
assert metrics.cpu_percent < 80.0
assert metrics.memory_percent < 80.0
assert metrics.thread_count < 100
assert metrics.file_descriptor_count < 500
# Property 2: Check resource stress
is_stressed = metrics.is_resource_stressed()
if is_stressed:
assert (metrics.cpu_percent > 90.0 or
metrics.memory_percent > 90.0 or
metrics.thread_count > 200 or
metrics.file_descriptor_count > 800)
@given(
resource_metrics_strategy,
st.integers(min_value=1, max_value=3600)
)
def test_responsiveness_checking(self, metrics, max_silence):
"""Property: Resource metrics are updated with timestamps."""
# ResourceMetrics has last_updated timestamp
now = datetime.now()
age = (now - metrics.last_updated).total_seconds()
# Property: Timestamp is reasonable (not in the future, not too old)
assert age >= 0 # Not in the future
assert age < 86400 # Less than a day old
class TestMessageProperties:
"""Property-based tests for messages."""
@given(message_strategy)
def test_message_properties(self, message):
"""Property: Valid messages meet constraints."""
# Property 1: Content is non-empty
assert len(message.content.strip()) > 0
# Property 2: Role is valid
assert message.role in {"user", "assistant", "system"}
# Property 3: Token count is positive when present
if message.token_count is not None:
assert message.token_count > 0
# Property 4: Processing time is non-negative when present
if message.processing_time_ms is not None:
assert message.processing_time_ms >= 0
def test_message_content_limits(self):
"""Property: Oversized content is rejected."""
# Create content larger than 100KB limit
oversized_content = "x" * 100_001
with pytest.raises(ValueError):
Message(
message_id=create_message_id(),
timestamp=datetime.now(),
role="user",
content=oversized_content,
token_count=100
)
@given(st.text().filter(lambda x: x.strip() == ""))
def test_message_empty_content_rejection(self, empty_content):
"""Property: Empty content is rejected."""
with pytest.raises(ValueError):
Message(
message_id=create_message_id(),
timestamp=datetime.now(),
role="user",
content=empty_content,
token_count=0
)
class TestAgentStateProperties:
"""Property-based tests for agent state management."""
def create_valid_agent_state(
self,
status=AgentStatus.ACTIVE,
specialization=AgentSpecialization.GENERAL,
error_count=0,
restart_count=0
):
"""Helper to create valid agent state for testing."""
# Create agent with specific number for consistency
agent_number = 1
agent_id = create_agent_id(agent_number)
session_id = create_session_id()
now = datetime.now()
created_time = now
return AgentState(
agent_id=agent_id,
session_id=session_id,
name=f"Agent_{agent_number}",
process_id=create_process_id(1234) if status != AgentStatus.CREATED else None,
iterm_tab_id=create_iterm_tab_id() if status != AgentStatus.CREATED else None,
status=status,
specialization=specialization,
system_prompt_suffix="",
claude_config=ClaudeConfig(),
created_at=created_time,
last_heartbeat=now,
last_status_change=created_time,
resource_usage=ResourceMetrics(
cpu_percent=10.0,
memory_mb=256.0,
memory_percent=25.0,
thread_count=50,
file_descriptor_count=50,
process_uptime_seconds=60
),
error_count=error_count,
restart_count=restart_count
)
@given(
agent_status_strategy,
agent_specialization_strategy,
st.integers(min_value=0, max_value=10),
st.integers(min_value=0, max_value=5)
)
def test_agent_state_creation_properties(self, status, specialization, error_count, restart_count):
"""Property: Agent states can be created with valid parameters."""
try:
agent_state = self.create_valid_agent_state(
status=status,
specialization=specialization,
error_count=error_count,
restart_count=restart_count
)
# Property 1: All fields are preserved
assert agent_state.status == status
assert agent_state.specialization == specialization
assert agent_state.error_count == error_count
assert agent_state.restart_count == restart_count
# Property 2: Timestamps are consistent
assert agent_state.last_heartbeat >= agent_state.created_at
# Property 3: Agent name is valid
assert agent_state.name == "Agent_1"
except ValueError:
# Some combinations might be invalid - that's expected
pass
@given(agent_status_strategy)
def test_agent_status_transitions(self, new_status):
"""Property: Status transitions update state correctly."""
agent_state = self.create_valid_agent_state(status=AgentStatus.ACTIVE)
# Status transition should always succeed (no validation in with_status)
new_state = agent_state.with_status(new_status)
assert new_state.status == new_status
# Last status change should be updated
assert new_state.last_status_change > agent_state.last_status_change
# Other fields should remain unchanged
assert new_state.agent_id == agent_state.agent_id
assert new_state.name == agent_state.name
assert new_state.error_count == agent_state.error_count
def test_agent_error_handling(self):
"""Property: Error handling preserves state consistency."""
agent_state = self.create_valid_agent_state()
# Test error increment
error_state = agent_state.with_error_increment()
# Property 1: Error count increments
assert error_state.error_count == agent_state.error_count + 1
# Property 2: Other fields remain unchanged
assert error_state.status == agent_state.status
assert error_state.agent_id == agent_state.agent_id
assert error_state.name == agent_state.name
def test_agent_health_checking_properties(self):
"""Property: Agent health checking is comprehensive."""
# Healthy agent
healthy_agent = self.create_valid_agent_state(status=AgentStatus.ACTIVE)
assert healthy_agent.is_healthy()
# Unhealthy status
error_agent = self.create_valid_agent_state(status=AgentStatus.ERROR)
assert not error_agent.is_healthy()
# Resource limits exceeded
overloaded_metrics = ResourceMetrics(
cpu_percent=95.0, # Over 90% stress threshold
memory_mb=256.0,
memory_percent=95.0, # Over 90% stress threshold
thread_count=50,
file_descriptor_count=50,
process_uptime_seconds=60
)
overloaded_agent = replace(healthy_agent, resource_usage=overloaded_metrics)
assert not overloaded_agent.is_healthy()
class AgentStateMachine(RuleBasedStateMachine):
"""
Stateful property-based testing for agent state transitions.
Tests complex agent lifecycle operations and ensures system invariants
are maintained across all state transitions.
"""
def __init__(self):
super().__init__()
self.agent_states = {}
self.agent_counter = 1
@initialize()
def setup_initial_agents(self):
"""Initialize with some agents in different states."""
for status in [AgentStatus.CREATED, AgentStatus.ACTIVE, AgentStatus.IDLE]:
agent_state = self.create_agent_state(status=status)
self.agent_states[agent_state.agent_id] = agent_state
self.agent_counter += 1
def create_agent_state(self, status=AgentStatus.CREATED):
"""Create agent state for testing."""
# Use counter to ensure unique agent numbers
agent_id = create_agent_id(self.agent_counter)
session_id = create_session_id()
now = datetime.now()
created_time = now
return AgentState(
agent_id=agent_id,
session_id=session_id,
name=f"Agent_{self.agent_counter}",
process_id=create_process_id(1000 + self.agent_counter) if status != AgentStatus.CREATED else None,
iterm_tab_id=create_iterm_tab_id() if status != AgentStatus.CREATED else None,
status=status,
specialization=AgentSpecialization.GENERAL,
system_prompt_suffix="",
claude_config=ClaudeConfig(),
created_at=created_time,
last_heartbeat=now,
last_status_change=created_time,
resource_usage=ResourceMetrics(
cpu_percent=10.0,
memory_mb=256.0,
memory_percent=25.0,
thread_count=50,
file_descriptor_count=50,
process_uptime_seconds=60
)
)
@rule(target_status=agent_status_strategy)
def transition_random_agent(self, target_status):
"""Rule: Transition random agent to new status."""
if not self.agent_states:
return
# Pick a random agent
import random
agent_id = random.choice(list(self.agent_states.keys()))
agent_state = self.agent_states[agent_id]
# Skip invalid transitions
# Can't transition to ACTIVE/IDLE without process and tab IDs
if target_status in [AgentStatus.ACTIVE, AgentStatus.IDLE]:
if not agent_state.process_id or not agent_state.iterm_tab_id:
return
# Always allow status transitions (no validation in with_status)
new_state = agent_state.with_status(target_status)
self.agent_states[agent_id] = new_state
@rule()
def add_error_to_random_agent(self):
"""Rule: Add error to random agent."""
if not self.agent_states:
return
# Pick a random agent
import random
agent_id = random.choice(list(self.agent_states.keys()))
agent_state = self.agent_states[agent_id]
# Increment error count
error_state = agent_state.with_error_increment()
self.agent_states[agent_id] = error_state
@rule()
def create_new_agent(self):
"""Rule: Create new agent."""
if len(self.agent_states) < 10: # Limit for testing
agent_state = self.create_agent_state()
self.agent_states[agent_state.agent_id] = agent_state
self.agent_counter += 1
@invariant()
def all_agents_have_valid_states(self):
"""Invariant: All agents maintain valid state."""
for agent_id, agent_state in self.agent_states.items():
# Check basic state validity
assert agent_state.agent_id == agent_id
assert agent_state.last_heartbeat >= agent_state.created_at
assert agent_state.error_count >= 0
assert agent_state.restart_count >= 0
@invariant()
def agent_health_consistency(self):
"""Invariant: Agent health status is consistent with state."""
for agent_state in self.agent_states.values():
if agent_state.is_healthy():
# Healthy agents should not be in error or terminated states
assert agent_state.status not in {AgentStatus.ERROR, AgentStatus.TERMINATED}
# Healthy agents should have low error count
assert agent_state.error_count <= 5
# Healthy agents should have healthy resource usage
assert agent_state.resource_usage.is_healthy()
if agent_state.status == AgentStatus.ERROR:
# Error agents should not be healthy
assert not agent_state.is_healthy()
if agent_state.status == AgentStatus.TERMINATED:
# Terminated agents should not be healthy
assert not agent_state.is_healthy()
# Test the state machine
TestAgentStateMachine = AgentStateMachine.TestCase
if __name__ == "__main__":
# Run property-based tests
pytest.main([__file__, "-v"])
