Spaces:

techprotrade
/

annator-command-center

Paused

annator-command-center / tests /property_tests /auto_dev /conftest.py

Deploy ATOM FastAPI command center runtime (part 8)

aef804e verified 3 months ago

8.3 kB

"""
Property-Based Testing Configuration for Auto-Dev
This module provides Hypothesis strategies and pytest fixtures for testing
Auto-Dev components (EventBus, FitnessService, CapabilityGate, ContainerSandbox,
MementoEngine, Database Models).
Property-based testing generates hundreds of test cases automatically to verify
invariants that must ALWAYS hold true.
"""

import sys
import os

# Add backend to path for imports
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../../..'))

from hypothesis import strategies as st
import pytest
from datetime import datetime, timedelta
from typing import Any

# =============================================================================
# Hypothesis Strategies for Auto-Dev Data Types
# =============================================================================

# Fitness score strategy (normalized to [0.0, 1.0])
fitness_scores = st.lists(
    st.floats(min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False),
    min_size=0,
    max_size=100
)

# Maturity level strategy
maturity_levels = st.sampled_from(['student', 'intern', 'supervised', 'autonomous'])

# Event data strategy (random dictionaries with string keys and mixed values)
event_data = st.dictionaries(
    keys=st.text(min_size=1, max_size=50, alphabet='abcdefghijklmnopqrstuvwxyz'),
    values=st.one_of(
        st.text(),
        st.integers(),
        st.floats(allow_nan=False, allow_infinity=False),
        st.booleans(),
        st.none()
    ),
    min_size=0,
    max_size=20
)

# Capability strategy
capabilities = st.sampled_from([
    'file_operations',
    'network_access',
    'system_commands',
    'database_write',
    'api_calls',
    'user_interaction',
    'auto_dev.memento_skills',
    'auto_dev.alpha_evolver',
    'auto_dev.background_evolution'
])

# Valid skill name strategy (alphanumeric with underscores)
valid_skill_names = st.text(min_size=1, max_size=50).filter(
    lambda x: x.replace('_', '').replace('-', '').isalnum() and len(x) > 0
)

# Agent ID strategy (UUID-like strings)
agent_ids = st.text(min_size=36, max_size=36, alphabet='abcdef0123456789-')

# Tenant ID strategy
tenant_ids = st.text(min_size=36, max_size=36, alphabet='abcdef0123456789-')

# Episode ID strategy
episode_ids = st.text(min_size=36, max_size=36, alphabet='abcdef0123456789-')

# Proxy signals strategy (for FitnessService)
proxy_signals = st.dictionaries(
    keys=st.sampled_from([
        'execution_success',
        'syntax_error',
        'execution_latency_ms',
        'user_approved_proposal',
        'expects_delayed_eval'
    ]),
    values=st.one_of(st.booleans(), st.floats(min_value=0.0, max_value=10000.0)),
    min_size=1,
    max_size=5
)

# External signals strategy (for FitnessService webhook evaluation)
external_signals = st.dictionaries(
    keys=st.sampled_from([
        'invoice_created',
        'crm_conversion',
        'conversion_success',
        'email_bounce',
        'error_signal',
        'conversion_value'
    ]),
    values=st.one_of(st.booleans(), st.floats(min_value=0.0, max_value=10000.0)),
    min_size=1,
    max_size=6
)

# Workspace settings strategy (for CapabilityGate)
workspace_settings = st.dictionaries(
    keys=st.sampled_from([
        'auto_dev',
        'max_mutations_per_day',
        'max_skill_candidates_per_day',
        'enabled'
    ]),
    values=st.one_of(
        st.booleans(),
        st.integers(min_value=1, max_value=100),
        st.dictionaries(
            keys=st.sampled_from(['enabled', 'memento_skills', 'alpha_evolver', 'background_evolution']),
            values=st.one_of(st.booleans(), st.integers(min_value=1, max_value=100)),
            min_size=1,
            max_size=4
        )
    ),
    min_size=0,
    max_size=4
)

