Bopytex/test/test_scheduler.py

from bopytex.tasks import Task
from bopytex.scheduler import Scheduler
import pytest


def test_schedule_append():
    scheduler = Scheduler()
    tasks = [
        Task(action="FOO", args={}, deps=["dep1", "dep2"], output="end1"),
        Task(action="FOO", args={}, deps=["dep1", "dep3"], output="end2"),
    ]
    scheduler.append(tasks)
    assert scheduler.tasks == tasks
    assert scheduler.all_deps == {"dep1", "dep2", "dep3"}
    assert scheduler.all_output == {"end1", "end2"}


def test_schedule_one_task():
    scheduler = Scheduler()
    tasks = [Task(action="FOO", args={}, deps=[], output="end")]
    scheduler.append(tasks)

    assert scheduler.doable_tasks == tasks

    result = scheduler.next_task()
    assert result == tasks[0]
    assert scheduler.tasks == []
    assert scheduler.done == ["end"]


def test_schedule_one_task_with_args():
    scheduler = Scheduler()
    tasks = [Task(action="FOO", args={"task": "one"}, deps=[], output="one")]
    scheduler.append(tasks)

    result = scheduler.next_task()

    assert result == tasks[0]
    assert scheduler.tasks == []
    assert scheduler.done == ["one"]


def test_schedule_multiple_tasks():
    scheduler = Scheduler()
    t1 = Task(action="FOO", args={"task": "one"}, deps=[], output="one")
    t2 = Task(action="FOO", args={"task": "two"}, deps=[], output="two")
    t3 = Task(action="FOO", args={"task": "three"}, deps=[], output="three")
    scheduler.append([t1, t2, t3])

    assert scheduler.doable_tasks == [t1, t2, t3]
    assert scheduler.is_finishable()

    result = scheduler.next_task()
    assert result == t1
    assert scheduler.tasks == [t2, t3]
    assert scheduler.done == ["one"]

    result = scheduler.next_task()
    assert result == t2
    assert scheduler.tasks == [t3]
    assert scheduler.done == ["one", "two"]

    result = scheduler.next_task()
    assert result == t3
    assert scheduler.tasks == []
    assert scheduler.done == ["one", "two", "three"]


def test_schedule_multiple_tasks_with_dependencies():
    scheduler = Scheduler()
    t1 = Task(action="FOO", args={"task": "one"}, deps=["three"], output="one")
    t2 = Task(action="FOO", args={"task": "two"}, deps=["one"], output="two")
    t3 = Task(action="FOO", args={"task": "three"}, deps=[], output="three")
    scheduler.append([t1, t2, t3])

    assert scheduler.doable_tasks == [t3]
    assert scheduler.is_finishable()

    result = scheduler.next_task()
    assert result == t3
    assert scheduler.tasks == [t1, t2]
    assert scheduler.done == ["three"]
    assert scheduler.doable_tasks == [t1]

    result = scheduler.next_task()
    assert result == t1
    assert scheduler.tasks == [t2]
    assert scheduler.done == ["three", "one"]
    assert scheduler.doable_tasks == [t2]

    result = scheduler.next_task()
    assert result == t2
    assert scheduler.tasks == []
    assert scheduler.done == ["three", "one", "two"]

def test_schedule_multiple_tasks_with_dependencies_loop():
    scheduler = Scheduler()
    t1 = Task(action="FOO", args={"task": "one"}, deps=["three"], output="one")
    t2 = Task(action="FOO", args={"task": "two"}, deps=["one"], output="two")
    t3 = Task(action="FOO", args={"task": "three"}, deps=[], output="three")
    scheduler.append([t1, t2, t3])

    ordered_tasks = []
    for task in scheduler.backlog():
        ordered_tasks.append(task)
    assert ordered_tasks == [t3, t1, t2]

def test_schedule_empty_task():
    scheduler = Scheduler()
    scheduler.append([])
    with pytest.raises(StopIteration):
        scheduler.next_task()


def test_schedule_multiple_tasks_with_undoable_dependencies():
    scheduler = Scheduler()
    t1 = Task(action="FOO", args={"task": "one"}, deps=["three"], output="one")
    t2 = Task(action="FOO", args={"task": "two"}, deps=[], output="two")
    scheduler.append([t1, t2])

    assert scheduler.doable_tasks == [t2]
    assert not scheduler.is_finishable()

    for _ in scheduler.backlog():
        pass

    assert scheduler.tasks == [t1]
    assert scheduler.done == ["two"]
    assert scheduler.doable_tasks == []