test_orchestrator_core.py

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

"""
Test for Orchestrator Core.

Проверяет базовый жизненный цикл Task.
"""

from orchestrator_core import (
    OrchestratorCore,
    TaskStatus,
    TaskPriority,
)


def main():

    print()
    print("====================================")
    print("Apckeyl Orchestrator Core Test")
    print("Version 1.0")
    print("====================================")
    print()

    # -------------------------------------------------
    # 1. Create Core
    # -------------------------------------------------

    core = OrchestratorCore()

    assert core.task_count() == 0
    assert core.queue_size() == 0

    print("TEST 1: Core initialization ........ OK")

    # -------------------------------------------------
    # 2. Create Task
    # -------------------------------------------------

    task = core.create_task(

        task_type="TEST",

        input_data="Hello Orchestrator",

        parameters={
            "test": True,
        },

        priority=TaskPriority.NORMAL,

    )

    assert task.status == TaskStatus.CREATED
    assert core.task_count() == 1

    print("TEST 2: Task creation .............. OK")

    # -------------------------------------------------
    # 3. Queue Task
    # -------------------------------------------------

    core.queue_task(
        task.task_id
    )

    assert task.status == TaskStatus.QUEUED
    assert core.queue_size() == 1

    print("TEST 3: Queue ...................... OK")

    # -------------------------------------------------
    # 4. Get Next Task
    # -------------------------------------------------

    next_task = core.get_next_task()

    assert next_task is not None
    assert next_task.task_id == task.task_id
    assert core.queue_size() == 0

    print("TEST 4: Queue retrieval ............ OK")

    # -------------------------------------------------
    # 5. Assign Worker
    # -------------------------------------------------

    task.status = TaskStatus.QUEUED

    core.assign_worker(

        task.task_id,

        "TEST_WORKER",

    )

    assert task.status == TaskStatus.ASSIGNED
    assert task.worker == "TEST_WORKER"

    print("TEST 5: Worker assignment .......... OK")

    # -------------------------------------------------
    # 6. Start Task
    # -------------------------------------------------

    core.start_task(
        task.task_id
    )

    assert task.status == TaskStatus.RUNNING
    assert task.started_at is not None

    print("TEST 6: Task start ................ OK")

    # -------------------------------------------------
    # 7. Complete Task
    # -------------------------------------------------

    core.complete_task(

        task.task_id,

        result="TEST_OK",

    )

    assert task.status == TaskStatus.COMPLETED
    assert task.result == "TEST_OK"
    assert task.completed_at is not None

    print("TEST 7: Task completion ........... OK")

    # -------------------------------------------------
    # 8. Final Status
    # -------------------------------------------------

    status = core.get_status(
        task.task_id
    )

    assert status == TaskStatus.COMPLETED

    print("TEST 8: Final status .............. OK")

    # -------------------------------------------------
    # Result
    # -------------------------------------------------

    print()
    print("------------------------------------")
    print("ALL ORCHESTRATOR CORE TESTS PASSED")
    print("------------------------------------")
    print()

    print(
        f"Task ID: {task.task_id}"
    )

    print(
        f"Task Type: {task.task_type}"
    )

    print(
        f"Worker: {task.worker}"
    )

    print(
        f"Status: {task.status.value}"
    )

    print(
        f"Result: {task.result}"
    )

    print()


if __name__ == "__main__":

    main()