beam: add test_task_scheduler.py (11 tests) covering exact-vs-critical-path and validation
This commit is contained in:
1 parent
71cdc6abed
commit
86238b639b
1 file changed
+78
@@ -0,0 +1,78 @@
|
||||
import random
|
||||
from task_scheduler import Task, makespan_exact, makespan_critical_path, _validate
|
||||
|
||||
|
||||
def test_empty():
|
||||
assert makespan_exact([]) == 0
|
||||
assert makespan_critical_path([]) == 0
|
||||
|
||||
|
||||
def test_single():
|
||||
assert makespan_exact([Task("a", 5)]) == 5
|
||||
|
||||
|
||||
def test_chain():
|
||||
assert makespan_exact([Task("a", 3), Task("b", 4, {"a"})]) == 7
|
||||
|
||||
|
||||
def test_parallel():
|
||||
assert makespan_exact([Task("a", 3), Task("b", 4)]) == 4
|
||||
|
||||
|
||||
def test_diamond():
|
||||
tasks = [Task("a", 3), Task("b", 4), Task("c", 2, {"a", "b"})]
|
||||
assert makespan_exact(tasks) == 6
|
||||
assert makespan_exact(tasks) == makespan_critical_path(tasks)
|
||||
|
||||
|
||||
def test_matches_critical_path_random():
|
||||
random.seed(7)
|
||||
for trial in range(500):
|
||||
n = random.randint(0, 8)
|
||||
tids = [f"t{i}" for i in range(n)]
|
||||
tasks = []
|
||||
for i, tid in enumerate(tids):
|
||||
dur = random.randint(1, 6)
|
||||
cand = tids[:i]
|
||||
preds = set(random.sample(cand, random.randint(0, len(cand))))
|
||||
tasks.append(Task(tid, dur, preds))
|
||||
a = makespan_exact(tasks)
|
||||
b = makespan_critical_path(tasks)
|
||||
assert a == b
|
||||
|
||||
|
||||
def test_validation_duplicate():
|
||||
try:
|
||||
_validate([Task("a", 1), Task("a", 2)])
|
||||
except ValueError:
|
||||
return
|
||||
raise AssertionError("expected ValueError")
|
||||
|
||||
|
||||
def test_validation_unknown_pred():
|
||||
try:
|
||||
_validate([Task("a", 1, {"z"})])
|
||||
except ValueError:
|
||||
return
|
||||
raise AssertionError("expected ValueError")
|
||||
|
||||
|
||||
def test_validation_self():
|
||||
try:
|
||||
_validate([Task("a", 1, {"a"})])
|
||||
except ValueError:
|
||||
return
|
||||
raise AssertionError("expected ValueError")
|
||||
|
||||
|
||||
def test_validation_cycle():
|
||||
try:
|
||||
_validate([Task("a", 1, {"b"}), Task("b", 1, {"a"})])
|
||||
except ValueError:
|
||||
return
|
||||
raise AssertionError("expected ValueError")
|
||||
|
||||
|
||||
def test_zero_duration():
|
||||
tasks = [Task("a", 0), Task("b", 2, {"a"})]
|
||||
assert makespan_exact(tasks) == 2
|
||||
Reference in new issue
Block a user