"""
Property-Based Tests for Communication Types - Agent Orchestration Platform
Comprehensive property-based testing for MCP tool results, message handling,
and status monitoring with focus on reliability and consistency.
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
import json
import secrets
from src.models.ids import create_agent_id, create_session_id, create_process_id, create_iterm_tab_id
from src.models.communication import (
MessageContent, MessageType, OperationStatus,
MessageResult, ConversationClearResult, ConversationStartResult
)
from src.models.mcp_results import (
MCPToolResult, AgentCreationResult, AgentDeletionResult,
SessionCreationResult, SessionDeletionResult, SessionStatusResult,
HealthStatus
)
# Strategy for generating valid MessageType values
message_type_strategy = st.sampled_from(MessageType)
# Strategy for generating valid OperationStatus values
operation_status_strategy = st.sampled_from(OperationStatus)
# Strategy for generating JSON-serializable data
json_serializable_strategy = st.recursive(
base=st.one_of(
st.none(),
st.booleans(),
st.integers(min_value=-1000000, max_value=1000000),
st.floats(allow_nan=False, allow_infinity=False, min_value=-1e6, max_value=1e6),
st.text(max_size=100)
),
extend=lambda children: st.one_of(
st.lists(children, max_size=10),
st.dictionaries(st.text(max_size=20), children, max_size=10)
),
max_leaves=20
)
# Strategy for generating valid MessageContent instances
message_content_strategy = st.builds(
MessageContent,
text=st.text(min_size=1, max_size=1000).filter(lambda x: len(x.strip()) > 0),
message_type=message_type_strategy,
metadata=st.dictionaries(
st.text(min_size=1, max_size=20),
st.one_of(st.text(max_size=100), st.integers(), st.floats(allow_nan=False, allow_infinity=False), st.booleans()),
max_size=5
)
)
# Strategy for generating valid MCPToolResult instances
@st.composite
def mcp_tool_result_strategy(draw):
operation_type = draw(st.text(min_size=1, max_size=50).filter(lambda x: len(x.strip()) > 0))
message = draw(st.text(min_size=1, max_size=500).filter(lambda x: len(x.strip()) > 0))
data = draw(st.one_of(st.none(), json_serializable_strategy))
operation_id = draw(st.one_of(st.none(), st.text(min_size=1, max_size=50)))
# Generate consistent success/status/error_code combination
success = draw(st.booleans())
if success:
status = OperationStatus.SUCCESS
error_code = None
error_details = None
else:
status = draw(st.sampled_from([OperationStatus.FAILURE, OperationStatus.TIMEOUT, OperationStatus.CANCELLED]))
error_code = draw(st.text(min_size=1, max_size=50).filter(lambda x: len(x.strip()) > 0))
error_details = draw(st.one_of(st.none(), st.text(max_size=1000)))
return MCPToolResult(
operation_type=operation_type,
status=status,
message=message,
success=success,
data=data,
error_code=error_code,
error_details=error_details,
operation_id=operation_id
)
class TestMessageContentProperties:
"""Property-based tests for message content."""
@given(message_content_strategy)
def test_message_content_validation_properties(self, message):
"""Property: Valid message content meets all constraints."""
# Property 1: Content is non-empty
assert len(message.text.strip()) > 0
# Property 2: Content is within size limits
assert len(message.text) <= 100000 # 100KB limit
# Property 3: Message type is valid
assert isinstance(message.message_type, MessageType)
# Property 4: Metadata values are serializable
for key, value in message.metadata.items():
assert isinstance(key, str)
assert isinstance(value, (str, int, float, bool, type(None)))
def test_message_content_size_limit_enforcement(self):
"""Property: Oversized content is rejected."""
from src.models.communication import MessageValidationError
oversized_content = "A" * 100001 # Over 100KB limit
with pytest.raises(MessageValidationError):
MessageContent(
text=oversized_content,
message_type=MessageType.USER_MESSAGE,
metadata={}
)
@given(st.text().filter(lambda x: x.strip() == ""))
def test_message_content_empty_rejection(self, empty_content):
"""Property: Empty content is rejected."""
from src.models.communication import MessageValidationError
with pytest.raises(MessageValidationError):
MessageContent(
text=empty_content,
message_type=MessageType.USER_MESSAGE,
metadata={}
)
@given(message_content_strategy)
def test_message_content_system_classification(self, message):
"""Property: System message classification is accurate."""
is_system = message.is_system_message()
expected_system = message.message_type in {
MessageType.SYSTEM,
MessageType.SYSTEM_COMMAND,
MessageType.ERROR_NOTIFICATION,
MessageType.HEARTBEAT
}
assert is_system == expected_system
@given(message_content_strategy)
def test_message_content_token_estimation(self, message):
"""Property: Token estimation is reasonable."""
estimated_tokens = message.get_estimated_tokens()
# Property 1: Token count is positive
assert estimated_tokens > 0
# Property 2: Token count is related to content length
expected_tokens = max(1, len(message.text) // 4)
assert estimated_tokens == expected_tokens
class TestMCPToolResultProperties:
"""Property-based tests for MCP tool results."""
@given(mcp_tool_result_strategy())
def test_mcp_tool_result_consistency_properties(self, result):
"""Property: MCP tool results maintain consistency."""
# Property 1: Message is non-empty
assert len(result.message.strip()) > 0
# Property 2: Success/error consistency
if result.success:
assert result.status in {OperationStatus.SUCCESS, OperationStatus.PARTIAL_SUCCESS}
else:
assert result.status != OperationStatus.SUCCESS
assert result.error_code is not None
# Property 3: Data is JSON-serializable when present
if result.data is not None:
try:
json.dumps(result.data)
except (TypeError, ValueError):
pytest.fail("Result data is not JSON-serializable")
# Property 4: Timestamp is reasonable
assert result.timestamp <= datetime.utcnow() + timedelta(minutes=1)
@given(
st.text(min_size=1, max_size=100),
json_serializable_strategy,
st.text(min_size=1, max_size=50)
)
def test_success_result_creation_properties(self, message, data, operation_id):
"""Property: Success results are created correctly."""
result = MCPToolResult.success_result(
message=message,
data=data,
operation_id=operation_id
)
# Property 1: Result indicates success
assert result.success
assert result.status == OperationStatus.SUCCESS
# Property 2: Message is preserved
assert result.message == message.strip()
# Property 3: Data is preserved
assert result.data == data
# Property 4: Operation ID is preserved
assert result.operation_id == operation_id
# Property 5: No error information
assert result.error_code is None
@given(
st.text(min_size=1, max_size=100),
st.text(min_size=1, max_size=50),
st.one_of(st.none(), st.text(max_size=500)),
st.one_of(st.none(), st.text(min_size=1, max_size=50))
)
def test_error_result_creation_properties(self, message, error_code, error_details, operation_id):
"""Property: Error results are created correctly."""
result = MCPToolResult.error_result(
message=message,
error_code=error_code,
error_details=error_details,
operation_id=operation_id
)
# Property 1: Result indicates failure
assert not result.success
assert result.status == OperationStatus.FAILURE
# Property 2: Message is preserved
assert result.message == message.strip()
# Property 3: Error code is preserved
if error_code and error_code.strip():
assert result.error_code == error_code.strip()
else:
assert result.error_code == error_code
# Property 4: Error details are preserved
if error_details:
assert result.error_details == {"details": error_details}
else:
assert result.error_details == {}
# Property 5: Operation ID is preserved
assert result.operation_id == operation_id
class TestSpecificResultTypeProperties:
"""Property-based tests for specific MCP result types."""
@given(
st.booleans(),
st.text(min_size=1, max_size=100),
st.one_of(st.none(), st.builds(create_agent_id)),
st.one_of(st.none(), st.text(min_size=1, max_size=50)),
st.one_of(st.none(), st.builds(create_iterm_tab_id)),
st.one_of(st.none(), st.builds(create_process_id, st.integers(min_value=1, max_value=65535)))
)
def test_agent_creation_result_properties(self, success, message, agent_id, agent_name, tab_id, process_id):
"""Property: Agent creation results maintain consistency."""
try:
if success:
# Successful creation should have agent_id
if agent_id is None:
agent_id = create_agent_id()
result = AgentCreationResult(
operation_type="agent_creation",
status=OperationStatus.SUCCESS,
message=message,
agent_id=agent_id,
agent_name=agent_name,
iterm_tab_id=tab_id,
process_id=process_id
)
else:
# Failed creation should not have agent_id
result = AgentCreationResult(
operation_type="agent_creation",
status=OperationStatus.FAILURE,
message=message,
error_code="creation_failed",
agent_id=None,
agent_name=None,
iterm_tab_id=None,
process_id=None
)
# Property 1: Success consistency
if result.is_success():
assert result.agent_id is not None
else:
assert result.agent_id is None
except ValueError:
# Some combinations might be invalid - that's expected
pass
@given(
st.booleans(),
st.text(min_size=1, max_size=100),
st.one_of(st.none(), st.builds(create_session_id)),
st.one_of(st.none(), st.text(min_size=1, max_size=50))
)
def test_session_creation_result_properties(self, success, message, session_id, session_name):
"""Property: Session creation results maintain consistency."""
try:
if success:
# Successful creation should have session_id
if session_id is None:
session_id = create_session_id()
result = SessionCreationResult(
operation_type="session_creation",
status=OperationStatus.SUCCESS,
message=message,
session_id=session_id,
session_name=session_name
)
else:
# Failed creation should not have session_id
result = SessionCreationResult(
operation_type="session_creation",
status=OperationStatus.FAILURE,
message=message,
error_code="creation_failed",
session_id=None,
session_name=None
)
# Property 1: Success consistency
if result.is_success():
assert result.session_id is not None
else:
assert result.session_id is None
except ValueError:
# Some combinations might be invalid - that's expected
pass
@given(
st.booleans(),
st.text(min_size=1, max_size=100),
st.lists(st.dictionaries(st.text(max_size=20), st.text(max_size=100), max_size=5), max_size=10),
st.integers(min_value=0, max_value=100),
st.integers(min_value=0, max_value=100),
st.integers(min_value=0, max_value=100)
)
def test_session_status_result_properties(self, success, message, session_details, session_count, total_agents, active_agents):
"""Property: Session status results maintain consistency."""
# Ensure active agents don't exceed total agents
active_agents = min(active_agents, total_agents)
try:
result = SessionStatusResult(
operation_type="session_status",
status=OperationStatus.SUCCESS if success else OperationStatus.FAILURE,
message=message,
error_code=None if success else "status_error",
session_details=session_details,
session_count=session_count,
total_agents=total_agents,
active_agents=active_agents
)
# Property 1: Agent counts are consistent
assert result.active_agents <= result.total_agents
# Property 2: All counts are non-negative
assert result.session_count >= 0
assert result.total_agents >= 0
assert result.active_agents >= 0
except ValueError:
# Some combinations might be invalid - that's expected
pass
class TestHealthStatusProperties:
"""Property-based tests for health status monitoring."""
@given(
st.sampled_from(["healthy", "warning", "critical", "unknown"]),
st.floats(min_value=0.0, max_value=800.0), # Up to 8 cores
st.floats(min_value=0.0, max_value=100.0),
st.floats(min_value=0.0, max_value=100.0),
st.integers(min_value=0, max_value=100),
st.integers(min_value=0, max_value=100),
st.integers(min_value=0, max_value=1000),
st.integers(min_value=0, max_value=86400)
)
def test_health_status_validation_properties(
self, status, cpu, memory, disk, sessions, agents, errors, uptime
):
"""Property: Valid health status meets all constraints."""
health = HealthStatus(
overall_status=status,
cpu_usage_percent=cpu,
memory_usage_percent=memory,
disk_usage_percent=disk,
active_sessions=sessions,
active_agents=agents,
error_count=errors,
uptime_seconds=uptime
)
# Property 1: Status is valid
assert health.overall_status in {"healthy", "warning", "critical", "unknown"}
# Property 2: All metrics are within bounds
assert 0.0 <= health.cpu_usage_percent <= 800.0
assert 0.0 <= health.memory_usage_percent <= 100.0
assert 0.0 <= health.disk_usage_percent <= 100.0
assert health.active_sessions >= 0
assert health.active_agents >= 0
assert health.error_count >= 0
assert health.uptime_seconds >= 0
@given(
st.floats(min_value=-100.0, max_value=-0.01),
st.floats(min_value=101.0, max_value=1000.0)
)
def test_health_status_invalid_ranges_rejection(self, negative_value, oversized_percentage):
"""Property: Invalid metric ranges are rejected."""
# Test negative CPU
with pytest.raises(ValueError):
HealthStatus(
overall_status="healthy",
cpu_usage_percent=negative_value,
memory_usage_percent=50.0,
disk_usage_percent=50.0,
active_sessions=1,
active_agents=1,
error_count=0,
uptime_seconds=100
)
# Test oversized memory percentage
with pytest.raises(ValueError):
HealthStatus(
overall_status="healthy",
cpu_usage_percent=50.0,
memory_usage_percent=oversized_percentage,
disk_usage_percent=50.0,
active_sessions=1,
active_agents=1,
error_count=0,
uptime_seconds=100
)
def test_health_status_classification_properties(self):
"""Property: Health classification is accurate."""
# Healthy system
healthy = HealthStatus(
overall_status="healthy",
cpu_usage_percent=50.0,
memory_usage_percent=60.0,
disk_usage_percent=70.0,
active_sessions=5,
active_agents=8,
error_count=2,
uptime_seconds=3600
)
assert healthy.is_healthy()
assert not healthy.needs_attention()
# Critical system
critical = HealthStatus(
overall_status="critical",
cpu_usage_percent=98.0,
memory_usage_percent=97.0,
disk_usage_percent=96.0,
active_sessions=5,
active_agents=8,
error_count=100,
uptime_seconds=3600
)
assert not critical.is_healthy()
assert critical.needs_attention()
class CommunicationStateMachine(RuleBasedStateMachine):
"""
Stateful property-based testing for communication operations.
Tests complex communication patterns and ensures consistency
across multiple operations and state changes.
"""
def __init__(self):
super().__init__()
self.messages = []
self.results = []
self.health_history = []
self.operation_counter = 0
@rule(
text=st.text(min_size=1, max_size=100).filter(lambda x: len(x.strip()) > 0),
message_type=st.sampled_from(MessageType)
)
def create_message(self, text, message_type):
"""Rule: Create new message."""
message = MessageContent(
text=text,
message_type=message_type,
metadata={}
)
self.messages.append(message)
@rule()
def create_success_result(self):
"""Rule: Create successful operation result."""
result = MCPToolResult.success_result(
message=f"Operation {self.operation_counter} succeeded",
data={"operation_id": self.operation_counter},
operation_id=f"op_{self.operation_counter}"
)
self.results.append(result)
self.operation_counter += 1
@rule()
def create_error_result(self):
"""Rule: Create error operation result."""
result = MCPToolResult.error_result(
message=f"Operation {self.operation_counter} failed",
error_code="operation_failed",
operation_id=f"op_{self.operation_counter}"
)
self.results.append(result)
self.operation_counter += 1
@rule(
overall_status=st.sampled_from(["healthy", "warning", "critical"]),
cpu_usage_percent=st.floats(min_value=0.0, max_value=100.0),
memory_usage_percent=st.floats(min_value=0.0, max_value=100.0),
disk_usage_percent=st.floats(min_value=0.0, max_value=100.0),
active_sessions=st.integers(min_value=0, max_value=20),
active_agents=st.integers(min_value=0, max_value=50),
error_count=st.integers(min_value=0, max_value=100),
uptime_seconds=st.integers(min_value=0, max_value=86400)
)
def record_health_status(self, overall_status, cpu_usage_percent, memory_usage_percent,
disk_usage_percent, active_sessions, active_agents, error_count, uptime_seconds):
"""Rule: Record system health status."""
health = HealthStatus(
overall_status=overall_status,
cpu_usage_percent=cpu_usage_percent,
memory_usage_percent=memory_usage_percent,
disk_usage_percent=disk_usage_percent,
active_sessions=active_sessions,
active_agents=active_agents,
error_count=error_count,
uptime_seconds=uptime_seconds
)
self.health_history.append(health)
@invariant()
def all_messages_are_valid(self):
"""Invariant: All messages maintain validity."""
for message in self.messages:
assert len(message.text.strip()) > 0
assert isinstance(message.message_type, MessageType)
@invariant()
def all_results_are_consistent(self):
"""Invariant: All results maintain consistency."""
for result in self.results:
assert len(result.message.strip()) > 0
if result.success:
assert result.status in {OperationStatus.SUCCESS, OperationStatus.PARTIAL_SUCCESS}
else:
assert result.status != OperationStatus.SUCCESS
assert result.error_code is not None
@invariant()
def health_history_is_valid(self):
"""Invariant: Health history maintains validity."""
for health in self.health_history:
assert health.overall_status in {"healthy", "warning", "critical", "unknown"}
assert 0.0 <= health.cpu_usage_percent <= 800.0
assert 0.0 <= health.memory_usage_percent <= 100.0
assert 0.0 <= health.disk_usage_percent <= 100.0
# Test the state machine
TestCommunicationStateMachine = CommunicationStateMachine.TestCase
class TestSerializationProperties:
"""Property-based tests for serialization and data integrity."""
@given(json_serializable_strategy)
def test_json_serialization_roundtrip(self, data):
"""Property: JSON serialization preserves data integrity."""
# Create result with data
result = MCPToolResult.success_result("test", data=data)
# Should be able to serialize and deserialize
try:
serialized = json.dumps(result.data)
deserialized = json.loads(serialized)
assert deserialized == data
except (TypeError, ValueError):
pytest.fail("Data should be JSON-serializable")
@given(message_content_strategy)
def test_message_content_metadata_serialization(self, message):
"""Property: Message metadata is JSON-serializable."""
try:
serialized = json.dumps(message.metadata)
deserialized = json.loads(serialized)
assert deserialized == message.metadata
except (TypeError, ValueError):
pytest.fail("Message metadata should be JSON-serializable")
if __name__ == "__main__":
# Run property-based tests
pytest.main([__file__, "-v"])
