Testing

pamiq_core.testing.connect_components

connect_components(
    agent: Agent[Any, Any] | None = None,
    trainers: Trainer | Mapping[str, Trainer] | None = None,
    buffers: Mapping[str, DataBuffer[Any]] | None = None,
    models: Mapping[str, TrainingModel[Any]] | None = None,
) -> ConnectedComponents

Connect PAMIQ Core components for testing or development.

This function wires together the core components (agent, trainers, buffers, models) by establishing the appropriate connection relationships between them. It handles the creation of data users from buffers, extraction of inference models from training models, and proper attachment of all related components.

PARAMETER	DESCRIPTION
agent	Optional agent to connect with models and data collectors TYPE: Agent[Any, Any] | None DEFAULT: None
trainers	Optional trainer or mapping of trainers to connect with models and data TYPE: Trainer | Mapping[str, Trainer] | None DEFAULT: None
buffers	Optional mapping of data buffers to create data users from TYPE: Mapping[str, DataBuffer[Any]] | None DEFAULT: None
models	Optional mapping of training models to connect with trainers and agent TYPE: Mapping[str, TrainingModel[Any]] | None DEFAULT: None

RETURNS	DESCRIPTION
ConnectedComponents	ConnectedComponents containing the connected component dictionaries

Source code in src/pamiq_core/testing.py

def connect_components(
    agent: Agent[Any, Any] | None = None,
    trainers: Trainer | Mapping[str, Trainer] | None = None,
    buffers: Mapping[str, DataBuffer[Any]] | None = None,
    models: Mapping[str, TrainingModel[Any]] | None = None,
) -> ConnectedComponents:
    """Connect PAMIQ Core components for testing or development.

    This function wires together the core components (agent, trainers, buffers, models)
    by establishing the appropriate connection relationships between them. It handles
    the creation of data users from buffers, extraction of inference models from
    training models, and proper attachment of all related components.

    Args:
        agent: Optional agent to connect with models and data collectors
        trainers: Optional trainer or mapping of trainers to connect with models and data
        buffers: Optional mapping of data buffers to create data users from
        models: Optional mapping of training models to connect with trainers and agent

    Returns:
        ConnectedComponents containing the connected component dictionaries
    """
    if buffers is None:
        buffers = {}
    if models is None:
        models = {}

    if isinstance(trainers, Trainer):
        trainers = {"trainer": trainers}

    data_users = DataUsersDict.from_data_buffers(buffers)
    data_collectors = data_users.data_collectors_dict

    training_models = TrainingModelsDict(models)
    inference_models = training_models.inference_models_dict

    if agent is not None:
        agent.attach_data_collectors(data_collectors)
        agent.attach_inference_models(inference_models)

    if trainers is not None:
        trainers_dict = TrainersDict(trainers)
        trainers_dict.attach_data_users(data_users)
        trainers_dict.attach_training_models(training_models)

    return ConnectedComponents(
        data_users=data_users,
        data_collectors=data_collectors,
        training_models=training_models,
        inference_models=inference_models,
    )

pamiq_core.testing.ConnectedComponents

Bases: NamedTuple

Container for connected PAMIQ components.

ATTRIBUTE	DESCRIPTION
data_users	Dictionary of data users created from buffers TYPE: DataUsersDict
data_collectors	Dictionary of data collectors associated with data users TYPE: DataCollectorsDict
training_models	Dictionary of training models TYPE: TrainingModelsDict
inference_models	Dictionary of inference models derived from training models TYPE: InferenceModelsDict

pamiq_core.testing.create_mock_models

create_mock_models(
    has_inference_model: bool = True, inference_thread_only: bool = False
) -> tuple[Mock, Mock]

Create mock training and inference models for testing.

Creates mocked instances of TrainingModel and InferenceModel with the specified configuration, properly connecting them when has_inference_model is True.

PARAMETER	DESCRIPTION
has_inference_model	Whether the training model should have an inference model TYPE: bool DEFAULT: True
inference_thread_only	Whether the model is for inference thread only TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
tuple[Mock, Mock]	A tuple containing (training_model, inference_model) mocks

Source code in src/pamiq_core/testing.py

def create_mock_models(
    has_inference_model: bool = True, inference_thread_only: bool = False
) -> tuple[Mock, Mock]:
    """Create mock training and inference models for testing.

    Creates mocked instances of TrainingModel and InferenceModel with the specified
    configuration, properly connecting them when has_inference_model is True.

    Args:
        has_inference_model: Whether the training model should have an inference model
        inference_thread_only: Whether the model is for inference thread only

    Returns:
        A tuple containing (training_model, inference_model) mocks
    """
    training_model = Mock(TrainingModel)
    inference_model = Mock(InferenceModel)
    training_model.has_inference_model = has_inference_model
    training_model.inference_thread_only = inference_thread_only
    if has_inference_model:
        training_model.inference_model = inference_model
    return training_model, inference_model

pamiq_core.testing.create_mock_buffer

create_mock_buffer(max_size: int = 1) -> MagicMock

Create a mock data buffer for testing.

Creates a MagicMock instance that mocks a DataBuffer with the specified max_size parameter. This is useful for testing components that depend on DataBuffer without implementing a full buffer.

PARAMETER	DESCRIPTION
max_size	Maximum size of the buffer. Defaults to 1. TYPE: int DEFAULT: 1

RETURNS	DESCRIPTION
MagicMock	A MagicMock object that mocks a DataBuffer.

Source code in src/pamiq_core/testing.py

def create_mock_buffer(max_size: int = 1) -> MagicMock:
    """Create a mock data buffer for testing.

    Creates a MagicMock instance that mocks a DataBuffer with the specified
    max_size parameter. This is useful for testing components that depend on
    DataBuffer without implementing a full buffer.

    Args:
        max_size: Maximum size of the buffer. Defaults to 1.

    Returns:
        A MagicMock object that mocks a DataBuffer.
    """
    buf = MagicMock(DataBuffer)
    buf.max_size = max_size
    return buf