601 lines
22 KiB
Python
601 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
spec_schedkit.py -- executable specification for schedkit.py.
|
|
|
|
Exit 0 == every contract held.
|
|
Exit 1 == at least one contract failed (the failure is printed).
|
|
|
|
Contract classes
|
|
A. fixture cases hand-computed waves, printed in full
|
|
B. invariant checks every output re-derived from first principles
|
|
C. oracle check find_cycles against an independent brute-force enumeration
|
|
D. randomized 1,000 DAGs x 40 permutations, checked against the oracle
|
|
E. determinism repeated calls, build-order shuffles, deep graphs
|
|
F. cycle handling exact CycleError messages, DOT/JSON refusal
|
|
G. scale 50,000 tasks, 100,000 edges, under a wall-clock budget
|
|
H. cost shape acyclic graphs must be linear; a 50,000-node ring must
|
|
be walked iteratively (the recursive build died at 2,000)
|
|
I. linear extension topological_order() must return the lexicographically
|
|
SMALLEST topological order -- the documented "min-id
|
|
first" contract -- checked against brute force over all
|
|
permutations, not against the module's own heap
|
|
|
|
The randomized section is the teeth: it compares this implementation against a
|
|
*different* algorithm (brute-force cycle enumeration over permutations), not
|
|
against itself. A tautological test suite cannot pass this file.
|
|
"""
|
|
|
|
import itertools
|
|
import json
|
|
import os
|
|
import random
|
|
import sys
|
|
import time
|
|
from typing import Dict, List, Sequence, Set, Tuple
|
|
|
|
# Import the schedkit *beside this file*, not one from a hard-coded path: the
|
|
# mutation harness copies both files into a scratch dir and mutates the copy.
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
import schedkit # noqa: E402
|
|
from schedkit import CycleError, Graph, find_cycles, iter_plan, plan # noqa: E402
|
|
|
|
FAILS: List[str] = []
|
|
CHECKS = 0
|
|
T_START = time.monotonic()
|
|
|
|
|
|
def check(name: str, got, want) -> None:
|
|
global CHECKS
|
|
CHECKS += 1
|
|
if got != want:
|
|
FAILS.append("%s\n got : %r\n want: %r" % (name, got, want))
|
|
|
|
|
|
def check_true(name: str, cond: bool, detail: str = "") -> None:
|
|
global CHECKS
|
|
CHECKS += 1
|
|
if not cond:
|
|
FAILS.append("%s %s" % (name, detail))
|
|
|
|
|
|
def section(title: str) -> None:
|
|
print("\n== %s" % title)
|
|
|
|
|
|
# =========================================================== A. fixture cases
|
|
def canon_ids(cyc: Sequence[str]) -> Tuple[str, ...]:
|
|
"""Rotate a cycle so its lexicographically smallest id comes first.
|
|
|
|
Mirrors schedkit._canon, but written here from the *contract* (a cycle is a
|
|
ring, so any rotation names it) rather than copied from the module.
|
|
"""
|
|
if not cyc:
|
|
return tuple(cyc)
|
|
i = min(range(len(cyc)), key=lambda k: (cyc[k], k))
|
|
return tuple(cyc[i:]) + tuple(cyc[:i])
|
|
|
|
|
|
def build(edges, extra_tasks=()):
|
|
g = Graph()
|
|
for t in extra_tasks:
|
|
g.add_task(t)
|
|
return g.add_edges(edges)
|
|
|
|
|
|
def fixtures() -> None:
|
|
section("A. fixtures")
|
|
|
|
# A1: diamond. a -> b, a -> c, b -> d, c -> d
|
|
g = build([("a", "b"), ("a", "c"), ("b", "d"), ("c", "d")])
|
|
check("A1 diamond waves", plan(g), (("a",), ("b", "c"), ("d",)))
|
|
|
|
# A2: independent chains do not wait on each other
|
|
g = build([("a1", "a2"), ("b1", "b2"), ("a2", "a3")])
|
|
check("A2 two chains", plan(g), (("a1", "b1"), ("a2", "b2"), ("a3",)))
|
|
|
|
# A3: empty graph
|
|
check("A3 empty", plan(Graph()), ())
|
|
|
|
# A4: single task, no deps
|
|
check("A4 single", plan(build([], ["solo"])), (("solo",),))
|
|
|
|
# A5: lexicographic order inside a wave (insertion order is reversed)
|
|
g = Graph()
|
|
for t in ["zeta", "alpha", "mike", "bravo"]:
|
|
g.add_task(t)
|
|
check("A5 wave order", plan(g), (("alpha", "bravo", "mike", "zeta"),))
|
|
|
|
# A6: a long chain gives one wave per task
|
|
g = build([(chr(97 + i), chr(98 + i)) for i in range(5)])
|
|
check("A6 chain of 6", plan(g), tuple((c,) for c in "abcdef"))
|
|
|
|
# A7: wave depth is prerequisite-depth, not "connected component"
|
|
g = build([("root", "x"), ("root", "y"), ("x", "z"), ("y", "z"), ("unrelated", "u2")])
|
|
check("A7 depth", plan(g), (("root", "unrelated"), ("u2", "x", "y"), ("z",)))
|
|
|
|
# A8: the graph is untouched by planning (no mutation)
|
|
g = build([("a", "b"), ("b", "c")])
|
|
before = g.edges()
|
|
plan(g)
|
|
check("A8 no mutation", g.edges(), before)
|
|
|
|
# A9: duplicate edge is idempotent
|
|
g = Graph()
|
|
g.add_edge("a", "b").add_edge("a", "b").add_edge("a", "b")
|
|
check("A9 idempotent edges", g.edges(), (("a", "b"),))
|
|
|
|
# A10: self-loop is a cycle, not a crash
|
|
g = Graph()
|
|
g.add_task("self", ["self"])
|
|
check("A10 self loop", find_cycles(g), (("self",),))
|
|
|
|
|
|
# ====================================================== B. invariant checks
|
|
def check_invariants(name: str, g: Graph) -> None:
|
|
waves = plan(g)
|
|
flat = tuple(iter_plan(waves))
|
|
|
|
# B1 partition: every task exactly once, in the graph
|
|
check("%s: partition" % name, tuple(sorted(flat)), g.tasks())
|
|
|
|
# B2 waves sorted ascending
|
|
for i, w in enumerate(waves):
|
|
check("%s: wave %d sorted" % (name, i), list(w), sorted(w))
|
|
|
|
# B3 acyclicity of the emitted plan order
|
|
pos = {t: i for i, t in enumerate(flat)}
|
|
for pre, post in g.edges():
|
|
check_true(
|
|
"%s: order respects %s->%s" % (name, pre, post),
|
|
pos[pre] < pos[post],
|
|
"(pos %d !< %d)" % (pos[pre], pos[post]),
|
|
)
|
|
|
|
# B4 wave of t == 1 + max(wave of deps); 0 if no deps
|
|
rank = {t: i for i, w in enumerate(waves) for t in w}
|
|
for t in g.tasks():
|
|
want = 1 + max((rank[d] for d in g.deps(t)), default=-1)
|
|
check("%s: rank(%s)" % (name, t), rank[t], want)
|
|
|
|
# B5 waves are contiguous and non-empty
|
|
check_true("%s: non-empty waves" % name, all(waves), str(waves))
|
|
|
|
# B6 plan_json is valid JSON and agrees with plan()
|
|
doc = json.loads(schedkit.plan_json(g))
|
|
check("%s: json waves" % name, tuple(tuple(w) for w in doc["waves"]), waves)
|
|
check("%s: json n_tasks" % name, doc["validation"]["n_tasks"], len(g))
|
|
|
|
|
|
def invariants() -> None:
|
|
section("B. invariants")
|
|
cases = {
|
|
"diamond": build([("a", "b"), ("a", "c"), ("b", "d"), ("c", "d")]),
|
|
"star": build([("h", "a"), ("h", "b"), ("h", "c"), ("h", "d")]),
|
|
"empty": Graph(),
|
|
"single": build([], ["only"]),
|
|
"mesh": build(
|
|
[("a", "b"), ("a", "c"), ("b", "d"), ("c", "d"), ("a", "d"), ("b", "c")]
|
|
),
|
|
"two-roots": build([("r1", "x"), ("r2", "x"), ("x", "y")]),
|
|
"fan-out-then-fan-in": build(
|
|
[("r", "a"), ("r", "b"), ("r", "c"), ("a", "z"), ("b", "z"), ("c", "z")]
|
|
),
|
|
}
|
|
for name, g in cases.items():
|
|
check_invariants(name, g)
|
|
|
|
|
|
# ============================================================= C. oracle check
|
|
def oracle_cycles(g: Graph) -> Tuple[Tuple[str, ...], ...]:
|
|
"""Independent brute force: DFS from every node, enumerate simple cycles.
|
|
|
|
Deliberately a different algorithm from schedkit.find_cycles (plain DFS with
|
|
path tracking rather than Johnson + Tarjan), so agreement is real evidence.
|
|
"""
|
|
adj = {t: sorted(g.dependents(t)) for t in g.tasks()}
|
|
found: Set[Tuple[str, ...]] = set()
|
|
|
|
def canon(cyc: List[str]) -> Tuple[str, ...]:
|
|
i = min(range(len(cyc)), key=lambda k: cyc[k])
|
|
return tuple(cyc[i:] + cyc[:i])
|
|
|
|
for start in sorted(adj):
|
|
stack: List[Tuple[str, List[str]]] = [(start, [start])]
|
|
while stack:
|
|
node, path = stack.pop()
|
|
for nxt in adj[node]:
|
|
if nxt == start and len(path) >= 1:
|
|
found.add(canon(path))
|
|
elif nxt not in path:
|
|
stack.append((nxt, path + [nxt]))
|
|
return tuple(sorted(found))
|
|
|
|
|
|
def oracle() -> None:
|
|
section("C. oracle: find_cycles vs brute-force DFS")
|
|
cases = {
|
|
"acyclic": build([("a", "b"), ("b", "c"), ("a", "c")]),
|
|
"one-cycle": build([("a", "b"), ("b", "c"), ("c", "a")]),
|
|
"two-cycles": build(
|
|
[("a", "b"), ("b", "a"), ("x", "y"), ("y", "x"), ("a", "x")]
|
|
),
|
|
"nested": build(
|
|
[("a", "b"), ("b", "c"), ("c", "a"), ("b", "d"), ("d", "a")]
|
|
),
|
|
"self-loop": (lambda: (lambda g: (g.add_task("s", ["s"]), g)[1])(Graph()))(),
|
|
"figure-eight": build(
|
|
[("a", "b"), ("b", "a"), ("b", "c"), ("c", "b"), ("a", "d"), ("d", "a")]
|
|
),
|
|
"chorded": build(
|
|
[("a", "b"), ("b", "c"), ("c", "d"), ("d", "a"), ("a", "c"), ("b", "d")]
|
|
),
|
|
"empty": Graph(),
|
|
}
|
|
for name, g in cases.items():
|
|
check("C oracle %s" % name, find_cycles(g), oracle_cycles(g))
|
|
|
|
|
|
# ============================================================ D. randomized
|
|
def random_dag(rng: random.Random, n: int, p: float) -> Graph:
|
|
"""Random DAG: edge i->j only when i<j, so acyclicity is by construction."""
|
|
g = Graph()
|
|
ids = ["t%03d" % i for i in range(n)]
|
|
for t in ids:
|
|
g.add_task(t)
|
|
for i in range(n):
|
|
for j in range(i + 1, n):
|
|
if rng.random() < p:
|
|
g.add_edge(ids[i], ids[j])
|
|
return g
|
|
|
|
|
|
def randomized(trials: int = 1000, perms: int = 40, seed: int = 20261021) -> None:
|
|
section("D. randomized differential (%d DAGs x %d permutations)" % (trials, perms))
|
|
rng = random.Random(seed)
|
|
t0 = time.monotonic()
|
|
oracle_calls = 0
|
|
|
|
for trial in range(trials):
|
|
n = rng.randint(0, 7)
|
|
p = rng.choice([0.0, 0.05, 0.15, 0.3, 0.6, 1.0])
|
|
g = random_dag(rng, n, p)
|
|
ids = g.tasks()
|
|
base = plan(g)
|
|
base_cycles = find_cycles(g)
|
|
|
|
# D1: oracle agreement on cycles (small graphs only -- brute force)
|
|
if n <= 6:
|
|
check("D1 trial %d cycles" % trial, base_cycles, oracle_cycles(g))
|
|
oracle_calls += 1
|
|
|
|
# D2: the plan is a permutation of the tasks
|
|
check("D2 trial %d partition" % trial, tuple(sorted(iter_plan(base))), ids)
|
|
|
|
# D3: isomorphism invariance under `perms` random relabellings.
|
|
# The relabelled plan must equal the original plan with every id
|
|
# substituted and the wave re-sorted -- a different id set is the whole
|
|
# point of the check, and wave order is defined by id order, not by the
|
|
# order the original ids happened to sit in.
|
|
for k in range(perms):
|
|
perm = list(ids)
|
|
rng.shuffle(perm)
|
|
mapping = {old: "n%03d" % i for i, old in enumerate(perm)}
|
|
g2 = g.relabel(mapping)
|
|
want_plan = tuple(
|
|
tuple(sorted(mapping[t] for t in w)) for w in base
|
|
)
|
|
check("D3 trial %d perm %d" % (trial, k), plan(g2), want_plan)
|
|
want_cycles = tuple(
|
|
sorted(canon_ids([mapping[t] for t in c]) for c in base_cycles)
|
|
)
|
|
check("D3c trial %d perm %d cycles" % (trial, k), find_cycles(g2), want_cycles)
|
|
check("D3b trial %d perm %d task count" % (trial, k), len(g2), len(g))
|
|
|
|
# D4: every edge is respected by the emitted order
|
|
flat = list(iter_plan(base))
|
|
pos = {t: i for i, t in enumerate(flat)}
|
|
for pre, post in g.edges():
|
|
check_true(
|
|
"D4 trial %d edge %s->%s" % (trial, pre, post),
|
|
pos[pre] < pos[post],
|
|
)
|
|
|
|
dt = time.monotonic() - t0
|
|
print(" %d DAGs, %d oracle comparisons, %.3fs" % (trials, oracle_calls, dt))
|
|
|
|
|
|
# ============================================================== E. determinism
|
|
def determinism() -> None:
|
|
section("E. determinism")
|
|
edges = [("a", "b"), ("b", "c"), ("a", "c"), ("d", "e"), ("e", "c"), ("f", "b")]
|
|
|
|
g = build(edges)
|
|
runs = [plan(g) for _ in range(200)]
|
|
check_true("E1 200 runs identical", all(r == runs[0] for r in runs))
|
|
|
|
# build-order shuffle must not change the plan
|
|
rng = random.Random(7)
|
|
outs = set()
|
|
for _ in range(50):
|
|
shuffled = list(edges)
|
|
rng.shuffle(shuffled)
|
|
gg = Graph()
|
|
for pre, post in shuffled:
|
|
gg.add_edge(pre, post)
|
|
outs.add(plan(gg))
|
|
check("E2 build order irrelevant", outs, {plan(build(edges))})
|
|
|
|
# deep graph must not blow the interpreter stack
|
|
deep = Graph()
|
|
N = 50000
|
|
for i in range(N - 1):
|
|
deep.add_edge("d%06d" % i, "d%06d" % (i + 1))
|
|
t0 = time.monotonic()
|
|
waves = plan(deep)
|
|
dt = time.monotonic() - t0
|
|
check("E3 deep chain waves", len(waves), N)
|
|
check("E3b deep chain last", waves[-1], ("d%06d" % (N - 1),))
|
|
print(" deep chain: %d tasks, %d waves, %.3fs" % (N, len(waves), dt))
|
|
|
|
# wide graph: one root, 20k leaves
|
|
wide = Graph()
|
|
for i in range(20000):
|
|
wide.add_edge("root", "leaf%05d" % i)
|
|
t0 = time.monotonic()
|
|
w = plan(wide)
|
|
dt = time.monotonic() - t0
|
|
check("E4 wide waves", (len(w), len(w[0]), len(w[1])), (2, 1, 20000))
|
|
print(" wide graph: 20001 tasks, %d waves, %.3fs" % (len(w), dt))
|
|
|
|
# build-order shuffle of a deep chain (insertion order reversed)
|
|
rev = Graph()
|
|
for i in range(N - 1, 0, -1):
|
|
rev.add_edge("d%06d" % (i - 1), "d%06d" % i)
|
|
check("E5 deep reversed identical", plan(rev), waves)
|
|
|
|
|
|
# =========================================================== F. cycle handling
|
|
def cycle_handling() -> None:
|
|
section("F. cycle handling")
|
|
g = build([("a", "b"), ("b", "c"), ("c", "a")])
|
|
try:
|
|
plan(g)
|
|
check_true("F1 CycleError raised", False, "no exception")
|
|
except CycleError as e:
|
|
check("F1 CycleError cycles", e.cycles, (("a", "b", "c"),))
|
|
check("F1b message", str(e), "dependency cycle detected: a -> b -> c -> a")
|
|
|
|
# canonical rotation: same cycle entered from a different node
|
|
g2 = build([("c", "a"), ("a", "b"), ("b", "c")])
|
|
check("F2 canonical rotation", find_cycles(g2), (("a", "b", "c"),))
|
|
|
|
# DOT and JSON must refuse a cyclic graph, not emit a broken plan
|
|
for label, fn in (("DOT", schedkit.plan_graphviz), ("JSON", schedkit.plan_json)):
|
|
try:
|
|
fn(g)
|
|
check_true("F3 %s refuses cycles" % label, False, "no exception")
|
|
except CycleError:
|
|
check_true("F3 %s refuses cycles" % label, True)
|
|
|
|
# two disjoint cycles are both reported
|
|
g3 = build([("a", "b"), ("b", "a"), ("x", "y"), ("y", "x")])
|
|
check("F4 two cycles", find_cycles(g3), (("a", "b"), ("x", "y")))
|
|
|
|
# a cycle reachable from an acyclic region still blocks the whole plan
|
|
g4 = build([("root", "a"), ("a", "b"), ("b", "a")])
|
|
try:
|
|
plan(g4)
|
|
check_true("F5 reachable cycle blocks", False, "no exception")
|
|
except CycleError:
|
|
check_true("F5 reachable cycle blocks", True)
|
|
|
|
|
|
# ==================================================================== G. scale
|
|
def scale() -> None:
|
|
section("G. scale")
|
|
budget = 20.0
|
|
t0 = time.monotonic()
|
|
|
|
# layered graph: 50k tasks, 100k edges, 10 layers
|
|
L, W = 10, 5000
|
|
g = Graph()
|
|
layers = [["L%d_%05d" % (l, i) for i in range(W)] for l in range(L)]
|
|
for l in layers:
|
|
for t in l:
|
|
g.add_task(t)
|
|
for l in range(L - 1):
|
|
for i in range(W):
|
|
for k in range(2): # 2 edges per task -> 100k edges
|
|
g.add_edge(layers[l][i], layers[l + 1][(i + k) % W])
|
|
t1 = time.monotonic()
|
|
waves = plan(g)
|
|
t2 = time.monotonic()
|
|
check("G1 layered waves", (len(waves), tuple(len(w) for w in waves)), (L, tuple([W] * L)))
|
|
check("G2 edge count", len(g.edges()), W * 2 * (L - 1))
|
|
print(
|
|
" %d tasks, %d edges: build %.3fs, plan %.3fs"
|
|
% (len(g), len(g.edges()), t1 - t0, t2 - t1)
|
|
)
|
|
check_true("G3 under budget", (t2 - t0) < budget, "(%.2fs >= %.0fs)" % (t2 - t0, budget))
|
|
|
|
# relabeling 50k tasks must also be linear-ish
|
|
t3 = time.monotonic()
|
|
ids = g.tasks()
|
|
mapping = {t: "R" + t[1:] for t in ids}
|
|
g2 = g.relabel(mapping)
|
|
w2 = plan(g2)
|
|
t4 = time.monotonic()
|
|
check("G4 relabel plan", [tuple(sorted(mapping[t] for t in w)) for w in waves],
|
|
[tuple(sorted(t for t in w)) for w in w2])
|
|
print(" relabel+plan 50k: %.3fs" % (t4 - t3))
|
|
check_true("G5 relabel under budget", (t4 - t3) < budget, "(%.2fs)" % (t4 - t3))
|
|
|
|
|
|
# ========================================== H. cost shape (why the build was fixed)
|
|
def shapes() -> None:
|
|
section("H. cost shape: acyclic graphs linear, big SCCs iterative")
|
|
|
|
# H1: a 2,000-node chain contains no cycle at all. The previous build took
|
|
# 5.6651s here (measured, quadratic); a linear one takes milliseconds. This
|
|
# gate catches a relapse into re-scanning every adjacency list per vertex.
|
|
chain = Graph()
|
|
for i in range(1999):
|
|
chain.add_edge("d%06d" % i, "d%06d" % (i + 1))
|
|
t0 = time.monotonic()
|
|
check("H1 chain has no cycles", find_cycles(chain), ())
|
|
dt = time.monotonic() - t0
|
|
print(" chain 2000: %.4fs (quadratic build measured 5.6651s)" % dt)
|
|
check_true("H1b chain under 0.5s", dt < 0.5, "(%.3fs)" % dt)
|
|
|
|
# H2: one 50,000-node ring is one SCC. Exactly one cycle, and it is found
|
|
# without recursing -- the recursive build raised RecursionError at n=2000
|
|
# with the default limit of 1000.
|
|
n = 50000
|
|
ring = Graph()
|
|
for i in range(n):
|
|
ring.add_edge("r%06d" % i, "r%06d" % ((i + 1) % n))
|
|
t0 = time.monotonic()
|
|
cyc = find_cycles(ring)
|
|
dt = time.monotonic() - t0
|
|
check("H2 ring one cycle", (len(cyc), len(cyc[0])), (1, n))
|
|
check("H2b ring is the ring", cyc[0], tuple("r%06d" % i for i in range(n)))
|
|
print(" ring 50000: %.3fs, recursionlimit %d" % (dt, sys.getrecursionlimit()))
|
|
check_true("H2c ring under 10s", dt < 10.0, "(%.2fs)" % dt)
|
|
|
|
# H3: 2,000 disjoint 5-rings -- every cycle reported, one SCC pass total.
|
|
many = Graph()
|
|
for r in range(2000):
|
|
ids = ["g%04d_%d" % (r, i) for i in range(5)]
|
|
for i in range(5):
|
|
many.add_edge(ids[i], ids[(i + 1) % 5])
|
|
t0 = time.monotonic()
|
|
c = find_cycles(many)
|
|
dt = time.monotonic() - t0
|
|
check("H3 disjoint rings", len(c), 2000)
|
|
print(" 2000 disjoint 5-rings: %d cycles, %.3fs" % (len(c), dt))
|
|
|
|
|
|
# ========================================= I. the linear extension's own contract
|
|
def smallest_topo_oracle(g: Graph) -> Tuple[str, ...]:
|
|
"""The lexicographically smallest topological order, by brute force.
|
|
|
|
Independent of the module and of any heap: walk every permutation of the
|
|
task ids in lexicographic order (itertools yields them sorted when the input
|
|
is sorted) and return the first one that respects every edge. Small graphs
|
|
only -- n! is the cost.
|
|
"""
|
|
ids = g.tasks()
|
|
deps = {t: set(g.deps(t)) for t in ids}
|
|
for perm in itertools.permutations(ids):
|
|
seen: Set[str] = set()
|
|
for t in perm:
|
|
if not deps[t] <= seen:
|
|
break
|
|
seen.add(t)
|
|
else:
|
|
return perm
|
|
return ()
|
|
|
|
|
|
def is_linear_extension(order: Sequence[str], g: Graph) -> bool:
|
|
if sorted(order) != list(g.tasks()):
|
|
return False
|
|
pos = {t: i for i, t in enumerate(order)}
|
|
return all(pos[pre] < pos[post] for pre, post in g.edges())
|
|
|
|
|
|
def linear_extension() -> None:
|
|
section("I. topological_order is the SMALLEST linear extension")
|
|
|
|
# I1: hand-computed. Kahn with a min-heap emits a, b, c, d; a plain stack
|
|
# (`ready.pop()`) emits a, c, b, d -- the same valid order, but not the
|
|
# smallest one, which is what the module's docstring promises.
|
|
g = build([("a", "b"), ("a", "c"), ("b", "d"), ("c", "d")])
|
|
check("I1 diamond topo", g.topological_order(), ("a", "b", "c", "d"))
|
|
|
|
# I2: tie-break is by id, not by insertion order or by stack position
|
|
g = build([("root", "zeta"), ("root", "alpha"), ("root", "mike")])
|
|
check("I2 tie-break by id", g.topological_order(), ("root", "alpha", "mike", "zeta"))
|
|
|
|
# I3: two ready roots -- the smaller one goes first, its whole chain too
|
|
g = build([("b1", "b2"), ("a1", "a2")])
|
|
check("I3 two roots", g.topological_order(), ("a1", "a2", "b1", "b2"))
|
|
|
|
# I4: chain and single task
|
|
g = build([(chr(97 + i), chr(98 + i)) for i in range(5)])
|
|
check("I4 chain topo", g.topological_order(), tuple("abcdef"))
|
|
check("I4b single", build([], ["solo"]).topological_order(), ("solo",))
|
|
check("I4c empty", Graph().topological_order(), ())
|
|
|
|
# I5: differential against brute force over all permutations, 300 random DAGs
|
|
rng = random.Random(4242)
|
|
checked = 0
|
|
for trial in range(300):
|
|
n = rng.randint(0, 6)
|
|
p = rng.choice([0.0, 0.1, 0.25, 0.5, 0.8, 1.0])
|
|
gg = random_dag(rng, n, p)
|
|
want = smallest_topo_oracle(gg)
|
|
check("I5 trial %d smallest linear extension" % trial, gg.topological_order(), want)
|
|
checked += 1
|
|
print(" %d random DAGs compared against all-permutations brute force" % checked)
|
|
|
|
# I6: insertion order must not matter -- ids alone decide the order
|
|
edges = [("a", "b"), ("a", "c"), ("c", "d"), ("b", "e"), ("d", "e"), ("z", "e")]
|
|
outs = set()
|
|
for _ in range(50):
|
|
shuffled = list(edges)
|
|
rng.shuffle(shuffled)
|
|
outs.add(build(shuffled).topological_order())
|
|
check("I6 build order irrelevant", outs, {build(edges).topological_order()})
|
|
|
|
# I7: still a valid linear extension at a size brute force cannot reach
|
|
big = Graph()
|
|
for i in range(3000):
|
|
big.add_task("t%04d" % i)
|
|
for i in range(0, 2999, 3):
|
|
big.add_edge("t%04d" % i, "t%04d" % (i + 1))
|
|
order = big.topological_order()
|
|
check_true("I7 3000-node order valid", is_linear_extension(order, big))
|
|
check("I7b 3000-node order size", len(order), 3000)
|
|
|
|
# I8: a cyclic graph must raise, not silently truncate the order
|
|
cyc = build([("a", "b"), ("b", "a")])
|
|
try:
|
|
cyc.topological_order()
|
|
check_true("I8 cyclic raises", False, "no exception")
|
|
except CycleError:
|
|
check_true("I8 cyclic raises", True)
|
|
|
|
|
|
# ======================================================================= main
|
|
def main() -> int:
|
|
print("schedkit spec -- %s" % schedkit.__file__)
|
|
print("python %s" % sys.version.split()[0])
|
|
|
|
fixtures()
|
|
invariants()
|
|
oracle()
|
|
randomized()
|
|
determinism()
|
|
cycle_handling()
|
|
scale()
|
|
shapes()
|
|
linear_extension()
|
|
|
|
dt = time.monotonic() - T_START
|
|
print("\n%d checks, %d failed, %.2fs" % (CHECKS, len(FAILS), dt))
|
|
if FAILS:
|
|
print("\nFAILURES (%d):" % len(FAILS))
|
|
for f in FAILS[:15]:
|
|
print(" - " + f)
|
|
if len(FAILS) > 15:
|
|
print(" ... and %d more" % (len(FAILS) - 15))
|
|
return 1
|
|
print("SPEC OK: all %d checks held" % CHECKS)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|