lafrite/Bopytex
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

feat: dispatch transmit args to actions

Browse Source
This commit is contained in:
Bertrand Benjamin 2022-04-08 21:35:55 +02:00
parent 480ced3259
commit 4dbfe7a82c
2 changed files with 15 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
bopytex/scheduler.py
View File

@ -20,7 +20,7 @@ class Scheduler:
def dispatch(self, task): def dispatch(self, task):
""" Do a task """ """ Do a task """
ans = self.actions[task.action](task.deps) ans = self.actions[task.action](task.deps, task.args)
return ans return ans
def __next__(self): def __next__(self):

17
test/test_scheduler.py
View File

@ -3,14 +3,15 @@ from bopytex.scheduler import Scheduler
def test_schedule_append(): def test_schedule_append():
actions = {"DO": lambda _: "done"} actions = {"DO": lambda deps, args: "done"}
scheduler = Scheduler(actions) scheduler = Scheduler(actions)
tasks = [Task(action="DO", args={}, deps=[])] tasks = [Task(action="DO", args={}, deps=[])]
scheduler.append(tasks) scheduler.append(tasks)
assert scheduler.tasks == tasks assert scheduler.tasks == tasks
def test_schedule_dispatch(): def test_schedule_dispatch():
actions = {"DO": lambda _: "done"} actions = {"DO": lambda deps, args: "done"}
scheduler = Scheduler(actions) scheduler = Scheduler(actions)
task = Task(action="DO", args={}, deps=[]) task = Task(action="DO", args={}, deps=[])
result = scheduler.dispatch(task) result = scheduler.dispatch(task)
@ -18,10 +19,20 @@ def test_schedule_dispatch():
def test_schedule_one_task(): def test_schedule_one_task():
actions = {"DO": lambda _: "done"} actions = {"DO": lambda deps, args: "done"}
scheduler = Scheduler(actions) scheduler = Scheduler(actions)
scheduler.append([Task(action="DO", args={}, deps=[])]) scheduler.append([Task(action="DO", args={}, deps=[])])
result = scheduler.__next__() result = scheduler.__next__()
assert result == "done" assert result == "done"
assert scheduler.tasks == [] assert scheduler.tasks == []
assert scheduler.done == ["done"] 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"]