Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
164 changes: 144 additions & 20 deletions pyosmo/config.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import logging
from random import Random, randint
from typing import Any

from pyosmo.algorithm import RandomAlgorithm
from pyosmo.algorithm.base import OsmoAlgorithm
Expand All @@ -14,6 +15,99 @@
DEFAULT_SEED_MAX = 10000


class ConfigurationError(ValueError):
"""Raised when configuration validation fails."""

pass


class ConfigValidator:
"""Validates configuration values with comprehensive error messages."""

@staticmethod
def validate_algorithm(algorithm: Any) -> None:
"""Validate algorithm configuration.

Args:
algorithm: Algorithm to validate

Raises:
ConfigurationError: If algorithm is invalid
"""
if algorithm is None:
raise ConfigurationError('Algorithm cannot be None. Please provide a valid OsmoAlgorithm instance.')

if not isinstance(algorithm, OsmoAlgorithm):
raise ConfigurationError(
f'Algorithm must be an instance of OsmoAlgorithm, '
f'got {type(algorithm).__name__}. '
f'Available algorithms: RandomAlgorithm, WeightedAlgorithm, BalancingAlgorithm.'
)

@staticmethod
def validate_end_condition(condition: Any, name: str = 'End condition') -> None:
"""Validate end condition configuration.

Args:
condition: End condition to validate
name: Name of the condition for error messages

Raises:
ConfigurationError: If end condition is invalid
"""
if condition is None:
raise ConfigurationError(f'{name} cannot be None. Please provide a valid OsmoEndCondition instance.')

if not isinstance(condition, OsmoEndCondition):
raise ConfigurationError(
f'{name} must be an instance of OsmoEndCondition, '
f'got {type(condition).__name__}. '
f'Available conditions: Length, Time, StepCoverage, Endless, And, Or.'
)

@staticmethod
def validate_error_strategy(strategy: Any, name: str = 'Error strategy') -> None:
"""Validate error strategy configuration.

Args:
strategy: Error strategy to validate
name: Name of the strategy for error messages

Raises:
ConfigurationError: If error strategy is invalid
"""
if strategy is None:
raise ConfigurationError(f'{name} cannot be None. Please provide a valid OsmoErrorStrategy instance.')

if not isinstance(strategy, OsmoErrorStrategy):
raise ConfigurationError(
f'{name} must be an instance of OsmoErrorStrategy, '
f'got {type(strategy).__name__}. '
f'Available strategies: AlwaysRaise, AlwaysIgnore, IgnoreAsserts, AllowCount.'
)

@staticmethod
def validate_seed(seed: Any) -> None:
"""Validate random seed value.

Args:
seed: Seed value to validate

Raises:
ConfigurationError: If seed is invalid
"""
if not isinstance(seed, int):
raise ConfigurationError(f'Seed must be an integer, got {type(seed).__name__}.')

if seed < 0:
raise ConfigurationError(f'Seed must be non-negative, got {seed}.')

if seed > 2**32 - 1:
raise ConfigurationError(
f'Seed must fit in 32 bits (max {2**32 - 1}), got {seed}. Use a smaller seed value for reproducibility.'
)


class OsmoConfig:
"""Osmo run configuration object"""

Expand All @@ -35,52 +129,82 @@ def algorithm(self) -> OsmoAlgorithm:
return self._algorithm

@algorithm.setter
def algorithm(self, value: OsmoAlgorithm):
"""Set test generation algorithm"""
if not isinstance(value, OsmoAlgorithm):
raise AttributeError('algorithm needs to be OsmoAlgorithm')
def algorithm(self, value: OsmoAlgorithm) -> None:
"""Set test generation algorithm with validation.

Args:
value: Algorithm instance

Raises:
ConfigurationError: If algorithm is invalid
"""
ConfigValidator.validate_algorithm(value)
self._algorithm = value

@property
def test_end_condition(self) -> OsmoEndCondition:
return self._test_end_condition

@test_end_condition.setter
def test_end_condition(self, value: OsmoEndCondition):
"""Set test generation test_end_condition"""
if not isinstance(value, OsmoEndCondition):
raise AttributeError('test_end_condition needs to be OsmoEndCondition')
def test_end_condition(self, value: OsmoEndCondition) -> None:
"""Set test end condition with validation.

Args:
value: End condition instance

Raises:
ConfigurationError: If end condition is invalid
"""
ConfigValidator.validate_end_condition(value, 'Test end condition')
self._test_end_condition = value

@property
def test_suite_end_condition(self) -> OsmoEndCondition:
return self._test_suite_end_condition

@test_suite_end_condition.setter
def test_suite_end_condition(self, value: OsmoEndCondition):
"""Set test generation test_suite_end_condition"""
if not isinstance(value, OsmoEndCondition):
raise AttributeError('test_suite_end_condition needs to be OsmoEndCondition')
def test_suite_end_condition(self, value: OsmoEndCondition) -> None:
"""Set test suite end condition with validation.

Args:
value: End condition instance

Raises:
ConfigurationError: If end condition is invalid
"""
ConfigValidator.validate_end_condition(value, 'Test suite end condition')
self._test_suite_end_condition = value