# =============================================================================
# Pytest Fixtures for Auto-Dev Components
# =============================================================================

@pytest.fixture
def event_bus():
    """Provide EventBus instance for testing."""
    from core.auto_dev.event_hooks import EventBus
    bus = EventBus()
    yield bus
    bus.clear()  # Clean up after test


@pytest.fixture
def fitness_service(db_session):
    """Provide FitnessService instance for testing."""
    from core.auto_dev.fitness_service import FitnessService
    return FitnessService(db_session)


@pytest.fixture
def capability_gate(db_session):
    """Provide AutoDevCapabilityService instance for testing."""
    from core.auto_dev.capability_gate import AutoDevCapabilityService
    return AutoDevCapabilityService(db_session)


@pytest.fixture
def container_sandbox():
    """Provide ContainerSandbox instance for testing."""
    from core.auto_dev.container_sandbox import ContainerSandbox
    return ContainerSandbox()


@pytest.fixture
def memento_engine(db_session):
    """Provide MementoEngine instance for testing."""
    from core.auto_dev.memento_engine import MementoEngine
    return MementoEngine(db_session)


@pytest.fixture
def db_session():
    """
    Provide a test database session with automatic rollback.
    This fixture creates a new database session for each test and ensures
    all changes are rolled back after the test completes.
    """
    from core.models import SessionLocal

    session = SessionLocal()
    try:
        yield session
    finally:
        session.rollback()
        session.close()


@pytest.fixture
def sample_agent(db_session):
    """Create a sample agent for testing."""
    from core.models import AgentRegistry
    import uuid

    agent = AgentRegistry(
        id=str(uuid.uuid4()),
        name="test_agent",
        role="assistant",
        maturity="student",
        created_at=datetime.utcnow(),
        updated_at=datetime.utcnow()
    )
    db_session.add(agent)
    db_session.commit()
    return agent


@pytest.fixture
def sample_workspace_settings():
    """Provide sample workspace settings for testing."""
    return {
        "auto_dev": {
            "enabled": True,
            "memento_skills": True,
            "alpha_evolver": True,
            "background_evolution": False,
            "max_mutations_per_day": 10,
            "max_skill_candidates_per_day": 5
        }
    }


# =============================================================================
# Test Helpers
# =============================================================================

def create_task_event(
    episode_id: str = None,
    agent_id: str = None,
    tenant_id: str = None,
    outcome: str = "success"
) -> Any:
    """Helper to create TaskEvent objects for testing."""
    from core.auto_dev.event_hooks import TaskEvent
    import uuid

    return TaskEvent(
        episode_id=episode_id or str(uuid.uuid4()),
        agent_id=agent_id or str(uuid.uuid4()),
        tenant_id=tenant_id or str(uuid.uuid4()),
        task_description="Test task",
        outcome=outcome,
        metadata={"test": True}
    )


def create_skill_execution_event(
    execution_id: str = None,
    agent_id: str = None,
    tenant_id: str = None,
    success: bool = True
) -> Any:
    """Helper to create SkillExecutionEvent objects for testing."""
    from core.auto_dev.event_hooks import SkillExecutionEvent
    import uuid

    return SkillExecutionEvent(
        execution_id=execution_id or str(uuid.uuid4()),
        agent_id=agent_id or str(uuid.uuid4()),
        tenant_id=tenant_id or str(uuid.uuid4()),
        skill_id="test_skill",
        skill_name="Test Skill",
        execution_seconds=1.5,
        token_usage=100,
        success=success,
        output="Test output"
    )


# =============================================================================
# Pytest Configuration
# =============================================================================

