test_compute_plane_executor.py

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

"""
Tests for Compute Plane Executor.

Version 1.0
"""

from compute_plane_executor import (
    ComputePlaneExecutor,
)

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
    )

    provider_executor = (
        ComputeProviderExecutor(
            manager
        )
    )

    executor = ComputePlaneExecutor(
        provider_executor
    )

    assert executor is not None

    assert (
        executor.provider_executor
        is provider_executor
    )


# =====================================================
# Test Missing Provider Executor
# =====================================================

def test_executor_requires_provider_executor():

    try:

        ComputePlaneExecutor(
            None
        )

        assert False

    except ValueError as error:

        assert str(error) == (
            "Provider executor is required"
        )


# =====================================================
# Test Execute Task
# =====================================================

def test_execute_task():

    registry = ComputeProviderRegistry()

    manager = ComputeProviderManager(
        registry
    )

    provider = ExampleComputeProvider()

    manager.register_provider(
        provider
    )

    provider_executor = (
        ComputeProviderExecutor(
            manager
        )
    )

    executor = ComputePlaneExecutor(
        provider_executor
    )

    task = {
        "task_id": "compute_plane_test_001",
        "task_type": "example_task",
        "payload": {
            "message": "compute plane test",
        },
    }

    result = executor.execute(
        task
    )

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

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

    assert result["result"] is not None

    assert result["error"] is None


# =====================================================
# Test Error Propagation
# =====================================================

def test_execute_error():

    registry = ComputeProviderRegistry()

    manager = ComputeProviderManager(
        registry
    )

    provider_executor = (
        ComputeProviderExecutor(
            manager
        )
    )

    executor = ComputePlaneExecutor(
        provider_executor
    )

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

    result = executor.execute(
        task
    )

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

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

    assert result["result"] is None

    assert result["error"] is not None


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

def test_execute_none_task():

    registry = ComputeProviderRegistry()

    manager = ComputeProviderManager(
        registry
    )

    provider_executor = (
        ComputeProviderExecutor(
            manager
        )
    )

    executor = ComputePlaneExecutor(
        provider_executor
    )

    result = executor.execute(
        None
    )

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

    assert result["result"] is None


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