test_example_compute_provider.py

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

"""
Tests for Example Compute Provider.

Version 1.0
"""

from example_compute_provider import (
    ExampleComputeProvider,
)


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

def test_get_provider_info():

    provider = ExampleComputeProvider()

    info = provider.get_provider_info()

    assert info["provider_id"] == (
        "example_compute_provider"
    )

    assert info["provider_name"] == (
        "Example Compute Provider"
    )

    assert info["provider_version"] == "1.0"

    assert info["status"] == "available"

    assert "example_task" in (
        info["supported_tasks"]
    )


# =====================================================
# Test Health Check
# =====================================================

def test_health_check():

    provider = ExampleComputeProvider()

    health = provider.health_check()

    assert health["provider_id"] == (
        "example_compute_provider"
    )

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


# =====================================================
# Test Supported Task
# =====================================================

def test_supports_task():

    provider = ExampleComputeProvider()

    assert provider.supports_task(
        "example_task"
    ) is True

    assert provider.supports_task(
        "unknown_task"
    ) is False


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

def test_execute_success():

    provider = ExampleComputeProvider()

    task = {
        "task_id": "task_001",
        "task_type": "example_task",
        "payload": {
            "message": "hello"
        },
    }

    result = provider.execute(task)

    assert result["task_id"] == "task_001"

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

    assert result["error"] is None

    assert result["result"]["message"] == "hello"

    assert result["result"]["provider"] == (
        "example_compute_provider"
    )

    assert result["result"]["processed"] is True


# =====================================================
# Test Missing Task ID
# =====================================================

def test_execute_missing_task_id():

    provider = ExampleComputeProvider()

    task = {
        "task_type": "example_task",
        "payload": {
            "message": "hello"
        },
    }

    result = provider.execute(task)

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

    assert result["result"] is None

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


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

def test_execute_missing_task_type():

    provider = ExampleComputeProvider()

    task = {
        "task_id": "task_002",
        "payload": {
            "message": "hello"
        },
    }

    result = provider.execute(task)

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

    assert result["result"] is None

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


# =====================================================
# Test Unsupported Task
# =====================================================

def test_execute_unsupported_task():

    provider = ExampleComputeProvider()

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

    result = provider.execute(task)

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

    assert result["result"] is None

    assert result["error"] == (
        "Unsupported task type: unknown_task"
    )


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