def pytest_configure(config):
    """Configure pytest with custom markers."""
    config.addinivalue_line(
        "markers",
        "property: Mark test as property-based test using Hypothesis"
    )
    config.addinivalue_line(
        "markers",
        "slow: Mark test as slow-running (integration tests, Docker-dependent)"
    )
    config.addinivalue_line(
        "markers",
        "docker_required: Mark test as requiring Docker (ContainerSandbox tests)"
    )

	"""
	Property-Based Testing Configuration for Auto-Dev

	This module provides Hypothesis strategies and pytest fixtures for testing
	Auto-Dev components (EventBus, FitnessService, CapabilityGate, ContainerSandbox,
	MementoEngine, Database Models).

	Property-based testing generates hundreds of test cases automatically to verify
	invariants that must ALWAYS hold true.
	"""

	import sys
	import os

	# Add backend to path for imports
	sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../../..'))

	from hypothesis import strategies as st
	import pytest
	from datetime import datetime, timedelta
	from typing import Any

	# =============================================================================
	# Hypothesis Strategies for Auto-Dev Data Types
	# =============================================================================

	# Fitness score strategy (normalized to [0.0, 1.0])
	fitness_scores = st.lists(
	st.floats(min_value=0.0, max_value=1.0, allow_nan=False, allow_infinity=False),
	min_size=0,
	max_size=100
	)

	# Maturity level strategy
	maturity_levels = st.sampled_from(['student', 'intern', 'supervised', 'autonomous'])

	# Event data strategy (random dictionaries with string keys and mixed values)
	event_data = st.dictionaries(
	keys=st.text(min_size=1, max_size=50, alphabet='abcdefghijklmnopqrstuvwxyz'),
	values=st.one_of(
	st.text(),
	st.integers(),
	st.floats(allow_nan=False, allow_infinity=False),
	st.booleans(),
	st.none()
	),
	min_size=0,
	max_size=20
	)

	# Capability strategy
	capabilities = st.sampled_from([
	'file_operations',
	'network_access',
	'system_commands',
	'database_write',
	'api_calls',
	'user_interaction',
	'auto_dev.memento_skills',
	'auto_dev.alpha_evolver',
	'auto_dev.background_evolution'
	])

	# Valid skill name strategy (alphanumeric with underscores)
	valid_skill_names = st.text(min_size=1, max_size=50).filter(
	lambda x: x.replace('_', '').replace('-', '').isalnum() and len(x) > 0
	)

	# Agent ID strategy (UUID-like strings)
	agent_ids = st.text(min_size=36, max_size=36, alphabet='abcdef0123456789-')

	# Tenant ID strategy
	tenant_ids = st.text(min_size=36, max_size=36, alphabet='abcdef0123456789-')

	# Episode ID strategy
	episode_ids = st.text(min_size=36, max_size=36, alphabet='abcdef0123456789-')

	# Proxy signals strategy (for FitnessService)
	proxy_signals = st.dictionaries(
	keys=st.sampled_from([
	'execution_success',
	'syntax_error',
	'execution_latency_ms',
	'user_approved_proposal',
	'expects_delayed_eval'
	]),
	values=st.one_of(st.booleans(), st.floats(min_value=0.0, max_value=10000.0)),
	min_size=1,
	max_size=5
	)

	# External signals strategy (for FitnessService webhook evaluation)
	external_signals = st.dictionaries(
	keys=st.sampled_from([
	'invoice_created',
	'crm_conversion',
	'conversion_success',
	'email_bounce',
	'error_signal',
	'conversion_value'
	]),
	values=st.one_of(st.booleans(), st.floats(min_value=0.0, max_value=10000.0)),
	min_size=1,
	max_size=6
	)

	# Workspace settings strategy (for CapabilityGate)
	workspace_settings = st.dictionaries(
	keys=st.sampled_from([
	'auto_dev',
	'max_mutations_per_day',
	'max_skill_candidates_per_day',
	'enabled'
	]),
	values=st.one_of(
	st.booleans(),
	st.integers(min_value=1, max_value=100),
	st.dictionaries(
	keys=st.sampled_from(['enabled', 'memento_skills', 'alpha_evolver', 'background_evolution']),
	values=st.one_of(st.booleans(), st.integers(min_value=1, max_value=100)),
	min_size=1,
	max_size=4
	)
	),
	min_size=0,
	max_size=4
	)

	# =============================================================================
	# Pytest Fixtures for Auto-Dev Components
	# =============================================================================

	@pytest.fixture
	def event_bus():
	"""Provide EventBus instance for testing."""
	from core.auto_dev.event_hooks import EventBus
	bus = EventBus()
	yield bus
	bus.clear() # Clean up after test


	@pytest.fixture
	def fitness_service(db_session):
	"""Provide FitnessService instance for testing."""
	from core.auto_dev.fitness_service import FitnessService
	return FitnessService(db_session)


	@pytest.fixture
	def capability_gate(db_session):
	"""Provide AutoDevCapabilityService instance for testing."""
	from core.auto_dev.capability_gate import AutoDevCapabilityService
	return AutoDevCapabilityService(db_session)


	@pytest.fixture
	def container_sandbox():
	"""Provide ContainerSandbox instance for testing."""
	from core.auto_dev.container_sandbox import ContainerSandbox
	return ContainerSandbox()


	@pytest.fixture
	def memento_engine(db_session):
	"""Provide MementoEngine instance for testing."""
	from core.auto_dev.memento_engine import MementoEngine
	return MementoEngine(db_session)


	@pytest.fixture
	def db_session():
	"""
	Provide a test database session with automatic rollback.

	This fixture creates a new database session for each test and ensures
	all changes are rolled back after the test completes.
	"""
	from core.models import SessionLocal

	session = SessionLocal()
	try:
	yield session
	finally:
	session.rollback()
	session.close()


	@pytest.fixture
	def sample_agent(db_session):
	"""Create a sample agent for testing."""
	from core.models import AgentRegistry
	import uuid

	agent = AgentRegistry(
	id=str(uuid.uuid4()),
	name="test_agent",
	role="assistant",
	maturity="student",
	created_at=datetime.utcnow(),
	updated_at=datetime.utcnow()
	)
	db_session.add(agent)
	db_session.commit()
	return agent


	@pytest.fixture
	def sample_workspace_settings():
	"""Provide sample workspace settings for testing."""
	return {
	"auto_dev": {
	"enabled": True,
	"memento_skills": True,
	"alpha_evolver": True,
	"background_evolution": False,
	"max_mutations_per_day": 10,
	"max_skill_candidates_per_day": 5
	}
	}


	# =============================================================================
	# Test Helpers
	# =============================================================================

	def create_task_event(
	episode_id: str = None,
	agent_id: str = None,
	tenant_id: str = None,
	outcome: str = "success"
	) -> Any:
	"""Helper to create TaskEvent objects for testing."""
	from core.auto_dev.event_hooks import TaskEvent
	import uuid

	return TaskEvent(
	episode_id=episode_id or str(uuid.uuid4()),
	agent_id=agent_id or str(uuid.uuid4()),
	tenant_id=tenant_id or str(uuid.uuid4()),
	task_description="Test task",
	outcome=outcome,
	metadata={"test": True}
	)


	def create_skill_execution_event(
	execution_id: str = None,
	agent_id: str = None,
	tenant_id: str = None,
	success: bool = True
	) -> Any:
	"""Helper to create SkillExecutionEvent objects for testing."""
	from core.auto_dev.event_hooks import SkillExecutionEvent
	import uuid

	return SkillExecutionEvent(
	execution_id=execution_id or str(uuid.uuid4()),
	agent_id=agent_id or str(uuid.uuid4()),
	tenant_id=tenant_id or str(uuid.uuid4()),
	skill_id="test_skill",
	skill_name="Test Skill",
	execution_seconds=1.5,
	token_usage=100,
	success=success,
	output="Test output"
	)


	# =============================================================================
	# Pytest Configuration
	# =============================================================================

	def pytest_configure(config):
	"""Configure pytest with custom markers."""
	config.addinivalue_line(
	"markers",
	"property: Mark test as property-based test using Hypothesis"
	)
	config.addinivalue_line(
	"markers",
	"slow: Mark test as slow-running (integration tests, Docker-dependent)"
	)
	config.addinivalue_line(
	"markers",
	"docker_required: Mark test as requiring Docker (ContainerSandbox tests)"
	)