107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
from bopytex.planner import Task
|
|
from bopytex.scheduler import Scheduler
|
|
import pytest
|
|
|
|
|
|
def test_schedule_append():
|
|
actions = {"DO": lambda deps, args: "done"}
|
|
scheduler = Scheduler(actions)
|
|
tasks = [Task(action="DO", args={}, deps=[])]
|
|
scheduler.append(tasks)
|
|
assert scheduler.tasks == tasks
|
|
|
|
|
|
def test_schedule_dispatch():
|
|
actions = {"DO": lambda deps, args: "done"}
|
|
scheduler = Scheduler(actions)
|
|
task = Task(action="DO", args={}, deps=[])
|
|
result = scheduler.dispatch(task)
|
|
assert result == "done"
|
|
|
|
|
|
def test_schedule_one_task():
|
|
actions = {"DO": lambda deps, args: "done"}
|
|
scheduler = Scheduler(actions)
|
|
scheduler.append([Task(action="DO", args={}, deps=[])])
|
|
result = scheduler.next()
|
|
assert result == "done"
|
|
assert scheduler.tasks == []
|
|
assert scheduler.done == ["done"]
|
|
|
|
|
|
def test_schedule_one_task_with_args():
|
|
actions = {"DO": lambda deps, args: f"{args['task']} done"}
|
|
scheduler = Scheduler(actions)
|
|
scheduler.append([Task(action="DO", args={"task": "one"}, deps=[])])
|
|
result = scheduler.next()
|
|
assert result == "one done"
|
|
assert scheduler.tasks == []
|
|
assert scheduler.done == ["one done"]
|
|
|
|
|
|
def test_schedule_multiple_tasks():
|
|
actions = {"DO": lambda deps, args: f"{args['task']} done"}
|
|
scheduler = Scheduler(actions)
|
|
t1 = Task(action="DO", args={"task": "one"}, deps=[])
|
|
t2 = Task(action="DO", args={"task": "two"}, deps=[])
|
|
t3 = Task(action="DO", args={"task": "three"}, deps=[])
|
|
scheduler.append([t1, t2, t3])
|
|
result = scheduler.next()
|
|
assert result == "one done"
|
|
assert scheduler.tasks == [t2, t3]
|
|
assert scheduler.done == ["one done"]
|
|
|
|
result = scheduler.next()
|
|
assert result == "two done"
|
|
assert scheduler.tasks == [t3]
|
|
assert scheduler.done == ["one done", "two done"]
|
|
|
|
result = scheduler.next()
|
|
assert result == "three done"
|
|
assert scheduler.tasks == []
|
|
assert scheduler.done == ["one done", "two done", "three done"]
|
|
|
|
|
|
def test_schedule_multiple_tasks_with_dependencies():
|
|
actions = {"DO": lambda deps, args: f"{args['task']} done"}
|
|
scheduler = Scheduler(actions)
|
|
t1 = Task(action="DO", args={"task": "one"}, deps=["three done"])
|
|
t2 = Task(action="DO", args={"task": "two"}, deps=["one done"])
|
|
t3 = Task(action="DO", args={"task": "three"}, deps=[])
|
|
scheduler.append([t1, t2, t3])
|
|
|
|
result = scheduler.next()
|
|
assert result == "three done"
|
|
assert scheduler.tasks == [t1, t2]
|
|
assert scheduler.done == ["three done"]
|
|
|
|
result = scheduler.next()
|
|
assert result == "one done"
|
|
assert scheduler.tasks == [t2]
|
|
assert scheduler.done == ["three done", "one done"]
|
|
|
|
result = scheduler.next()
|
|
assert result == "two done"
|
|
assert scheduler.tasks == []
|
|
assert scheduler.done == ["three done", "one done", "two done"]
|
|
|
|
|
|
def test_schedule_empty_task():
|
|
actions = {"DO": lambda deps, args: f"{args['task']} done"}
|
|
scheduler = Scheduler(actions)
|
|
scheduler.append([])
|
|
with pytest.raises(StopIteration):
|
|
scheduler.next()
|
|
|
|
|
|
def test_schedule_multiple_tasks_with_undoable_dependencies():
|
|
actions = {"DO": lambda deps, args: f"{args['task']} done"}
|
|
scheduler = Scheduler(actions)
|
|
t1 = Task(action="DO", args={"task": "one"}, deps=["three done"])
|
|
t2 = Task(action="DO", args={"task": "two"}, deps=[])
|
|
scheduler.append([t1, t2])
|
|
scheduler.run()
|
|
assert scheduler.tasks == [t1]
|
|
assert scheduler.done == ["two done"]
|
|
|