@property
def test_error_strategy(self) -> OsmoErrorStrategy:
return self._test_error_strategy

@test_error_strategy.setter
def test_error_strategy(self, value: OsmoErrorStrategy):
"""Set test generation test_suite_end_condition"""
if not isinstance(value, OsmoErrorStrategy):
raise AttributeError('test_error_strategy needs to be OsmoErrorStrategy')
def test_error_strategy(self, value: OsmoErrorStrategy) -> None:
"""Set test error strategy with validation.

Args:
value: Error strategy instance

Raises:
ConfigurationError: If error strategy is invalid
"""
ConfigValidator.validate_error_strategy(value, 'Test error strategy')
self._test_error_strategy = value

@property
def test_suite_error_strategy(self) -> OsmoErrorStrategy:
return self._test_suite_error_strategy

@test_suite_error_strategy.setter
def test_suite_error_strategy(self, value: OsmoErrorStrategy):
"""Set test generation test_suite_end_condition"""
if not isinstance(value, OsmoErrorStrategy):
raise AttributeError('test_suite_error_strategy needs to be OsmoErrorStrategy')
def test_suite_error_strategy(self, value: OsmoErrorStrategy) -> None:
"""Set test suite error strategy with validation.

Args:
value: Error strategy instance

Raises:
ConfigurationError: If error strategy is invalid
"""
ConfigValidator.validate_error_strategy(value, 'Test suite error strategy')
self._test_suite_error_strategy = value
18 changes: 18 additions & 0 deletions pyosmo/discovery/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
"""Model discovery strategies for PyOsmo.

This module provides extensible discovery mechanisms for finding test steps,
guards, and weights in model classes.
"""

from pyosmo.discovery.base import DiscoveryStrategy, ModelMetadata
from pyosmo.discovery.decorator import DecoratorBasedDiscovery
from pyosmo.discovery.naming import NamingConventionDiscovery
from pyosmo.discovery.orchestrator import ModelDiscovery

__all__ = [
'DiscoveryStrategy',
'ModelMetadata',
'DecoratorBasedDiscovery',
'NamingConventionDiscovery',
'ModelDiscovery',
]
67 changes: 67 additions & 0 deletions pyosmo/discovery/base.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,67 @@
"""Base classes for model discovery strategies."""

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any


@dataclass
class StepMetadata:
"""Metadata for a discovered test step."""

name: str # Step name (without 'step_' prefix)
function_name: str # Full function name
method: Any # The actual method object
is_decorator_based: bool = False
metadata: dict[str, Any] = field(default_factory=dict)


@dataclass
class ModelMetadata:
"""Complete metadata for a discovered model."""

steps: list[StepMetadata]
model: object

def get_step_names(self) -> list[str]:
"""Get list of step names."""
return [step.name for step in self.steps]

def get_step_by_name(self, name: str) -> StepMetadata | None:
"""Get step metadata by name."""
for step in self.steps:
if step.name == name:
return step
return None


class DiscoveryStrategy(ABC):
"""Base class for model discovery strategies.

Discovery strategies are responsible for finding test steps,
guards, and weights in model classes using different mechanisms
(naming conventions, decorators, annotations, etc.).
"""

@abstractmethod
def discover_steps(self, model: object) -> list[StepMetadata]:
"""Discover test steps in the model.

Args:
model: Model instance to inspect

Returns:
List of discovered step metadata
"""
pass

def get_priority(self) -> int:
"""Get priority for this discovery strategy.

Lower numbers = higher priority (checked first).
Default is 100.

Returns:
Priority value
"""
return 100
51 changes: 51 additions & 0 deletions pyosmo/discovery/decorator.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
"""Decorator-based discovery strategy."""

import inspect

from pyosmo.discovery.base import DiscoveryStrategy, StepMetadata


class DecoratorBasedDiscovery(DiscoveryStrategy):
"""Discover steps via @step, @guard decorators.

This strategy finds methods decorated with @step and extracts
their metadata. It has higher priority than naming convention
to allow explicit override of naming-based discovery.
"""

def discover_steps(self, model: object) -> list[StepMetadata]:
"""Discover steps via decorators.

Args:
model: Model instance to inspect

Returns:
List of step metadata for decorator-based steps
"""
steps = []

for attr_name, method in inspect.getmembers(model, predicate=callable):
# Skip private/protected methods
if attr_name.startswith('_'):
continue

# Check for @step decorator
if hasattr(method, '_osmo_step'):
step_name = getattr(method, '_osmo_step_name', attr_name)
metadata = getattr(method, '_osmo_metadata', {})

steps.append(
StepMetadata(
name=step_name,
function_name=attr_name,
method=method,
is_decorator_based=True,
metadata=metadata,
)
)

return steps

def get_priority(self) -> int:
"""Decorator-based discovery has high priority (10)."""
return 10
Loading