test_compute_provider_contract.py

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

"""
Tests for Compute Provider Contract.

Version 1.0
"""

from example_compute_provider import (
    ExampleComputeProvider,
)


# =====================================================
# Test Provider Interface
# =====================================================

def test_provider_has_get_provider_info():

    provider = ExampleComputeProvider()

    assert callable(
        provider.get_provider_info
    )


def test_provider_has_health_check():

    provider = ExampleComputeProvider()

    assert callable(
        provider.health_check
    )


def test_provider_has_supports_task():

    provider = ExampleComputeProvider()

    assert callable(
        provider.supports_task
    )


def test_provider_has_execute():

    provider = ExampleComputeProvider()

    assert callable(
        provider.execute
    )


# =====================================================
# Test Provider Information Contract
# =====================================================

def test_provider_information_contract():

    provider = ExampleComputeProvider()

    info = provider.get_provider_info()

    required_fields = [
        "provider_id",
        "provider_name",
        "provider_version",
        "status",
        "supported_tasks",
    ]

    for field in required_fields:

        assert field in info


# =====================================================
# Test Health Contract
# =====================================================

def test_health_contract():

    provider = ExampleComputeProvider()

    health = provider.health_check()

    assert "provider_id" in health

    assert "status" in health

    assert health["status"] == "healthy"


# =====================================================
# Test Task Support Contract
# =====================================================

def test_task_support_contract():

    provider = ExampleComputeProvider()

    assert provider.supports_task(
        "example_task"
    ) is True

    assert provider.supports_task(
        "unknown_task"
    ) is False


# =====================================================
# Test Execution Contract
# =====================================================

def test_execution_contract():

    provider = ExampleComputeProvider()

    task = {
        "task_id": "contract_test_001",
        "task_type": "example_task",
        "payload": {
            "message": "contract test",
        },
    }

    result = provider.execute(task)

    required_fields = [
        "task_id",
        "status",
        "result",
        "error",
    ]

    for field in required_fields:

        assert field in result

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

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

    assert result["error"] is None


# =====================================================
# Test Error Contract
# =====================================================

def test_error_contract():

    provider = ExampleComputeProvider()

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

    result = provider.execute(task)

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

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

    assert result["result"] is None

    assert result["error"] is not None


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