Create test_compute_provider_contract.py

test_compute_provider_contract.py

@@ -0,0 +1,182 @@
+# =====================================================
+# 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
+# =====================================================