test_compute_provider_executor.py

# =====================================================
# Apckeyl Framework
# Version 1.0
# test_compute_provider_executor.py
# =====================================================

"""
Tests for Compute Provider Executor.

Version 1.0
"""

from compute_provider_executor import (
    ComputeProviderExecutor,
)

from compute_provider_manager import (
    ComputeProviderManager,
)

from compute_provider_registry import (
    ComputeProviderRegistry,
)

from example_compute_provider import (
    ExampleComputeProvider,
)


# =====================================================
# Test Executor Creation
# =====================================================

def test_executor_creation():

    registry = ComputeProviderRegistry()

    manager = ComputeProviderManager(
        registry
    )

    executor = ComputeProviderExecutor(
        manager
    )

    assert executor is not None

    assert executor.manager is manager


# =====================================================
# Test Missing Manager
# =====================================================

def test_executor_requires_manager():

    try:

        ComputeProviderExecutor(
            None
        )

        assert False

    except ValueError as error:

        assert str(error) == (
            "Manager is required"
        )


# =====================================================
# Test None Task
# =====================================================

def test_execute_none_task():

    registry = ComputeProviderRegistry()

    manager = ComputeProviderManager(
        registry
    )

    executor = ComputeProviderExecutor(
        manager
    )

    result = executor.execute(
        None
    )

    assert result["task_id"] is None

    assert result["status"] == "error"

    assert result["result"] is None

    assert result["error"] == (
        "Task is required"
    )


# =====================================================
# Test Missing Task Type
# =====================================================

def test_execute_missing_task_type():

    registry = ComputeProviderRegistry()

    manager = ComputeProviderManager(
        registry
    )

    executor = ComputeProviderExecutor(
        manager
    )

    task = {
        "task_id": "executor_test_001",
        "payload": {},
    }

    result = executor.execute(
        task
    )

    assert result["task_id"] == (
        "executor_test_001"
    )

    assert result["status"] == "error"

    assert result["result"] is None

    assert result["error"] == (
        "Missing task_type"
    )


# =====================================================
# Test No Provider
# =====================================================

def test_execute_without_provider():

    registry = ComputeProviderRegistry()

    manager = ComputeProviderManager(
        registry
    )

    executor = ComputeProviderExecutor(
        manager
    )

    task = {
        "task_id": "executor_test_002",
        "task_type": "unknown_task",
        "payload": {},
    }

    result = executor.execute(
        task
    )

    assert result["task_id"] == (
        "executor_test_002"
    )

    assert result["status"] == "error"

    assert result["result"] is None

    assert result["error"] == (
        "No provider available "
        "for task type: unknown_task"
    )


# =====================================================
# Test Successful Execution
# =====================================================

def test_execute_success():

    registry = ComputeProviderRegistry()

    manager = ComputeProviderManager(
        registry
    )

    provider = ExampleComputeProvider()

    manager.register_provider(
        provider
    )

    executor = ComputeProviderExecutor(
        manager
    )

    task = {
        "task_id": "executor_test_003",
        "task_type": "example_task",
        "payload": {
            "message": "executor test",
        },
    }

    result = executor.execute(
        task
    )

    assert result["task_id"] == (
        "executor_test_003"
    )

    assert result["status"] == "completed"

    assert result["result"] is not None

    assert result["error"] is None


# =====================================================
# End of File
# =====================================================