stackchain/troupe-standup
2
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

null: stratified Datalog engine (parser + Tarjan strata + semi-naive eval) verified by 600-program differential oracle

Browse Source
This commit is contained in:
null committed 2026-10-24 22:01:36 +00:00
1 parent b293175382
commit 7cf828a78c
6 files changed
+1651
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

wizard-work/null/datalogkit/EVIDENCE.txt
+82
View File
@@ -0,0 +1,82 @@
[differential] programs evaluated: 600 (unstratified: 61, predicate-relations compared: 3234)
== 1. full unittest suite (the CHECK command runs this same suite)
........................................
----------------------------------------------------------------------
Ran 40 tests in 1.022s
OK
testsRun=40 failures=0 errors=0
== 2. semi-naive vs naive work on a chain transitive closure
n |path| semi-naive naive speedup rounds
6 15 105 300 2.9x 5
12 66 858 5687 6.6x 11
16 120 2040 18825 9.2x 15
20 190 3990 47291 11.9x 19
== 3. fresh-seed differential run: 400 programs, engine vs oracle
programs=400 unstratified=43 predicate-relations compared=2142 mismatches=0
== 4. strata assignment + refusal of a non-stratified program
strata: [['edge'], ['node'], ['source'], ['reach'], ['unreach']]
reach: [('a',), ('b',)] unreach: [('c',)]
refused: negation is not stratified: p negates q but they are mutually recursive (SCC: p, q)
ALL SECTIONS CLEAN
== 5. CLI on examples/reach.dl (appended by hand, not by evidence_run.py)
$ python3 datalog.py examples/reach.dl --text --stats
edge: 2 tuple(s)
edge(a, b).
edge(b, c).
node: 4 tuple(s)
node(a).
node(b).
node(c).
node(d).
reach: 3 tuple(s)
reach(a).
reach(b).
reach(c).
sink: 2 tuple(s)
sink(c).
sink(d).
source: 1 tuple(s)
source(a).
unreachable: 1 tuple(s)
unreachable(d).
{"bindings": 20, "derived": 6, "rounds": 5, "rule_runs": 6}
exit=0
$ python3 datalog.py examples/reach.dl -q unreachable
{
"predicates": {
"unreachable": [
[
"d"
]
]
},
"strata": [
[
"edge"
],
[
"node"
],
[
"sink"
],
[
"source"
],
[
"reach"
],
[
"unreachable"
]
]
}
exit=0
wizard-work/null/datalogkit/README.md
+79
View File
@@ -0,0 +1,79 @@
# datalogkit — a stratified Datalog engine, differentially verified
Wizard: **null** · Lane: Codeforge (hard coding and agentic reasoning)
`datalog.py` is a dependency-free Datalog engine: hand-written parser, static
checks (arity, rule safety, stratification), Tarjan-SCC stratification, and
**semi-naive** bottom-up evaluation with stratified negation. No imports beyond
the standard library.
## Files
datalog.py the engine (module + JSON/text CLI)
test_datalog.py the verification suite — its exit code is the verdict
evidence_run.py reproduces every number in EVIDENCE.txt
examples/reach.dl CLI demo program
EVIDENCE.txt captured output of the above (measured, not claimed)
## Run it
python3 datalog.py examples/reach.dl --text --stats
python3 datalog.py examples/reach.dl -q unreachable # JSON, one predicate
python3 test_datalog.py # 0 = all checks pass
## Language
% comment
edge(a, b). % fact
path(X, Y) :- edge(X, Y). % rule
path(X, Y) :- edge(X, Z), path(Z, Y).
unreach(X) :- node(X), not reach(X). % '!' is accepted as 'not'
Terms: `?X` variable, `?_` anonymous (independent per occurrence), bare lowercase
atom, `"quoted"` string, integer. Head and negated-literal variables must be
bound by a positive body literal; a rule with no positive body literal is
rejected. Negation must be stratified — a negated literal may not sit inside a
recursive cycle.
## Verification (this is the point of the artifact)
`test_datalog.py` = 40 tests, 0 failures. The load-bearing part is
**differential testing against an independent oracle** written from scratch in
the test file: stratification via reflexive-transitive reachability closure plus
Kahn waves (the engine uses iterative Tarjan SCC), evaluation via
cartesian-product naive fixpoint (the engine uses delta-driven semi-naive).
The two share only the parser.
| check | result |
|---|---|
| unittest suite | 40 tests, 0 failures, 0 errors |
| random programs, engine vs oracle (seeds 20261024 / 20261025) | 600 programs, 61 of them unstratified — verdicts and 3234 predicate-relations identical, 0 mismatches |
| fresh-seed re-run (evidence_run.py, seed 999) | 400 programs, 43 unstratified, 2142 relations, 0 mismatches |
| semi-naive actually saves work (chain transitive closure, 12 nodes) | 858 vs 5687 body bindings enumerated — 6.6x; at n=20, 3990 vs 47291 — 11.9x |
| KATs | transitive closure, non-linear same-generation recursion, stratified negation, anonymous vars in negation, zero-arity, repeated variables, order independence |
Numbers above are read off `EVIDENCE.txt`, which `evidence_run.py` regenerates
in one run.
## Bug this suite actually caught
The first run of the differential test failed on 1 of 300 random programs:
the engine derived `p0(c, c)` where the oracle derived nothing. Root cause: the
join reorders positive body literals smallest-relation-first, so a literal like
`p1(X, X)` can be joined *first*; the unifier bound `X` from the first argument
into a private dict while checking the second argument against the *original*
environment, so a repeated variable inside a single literal did not constrain
both positions. Fixed in `_match` (bindings are now visible to later arguments
of the same literal); the same latent flaw was removed from the oracle, and the
exact failing program is now a regression KAT
(`test_repeated_variable_inside_one_literal`).
## Stated limits (not hidden)
* String constants compare as written — no unescaping, so two spellings of the
same escaped string are distinct terms.
* `-` is a token only as part of an integer literal; no arithmetic or builtins.
* No functors, no aggregation, no `$ref`-style imports.
* The engine reports `rounds`/`bindings`/`derived`/`rule_runs`; it is not tuned
for large relations (semi-naive delta is tracked per predicate, joins are
unindexed).
wizard-work/null/datalogkit/datalog.py
+691
View File
@@ -0,0 +1,691 @@
#!/usr/bin/env python3
"""datalog.py -- a self-contained Datalog engine.
Wizard: null Lane: Codeforge (hard coding and agentic reasoning)
What this is
------------
A bottom-up Datalog engine with *stratified negation*, written against the
standard semantics and nothing else:
* a hand-written tokenizer + recursive-descent parser for facts and rules,
* static checks: arity consistency, rule safety (range restriction),
and stratification (no negation may sit inside a recursive cycle),
* stratification by Tarjan SCC + Kahn topological order on the condensation,
* evaluation by *semi-naive* fixpoint iteration inside each stratum, with
negation resolved against strictly lower (already final) strata,
* a JSON / text CLI.
Term representation
-------------------
A term is its canonical source text, one Python string:
'?X' variable X (leading '?' marks a variable)
'?_' anonymous variable (each occurrence is independent)
'foo' atom constant
'"a"' string constant (kept verbatim, quotes included)
'-7' integer constant (kept verbatim)
Variables in heads and in negated literals must also occur in a positive body
literal (range restriction); a rule with no positive body literal is rejected.
Anonymous variables are allowed in bodies (including negated literals) but not
in rule heads.
Semantics notes
---------------
* Negation in a rule body is a *pattern match*: `not edge(X, _)` succeeds iff no
tuple of `edge` matches the pattern with the already-bound variables. With all
variables bound (guaranteed by the safety check) this is plain tuple
non-membership.
* Duplicate facts and duplicate derivations collapse: relations are sets.
* The result is independent of the order of facts and rules.
Public API
----------
parse_program(text) -> Program
load(path) -> Program
check_arity(program) / check_safety(program) / stratify(program)
evaluate(program, stats=None) -> dict[pred, frozenset[tuple[str, ...]]]
format_relations(rels, preds=None) -> str
Errors: ParseError, ArityError, SafetyError, StratificationError, all
subclasses of DatalogError.
Known limits (stated, not hidden): string constants are compared as written
(no unescaping), so two spellings of the same escaped string are distinct;
`-` is only a token as part of an integer literal; functors and builtins
(arithmetic, comparison) are not implemented.
"""
from __future__ import annotations
import argparse
import json
import os
import re
import sys
from dataclasses import dataclass, field
__all__ = [
"DatalogError",
"ParseError",
"ArityError",
"SafetyError",
"StratificationError",
"Literal",
"Rule",
"Program",
"parse_program",
"load",
"check_arity",
"check_safety",
"stratify",
"evaluate",
"format_relations",
"main",
]
class DatalogError(Exception):
"""Base class for every error this engine raises."""
class ParseError(DatalogError):
pass
class ArityError(DatalogError):
pass
class SafetyError(DatalogError):
pass
class StratificationError(DatalogError):
pass
# --------------------------------------------------------------------------
# Terms
# --------------------------------------------------------------------------
_ANON = "?_"
def is_var(term: str) -> bool:
return term.startswith("?")
def is_anon(term: str) -> bool:
return term == _ANON
def var_name(term: str) -> str:
return term[1:]
# --------------------------------------------------------------------------
# Syntax tree
# --------------------------------------------------------------------------
@dataclass(frozen=True)
class Literal:
pred: str
args: tuple
negated: bool = False
def __str__(self) -> str:
head = "not " if self.negated else ""
return f"{head}{self.pred}({', '.join(self.args)})"
@dataclass(frozen=True)
class Rule:
head_pred: str
head_args: tuple
body: tuple
@property
def positive(self) -> tuple:
return tuple(lit for lit in self.body if not lit.negated)
@property
def negated(self) -> tuple:
return tuple(lit for lit in self.body if lit.negated)
def __str__(self) -> str:
head = f"{self.head_pred}({', '.join(self.head_args)})"
body = ", ".join(str(lit) for lit in self.body)
return f"{head} :- {body}."
@dataclass
class Program:
facts: dict = field(default_factory=dict) # pred -> set of tuples
rules: list = field(default_factory=list) # list[Rule]
arities: dict = field(default_factory=dict) # pred -> int
def predicates(self) -> set:
preds = set(self.facts)
preds.update(self.arities)
for rule in self.rules:
preds.add(rule.head_pred)
for lit in rule.body:
preds.add(lit.pred)
return preds
def add_fact(self, pred: str, args: tuple) -> None:
self.facts.setdefault(pred, set()).add(tuple(args))
def add_rule(self, rule: Rule) -> None:
self.rules.append(rule)
def text(self) -> str:
lines = []
for pred in sorted(self.facts):
for tup in sorted(self.facts[pred]):
lines.append(f"{pred}({', '.join(tup)}).")
lines.extend(str(r) for r in self.rules)
return "\n".join(lines) + "\n"
# --------------------------------------------------------------------------
# Tokenizer + parser
# --------------------------------------------------------------------------
_TOKEN_RE = re.compile(
r"(?P<ws>[ \t\r\n]+)"
r"|(?P<comment>%[^\n]*)"
r"|(?P<string>\"(?:[^\"\\]|\\.)*\")"
r"|(?P<int>-?\d+)"
r"|(?P<name>[A-Za-z_][A-Za-z0-9_]*)"
r"|(?P<punct>:-|[!(),.])"
)
def tokenize(text: str) -> list:
tokens = []
pos = 0
n = len(text)
while pos < n:
m = _TOKEN_RE.match(text, pos)
if m is None:
line = text.count("\n", 0, pos) + 1
raise ParseError(
f"unexpected character {text[pos]!r} at line {line}, column "
f"{pos - text.rfind(chr(10), 0, pos)}"
)
kind = m.lastgroup
if kind not in ("ws", "comment"):
tokens.append((kind, m.group()))
pos = m.end()
return tokens
class _Parser:
def __init__(self, tokens):
self.toks = tokens
self.i = 0
def peek(self, k: int = 0):
j = self.i + k
return self.toks[j] if j < len(self.toks) else (None, None)
def next(self):
tok = self.peek()
self.i += 1
return tok
def expect(self, kind, value=None):
k, v = self.next()
if k != kind or (value is not None and v != value):
want = value if value is not None else kind
raise ParseError(f"expected {want!r} but found {v!r}")
return v
# ---- grammar ------------------------------------------------------
def parse_term(self):
kind, value = self.next()
if kind in ("string", "int"):
return value
if kind == "name":
if value in ("not", "!"):
raise ParseError(f"{value!r} cannot be used as a term")
if value[0].isupper() or value[0] == "_":
return "?" + value
return value
raise ParseError(f"expected a term but found {value!r}")
def parse_literal(self):
negated = False
kind, value = self.peek()
if (kind == "name" and value == "not") or (kind == "punct" and value == "!"):
self.next()
negated = True
pred = self.expect("name")
if pred in ("not", "!"):
raise ParseError(f"{pred!r} cannot be used as a predicate name")
self.expect("punct", "(")
args = []
if self.peek() == ("punct", ")"):
self.next()
else:
while True:
args.append(self.parse_term())
kind, value = self.peek()
if (kind, value) == ("punct", ","):
self.next()
continue
break
self.expect("punct", ")")
return Literal(pred, tuple(args), negated)
def parse_program(self) -> Program:
program = Program()
while self.peek()[0] is not None:
head = self.parse_literal()
if head.negated:
raise ParseError("a clause head cannot be negated")
if self.peek() == ("punct", ":-"):
self.next()
body = [self.parse_literal()]
while self.peek() == ("punct", ","):
self.next()
body.append(self.parse_literal())
self.expect("punct", ".")
program.add_rule(Rule(head.pred, head.args, tuple(body)))
else:
self.expect("punct", ".")
for term in head.args:
if is_var(term):
raise ParseError(
f"fact {head.pred}/{len(head.args)} contains the "
f"variable {term}; facts must be ground"
)
program.add_fact(head.pred, head.args)
return program
def parse_program(text: str) -> Program:
"""Parse Datalog source text into a Program."""
return _Parser(tokenize(text)).parse_program()
def load(path: str) -> Program:
with open(path, "r", encoding="utf-8") as fh:
return parse_program(fh.read())
# --------------------------------------------------------------------------
# Static checks
# --------------------------------------------------------------------------
def check_arity(program: Program) -> None:
"""Every predicate must be used with one arity throughout."""
seen = dict(program.arities)
def note(pred, n, where):
if pred in seen and seen[pred] != n:
raise ArityError(
f"predicate {pred!r} used with arity {n} in {where} "
f"but arity {seen[pred]} elsewhere"
)
seen[pred] = n
for pred, tuples in program.facts.items():
for tup in tuples:
note(pred, len(tup), f"a fact of {pred!r}")
for rule in program.rules:
note(rule.head_pred, len(rule.head_args), f"the head of {rule}")
for lit in rule.body:
note(lit.pred, len(lit.args), f"the body of {rule}")
def check_safety(program: Program) -> None:
"""Range restriction: head and negated-literal variables must be bound by a
positive body literal, and every rule needs at least one positive literal."""
for rule in program.rules:
if not rule.positive:
raise SafetyError(f"rule {rule} has no positive body literal (unsafe)")
bound = set()
for lit in rule.positive:
for term in lit.args:
if is_var(term) and not is_anon(term):
bound.add(term)
for term in rule.head_args:
if is_anon(term):
raise SafetyError(f"anonymous variable in the head of {rule}")
if is_var(term) and term not in bound:
raise SafetyError(
f"head variable {term} of {rule} is not bound by a positive "
f"body literal (unsafe)"
)
for lit in rule.negated:
for term in lit.args:
if is_var(term) and not is_anon(term) and term not in bound:
raise SafetyError(
f"variable {term} in the negated literal {lit} of {rule} "
f"is not bound by a positive body literal (unsafe)"
)
def _tarjan_scc(nodes, adj) -> list:
"""Iterative Tarjan SCC. Returns components as sorted lists, deterministic."""
index: dict = {}
low: dict = {}
on_stack: dict = {}
stack: list = []
out: list = []
counter = [0]
for root in nodes:
if root in index:
continue
index[root] = low[root] = counter[0]
counter[0] += 1
stack.append(root)
on_stack[root] = True
work = [(root, iter(sorted(adj.get(root, ()))))]
while work:
node, it = work[-1]
advanced = False
for nxt in it:
if nxt not in index:
index[nxt] = low[nxt] = counter[0]
counter[0] += 1
stack.append(nxt)
on_stack[nxt] = True
work.append((nxt, iter(sorted(adj.get(nxt, ())))))
advanced = True
break
if on_stack.get(nxt):
low[node] = min(low[node], index[nxt])
if advanced:
continue
work.pop()
if work:
parent = work[-1][0]
low[parent] = min(low[parent], low[node])
if low[node] == index[node]:
comp = []
while True:
top = stack.pop()
on_stack[top] = False
comp.append(top)
if top == node:
break
out.append(sorted(comp))
return out
def stratify(program: Program) -> list:
"""Split predicates into strata (a list of frozensets, dependencies first).
Raises StratificationError when a negated literal sits inside a recursive
cycle, i.e. when head and negated body predicate share an SCC.
"""
preds = sorted(program.predicates())
adj = {p: set() for p in preds}
neg_edges = []
for rule in program.rules:
for lit in rule.body:
adj[rule.head_pred].add(lit.pred)
if lit.negated:
neg_edges.append((rule.head_pred, lit.pred))
comps = _tarjan_scc(preds, adj)
comp_of = {}
for idx, comp in enumerate(comps):
for pred in comp:
comp_of[pred] = idx
for head, body in neg_edges:
if comp_of[head] == comp_of[body]:
members = ", ".join(comps[comp_of[head]])
raise StratificationError(
f"negation is not stratified: {head} negates {body} but they are "
f"mutually recursive (SCC: {members})"
)
# condensation: edge comp(p) -> comp(q) means "p depends on q", so q must be
# computed first. Kahn over the reversed relation.
must_precede = {i: set() for i in range(len(comps))}
for pred in preds:
for dep in adj[pred]:
a, b = comp_of[pred], comp_of[dep]
if a != b:
must_precede[a].add(b)
indeg = {i: len(must_precede[i]) for i in must_precede}
dependents = {i: set() for i in must_precede}
for i, deps in must_precede.items():
for dep in deps:
dependents[dep].add(i)
ready = sorted((i for i, d in indeg.items() if d == 0), key=lambda i: comps[i][0])
order = []
while ready:
cur = ready.pop(0)
order.append(cur)
for nxt in sorted(dependents[cur], key=lambda i: comps[i][0]):
indeg[nxt] -= 1
if indeg[nxt] == 0:
ready.append(nxt)
ready.sort(key=lambda i: comps[i][0])
if len(order) != len(comps): # pragma: no cover - condensation is a DAG
raise StratificationError("internal error: condensation is not a DAG")
return [frozenset(comps[i]) for i in order]
# --------------------------------------------------------------------------
# Evaluation
# --------------------------------------------------------------------------
def _match(pattern, tup, env):
"""Unify a literal's argument pattern against a tuple. Returns a new env or
None. Anonymous variables never bind. Bindings made while walking this
literal are visible to its later arguments, so `p(X, X)` really does mean
"both arguments equal"."""
cur = env
copied = False
for term, value in zip(pattern, tup):
if is_var(term):
if is_anon(term):
continue
name = var_name(term)
if name in cur:
if cur[name] != value:
return None
else:
if not copied:
cur = dict(env)
copied = True
cur[name] = value
elif term != value:
return None
return cur
def _subst(term, env):
if is_var(term):
if is_anon(term):
raise SafetyError("anonymous variable cannot be substituted")
return env[var_name(term)]
return term
def _negation_fails(lit, env, rels):
"""True when the negated literal matches some tuple (so the body fails)."""
rel = rels.get(lit.pred, ())
if not rel:
return False
for tup in rel:
if _match(lit.args, tup, env) is not None:
return True
return False
def _order_pos(rule, R, rels):
"""Positive body literals, smallest relation first (deterministic)."""
def size(lit):
if lit.pred in R:
return len(R[lit.pred])
return len(rels.get(lit.pred, ()))
return sorted(rule.positive, key=lambda lit: (size(lit), lit.pred))
def _join(rule, pos, R, rels, override, stats):
"""Enumerate satisfying environments for the positive body literals."""
def relation_for(idx, lit):
if override and idx in override:
return override[idx]
if lit.pred in R:
return R[lit.pred]
return rels.get(lit.pred, ())
def rec(idx, env):
if idx == len(pos):
for lit in rule.negated:
if _negation_fails(lit, env, rels):
return
yield env
return
lit = pos[idx]
for tup in sorted(relation_for(idx, lit)):
stats["bindings"] += 1
env2 = _match(lit.args, tup, env)
if env2 is not None:
yield from rec(idx + 1, env2)
yield from rec(0, {})
def _derive_one(rule, pos, R, rels, override, out, stats):
head_pred = rule.head_pred
for env in _join(rule, pos, R, rels, override, stats):
stats["rule_runs"] += 1
tup = tuple(_subst(term, env) for term in rule.head_args)
if tup in R[head_pred] or tup in out[head_pred]:
continue
out[head_pred].add(tup)
stats["derived"] += 1
def _derive(rules_here, R, rels, wave, override, stats):
out = {pred: set() for pred in wave}
for rule in rules_here:
pos = _order_pos(rule, R, rels)
_derive_one(rule, pos, R, rels, override, out, stats)
return out
def _eval_stratum(program, wave, rels, stats):
rules_here = [r for r in program.rules if r.head_pred in wave]
if not rules_here:
return
R = {pred: rels[pred] for pred in wave} # live sets, mutated in place
# Round 0 is naive: it also fires rules whose positive body lives entirely in
# lower strata (such a rule can never see a same-stratum delta).
new = _derive(rules_here, R, rels, wave, None, stats)
for pred, added in new.items():
R[pred] |= added
delta = {pred: set(added) for pred, added in new.items()}
while any(delta.values()):
stats["rounds"] += 1
new = {pred: set() for pred in wave}
for rule in rules_here:
pos = _order_pos(rule, R, rels)
for idx, lit in enumerate(pos):
if lit.pred in wave and delta.get(lit.pred):
_derive_one(
rule, pos, R, rels, {idx: delta[lit.pred]}, new, stats
)
for pred, added in new.items():
R[pred] |= added
delta = new
def evaluate(program: Program, stats: dict = None, check: bool = True) -> dict:
"""Semi-naive bottom-up evaluation. Returns {pred: frozenset(tuples)}."""
if check:
check_arity(program)
check_safety(program)
strata = stratify(program)
if stats is None:
stats = {}
stats.update(rounds=0, bindings=0, derived=0, rule_runs=0)
rels = {pred: set(program.facts.get(pred, ())) for pred in program.predicates()}
for wave in strata:
_eval_stratum(program, wave, rels, stats)
return {pred: frozenset(tuples) for pred, tuples in rels.items()}
# --------------------------------------------------------------------------
# Output
# --------------------------------------------------------------------------
def format_relations(rels: dict, preds=None) -> str:
lines = []
for pred in sorted(rels) if preds is None else list(preds):
tuples = sorted(rels.get(pred, ()))
lines.append(f"{pred}: {len(tuples)} tuple(s)")
for tup in tuples:
lines.append(f" {pred}({', '.join(tup)}).")
return "\n".join(lines)
def main(argv=None) -> int:
ap = argparse.ArgumentParser(
prog="datalog.py",
description="Evaluate a Datalog program (stratified negation, semi-naive).",
)
ap.add_argument("program", help="path to a Datalog source file")
ap.add_argument("-q", "--query", action="append", default=None,
help="only report this predicate (repeatable)")
ap.add_argument("--text", action="store_true", help="human-readable output")
ap.add_argument("--stats", action="store_true", help="include run statistics")
args = ap.parse_args(argv)
try:
program = load(args.program)
stats = {}
rels = evaluate(program, stats)
strata = stratify(program)
except DatalogError as exc:
print(f"error: {type(exc).__name__}: {exc}", file=sys.stderr)
return 2
except OSError as exc:
print(f"error: {exc}", file=sys.stderr)
return 2
if args.text:
print(format_relations(rels, args.query))
if args.stats:
print(json.dumps(stats, sort_keys=True))
return 0
payload = {
"predicates": {
pred: [list(t) for t in sorted(rels[pred])]
for pred in (sorted(rels) if args.query is None else args.query)
},
"strata": [sorted(wave) for wave in strata],
}
if args.stats:
payload["stats"] = stats
print(json.dumps(payload, sort_keys=True, indent=2))
return 0
if __name__ == "__main__": # pragma: no cover
sys.exit(main())
wizard-work/null/datalogkit/evidence_run.py
+83
View File
@@ -0,0 +1,83 @@
#!/usr/bin/env python3
"""evidence_run.py -- reproduce every number quoted in EVIDENCE.txt, in one run.
python3 evidence_run.py (exit 0 = all sections ran clean)
Wizard: null Lane: Codeforge (hard coding and agentic reasoning)
"""
from __future__ import annotations
import io
import os
import random
import sys
import unittest
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import datalog # noqa: E402
import test_datalog as T # noqa: E402
def main() -> int:
print("== 1. full unittest suite (the CHECK command runs this same suite)")
suite = unittest.defaultTestLoader.loadTestsFromModule(T)
stream = io.StringIO()
result = unittest.TextTestRunner(stream=stream, verbosity=1).run(suite)
print(stream.getvalue().strip())
print(f"testsRun={result.testsRun} failures={len(result.failures)} "
f"errors={len(result.errors)}")
print("\n== 2. semi-naive vs naive work on a chain transitive closure")
print(f"{'n':>3} {'|path|':>7} {'semi-naive':>11} {'naive':>10} {'speedup':>8} {'rounds':>7}")
for n in (6, 12, 16, 20):
program = T.WorkReductionTests.chain_program(n)
engine_stats, oracle_stats = {}, {}
got = datalog.evaluate(program, engine_stats)
want = T.oracle_evaluate(program, oracle_stats)
assert got == want, f"engine/oracle disagree at n={n}"
ratio = oracle_stats["bindings"] / max(engine_stats["bindings"], 1)
print(f"{n:>3} {len(got['path']):>7} {engine_stats['bindings']:>11} "
f"{oracle_stats['bindings']:>10} {ratio:>7.1f}x {engine_stats['rounds']:>7}")
print("\n== 3. fresh-seed differential run: 400 programs, engine vs oracle")
rng = random.Random(999)
programs = ([T.gen_program(rng, constrained=True) for _ in range(200)]
+ [T.gen_program(rng, constrained=False) for _ in range(200)])
counters = {"programs": 0, "unstratified": 0, "predicates": 0}
failures = T.differential_run(programs, counters)
print(f"programs={counters['programs']} unstratified={counters['unstratified']} "
f"predicate-relations compared={counters['predicates']} "
f"mismatches={len(failures)}")
for idx, text, why in failures[:3]:
print(f"MISMATCH #{idx}: {why}\n{text}")
print("\n== 4. strata assignment + refusal of a non-stratified program")
program = datalog.parse_program(
"node(a). node(b). node(c). source(a). edge(a,b).\n"
"reach(X) :- source(X).\n"
"reach(X) :- edge(Y,X), reach(Y).\n"
"unreach(X) :- node(X), not reach(X).\n"
)
print("strata:", [sorted(w) for w in datalog.stratify(program)])
rels = datalog.evaluate(program)
print("reach:", sorted(rels["reach"]), " unreach:", sorted(rels["unreach"]))
try:
datalog.stratify(datalog.parse_program(
"dom(a). p(X) :- dom(X), not q(X). q(X) :- p(X).\n"
))
print("ERROR: unstratified program was accepted")
return 1
except datalog.StratificationError as exc:
print("refused:", exc)
ok = (len(result.failures) == 0 and len(result.errors) == 0
and not failures and result.testsRun >= 40)
print("\nALL SECTIONS CLEAN" if ok else "\nSOMETHING FAILED")
return 0 if ok else 1
if __name__ == "__main__":
sys.exit(main())
wizard-work/null/datalogkit/examples/reach.dl
+10
View File
@@ -0,0 +1,10 @@
% examples/reach.dl -- stratified negation + recursion, the CLI demo program.
node(a). node(b). node(c). node(d).
source(a).
edge(a, b). edge(b, c).
reach(X) :- source(X).
reach(X) :- edge(Y, X), reach(Y).
unreachable(X) :- node(X), not reach(X).
sink(X) :- node(X), not edge(X, _).
wizard-work/null/datalogkit/test_datalog.py
+706
View File
@@ -0,0 +1,706 @@
#!/usr/bin/env python3
"""test_datalog.py -- verification suite for datalog.py.
Wizard: null Lane: Codeforge (hard coding and agentic reasoning)
Exit code is the verdict: 0 = every check passed, 1 = at least one failed.
What is verified, and how
-------------------------
1. Parser KATs -- facts, rules, comments, '!', zero-arity, int/string
constants, and six malformed inputs that must raise.
2. Static checks -- arity conflicts, rule safety, stratification.
3. Evaluation KATs -- transitive closure, the non-linear same-generation
program, stratified negation, anonymous variables.
4. Order independence -- shuffled fact/rule order gives identical relations.
5. Differential testing -- 600 randomly generated programs are evaluated by the
engine AND by an *independent oracle* written from
scratch in this file (reachability-closure SCCs +
Kahn waves for stratification; cartesian-product
naive fixpoint for evaluation). Both the stratified/
unstratified verdict and every relation must agree.
6. Work reduction -- semi-naive evaluation must enumerate strictly fewer
body bindings than the naive oracle on a chain
transitive closure (i.e. the delta rule is doing real
work, not just returning the right answer by luck).
7. CLI end-to-end -- subprocess run of datalog.py, JSON contract, exit
codes for success and for an unstratified program.
The oracle deliberately shares NOTHING with the engine except the parser and the
term representation; it re-derives stratification and evaluation itself.
"""
from __future__ import annotations
import itertools
import json
import os
import random
import subprocess
import sys
import tempfile
import unittest
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import datalog # noqa: E402
from datalog import ( # noqa: E402
ArityError,
Literal,
ParseError,
Program,
Rule,
SafetyError,
StratificationError,
check_arity,
check_safety,
evaluate,
parse_program,
stratify,
)
DIFFERENTIAL_RUNS = {"programs": 0, "unstratified": 0, "predicates": 0}
# ==========================================================================
# Independent oracle
# ==========================================================================
def oracle_dependency_graph(program):
preds = sorted(program.predicates())
adj = {p: set() for p in preds}
for rule in program.rules:
for lit in rule.body:
adj[rule.head_pred].add(lit.pred)
return preds, adj
def oracle_reachability(program):
preds, adj = oracle_dependency_graph(program)
reach = {p: set(adj[p]) | {p} for p in preds}
changed = True
while changed:
changed = False
for p in preds:
grown = set(reach[p])
for q in list(reach[p]):
grown |= reach[q]
if grown != reach[p]:
reach[p] = grown
changed = True
return preds, adj, reach
def oracle_stratify(program):
"""Stratify via reachability closure. Returns waves (list[list[str]]) or
None when the program is not stratified."""
preds, adj, reach = oracle_reachability(program)
neg = {(r.head_pred, lit.pred) for r in program.rules for lit in r.negated}
sccs, seen = [], set()
for p in preds:
if p in seen:
continue
comp = sorted(q for q in preds if q in reach[p] and p in reach[q])
seen.update(comp)
sccs.append(comp)
comp_of = {p: i for i, comp in enumerate(sccs) for p in comp}
for head, body in neg:
if comp_of[head] == comp_of[body]:
return None
before = {i: set() for i in range(len(sccs))}
for p in preds:
for q in adj[p]:
if comp_of[p] != comp_of[q]:
before[comp_of[p]].add(comp_of[q])
waves, assigned = [], set()
while len(assigned) < len(sccs):
ready = sorted(
(i for i in range(len(sccs)) if i not in assigned and before[i] <= assigned),
key=lambda i: sccs[i][0],
)
if not ready:
return None
assigned.update(ready)
waves.append(sorted(p for i in ready for p in sccs[i]))
return waves
def oracle_unify(pattern, tup, env):
cur = env
copied = False
for term, value in zip(pattern, tup):
if datalog.is_var(term):
if datalog.is_anon(term):
continue
name = term[1:]
if name in cur:
if cur[name] != value:
return None
else:
if not copied:
cur = dict(env)
copied = True
cur[name] = value
elif term != value:
return None
return cur
def oracle_neg_fails(lit, env, rels):
for tup in rels[lit.pred]:
if oracle_unify(lit.args, tup, env) is not None:
return True
return False
def oracle_evaluate(program, stats=None):
"""Naive (product-based) fixpoint per wave. Returns None if unstratified."""
waves = oracle_stratify(program)
if waves is None:
return None
if stats is None:
stats = {}
stats.setdefault("bindings", 0)
rels = {p: set(program.facts.get(p, ())) for p in program.predicates()}
for wave in waves:
wave = set(wave)
changed = True
while changed:
changed = False
for rule in program.rules:
if rule.head_pred not in wave:
continue
pools = [sorted(rels[lit.pred]) for lit in rule.positive]
for combo in itertools.product(*pools):
stats["bindings"] += 1
env = {}
ok = True
for lit, tup in zip(rule.positive, combo):
env = oracle_unify(lit.args, tup, env)
if env is None:
ok = False
break
if not ok:
continue
for lit in rule.negated:
if oracle_neg_fails(lit, env, rels):
ok = False
break
if not ok:
continue
head = tuple(
env[t[1:]] if datalog.is_var(t) else t for t in rule.head_args
)
if head not in rels[rule.head_pred]:
rels[rule.head_pred].add(head)
changed = True
return {p: frozenset(v) for p, v in rels.items()}
# ==========================================================================
# Random program generators
# ==========================================================================
CONSTS = ["a", "b", "c"]
VARS = ["?X", "?Y", "?Z"]
EDB2 = ["e0", "e1"]
IDB2 = ["p0", "p1"]
IDB1 = ["q0", "q1"]
def _random_facts(rng):
facts = {}
for pred in EDB2:
fs = set()
for a in CONSTS:
for b in CONSTS:
if rng.random() < 0.30:
fs.add((a, b))
facts[pred] = fs
return facts
def gen_program(rng, constrained=True):
"""Random program. constrained=True keeps negation strictly lower-stratum,
so the program is stratified by construction; constrained=False lets
negation point anywhere, which yields unstratified programs too."""
arity = {p: 2 for p in EDB2 + IDB2}
arity.update({p: 1 for p in IDB1})
levels = {p: 0 for p in EDB2}
for p in IDB2 + IDB1:
levels[p] = rng.choice([1, 1, 2])
program = Program(facts=_random_facts(rng), rules=[], arities=dict(arity))
for _ in range(rng.randint(1, 5)):
head_pred = rng.choice(IDB2 + IDB1)
hl = levels[head_pred]
if constrained:
cands = [p for p in arity if levels[p] <= hl]
else:
cands = list(arity)
positives, used_vars = [], []
for _ in range(rng.randint(1, 2)):
bp = rng.choice(cands)
args = []
for _ in range(arity[bp]):
if rng.random() < 0.15:
args.append(rng.choice(CONSTS))
else:
args.append(rng.choice(VARS))
positives.append(Literal(bp, tuple(args), False))
used_vars.extend(a for a in args if datalog.is_var(a))
def pick():
if used_vars and rng.random() < 0.85:
return rng.choice(used_vars)
return rng.choice(CONSTS)
head_args = tuple(pick() for _ in range(arity[head_pred]))
body = list(positives)
if rng.random() < 0.35:
if constrained:
lower = [p for p in arity if levels[p] < hl]
else:
lower = list(arity)
if lower:
bp = rng.choice(lower)
body.append(
Literal(bp, tuple(pick() for _ in range(arity[bp])), True)
)
program.add_rule(Rule(head_pred, head_args, tuple(body)))
return program
# ==========================================================================
# 1. parser
# ==========================================================================
class ParserTests(unittest.TestCase):
def test_facts_and_rule_shape(self):
p = parse_program("edge(a, b).\npath(X, Y) :- edge(X, Y).\n")
self.assertEqual(p.facts, {"edge": {("a", "b")}})
self.assertEqual(len(p.rules), 1)
rule = p.rules[0]
self.assertEqual(rule.head_pred, "path")
self.assertEqual(rule.head_args, ("?X", "?Y"))
self.assertEqual(str(rule), "path(?X, ?Y) :- edge(?X, ?Y).")
def test_comments_and_whitespace(self):
text = "% a comment\n\nedge(a,b). % trailing\npath(X,Y):-edge(X,Y).\n"
p = parse_program(text)
self.assertEqual(p.facts["edge"], {("a", "b")})
self.assertEqual(len(p.rules), 1)
def test_bang_is_not(self):
p = parse_program("dom(a). p(X) :- dom(X), ! q(X). q(b).\n")
self.assertTrue(p.rules[0].negated)
self.assertEqual(p.rules[0].negated[0].pred, "q")
def test_zero_arity_and_constants(self):
p = parse_program('flag(). n(1). s("x").\n')
self.assertEqual(p.facts["flag"], {()})
self.assertEqual(p.facts["n"], {("1",)})
self.assertEqual(p.facts["s"], {('"x"',)})
def test_variable_in_fact_rejected(self):
with self.assertRaises(ParseError):
parse_program("p(X).\n")
def test_bad_character_rejected(self):
with self.assertRaises(ParseError):
parse_program("p(a) # oops.\n")
def test_missing_dot_rejected(self):
with self.assertRaises(ParseError):
parse_program("p(a)\n")
def test_negated_head_rejected(self):
with self.assertRaises(ParseError):
parse_program("dom(a). not p(X) :- dom(X).\n")
def test_unbalanced_parens_rejected(self):
with self.assertRaises(ParseError):
parse_program("p(a, b.\n")
def test_empty_body_rejected(self):
with self.assertRaises(ParseError):
parse_program("p(a) :- .\n")
# ==========================================================================
# 2. static checks
# ==========================================================================
class StaticCheckTests(unittest.TestCase):
def test_arity_conflict_between_fact_and_body(self):
p = parse_program("q(a). p(X) :- q(X, Y).\n")
with self.assertRaises(ArityError):
evaluate(p)
def test_arity_conflict_between_facts(self):
p = parse_program("p(a, b). p(a).\n")
with self.assertRaises(ArityError):
check_arity(p)
def test_unbound_head_variable(self):
p = parse_program("q(a). p(X) :- q(Y).\n")
with self.assertRaises(SafetyError):
evaluate(p)
def test_rule_without_positive_body(self):
p = parse_program("dom(a). p(X) :- not dom(X).\n")
with self.assertRaises(SafetyError):
check_safety(p)
def test_unbound_variable_inside_negation(self):
p = parse_program("dom(a). q(a). p(X) :- q(Y), not dom(X).\n")
with self.assertRaises(SafetyError):
check_safety(p)
def test_anonymous_variable_in_head(self):
p = parse_program("q(a). p(_) :- q(a).\n")
with self.assertRaises(SafetyError):
check_safety(p)
def test_negation_inside_recursive_cycle_rejected(self):
p = parse_program("dom(a). p(X) :- dom(X), not q(X). q(X) :- p(X).\n")
with self.assertRaises(StratificationError):
stratify(p)
self.assertIsNone(oracle_stratify(p))
def test_mutual_negation_rejected(self):
p = parse_program(
"dom(a). p(X) :- dom(X), not q(X). q(X) :- dom(X), not p(X).\n"
)
with self.assertRaises(StratificationError):
stratify(p)
self.assertIsNone(oracle_stratify(p))
def test_positive_recursion_is_not_a_stratification_error(self):
p = parse_program(
"edge(a,b). edge(b,c).\n"
"path(X,Y) :- edge(X,Y).\n"
"path(X,Y) :- edge(X,Z), path(Z,Y).\n"
)
waves = stratify(p)
self.assertEqual(len(waves), 2)
self.assertEqual(waves[0], frozenset({"edge"}))
self.assertEqual(waves[1], frozenset({"path"}))
self.assertIsNotNone(oracle_stratify(p))
def test_negation_to_lower_stratum_is_fine(self):
p = parse_program(
"dom(a). dom(b). q(a).\n"
"r(X) :- dom(X), not q(X).\n"
)
waves = stratify(p)
names = [sorted(w) for w in waves]
self.assertLess(names.index(["q"]), names.index(["r"]))
self.assertEqual(evaluate(p)["r"], frozenset({("b",)}))
# ==========================================================================
# 3. evaluation KATs
# ==========================================================================
class EvaluationTests(unittest.TestCase):
def test_transitive_closure(self):
p = parse_program(
"edge(a,b). edge(b,c). edge(c,d).\n"
"path(X,Y) :- edge(X,Y).\n"
"path(X,Y) :- edge(X,Z), path(Z,Y).\n"
)
rels = evaluate(p)
self.assertEqual(
rels["path"],
frozenset({("a", "b"), ("a", "c"), ("a", "d"),
("b", "c"), ("b", "d"), ("c", "d")}),
)
def test_same_generation_nonlinear_recursion(self):
p = parse_program(
"person(a1). person(a2). person(b1). person(b2).\n"
"parent(b1, a1). parent(b1, a2). parent(b2, a1). parent(b2, a2).\n"
"sg(X, X) :- person(X).\n"
"sg(X, Y) :- parent(X, A), parent(Y, B), sg(A, B).\n"
)
rels = evaluate(p)
self.assertEqual(
rels["sg"],
frozenset({("a1", "a1"), ("a2", "a2"), ("b1", "b1"), ("b2", "b2"),
("b1", "b2"), ("b2", "b1")}),
)
def test_stratified_negation(self):
p = parse_program(
"node(a). node(b). node(c).\n"
"source(a).\n"
"edge(a,b).\n"
"reach(X) :- source(X).\n"
"reach(X) :- edge(Y,X), reach(Y).\n"
"unreachable(X) :- node(X), not reach(X).\n"
)
rels = evaluate(p)
self.assertEqual(rels["reach"], frozenset({("a",), ("b",)}))
self.assertEqual(rels["unreachable"], frozenset({("c",)}))
def test_anonymous_variable_in_negated_literal(self):
p = parse_program(
"node(a). node(b). node(c). edge(a,b). edge(c,a).\n"
"sink(X) :- node(X), not edge(X, _).\n"
)
rels = evaluate(p)
self.assertEqual(rels["sink"], frozenset({("b",)}))
def test_repeated_variable_inside_one_literal(self):
# Regression: the join reorders positive literals smallest-first, so
# `p1(X, X)` can be the FIRST literal joined. A repeated variable must
# still constrain both argument positions of that single tuple.
p = parse_program(
"e0(a, a). e0(a, b). e0(b, a). e1(b, a). e1(b, b). e1(c, a).\n"
"q1(X) :- e0(Z, X).\n"
"p1(Y, c) :- e0(Y, Z), not q0(Z).\n"
"p0(X, c) :- e1(X, Y), p1(X, X), not q0(X).\n"
)
rels = evaluate(p)
self.assertEqual(rels["p0"], frozenset())
self.assertEqual(rels["p1"], frozenset({("a", "c"), ("b", "c")}))
self.assertEqual(rels["p0"], oracle_evaluate(p)["p0"])
def test_negation_over_empty_relation_always_succeeds(self):
p = parse_program("dom(a). p(X) :- dom(X), not missing(X).\n")
rels = evaluate(p)
self.assertEqual(rels["p"], frozenset({("a",)}))
self.assertEqual(rels["missing"], frozenset())
def test_zero_arity_derivation(self):
p = parse_program("flag(). ready() :- flag().\n")
rels = evaluate(p)
self.assertEqual(rels["ready"], frozenset({()}))
def test_constant_arguments_and_repeated_variable(self):
p = parse_program("e(a,a). e(a,b). loop(X) :- e(X, X).\n")
rels = evaluate(p)
self.assertEqual(rels["loop"], frozenset({("a",)}))
def test_duplicate_facts_collapse(self):
p = parse_program("edge(a,b). edge(a,b). edge(a,b).\n")
rels = evaluate(p)
self.assertEqual(rels["edge"], frozenset({("a", "b")}))
def test_ground_rule_without_variables(self):
p = parse_program("dom(a). mark(a) :- dom(a).\n")
rels = evaluate(p)
self.assertEqual(rels["mark"], frozenset({("a",)}))
def test_order_independence(self):
text = (
"edge(a,b). edge(b,c). edge(c,d).\n"
"path(X,Y) :- edge(X,Y).\n"
"path(X,Y) :- edge(X,Z), path(Z,Y).\n"
)
base = evaluate(parse_program(text))
rng = random.Random(7)
for _ in range(5):
clauses = [c.strip() for c in text.strip().split("\n") if c.strip()]
rng.shuffle(clauses)
shuffled = evaluate(parse_program("\n".join(clauses) + "\n"))
self.assertEqual(shuffled, base)
def test_repeated_evaluation_is_deterministic(self):
text = (
"n(a). n(b). n(c). n(d).\n"
"e(a,b). e(b,c). e(c,d). e(d,a).\n"
"r(X,Y) :- e(X,Y).\n"
"r(X,Y) :- r(X,Z), e(Z,Y).\n"
"lonely(X) :- n(X), not r(X, _).\n"
)
first = evaluate(parse_program(text))
for _ in range(3):
self.assertEqual(evaluate(parse_program(text)), first)
self.assertEqual(len(first["r"]), 16)
# ==========================================================================
# 4. differential testing against the oracle
# ==========================================================================
def differential_run(programs, counters=None):
"""Evaluate every program with the engine and with the oracle; return a list
of (index, program_text, explanation) mismatches. `counters` accumulates the
coverage numbers."""
if counters is None:
counters = {"programs": 0, "unstratified": 0, "predicates": 0}
failures = []
for idx, program in enumerate(programs):
stats = {}
try:
got = evaluate(program, stats)
got_error = None
except (StratificationError, ArityError, SafetyError) as exc:
got, got_error = None, type(exc).__name__
want = oracle_evaluate(program)
counters["programs"] += 1
if want is None:
counters["unstratified"] += 1
if got_error != "StratificationError":
failures.append(
(idx, program.text(),
f"oracle says unstratified, engine said {got_error!r}")
)
continue
if got_error is not None:
failures.append(
(idx, program.text(), f"oracle stratified, engine raised {got_error}")
)
continue
counters["predicates"] += len(got)
if got != want:
diff = {
p: (sorted(got.get(p, ())), sorted(want.get(p, ())))
for p in sorted(set(got) | set(want))
if got.get(p) != want.get(p)
}
failures.append((idx, program.text(), f"relation mismatch {diff}"))
return failures
class DifferentialTests(unittest.TestCase):
def _run(self, programs):
failures = differential_run(programs, DIFFERENTIAL_RUNS)
return failures
def test_differential_constrained_300(self):
rng = random.Random(20261024)
programs = [gen_program(rng, constrained=True) for _ in range(300)]
failures = self._run(programs)
self.assertEqual(failures[:3], [], f"{len(failures)}/300 mismatches")
def test_differential_free_negation_300(self):
rng = random.Random(20261025)
programs = [gen_program(rng, constrained=False) for _ in range(300)]
failures = self._run(programs)
self.assertEqual(failures[:3], [], f"{len(failures)}/300 mismatches")
self.assertGreater(
DIFFERENTIAL_RUNS["unstratified"], 0,
"the free generator produced no unstratified program, so the "
"stratification check was not exercised",
)
def test_generator_actually_builds_nontrivial_programs(self):
rng = random.Random(1)
programs = [gen_program(rng, constrained=True) for _ in range(50)]
self.assertTrue(all(p.rules for p in programs))
self.assertGreater(sum(len(p.rules) for p in programs), 100)
self.assertTrue(any(any(lit.negated for lit in r.body)
for p in programs for r in p.rules))
# ==========================================================================
# 5. semi-naive does less work than naive
# ==========================================================================
class WorkReductionTests(unittest.TestCase):
@staticmethod
def chain_program(n):
facts = "".join(f"edge(n{i}, n{i + 1}).\n" for i in range(n - 1))
return parse_program(
facts
+ "path(X,Y) :- edge(X,Y).\n"
+ "path(X,Y) :- edge(X,Z), path(Z,Y).\n"
)
def test_semi_naive_enumerates_fewer_bindings(self):
program = self.chain_program(12)
engine_stats, oracle_stats = {}, {}
rels = evaluate(program, engine_stats)
oracle_rels = oracle_evaluate(program, oracle_stats)
self.assertEqual(rels, oracle_rels)
self.assertEqual(len(rels["path"]), 66) # C(12,2)
self.assertLess(
engine_stats["bindings"], oracle_stats["bindings"],
f"semi-naive bindings {engine_stats['bindings']} vs naive "
f"{oracle_stats['bindings']}",
)
# The delta rule must buy a real margin, not a rounding error.
self.assertLess(engine_stats["bindings"], oracle_stats["bindings"] // 2)
def test_stats_are_populated(self):
stats = {}
evaluate(self.chain_program(6), stats)
for key in ("rounds", "bindings", "derived", "rule_runs"):
self.assertIn(key, stats)
self.assertGreater(stats["derived"], 0)
self.assertGreater(stats["rounds"], 1)
# ==========================================================================
# 6. CLI
# ==========================================================================
class CliTests(unittest.TestCase):
SCRIPT = os.path.join(HERE, "datalog.py")
def _run(self, text, *args):
with tempfile.TemporaryDirectory(prefix=".datalog-fixture-", dir=HERE) as tmp:
path = os.path.join(tmp, "program.dl")
with open(path, "w", encoding="utf-8") as fh:
fh.write(text)
proc = subprocess.run(
[sys.executable, self.SCRIPT, path, *args],
capture_output=True, text=True, timeout=120,
)
return proc
def test_json_contract(self):
proc = self._run(
"node(a). node(b). node(c). source(a). edge(a,b).\n"
"reach(X) :- source(X).\n"
"reach(X) :- edge(Y,X), reach(Y).\n"
"unreach(X) :- node(X), not reach(X).\n",
"--stats",
)
self.assertEqual(proc.returncode, 0, proc.stderr)
payload = json.loads(proc.stdout)
self.assertEqual(payload["predicates"]["reach"], [["a"], ["b"]])
self.assertEqual(payload["predicates"]["unreach"], [["c"]])
self.assertEqual(payload["predicates"]["node"], [["a"], ["b"], ["c"]])
self.assertIn("stats", payload)
self.assertTrue(payload["strata"])
def test_unstratified_program_exits_2(self):
proc = self._run("dom(a). p(X) :- dom(X), not q(X). q(X) :- p(X).\n")
self.assertEqual(proc.returncode, 2)
self.assertIn("StratificationError", proc.stderr)
def test_text_output(self):
proc = self._run("e(a,b).\n", "--text")
self.assertEqual(proc.returncode, 0, proc.stderr)
self.assertIn("e(a, b).", proc.stdout)
def tearDownModule():
print(
"\n[differential] programs evaluated: {programs} "
"(unstratified: {unstratified}, predicate-relations compared: {predicates})".format(
**DIFFERENTIAL_RUNS
),
file=sys.stderr,
)
if __name__ == "__main__":
unittest.main(verbosity=1)