test_compute_adapter.py

# =====================================================
# Apckeyl Framework
# Version 1.1
# test_compute_adapter.py
# =====================================================

"""
Tests for Compute Adapter.

Version 1.1
"""

from compute_adapter import (
    ComputeAdapter,
    create_compute_adapter,
)

from compute_module import (
    ExampleComputeModule,
)


# =====================================================
# Test 1 — Adapter Creation
# =====================================================

def test_adapter_creation():

    module = ExampleComputeModule()

    adapter = ComputeAdapter(
        module
    )

    assert adapter is not None


# =====================================================
# Test 2 — Adapter Factory
# =====================================================

def test_adapter_factory():

    module = ExampleComputeModule()

    adapter = create_compute_adapter(
        module
    )

    assert isinstance(
        adapter,
        ComputeAdapter,
    )


# =====================================================
# Test 3 — Module Information
# =====================================================

def test_module_info():

    module = ExampleComputeModule()

    adapter = ComputeAdapter(
        module
    )

    info = adapter.get_module_info()

    assert info["module_id"] == (
        "example"
    )

    assert info["module_name"] == (
        "Example Compute Module"
    )

    assert info["module_version"] == (
        "1.0"
    )

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


# =====================================================
# Test 4 — Health Check
# =====================================================

def test_health_check():

    module = ExampleComputeModule()

    adapter = ComputeAdapter(
        module
    )

    health = adapter.health_check()

    assert health["module_id"] == (
        "example"
    )

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


# =====================================================
# Test 5 — Supported Task
# =====================================================

def test_supported_task():

    module = ExampleComputeModule()

    adapter = ComputeAdapter(
        module
    )

    assert adapter.supports_task(
        "example_task"
    )


# =====================================================
# Test 6 — Unsupported Task
# =====================================================

def test_unsupported_task():

    module = ExampleComputeModule()

    adapter = ComputeAdapter(
        module
    )

    assert not adapter.supports_task(
        "image_upscale"
    )


# =====================================================
# Test 7 — Execute Task
# =====================================================

def test_execute_task():

    module = ExampleComputeModule()

    adapter = ComputeAdapter(
        module
    )

    task = {

        "task_id": "adapter_001",

        "task_type": "example_task",

        "payload": {

            "message": "hello",

        },
    }

    result = adapter.execute(
        task
    )

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

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

    assert "result" in result


# =====================================================
# Test 8 — None Task Rejected
# =====================================================

def test_none_task_rejected():

    module = ExampleComputeModule()

    adapter = ComputeAdapter(
        module
    )

    try:

        adapter.execute(None)

    except ValueError:

        return

    raise AssertionError(
        "None task was not rejected"
    )


# =====================================================
# Test 9 — Invalid Task Rejected
# =====================================================

def test_invalid_task_rejected():

    module = ExampleComputeModule()

    adapter = ComputeAdapter(
        module
    )

    try:

        adapter.execute(
            "invalid_task"
        )

    except TypeError:

        return

    raise AssertionError(
        "Invalid task was not rejected"
    )


# =====================================================
# Test 10 — Missing Task ID
# =====================================================

def test_missing_task_id():

    module = ExampleComputeModule()

    adapter = ComputeAdapter(
        module
    )

    task = {

        "task_type": "example_task",

    }

    try:

        adapter.execute(
            task
        )

    except ValueError:

        return

    raise AssertionError(
        "Task without task_id "
        "was not rejected"
    )


# =====================================================
# Test 11 — Missing Task Type
# =====================================================

def test_missing_task_type():

    module = ExampleComputeModule()

    adapter = ComputeAdapter(
        module
    )

    task = {

        "task_id": "adapter_002",

    }

    try:

        adapter.execute(
            task
        )

    except ValueError:

        return

    raise AssertionError(
        "Task without task_type "
        "was not rejected"
    )


# =====================================================
# Test 12 — Unsupported Task Rejected
# =====================================================

def test_unsupported_task_rejected():

    module = ExampleComputeModule()

    adapter = ComputeAdapter(
        module
    )

    task = {

        "task_id": "adapter_003",

        "task_type": (
            "unsupported_task"
        ),

    }

    try:

        adapter.execute(
            task
        )

    except ValueError:

        return

    raise AssertionError(
        "Unsupported task "
        "was not rejected"
    )


# =====================================================
# Test 13 — Adapter Requires Module
# =====================================================

def test_adapter_requires_module():

    try:

        ComputeAdapter(
            None
        )

    except ValueError:

        return

    raise AssertionError(
        "Adapter accepted None module"
    )


# =====================================================
# Test 14 — Adapter Requires Correct Module Type
# =====================================================

def test_adapter_requires_correct_module_type():

    try:

        ComputeAdapter(
            "invalid_module"
        )

    except TypeError:

        return

    raise AssertionError(
        "Adapter accepted invalid "
        "module type"
    )


# =====================================================
# Test Runner
# =====================================================

def main():

    tests = [

        test_adapter_creation,

        test_adapter_factory,

        test_module_info,

        test_health_check,

        test_supported_task,

        test_unsupported_task,

        test_execute_task,

        test_none_task_rejected,

        test_invalid_task_rejected,

        test_missing_task_id,

        test_missing_task_type,

        test_unsupported_task_rejected,

        test_adapter_requires_module,

        test_adapter_requires_correct_module_type,

    ]

    passed = 0

    failed = 0

    print(
        "===================================="
    )

    print(
        "Apckeyl Compute Adapter Test"
    )

    print(
        "Version 1.1"
    )

    print(
        "===================================="
    )

    for test in tests:

        try:

            test()

            print(
                f"PASS: {test.__name__}"
            )

            passed += 1

        except Exception as error:

            print(
                f"FAIL: {test.__name__}"
            )

            print(
                f"ERROR: {error}"
            )

            failed += 1

    print(
        "===================================="
    )

    print(
        f"PASSED: {passed}"
    )

    print(
        f"FAILED: {failed}"
    )

    print(
        "===================================="
    )

    if failed > 0:

        raise RuntimeError(
            "Compute Adapter Test FAILED"
        )

    print(
        "COMPUTE ADAPTER TEST: PASSED"
    )


# =====================================================
# Entry Point
# =====================================================

if __name__ == "__main__":

    main()