130 lines
4.0 KiB
Python
130 lines
4.0 KiB
Python
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 == []
|