Create test_dispatcher.py

test_dispatcher.py

@@ -0,0 +1,332 @@
+# =====================================================
+# Apckeyl Framework
+# Version 1.0
+# test_dispatcher.py
+# =====================================================
+
+"""
+Tests for Apckeyl Dispatcher.
+
+Version 1.0
+"""
+
+from task_manager import TaskManager
+from dispatcher import Dispatcher
+
+
+# =====================================================
+# Test 1 — Dispatcher Exists
+# =====================================================
+
+def test_dispatcher_exists():
+
+    manager = TaskManager()
+
+    dispatcher = Dispatcher(
+        task_manager=manager
+    )
+
+    assert dispatcher is not None
+
+
+# =====================================================
+# Test 2 — Prepare Dispatch
+# =====================================================
+
+def test_prepare_dispatch():
+
+    manager = TaskManager()
+
+    dispatcher = Dispatcher(
+        task_manager=manager
+    )
+
+    task = manager.create_task(
+        task_type="image_upscale"
+    )
+
+    request = dispatcher.prepare_dispatch(
+        task["task_id"]
+    )
+
+    assert request is not None
+
+    assert request["task_id"] == (
+        task["task_id"]
+    )
+
+    assert request["task_type"] == (
+        "image_upscale"
+    )
+
+    assert request["module_id"] == (
+        "realesrgan"
+    )
+
+
+# =====================================================
+# Test 3 — Status Becomes Dispatched
+# =====================================================
+
+def test_dispatch_status():
+
+    manager = TaskManager()
+
+    dispatcher = Dispatcher(
+        task_manager=manager
+    )
+
+    task = manager.create_task(
+        task_type="image_upscale"
+    )
+
+    dispatcher.prepare_dispatch(
+        task["task_id"]
+    )
+
+    updated_task = manager.get_task(
+        task["task_id"]
+    )
+
+    assert updated_task["status"] == (
+        "dispatched"
+    )
+
+
+# =====================================================
+# Test 4 — Payload Preserved
+# =====================================================
+
+def test_payload_preserved():
+
+    manager = TaskManager()
+
+    dispatcher = Dispatcher(
+        task_manager=manager
+    )
+
+    payload = {
+
+        "filename": "test.png",
+
+        "scale": 4,
+    }
+
+    task = manager.create_task(
+
+        task_type="image_upscale",
+
+        payload=payload,
+    )
+
+    request = dispatcher.prepare_dispatch(
+        task["task_id"]
+    )
+
+    assert request["payload"] == (
+        payload
+    )
+
+
+# =====================================================
+# Test 5 — Unknown Task Rejected
+# =====================================================
+
+def test_unknown_task_rejected():
+
+    manager = TaskManager()
+
+    dispatcher = Dispatcher(
+        task_manager=manager
+    )
+
+    try:
+
+        dispatcher.prepare_dispatch(
+            "task_unknown"
+        )
+
+    except KeyError:
+
+        return
+
+    raise AssertionError(
+        "Unknown task was not rejected"
+    )
+
+
+# =====================================================
+# Test 6 — Invalid Status Rejected
+# =====================================================
+
+def test_invalid_status_rejected():
+
+    manager = TaskManager()
+
+    dispatcher = Dispatcher(
+        task_manager=manager
+    )
+
+    task = manager.create_task(
+        task_type="image_upscale"
+    )
+
+    manager.update_status(
+        task["task_id"],
+        "processing"
+    )
+
+    try:
+
+        dispatcher.prepare_dispatch(
+            task["task_id"]
+        )
+
+    except RuntimeError:
+
+        return
+
+    raise AssertionError(
+        "Invalid task status was not rejected"
+    )
+
+
+# =====================================================
+# Test 7 — Queued Task Can Be Dispatched
+# =====================================================
+
+def test_queued_task_dispatch():
+
+    manager = TaskManager()
+
+    dispatcher = Dispatcher(
+        task_manager=manager
+    )
+
+    task = manager.create_task(
+        task_type="image_upscale"
+    )
+
+    manager.update_status(
+        task["task_id"],
+        "queued"
+    )
+
+    request = dispatcher.prepare_dispatch(
+        task["task_id"]
+    )
+
+    assert request["task_id"] == (
+        task["task_id"]
+    )
+
+    updated_task = manager.get_task(
+        task["task_id"]
+    )
+
+    assert updated_task["status"] == (
+        "dispatched"
+    )
+
+
+# =====================================================
+# Test Runner
+# =====================================================
+
+def main():
+
+    tests = [
+
+        test_dispatcher_exists,
+
+        test_prepare_dispatch,
+
+        test_dispatch_status,
+
+        test_payload_preserved,
+
+        test_unknown_task_rejected,
+
+        test_invalid_status_rejected,
+
+        test_queued_task_dispatch,
+    ]
+
+    passed = 0
+
+    failed = 0
+
+    print(
+        "===================================="
+    )
+
+    print(
+        "Apckeyl Dispatcher 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(
+            "Dispatcher Test FAILED"
+        )
+
+    print(
+        "DISPATCHER TEST: PASSED"
+    )
+
+
+# =====================================================
+# Entry Point
+# =====================================================
+
+if __name__ == "__main__":
+
+    main()