Frank14f/CelerisLab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: add scaled N256 Taylor-Green diagnostic

Browse Source
This commit is contained in:
Codex
2026-10-18 17:00:55 -07:00
parent eec4677db1
commit 92b8afeeed
3 changed files with 249 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/validation/taylor_green/assess_projected_pairs.py
+105 -14
View File
@@ -1,6 +1,7 @@
"""New-source four-case repeat of 11901a7; frozen oracle and gates are unchanged."""
import argparse
import hashlib
import json
import math
from pathlib import Path
@@ -8,16 +9,25 @@ from pathlib import Path
from assess import BASELINE_FACTOR, FLOORS
from oracle import LIMITS, TIMES, U, rectangles
from run_case import (INITIALIZATION, sample_steps, NUMERICAL_REVISION, SOURCE_SHA256,
BUILD_CAPTURE_SHA256, REFERENCE_HARNESS)
BUILD_CAPTURE_SHA256, REFERENCE_HARNESS, SCALED_DIAGNOSTIC)
# Host-generated snapshots of a135252 at the frozen two local configurations.
# Host-generated snapshots of a135252 at the frozen local configurations.
# These describe expected inputs; actual sidecars are captured before cleanup.
BUILD_SOURCES = {
128: "803177416b7599ab89cc5afd35cf4a28a673b2f4bb8e8a6a1abc0b6d77610873",
256: "17ec5e60f87b840d3d13b8778bc8a3aecf485469c5a061af46c5cc19b73396cf",
512: "4e85e38098858ec95c137d1b1e96846a58d9cf3cc242b349a3ad7c8b8e3ce0aa",
}
CASES = ((100, 128, 1, "constant"), (100, 256, 1, "constant"),
(100, 128, 2, "constant"), (100, 128, 2, "projected_linear"))
REFERENCE_REVISION = "2cf9853952e52f59bea63cf5c1d3a1cd76e53eed"
# Audited 2cf9853 scalar receipts, archived at evidence commit 90849a55cad4a0da9b72747da7bfe6eb75962a44.
REFERENCE_RECEIPTS = {
"uniform128": "724efafc267eaa59dcc59e8d4c16177b67f6c6a08fd2732a481833237bb19cee",
"uniform256": "3c312a99ebf747a135674dd481b77455a3d2c9a406ca388f30524b37dad9bbe9",
"constant128": "2ee28efd1e81b2e28b7902b6c4f95bb2c1ea4e5f0ec484638fa5ae83dca34bba",
"projected128": "ee67b1d4d90a048c0d3d1a16447c6eab3d89c94d151f791306e373f96dd8b963",
}
NUMERICS = dict(lattice_model="D2Q9", collision="SRT", store_precision="FP32",
ddf_shifting=True, streaming="double_buffer", les_enabled=False,
boundary="periodic", block_shape=[8, 8],
@@ -73,7 +83,11 @@ def valid_builds(record, n, depth):
return True
def assess(records, expected_revision=None):
def assess(records, expected_revision=None, *, diagnostic=None):
if diagnostic not in (None, SCALED_DIAGNOSTIC):
raise ValueError("Unknown diagnostic")
base_n = 256 if diagnostic else 128
cases = tuple((re, n * base_n // 128, depth, scheme) for re, n, depth, scheme in CASES)
indexed, duplicates = {}, []
for record in records:
key = tuple(record.get(k) for k in ("re_L", "n", "depth", "explosion"))
@@ -81,9 +95,11 @@ def assess(records, expected_revision=None):
duplicates.append(str(key))
else:
indexed[key] = record
missing = [str(k) for k in CASES if k not in indexed]
checks = dict(complete_matrix=not missing, no_extra_cases=set(indexed) <= set(CASES),
missing = [str(k) for k in cases if k not in indexed]
checks = dict(complete_matrix=not missing, no_extra_cases=set(indexed) <= set(cases),
no_duplicate_cases=not duplicates, finite_receipts=finite_tree(records))
if diagnostic or any("diagnostic" in r for r in records):
checks["diagnostic_identity"] = all(r.get("diagnostic") == diagnostic for r in records)
sources = [r.get("source", {}) for r in records]
checks["same_source_and_harness"] = bool(sources) and len({
json.dumps(s, sort_keys=True) for s in sources}) == 1
@@ -106,9 +122,11 @@ def assess(records, expected_revision=None):
) and len(set(devices)) == 1
enabled = [r.get("early_pressure", {}).get("enabled") for r in records]
checks["same_early_sampling"] = bool(enabled) and all(type(v) is bool for v in enabled) and len(set(enabled)) == 1
if diagnostic:
checks["diagnostic_early_sampling"] = all(v is True for v in enabled)
shared_valid = all(checks.values())
complete, case_valid, early = {}, {}, []
for key in CASES:
for key in cases:
if key not in indexed:
continue
r = indexed[key]
@@ -135,7 +153,7 @@ def assess(records, expected_revision=None):
(("module", "__init__.py"), ("compiler", "cuda/compiler_v2.py"), ("kernel_root", "lbm/kernels")))
expected_patch = [] if depth == 1 else rectangles(n, depth)
checks[label + " patch"] = json.dumps(r.get("refinements")) == json.dumps(expected_patch)
cells = [n*n] if depth == 1 else [8704, 30720]
cells = [n*n] if depth == 1 else [17*n*n//32, 15*n*n//8]
hops = [16*n * 2**l for l in range(depth)]
work = r.get("workload", {})
checks[label + " workload"] = (
@@ -168,19 +186,22 @@ def assess(records, expected_revision=None):
comparisons, scheme_results = [], {}
for scheme in ("constant", "projected_linear"):
key = (100, 128, 2, scheme)
key = (100, base_n, 2, scheme)
for metric, floor in FLOORS.items():
amr, coarse, fine = (worst(k, metric) for k in (key, CASES[0], CASES[1]))
amr, coarse, fine = (worst(k, metric) for k in (key, cases[0], cases[1]))
budget = None if coarse is None else max(BASELINE_FACTOR * coarse, floor)
budget = budget if finite(budget) else None
ratio = None if amr is None or fine is None else amr / max(fine, 1e-30)
passed = amr is not None and budget is not None and amr <= budget
checks[f"baseline {scheme}/{metric}"] = passed
comparisons.append(dict(explosion=scheme, metric=metric, amr=amr,
uniform128=coarse, baseline_budget=budget, passed=passed,
uniform256=fine, amr_to_uniform256=ratio if finite(ratio) else None))
baseline_budget=budget, passed=passed,
**{f"uniform{base_n}": coarse, f"uniform{2*base_n}": fine,
f"amr_to_uniform{2*base_n}": ratio if finite(ratio) else None}))
if diagnostic:
comparisons[-1]["floor_dominates"] = None if coarse is None else floor >= BASELINE_FACTOR * coarse
baseline_passed = all(checks[f"baseline {scheme}/{metric}"] for metric in FLOORS)
valid = shared_valid and all(case_valid.get(k, False) for k in (CASES[0], CASES[1], key))
valid = shared_valid and all(case_valid.get(k, False) for k in (cases[0], cases[1], key))
scheme_results[scheme] = dict(passed=valid and baseline_passed,
valid_case_and_controls=valid, baseline_passed=baseline_passed)
return dict(passed=all(checks.values()), missing=missing, duplicates=duplicates, checks=checks,
@@ -190,19 +211,89 @@ def assess(records, expected_revision=None):
floors=FLOORS, baseline_factor=BASELINE_FACTOR,
main_statistic="maximum of four post-initial samples at t*=.12,.24,.36,.48",
early_statistic="sampled pressure maximum only, t*=k*.001875, k=0..64; not an all-step or continuous maximum",
finest_control="uniform256 and AMR/ uniform256 ratios are descriptive, not acceptance gates",
finest_control=f"uniform{2*base_n} and AMR/ uniform{2*base_n} ratios are descriptive, not acceptance gates",
scope="2D periodic Re100 smooth-decay diagnostic; no new algorithm, convergence-order or general AMR qualification")
def load_reference(paths):
"""Use the four audited historical files verbatim, never relabel them fresh controls."""
blobs = [path.read_bytes() for path in paths]
hashes = [hashlib.sha256(blob).hexdigest() for blob in blobs]
if len(hashes) != 4 or set(hashes) != set(REFERENCE_RECEIPTS.values()):
raise ValueError("Historical N128 receipt hashes differ from the audited 2cf9853 evidence")
return [json.loads(blob) for blob in blobs]
def resolution_trend(records, reference, expected_revision):
"""Signed changes are descriptive; they never replace the frozen acceptance budgets."""
old = assess(reference, REFERENCE_REVISION)
new = assess(records, expected_revision, diagnostic=SCALED_DIAGNOSTIC)
# Numerical screen failures still have useful raw errors; only malformed inputs block trends.
valid = all(v for result in (old, new) for k, v in result["checks"].items()
if not k.startswith("baseline ") and not k.endswith(" absolute"))
valid = valid and all(maximum(r["rows"], m) is not None for r in reference + records for m in LIMITS)
oracle = reference[0]["source"]["harness_sha256"].get("oracle.py") if valid else None
valid = valid and bool(oracle) and records[0]["source"]["harness_sha256"].get("oracle.py") == oracle
if not valid:
return dict(valid=False, per_time=[], worst_post_initial=[], uniform256_bridge=[])
key = lambda r: (r["re_L"], r["n"], r["depth"], r["explosion"])
before, after = {key(r): r for r in reference}, {key(r): r for r in records}
def change(role, metric, low, high, **extra):
ratio = high / low if low else None
percent = 100*(ratio-1) if ratio is not None else None
return dict(role=role, metric=metric, n128_campaign_error=low, n256_campaign_error=high,
signed_change=high-low, new_to_old=ratio if finite(ratio) else None,
ratio_unavailable_reason=None if finite(ratio) else "zero historical error" if not low else "overflow",
change_percent=percent if finite(percent) else None, **extra)
per_time, worst, bridge = [], [], []
for old_key, role in zip(CASES, ("uniform_root", "uniform_fine", "constant", "projected_linear")):
re, n, depth, scheme = old_key
low, high = before[old_key], after[(re, 2*n, depth, scheme)]
for metric in LIMITS:
per_time.extend(change(role, metric, a[metric], b[metric], tstar=a["tstar"])
for a, b in zip(low["rows"], high["rows"]))
worst.append(change(role, metric, maximum(low["rows"][1:], metric),
maximum(high["rows"][1:], metric)))
for a, b in zip(before[CASES[1]]["rows"], after[CASES[1]]["rows"]):
bridge.extend(dict(metric=m, tstar=a["tstar"], historical=a[m], fresh=b[m], signed_change=b[m]-a[m])
for m in LIMITS)
low, high = before[CASES[1]], after[CASES[1]]
bridge_identity = dict(main_rows_identical=low["rows"] == high["rows"],
early_pressure_rows_identical=low["early_pressure"]["rows"] == high["early_pressure"]["rows"])
same_runtime = reference[0]["runtime_identity"] == records[0]["runtime_identity"]
return dict(valid=True, reference_revision=REFERENCE_REVISION, reference_receipt_sha256=REFERENCE_RECEIPTS,
reference_scheme_results=old["scheme_results"], per_time=per_time, worst_post_initial=worst,
uniform256_bridge=bridge, uniform256_bridge_identity=bridge_identity,
same_recorded_runtime=same_runtime,
trend_interpretation_ready=all(bridge_identity.values()) and same_runtime,
scope="Historical same-production/oracle reference versus fresh controls; same-placement two-resolution trend only, not convergence order or phase robustness")
if __name__ == "__main__":
p = argparse.ArgumentParser(description=__doc__)
p.add_argument("files", type=Path, nargs="+")
p.add_argument("--output", type=Path, required=True)
p.add_argument("--expected-revision", required=True)
p.add_argument("--diagnostic", choices=[SCALED_DIAGNOSTIC])
p.add_argument("--reference-files", type=Path, nargs=4)
args = p.parse_args()
if args.output.exists():
p.error("Output already exists; choose a new path to retain prior evidence")
result = assess([json.loads(path.read_text()) for path in args.files], args.expected_revision)
if bool(args.diagnostic) != bool(args.reference_files):
p.error("--diagnostic scaled_n256 requires --reference-files; references are diagnostic-only")
records = [json.loads(path.read_text()) for path in args.files]
result = assess(records, args.expected_revision, diagnostic=args.diagnostic)
if args.diagnostic:
try:
reference = load_reference(args.reference_files)
except ValueError as exc:
p.error(str(exc))
result["diagnostic"] = args.diagnostic
result["resolution_trend"] = resolution_trend(records, reference, args.expected_revision)
result["checks"]["valid_historical_reference_and_trend"] = result["resolution_trend"]["valid"]
result["passed"] = all(result["checks"].values())
args.output.write_text(json.dumps(result, indent=2, allow_nan=False) + "\n")
print(json.dumps(result, indent=2, allow_nan=False))
raise SystemExit(0 if result["passed"] else 1)
tests/validation/taylor_green/run_case.py
+15 -2
View File
@@ -18,6 +18,7 @@ SOURCE_SHA256 = "a5dbfd804befce868067d79a23e5fd47dce7f505b3f9b1bb2145fb57cb5f217
BUILD_CAPTURE_SHA256 = "3dbaa6547cb0a4cb1541054f3b64ad8b6e18cc70527ac6166e56b944e9bd464a"
REFERENCE_HARNESS = "11901a7f77557a901e017a171eeae1758d584fa5"
INITIALIZATION = "physical FP64 equilibrium G at actual cell centers; rho=1+3p"
SCALED_DIAGNOSTIC = "scaled_n256"
def sample_steps(n, early_pressure=False):
@@ -145,10 +146,12 @@ def main():
p.add_argument("--run-gpu", action="store_true")
p.add_argument("--dry-run", action="store_true")
p.add_argument("--re", type=int, choices=[100, 1000], required=True)
p.add_argument("--n", type=int, choices=[32, 64, 128, 256], required=True)
p.add_argument("--n", type=int, choices=[32, 64, 128, 256, 512], required=True)
p.add_argument("--depth", type=int, choices=[1, 2, 3], required=True)
p.add_argument("--output", type=Path, required=True)
p.add_argument("--explosion", choices=["constant", "projected_linear"], default="constant")
p.add_argument("--diagnostic", choices=[SCALED_DIAGNOSTIC],
help="explicit four-case N256/fine512 same-placement diagnostic")
p.add_argument("--early-pressure", action="store_true",
help="also save 65 scalar pressure samples at t*=k*.001875, k=0..64")
args = p.parse_args()
@@ -156,6 +159,13 @@ def main():
p.error("Output already exists; choose a new path to retain prior evidence")
if args.explosion != "constant" and args.depth != 2:
p.error("projected_linear requires depth 2")
if args.diagnostic:
cases = {(256, 1, "constant"), (512, 1, "constant"),
(256, 2, "constant"), (256, 2, "projected_linear")}
if args.re != 100 or (args.n, args.depth, args.explosion) not in cases or not args.early_pressure:
p.error("scaled_n256 requires one of its four Re100 cases and --early-pressure")
elif args.n == 512:
p.error("N512 requires --diagnostic scaled_n256")
selected = os.environ.get("CUDA_VISIBLE_DEVICES", "")
if not args.dry_run and (not args.run_gpu or not selected or "," in selected or selected == "-1"):
p.error("--run-gpu and exactly one explicit CUDA_VISIBLE_DEVICES are required")
@@ -164,7 +174,8 @@ def main():
source = verify_source(repo, args.expected_revision)
runtime = runtime_source(repo)
if args.dry_run:
print(json.dumps(dict(source=source, runtime_source=runtime, dry_run=True)))
print(json.dumps(dict(source=source, runtime_source=runtime, dry_run=True,
**({"diagnostic": args.diagnostic} if args.diagnostic else {}))))
return 0
from CelerisLab.cuda.context import CudaContext
from CelerisLab.lbm.blocks import BlockLayout2D
@@ -212,6 +223,8 @@ def main():
sampled_max_pressure_l2=None),
completed=False,
)
if args.diagnostic:
report["diagnostic"] = args.diagnostic
args.output.parent.mkdir(parents=True, exist_ok=True)
def save():
temporary = args.output.with_suffix(args.output.suffix + ".tmp")
tests/validation/taylor_green/test_projected_pairs.py
+129 -9
View File
@@ -17,6 +17,7 @@ from unittest.mock import patch
import numpy as np
import run_case
import assess_projected_pairs as assessor
from assess_projected_pairs import (BUILD_SOURCES, CASES, NUMERICS, NUMERICAL_REVISION,
SOURCE_SHA256, assess)
from oracle import LIMITS, TIMES, U, rectangles
@@ -31,7 +32,7 @@ RUNTIME = dict(device=dict(name="host mock", pci_bus_id="mock", capability=[7, 0
def source():
return dict(revision=MOCK_REVISION, dirty=False, source_tree_sha256=SOURCE_SHA256,
harness_sha256={"run_case.py": "a" * 64},
harness_sha256={"run_case.py": "a" * 64, "oracle.py": "c" * 64},
build_capture_sha256=run_case.BUILD_CAPTURE_SHA256)
@@ -42,12 +43,14 @@ def build_metadata(size):
arch="sm_70", flags=["--use_fast_math"], ptx_sha256="b" * 64)
def records(early=False):
def records(early=False, diagnostic=None):
from CelerisLab.lbm.hierarchy import build_hierarchy
from CelerisLab.lbm.refinement import ACTIVE
out = []
for re, n, depth, scheme in CASES:
if diagnostic:
n *= 2
cfg = run_case.config(n, re)
refinements = [] if depth == 1 else rectangles(n, depth)
h = build_hierarchy(cfg, refinements)
@@ -66,6 +69,8 @@ def records(early=False):
early_pressure=dict(enabled=early, rows=[
dict(step=s, tstar=s*U/n, pressure_l2=.01 if s not in main_steps else 0.0) for s in early_steps],
sampled_max_pressure_l2=.01 if early else None)))
if diagnostic:
out[-1]["diagnostic"] = diagnostic
return out
@@ -251,7 +256,7 @@ class ProjectedPairsTests(unittest.TestCase):
def test_pinned_build_sources_match_host_generated_inputs(self):
from CelerisLab.cuda import compiler_v2
for size in (128, 256):
for size in (128, 256, 512):
cfg = replace(run_case.config(size, 100), nx=10, ny=10)
cfg.validate()
root = compiler_v2.generate_config(cfg)
@@ -263,6 +268,111 @@ class ProjectedPairsTests(unittest.TestCase):
finally:
compiler_v2.cleanup_build_dir(root)
def test_scaled_geometry_work_and_frozen_gates(self):
data = records(True, run_case.SCALED_DIAGNOSTIC)
self.assertEqual(sum(r["workload"]["active_collision_updates"] for r in data), 4714397696)
self.assertEqual([r["viscosity"] for r in data], [.0768, .1536, .0768, .0768])
self.assertEqual(data[-1]["tau_by_level"], [.7303999999999999, .9608])
for r in data:
self.assertEqual([row["tstar"] for row in r["rows"]], TIMES)
self.assertEqual(len(r["early_pressure"]["rows"]), 65)
if r["depth"] == 2:
self.assertEqual(r["refinements"], [[(32, 32, 224, 192)]])
self.assertEqual(r["workload"]["active_cells_by_level"], [34816, 122880])
self.assertEqual(r["workload"]["level_hops"], [4096, 8192])
result = assess(data, MOCK_REVISION, diagnostic=run_case.SCALED_DIAGNOSTIC)
self.assertTrue(result["passed"])
self.assertEqual(result["floors"], dict(velocity_l2=.002, pressure_l2=.02, energy_relative=.004))
self.assertTrue(all(row["floor_dominates"] for row in result["comparisons"]))
data[-1]["rows"][-1]["velocity_l2"] = .00200001
self.assertFalse(assess(data, diagnostic=run_case.SCALED_DIAGNOSTIC)["passed"])
self.assertFalse(assess(data)["passed"])
data[-1]["rows"][-1]["velocity_l2"] = 0.0
for mutate in (lambda r: r.pop("diagnostic"), lambda r: r.update(diagnostic="wrong"),
lambda r: r["source"].update(revision="wrong"),
lambda r: r["early_pressure"].update(enabled=False),
lambda r: r.update(refinements=[[(32, 32, 224, 200)]])):
bad = deepcopy(data)
mutate(bad[-1])
self.assertFalse(assess(bad, MOCK_REVISION, diagnostic=run_case.SCALED_DIAGNOSTIC)["passed"])
def test_raw_resolution_changes_and_reference_identity(self):
old, new = records(True), records(True, run_case.SCALED_DIAGNOSTIC)
for r in old:
r["source"]["revision"] = assessor.REFERENCE_REVISION
old[-2]["rows"][-1]["velocity_l2"] = .004 # historical constant failure remains usable
old[-1]["rows"][-1]["velocity_l2"] = .001
new[-1]["rows"][-1]["velocity_l2"] = .0015 # floor pass, but worse raw error
result = assessor.resolution_trend(new, old, MOCK_REVISION)
self.assertTrue(result["valid"])
self.assertTrue(result["trend_interpretation_ready"])
self.assertFalse(result["reference_scheme_results"]["constant"]["passed"])
self.assertEqual((len(result["per_time"]), len(result["worst_post_initial"]),
len(result["uniform256_bridge"])), (80, 16, 20))
row = next(r for r in result["per_time"] if r["role"] == "projected_linear"
and r["metric"] == "velocity_l2" and r["tstar"] == .48)
self.assertEqual((row["signed_change"], row["new_to_old"], row["change_percent"]), (.0005, 1.5, 50.))
self.assertTrue(assess(new, diagnostic=run_case.SCALED_DIAGNOSTIC)["passed"])
self.assertIsNone(result["per_time"][0]["new_to_old"]) # zero denominator is explicit
self.assertEqual(result["per_time"][0]["ratio_unavailable_reason"], "zero historical error")
new[-1]["rows"][-1]["velocity_l2"] = .06 # retain raw errors even when absolute screen fails
self.assertTrue(assessor.resolution_trend(new, old, MOCK_REVISION)["valid"])
self.assertFalse(assess(new, diagnostic=run_case.SCALED_DIAGNOSTIC)["passed"])
new[-1]["rows"][-1]["velocity_l2"] = .0015
for sample in ("rows", "early_pressure"):
bad = deepcopy(new)
if sample == "rows":
bad[0]["rows"][-1]["velocity_l2"] = .0001
else:
bad[0]["early_pressure"]["rows"][1]["pressure_l2"] = .009
result = assessor.resolution_trend(bad, old, MOCK_REVISION)
self.assertTrue(result["valid"])
self.assertFalse(result["trend_interpretation_ready"])
for mutate in (lambda r: r["source"].update(revision=MOCK_REVISION),
lambda r: r["source"].update(source_tree_sha256="wrong"),
lambda r: r["source"]["harness_sha256"].update(**{"oracle.py": "wrong"}),
lambda r: r.update(diagnostic=run_case.SCALED_DIAGNOSTIC),
lambda r: r.update(refinements=[]), lambda r: r["rows"].pop()):
bad = deepcopy(old)
mutate(bad[-1])
self.assertFalse(assessor.resolution_trend(new, bad, MOCK_REVISION)["valid"])
with tempfile.TemporaryDirectory() as tmp:
paths = [Path(tmp) / f"{i}.json" for i in range(4)]
for path, receipt in zip(paths, old):
path.write_text(json.dumps(receipt))
with self.assertRaisesRegex(ValueError, "receipt hashes"):
assessor.load_reference(paths)
witnesses = {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in paths}
with patch.object(assessor, "REFERENCE_RECEIPTS", witnesses):
self.assertEqual(assessor.load_reference(paths), json.loads(json.dumps(old)))
with self.assertRaises(ValueError):
assessor.load_reference(paths[:-1] + paths[:1])
paths[-1].write_text(paths[-1].read_text() + " ")
with self.assertRaises(ValueError):
assessor.load_reference(paths)
def test_scaled_runner_requires_explicit_bounded_diagnostic(self):
with tempfile.TemporaryDirectory() as tmp:
base = ["run_case.py", "--repo", str(REPO), "--expected-revision", MOCK_REVISION,
"--output", str(Path(tmp) / "case.json"), "--dry-run"]
diagnostic = ["--diagnostic", run_case.SCALED_DIAGNOSTIC, "--early-pressure"]
for n, depth, re, extra, valid in (
(512, 1, 100, [], False), (512, 1, 100, diagnostic, True),
(256, 2, 100, diagnostic, True), (128, 2, 100, diagnostic, False),
(256, 3, 100, diagnostic, False), (256, 2, 1000, diagnostic, False),
(256, 2, 100, diagnostic[:2], False), (512, 2, 100, diagnostic, False)):
argv = base + ["--n", str(n), "--depth", str(depth), "--re", str(re)] + extra
with self.subTest(argv=argv), patch.object(sys, "argv", argv), \
patch.object(run_case, "verify_source", return_value=source()) as verify, \
redirect_stdout(io.StringIO()), patch("sys.stderr", io.StringIO()):
if valid:
self.assertEqual(run_case.main(), 0)
else:
with self.assertRaises(SystemExit) as caught:
run_case.main()
self.assertEqual(caught.exception.code, 2)
verify.assert_not_called()
def test_source_preflight_requires_clean_exact_revision_and_tree(self):
import oracle
from tests.validation.multigrid import check_projected_pairs as gate
@@ -349,9 +459,15 @@ class ProjectedPairsTests(unittest.TestCase):
def test_mocked_runner_default_early_and_failure(self):
# Replace the GPU context and steppers before import. No CUDA module or flow is used.
for early, depth, fail in ((False, 1, None), (False, 2, None), (True, 2, None),
cases = [(early, depth, fail, 128, None, "projected_linear" if early else "constant")
for early, depth, fail in
((False, 1, None), (False, 2, None), (True, 2, None),
(True, 2, "resource"), (True, 2, "nonfinite"),
(True, 2, "probe_nonfinite")):
(True, 2, "probe_nonfinite"))]
cases += [(True, depth, None, n, run_case.SCALED_DIAGNOSTIC, scheme)
for n, depth, scheme in ((256, 1, "constant"), (512, 1, "constant"),
(256, 2, "constant"), (256, 2, "projected_linear"))]
for early, depth, fail, n, diagnostic, scheme in cases:
calls, selectors = [], []
class FakeRun:
@@ -395,11 +511,14 @@ class ProjectedPairsTests(unittest.TestCase):
with tempfile.TemporaryDirectory() as tmp:
output = Path(tmp) / "case.json"
argv = ["run_case.py", "--repo", str(REPO), "--re", "100", "--n", "128",
argv = ["run_case.py", "--repo", str(REPO), "--re", "100", "--n", str(n),
"--depth", str(depth), "--output", str(output),
"--expected-revision", MOCK_REVISION, "--run-gpu"]
argv += ["--explosion", scheme]
if early:
argv += ["--early-pressure", "--explosion", "projected_linear"]
argv += ["--early-pressure"]
if diagnostic:
argv += ["--diagnostic", diagnostic]
with patch.dict(sys.modules, modules), patch.object(sys, "argv", argv), \
patch.dict("os.environ", CUDA_VISIBLE_DEVICES="mock"), \
patch.object(run_case, "verify_source", return_value=source()) as verify, \
@@ -413,15 +532,16 @@ class ProjectedPairsTests(unittest.TestCase):
self.assertEqual(run_case.main(), 0)
receipt = json.loads(output.read_text())
verify.assert_called_once_with(REPO, MOCK_REVISION)
self.assertEqual(selectors, ["projected_linear" if early else "constant"] if depth == 2 else [])
self.assertEqual(selectors, [scheme] if depth == 2 else [])
self.assertEqual(receipt["source"], source())
self.assertEqual(receipt["runtime_identity"], RUNTIME)
self.assertEqual(receipt["completed"], not fail)
self.assertEqual(receipt.get("diagnostic"), diagnostic)
self.assertEqual(calls[:2], [0, 1]) # READY probe is part of this trajectory
if not fail:
self.assertEqual([r["tstar"] for r in receipt["rows"]], TIMES)
self.assertEqual(len(receipt["early_pressure"]["rows"]), 65 if early else 0)
self.assertEqual(calls[-1], 2048)
self.assertEqual(calls[-1], 16*n)
else:
self.assertEqual(len(receipt["rows"]), 1)
self.assertEqual(len(receipt["early_pressure"]["rows"]),