Add Compute Adapter Test Version 1.0

test_compute_adapter.py

@@ -0,0 +1,369 @@
+# =====================================================
+# Apckeyl Framework
+# Version 1.0
+# test_compute_adapter.py
+# =====================================================
+
+"""
+Tests for Apckeyl Compute Adapter.
+
+Version 1.0
+"""
+
+from compute_adapter import (
+    ComputeAdapter,
+    RealESRGANAdapter,
+)
+
+
+# =====================================================
+# Test 1 — Base Adapter Exists
+# =====================================================
+
+def test_base_adapter_exists():
+
+    adapter = ComputeAdapter(
+        module_id="test_module",
+        module_name="Test Module",
+    )
+
+    assert adapter is not None
+
+
+# =====================================================
+# Test 2 — Adapter Information
+# =====================================================
+
+def test_adapter_info():
+
+    adapter = ComputeAdapter(
+
+        module_id="test_module",
+
+        module_name="Test Module",
+
+        endpoint="https://example.com",
+    )
+
+    info = adapter.get_info()
+
+    assert info["module_id"] == (
+        "test_module"
+    )
+
+    assert info["module_name"] == (
+        "Test Module"
+    )
+
+    assert info["endpoint"] == (
+        "https://example.com"
+    )
+
+
+# =====================================================
+# Test 3 — Dispatch Preparation
+# =====================================================
+
+def test_dispatch():
+
+    adapter = ComputeAdapter(
+
+        module_id="test_module",
+
+        module_name="Test Module",
+    )
+
+    task = {
+
+        "task_id": "task_000001",
+
+        "task_type": "image_upscale",
+
+        "payload": {
+
+            "filename": "test.png",
+
+            "scale": 4,
+        },
+    }
+
+    result = adapter.dispatch(
+        task
+    )
+
+    assert result["status"] == (
+        "prepared"
+    )
+
+    assert result["module_id"] == (
+        "test_module"
+    )
+
+    assert result["task_id"] == (
+        "task_000001"
+    )
+
+    assert result["task_type"] == (
+        "image_upscale"
+    )
+
+    assert result["payload"] == (
+        task["payload"]
+    )
+
+
+# =====================================================
+# Test 4 — None Task Rejected
+# =====================================================
+
+def test_none_task_rejected():
+
+    adapter = ComputeAdapter(
+
+        module_id="test_module",
+
+        module_name="Test Module",
+    )
+
+    try:
+
+        adapter.dispatch(None)
+
+    except ValueError:
+
+        return
+
+    raise AssertionError(
+        "None task was not rejected"
+    )
+
+
+# =====================================================
+# Test 5 — Invalid Task Type Rejected
+# =====================================================
+
+def test_invalid_task_type_rejected():
+
+    adapter = ComputeAdapter(
+
+        module_id="test_module",
+
+        module_name="Test Module",
+    )
+
+    try:
+
+        adapter.dispatch(
+            "invalid_task"
+        )
+
+    except TypeError:
+
+        return
+
+    raise AssertionError(
+        "Invalid task type was not rejected"
+    )
+
+
+# =====================================================
+# Test 6 — Task Without ID Rejected
+# =====================================================
+
+def test_missing_task_id_rejected():
+
+    adapter = ComputeAdapter(
+
+        module_id="test_module",
+
+        module_name="Test Module",
+    )
+
+    task = {
+
+        "task_type": "image_upscale",
+
+        "payload": {},
+    }
+
+    try:
+
+        adapter.dispatch(task)
+
+    except ValueError:
+
+        return
+
+    raise AssertionError(
+        "Task without task_id was not rejected"
+    )
+
+
+# =====================================================
+# Test 7 — RealESRGAN Adapter
+# =====================================================
+
+def test_realesrgan_adapter():
+
+    adapter = RealESRGANAdapter(
+
+        endpoint="https://example.com",
+    )
+
+    info = adapter.get_info()
+
+    assert info["module_id"] == (
+        "realesrgan"
+    )
+
+    assert info["module_name"] == (
+        "Apckeyl_RealESRGAN"
+    )
+
+    assert info["endpoint"] == (
+        "https://example.com"
+    )
+
+
+# =====================================================
+# Test 8 — RealESRGAN Dispatch
+# =====================================================
+
+def test_realesrgan_dispatch():
+
+    adapter = RealESRGANAdapter()
+
+    task = {
+
+        "task_id": "task_realesrgan_001",
+
+        "task_type": "image_upscale",
+
+        "payload": {
+
+            "filename": "image.png",
+
+            "scale": 4,
+        },
+    }
+
+    result = adapter.dispatch(
+        task
+    )
+
+    assert result["status"] == (
+        "prepared"
+    )
+
+    assert result["module_id"] == (
+        "realesrgan"
+    )
+
+    assert result["module_name"] == (
+        "Apckeyl_RealESRGAN"
+    )
+
+
+# =====================================================
+# Test Runner
+# =====================================================
+
+def main():
+
+    tests = [
+
+        test_base_adapter_exists,
+
+        test_adapter_info,
+
+        test_dispatch,
+
+        test_none_task_rejected,
+
+        test_invalid_task_type_rejected,
+
+        test_missing_task_id_rejected,
+
+        test_realesrgan_adapter,
+
+        test_realesrgan_dispatch,
+    ]
+
+    passed = 0
+
+    failed = 0
+
+    print(
+        "===================================="
+    )
+
+    print(
+        "Apckeyl Compute Adapter Test"
+    )
+
+    print(
+        "Version 1.0"
+    )
+
+    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()