From 02bde0f75187786a2ce10abc9a863f750b6343a8 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 19 Oct 2026 02:52:20 +0000 Subject: [PATCH] test: depth-2 twin of the nested shear preset (twin_shear_n128) Test-only opt-in. New shear_twin.py: same two L0->L1 boxes as shear_nested, no L2; leaves [7936,33792], hops [2048,4096], work 154664960. Four fresh cases U128/U256/constant R128 d2/projected R128 d2; physics is the unchanged depth-2/2N shear_wave.assess. run_case: --diagnostic twin_shear_n128 with shear_twin.py+shear_nested.py pins, no snapshot. assess_shear_wave: nested contract made preset-generic (PRESETS) with nested behaviour unchanged. Labels: twin vs nested = effect of adding this nested refinement (not abstract depth). shear_wave.py, shear_nested.py and oracle.py byte-identical. Co-Authored-By: Claude Opus 5.5 --- docs/validation/taylor_green.md | 33 ++++ .../taylor_green/assess_shear_wave.py | 85 ++++++---- tests/validation/taylor_green/run_case.py | 37 +++-- tests/validation/taylor_green/shear_twin.py | 36 ++++ .../taylor_green/test_shear_wave.py | 154 ++++++++++++++++++ 5 files changed, 304 insertions(+), 41 deletions(-) create mode 100644 tests/validation/taylor_green/shear_twin.py diff --git a/docs/validation/taylor_green.md b/docs/validation/taylor_green.md index 8fb7acd..fd2a981 100644 --- a/docs/validation/taylor_green.md +++ b/docs/validation/taylor_green.md @@ -385,3 +385,36 @@ Assessment: `assess_shear_wave.py U128.json U512.json R128c.json R128p.json identities and verdicts are untouched. - This preset changes placement and refined volume as well as depth. It gives no isolated cost of an extra level and no resolution trend. + +### Depth-2 twin of the nested preset (opt-in `twin_shear_n128`) + +`shear_twin.py` uses the same two L0→L1 boxes as `shear_nested.py`, with no +L2. Leaves are `[7936, 33792]`, hops `[2048, 4096]`, and work is 154664960 per +refined case (611319808 for all four). + +The campaign is four fresh runs: +`run_case.py --flow shear_wave --diagnostic twin_shear_n128 --re 100` with +- `--n 128 --depth 1` +- `--n 256 --depth 1` (the finest control is 2N) +- `--n 128 --depth 2` +- `--n 128 --depth 2 --explosion projected_linear` + +There is no snapshot option. Older exact controls serve only as consistency +checks; they are not reused as references. + +Assessment uses the same preset contract as the nested preset, through +`assess_shear_wave.py ... --diagnostic twin_shear_n128`: +- the physics is the unchanged depth-2 / 2N `shear_wave.assess`; +- receipts must pin `shear_twin.py`, `shear_nested.py` and `shear_wave.py`; +- `shear_nested.py` stays byte-identical (fa6cfe87…), so the 335624c nested + receipts still validate. + +Labels: +- *Twin vs nested* is the effect of adding this nested L2 refinement. + Removing L2 also changes the finest spacing and refined volume, so it does + not isolate abstract depth. +- *Twin vs the single-box depth-2 runs* is a placement/geometry contrast at + equal depth. +- Neither contrast is additive by construction. +- The thin-band/acoustic explanation for the nested L1 pressure is a + hypothesis, not established. diff --git a/tests/validation/taylor_green/assess_shear_wave.py b/tests/validation/taylor_green/assess_shear_wave.py index 452f4a6..252fe7e 100644 --- a/tests/validation/taylor_green/assess_shear_wave.py +++ b/tests/validation/taylor_green/assess_shear_wave.py @@ -19,8 +19,9 @@ from pathlib import Path import assess_projected_pairs as tg import shear_wave as sw import shear_nested as nested +import shear_twin as twin from run_case import (NESTED_SHEAR, SCALED_DIAGNOSTIC, SCALED_SHEAR, SHEAR_NESTED_SHA256, - SHEAR_WAVE_SHA256, sample_steps) + SHEAR_TWIN_SHA256, SHEAR_WAVE_SHA256, TWIN_SHEAR, sample_steps) # Audited de5b0cc N128 receipts (evidence commit cba6bca, shear-wave-de5b0cc zip). REFERENCE_REVISION = "de5b0cc080794d0ce44d37f845aa8e8dbf8c2016" @@ -120,15 +121,38 @@ CASE_AGNOSTIC = ("finite_receipts", "same_source_and_harness", "clean_a135252_nu "same_recorded_runtime", "same_explicit_device", "same_early_sampling") -def _nested_case_ok(r, n, depth): - """Per-case identity of the fixed nested preset (scales, builds, patch, work, pins).""" +# Fixed multi-box presets sharing one contract; only geometry, controls, pins differ. +PRESETS = { + NESTED_SHEAR: dict( + module=nested, physics=nested.assess, snapshot=True, + pins={"shear_nested_sha256": SHEAR_NESTED_SHA256}, + finest_control="uniform 4N (U512)", + scope="2D periodic Re100 oblique (2,1) shear wave; fixed three-level/two-region " + "placement; changes placement and refined volume as well as depth, so no " + "isolated-level cost or resolution trend"), + TWIN_SHEAR: dict( + module=twin, physics=sw.assess, snapshot=False, + pins={"shear_nested_sha256": SHEAR_NESTED_SHA256, + "shear_twin_sha256": SHEAR_TWIN_SHA256}, + finest_control="uniform 2N (U256)", + scope="2D periodic Re100 oblique (2,1) shear wave; depth-2 twin of the nested " + "preset (same two L0->L1 boxes, no L2). Twin vs nested is the effect of adding " + "this nested refinement; it also changes finest spacing and refined volume, so it " + "does not isolate abstract depth"), +} + + +def _nested_case_ok(r, n, depth, diagnostic=NESTED_SHEAR): + """Per-case identity of a fixed multi-box preset (scales, builds, patch, work, pins).""" + preset = PRESETS[diagnostic] + geometry = preset["module"] runtime = r.get("runtime_source", {}) package = Path(runtime.get("checkout", "")) / "src/CelerisLab" - cells = [n * n] if depth == 1 else nested.LEAVES + cells = [n * n] if depth == 1 else geometry.LEAVES hops = [16 * n * 2**level for level in range(depth)] work = r.get("workload", {}) snapshot = r.get("final_snapshot") - expected_patch = [] if depth == 1 else nested.REFINEMENTS + expected_patch = [] if depth == 1 else geometry.REFINEMENTS return { "complete_times": r.get("completed") is True and tg.at_steps( r.get("rows", []), sample_steps(n)[0], n), @@ -152,11 +176,11 @@ def _nested_case_ok(r, n, depth): and work.get("active_collision_updates") == sum( c * h for c, h in zip(cells, hops))), "shear_metadata": metadata_ok(r, n), - "nested_module_pin": (r.get("source") or {}).get("shear_nested_sha256") - == SHEAR_NESTED_SHA256, - # Optional, projected depth-3 only; bounded and hashed when present. + "nested_module_pin": all((r.get("source") or {}).get(key) == pin + for key, pin in preset["pins"].items()), + # Optional, nested projected depth-3 only; bounded and hashed when present. "snapshot": snapshot is None or ( - depth == nested.DEPTH and r.get("explosion") == "projected_linear" + preset["snapshot"] and depth == nested.DEPTH and r.get("explosion") == "projected_linear" and snapshot.get("kept") is True and isinstance(snapshot.get("sha256"), str) and len(snapshot["sha256"]) == 64 and 0 < snapshot.get("bytes", 0) <= nested.SNAPSHOT_MAX_BYTES @@ -165,8 +189,11 @@ def _nested_case_ok(r, n, depth): } -def assess_nested(records, expected_revision=None): - """Fixed nested_shear_n128 four-case contract (U128, U512, constant/projected R128 d3).""" +def assess_nested(records, expected_revision=None, diagnostic=NESTED_SHEAR): + """Fixed four-case preset contract: nested_shear_n128 (U128, U512, constant/projected + R128 depth 3) or twin_shear_n128 (U128, U256, constant/projected R128 depth 2).""" + preset = PRESETS[diagnostic] + geometry = preset["module"] if not isinstance(records, list) or not all(isinstance(r, dict) for r in records): raise ValueError("records must be a list of receipt objects") base_checks = tg.assess(records, expected_revision)["checks"] @@ -175,38 +202,36 @@ def assess_nested(records, expected_revision=None): for r in records: key = tuple(r.get(k) for k in ("re_L", "n", "depth", "explosion")) (duplicates.append(key) if key in indexed else indexed.__setitem__(key, r)) - checks.update(complete_matrix=all(k in indexed for k in nested.CASES), - no_extra_cases=set(indexed) <= set(nested.CASES), + checks.update(complete_matrix=all(k in indexed for k in geometry.CASES), + no_extra_cases=set(indexed) <= set(geometry.CASES), no_duplicate_cases=not duplicates, - shear_diagnostic_identity=all(r.get("diagnostic") == NESTED_SHEAR + shear_diagnostic_identity=all(r.get("diagnostic") == diagnostic for r in records)) - for key in nested.CASES: + for key in geometry.CASES: _, n, depth, scheme = key label = f"{n}/{depth}/{scheme}" - case = _nested_case_ok(indexed[key], n, depth) if key in indexed else {} + case = _nested_case_ok(indexed[key], n, depth, diagnostic) if key in indexed else {} for name in ("complete_times", "scales", "actual_builds", "reference", "runtime_source", "patch", "workload", "shear_metadata", "nested_module_pin", "snapshot"): checks[f"{label} {name}"] = bool(case.get(name, False)) contract_valid = all(checks.values()) - result = dict(checks=checks, contract_valid=contract_valid, diagnostic=NESTED_SHEAR, - cases=[list(k) for k in nested.CASES], expected_revision=expected_revision, - finest_control="uniform 4N (U512)", - scope="2D periodic Re100 oblique (2,1) shear wave; fixed three-level/two-region " - "placement; changes placement and refined volume as well as depth, so no " - "isolated-level cost or resolution trend") + result = dict(checks=checks, contract_valid=contract_valid, diagnostic=diagnostic, + cases=[list(k) for k in geometry.CASES], expected_revision=expected_revision, + finest_control=preset["finest_control"], scope=preset["scope"]) if not contract_valid: result.update(status="INVALID_SET", passed=False) return result try: - physics = nested.assess(indexed[nested.CASES[0]], indexed[nested.CASES[1]], - [indexed[nested.CASES[2]], indexed[nested.CASES[3]]]) + cases = geometry.CASES + physics = preset["physics"](indexed[cases[0]], indexed[cases[1]], + [indexed[cases[2]], indexed[cases[3]]]) except (KeyError, ValueError) as error: result.update(status="INVALID_SET", passed=False, reason=str(error)) return result result.update(physics=physics, status=physics["status"], passed=physics["status"] == "PASS") try: - # Role "uniform_fine" here is the 4N finest control. - result["signed"] = signed_summary([indexed[k] for k in nested.CASES]) + # Role "uniform_fine" is the preset's finest control (4N nested, 2N twin). + result["signed"] = signed_summary([indexed[k] for k in geometry.CASES]) except (KeyError, TypeError, ValueError) as error: result["signed_unavailable"] = str(error) return result @@ -281,16 +306,16 @@ if __name__ == "__main__": p.add_argument("files", type=Path, nargs=4) p.add_argument("--output", type=Path, required=True) p.add_argument("--expected-revision", required=True) - p.add_argument("--diagnostic", choices=[SCALED_SHEAR, NESTED_SHEAR]) + p.add_argument("--diagnostic", choices=[SCALED_SHEAR, NESTED_SHEAR, TWIN_SHEAR]) 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") - if args.diagnostic == NESTED_SHEAR: + if args.diagnostic in PRESETS: if args.reference_files: - p.error("nested_shear_n128 takes no reference files (no resolution trend)") + p.error(f"{args.diagnostic} takes no reference files (no resolution trend)") result = assess_nested([json.loads(path.read_text()) for path in args.files], - args.expected_revision) + args.expected_revision, args.diagnostic) text = json.dumps(result, indent=2, allow_nan=False) + "\n" args.output.write_text(text) print(text, end="") diff --git a/tests/validation/taylor_green/run_case.py b/tests/validation/taylor_green/run_case.py index 5889d1b..91d8b9a 100644 --- a/tests/validation/taylor_green/run_case.py +++ b/tests/validation/taylor_green/run_case.py @@ -23,6 +23,8 @@ SCALED_DIAGNOSTIC = "scaled_n256" SCALED_SHEAR = "scaled_shear_n256" # mutually exclusive with scaled_n256 (one flow each) NESTED_SHEAR = "nested_shear_n128" # fixed three-level/two-region shear preset SHEAR_NESTED_SHA256 = "fa6cfe87fb4df8c6d732c637c885276833c81973b950c7724d734f8fc350bcb0" +TWIN_SHEAR = "twin_shear_n128" # depth-2 twin: nested L0->L1 boxes without L2 +SHEAR_TWIN_SHA256 = "1cbf37f18e89aea577581631c4565f51e68d22a3739b698816ef64acbf2b2bb4" def flow_functions(flow="taylor_green"): @@ -111,14 +113,21 @@ def verify_source(repo, revision, flow="taylor_green"): if result["source_tree_sha256"] != SOURCE_SHA256: raise RuntimeError("Wrong numerical source fingerprint") result["build_capture_sha256"] = BUILD_CAPTURE_SHA256 - if flow == "shear_nested": + if flow == "shear_twin": + import shear_twin + module = harness / "shear_twin.py" + if (Path(shear_twin.__file__).resolve() != module + or hashlib.sha256(module.read_bytes()).hexdigest() != SHEAR_TWIN_SHA256): + raise RuntimeError("Wrong shear_twin module checkout or content") + result["shear_twin_sha256"] = SHEAR_TWIN_SHA256 + if flow in ("shear_nested", "shear_twin"): # the twin reuses the nested geometry import shear_nested module = harness / "shear_nested.py" if (Path(shear_nested.__file__).resolve() != module or hashlib.sha256(module.read_bytes()).hexdigest() != SHEAR_NESTED_SHA256): raise RuntimeError("Wrong shear_nested module checkout or content") result["shear_nested_sha256"] = SHEAR_NESTED_SHA256 - if flow in ("shear_wave", "shear_nested"): + if flow in ("shear_wave", "shear_nested", "shear_twin"): import shear_wave module = harness / "shear_wave.py" if (Path(shear_wave.__file__).resolve() != module @@ -191,10 +200,12 @@ def main(): 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, SCALED_SHEAR, NESTED_SHEAR], + p.add_argument("--diagnostic", + choices=[SCALED_DIAGNOSTIC, SCALED_SHEAR, NESTED_SHEAR, TWIN_SHEAR], help="explicit four-case N256/fine512 same-placement diagnostic " "(scaled_n256: Taylor-Green; scaled_shear_n256: shear wave) or the " - "fixed three-level/two-region shear preset (nested_shear_n128)") + "fixed three-level/two-region shear preset (nested_shear_n128) " + "or its depth-2 twin (twin_shear_n128)") p.add_argument("--final-snapshot", action="store_true", help="nested_shear_n128 projected depth-3 only: final macro snapshot") p.add_argument("--early-pressure", action="store_true", @@ -213,12 +224,15 @@ def main(): p.error("Output already exists; choose a new path to retain prior evidence") if args.explosion != "constant" and args.depth != 2 and not nested: p.error("projected_linear requires depth 2") - if nested: + twin = args.diagnostic == TWIN_SHEAR + if nested or twin: import shear_nested + import shear_twin + preset = shear_nested if nested else shear_twin if args.flow != "shear_wave": - p.error(f"--diagnostic {NESTED_SHEAR} requires --flow shear_wave") - if (args.re, args.n, args.depth, args.explosion) not in shear_nested.CASES: - p.error("nested_shear_n128 requires one of its four Re100 cases") + p.error(f"--diagnostic {args.diagnostic} requires --flow shear_wave") + if (args.re, args.n, args.depth, args.explosion) not in preset.CASES: + p.error(f"{args.diagnostic} requires one of its four Re100 cases") elif args.diagnostic: cases = {(256, 1, "constant"), (512, 1, "constant"), (256, 2, "constant"), (256, 2, "projected_linear")} @@ -238,7 +252,8 @@ def main(): p.error("--run-gpu and exactly one explicit CUDA_VISIBLE_DEVICES are required") repo = args.repo.resolve() sys.path[:0] = [str(repo / "src"), str(repo)] - source = verify_source(repo, args.expected_revision, "shear_nested" if nested else args.flow) + source = verify_source(repo, args.expected_revision, + "shear_nested" if nested else "shear_twin" if twin else args.flow) runtime = runtime_source(repo) if args.dry_run: print(json.dumps(dict(source=source, runtime_source=runtime, dry_run=True, @@ -254,8 +269,8 @@ def main(): cfg = config(args.n, args.re) if args.depth == 1: refinements = [] - elif nested: - refinements = [list(level) for level in shear_nested.REFINEMENTS] + elif nested or twin: + refinements = [list(level) for level in preset.REFINEMENTS] else: refinements = rectangles(args.n, args.depth) h = build_hierarchy(cfg, refinements) diff --git a/tests/validation/taylor_green/shear_twin.py b/tests/validation/taylor_green/shear_twin.py new file mode 100644 index 0000000..26ad08d --- /dev/null +++ b/tests/validation/taylor_green/shear_twin.py @@ -0,0 +1,36 @@ +"""Depth-2 twin of the nested shear preset (opt-in twin_shear_n128). + +Same two L0->L1 boxes as shear_nested, no L2. Four fresh cases: U128, U256 +(finest control = 2N), constant R128 depth 2, projected R128 depth 2; same +wave, phase, Re100/U=0.03, initialization, times, eps(root N), limits, floors. + +Twin vs nested measures the effect of adding this nested L2 refinement. +Removing L2 also changes the finest spacing and refined volume, so the +contrast does not isolate abstract depth. Physics is the unchanged depth-2, +2N-control shear_wave.assess; shear_nested.py stays byte-identical. +""" + +import numpy as np + +import shear_nested as nested + +ROOT_N = nested.ROOT_N +DEPTH = 2 +FINEST_FACTOR = 2 +REFINEMENTS = [nested.REFINEMENTS[0]] # L0 -> L1 only, root coordinates +CASES = ((100, ROOT_N, 1, "constant"), (100, FINEST_FACTOR * ROOT_N, 1, "constant"), + (100, ROOT_N, DEPTH, "constant"), (100, ROOT_N, DEPTH, "projected_linear")) +ROOT_STEPS = 16 * ROOT_N +LEAVES = [7936, 33792] +HOPS = [ROOT_STEPS * 2**level for level in range(DEPTH)] # [2048, 4096] +WORK = sum(c * h for c, h in zip(LEAVES, HOPS)) # 154664960 per refined case +NU = nested.NU[:DEPTH] +TAU = nested.TAU[:DEPTH] + + +def leaf_masks(): + """Independent numpy leaf masks (root complement, full L1 boxes).""" + child = np.zeros((ROOT_N, ROOT_N), bool) + for x0, y0, x1, y1 in REFINEMENTS[0]: + child[y0:y1, x0:x1] = True + return [~child, np.kron(child, np.ones((2, 2), bool))] diff --git a/tests/validation/taylor_green/test_shear_wave.py b/tests/validation/taylor_green/test_shear_wave.py index b739391..ba6f7de 100644 --- a/tests/validation/taylor_green/test_shear_wave.py +++ b/tests/validation/taylor_green/test_shear_wave.py @@ -776,3 +776,157 @@ class NestedPresetTests(unittest.TestCase): capped = nested.write_snapshot(Path(tmp) / "c.npz", arrays, masks, meta) self.assertFalse(capped["kept"]) self.assertFalse((Path(tmp) / "c.npz").exists()) + + +class TwinPresetTests(unittest.TestCase): + """Depth-2 twin (twin_shear_n128): nested L0->L1 boxes, no L2, fresh U128/U256.""" + + def setUp(self): + import assess_shear_wave + import shear_nested + import shear_twin + self.mod, self.nested, self.twin = assess_shear_wave, shear_nested, shear_twin + + def test_exact_geometry_work_and_nested_identity(self): + twin = self.twin + masks = twin.leaf_masks() + self.assertEqual([int(m.sum()) for m in masks], [7936, 33792]) + self.assertEqual((twin.LEAVES, twin.HOPS, twin.WORK), ([7936, 33792], [2048, 4096], 154664960)) + self.assertEqual(128 * 128 * 2048 + 256 * 256 * 4096 + 2 * twin.WORK, 611319808) + self.assertEqual(twin.REFINEMENTS, [self.nested.REFINEMENTS[0]]) + np.testing.assert_array_equal(masks[0], self.nested.leaf_masks()[0]) # same root leaves + from CelerisLab.lbm.hierarchy import build_hierarchy, leaf_partition + from CelerisLab.lbm.refinement import ACTIVE + h = build_hierarchy(run_case.config(128, 100), twin.REFINEMENTS) + np.testing.assert_array_equal(leaf_partition(h), np.ones((128, 128))) + for mine, kinds in zip(masks, h.kinds): + np.testing.assert_array_equal(mine, kinds == ACTIVE) + # Historical nested identity is unchanged; the twin is pinned separately. + self.assertEqual(hashlib.sha256((HERE / "shear_nested.py").read_bytes()).hexdigest(), + "fa6cfe87fb4df8c6d732c637c885276833c81973b950c7724d734f8fc350bcb0") + self.assertEqual(run_case.SHEAR_TWIN_SHA256, + hashlib.sha256((HERE / "shear_twin.py").read_bytes()).hexdigest()) + + def twin_set(self): + out = [] + for n, depth, scheme in ((128, 1, "constant"), (256, 1, "constant"), + (128, 2, "constant"), (128, 2, "projected_linear")): + r = receipt(n, depth, scheme, diagnostic=run_case.TWIN_SHEAR) + r["source"].update(shear_nested_sha256=run_case.SHEAR_NESTED_SHA256, + shear_twin_sha256=run_case.SHEAR_TWIN_SHA256) + if depth == 2: + cells, hops = self.twin.LEAVES, self.twin.HOPS + r["refinements"] = [list(level) for level in self.twin.REFINEMENTS] + r["workload"] = dict(active_cells_by_level=cells, level_hops=hops, root_intervals=2048, + active_collision_updates_by_level=[c * h for c, h in zip(cells, hops)], + active_collision_updates=self.twin.WORK) + out.append(r) + return out + + def test_twin_contract_and_rejections(self): + a = lambda records: self.mod.assess_nested(records, "r", run_case.TWIN_SHEAR) + result = a(self.twin_set()) + self.assertTrue(result["contract_valid"], [k for k, v in result["checks"].items() if not v]) + self.assertEqual(result["status"], "PASS") + self.assertIn("does not isolate abstract depth", result["scope"]) + self.assertEqual(result["finest_control"], "uniform 2N (U256)") + for mutate in ( + lambda s: s.__setitem__(1, receipt(512, 1, diagnostic=run_case.TWIN_SHEAR)), # 4N control + lambda s: s[2].update(refinements=[list(level) for level in self.nested.REFINEMENTS]), + lambda s: s[3].update(refinements=rectangles(128, 2)), # old single box + lambda s: s[2]["rows"][1].update(spurious_pressure_scaled_by_level=[0.0, 0.0, 0.0]), + lambda s: s[0].update(diagnostic=run_case.NESTED_SHEAR), + lambda s: s[2]["source"].pop("shear_twin_sha256"), + lambda s: s[3].update(final_snapshot=dict(kept=True, sha256="a" * 64, bytes=10, + leaf_counts=self.twin.LEAVES, step=2048)), + ): + records = self.twin_set() + mutate(records) + self.assertNotEqual(a(records)["status"], "PASS") + # Twin and nested sets never validate as each other. + self.assertEqual(self.mod.assess_nested(self.twin_set(), "r")["status"], "INVALID_SET") + + def test_twin_runner_options(self): + repo = Path(__file__).resolve().parents[3] + with tempfile.TemporaryDirectory() as tmp: + base = ["run_case.py", "--repo", str(repo), "--expected-revision", "f" * 40, + "--output", str(Path(tmp) / "case.json"), "--dry-run", "--re", "100"] + tw = ["--flow", "shear_wave", "--diagnostic", run_case.TWIN_SHEAR] + proj = ["--explosion", "projected_linear"] + for extra, valid in ( + (["--n", "128", "--depth", "1"] + tw, True), + (["--n", "256", "--depth", "1"] + tw, True), + (["--n", "128", "--depth", "2"] + tw, True), + (["--n", "128", "--depth", "2"] + proj + tw, True), + (["--n", "512", "--depth", "1"] + tw, False), + (["--n", "128", "--depth", "3"] + tw, False), + (["--n", "128", "--depth", "2"] + proj + tw + ["--final-snapshot"], False), + (["--n", "128", "--depth", "2", "--diagnostic", run_case.TWIN_SHEAR], False)): + argv = base + extra + with self.subTest(argv=argv), patch.object(sys, "argv", argv), \ + patch.object(run_case, "verify_source", return_value={}) as verify, \ + patch.object(run_case, "runtime_source", return_value={}), \ + redirect_stdout(io.StringIO()), patch("sys.stderr", io.StringIO()): + if valid: + self.assertEqual(run_case.main(), 0) + self.assertEqual(verify.call_args.args[2], "shear_twin") + else: + with self.assertRaises(SystemExit): + run_case.main() + + def test_mocked_twin_execution(self): + from types import SimpleNamespace + from contextlib import nullcontext + from CelerisLab.lbm.hierarchy import build_hierarchy + h = build_hierarchy(run_case.config(128, 100), self.twin.REFINEMENTS) + arrays = [sw.initial(k.shape, level, 128, U * 128 / sw.RE) for level, k in enumerate(h.kinds)] + selectors = [] + + class FakeRun: + def __init__(self, *args, **kwargs): + self.step_count = 0 + if "refinements" in kwargs: + selectors.append((kwargs["explosion"], kwargs["refinements"])) + + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + def initialize(self, *given): + assert len(given) == 2 + + def step(self, count): + self.step_count += count + + def download(self): + return arrays + + modules = {"CelerisLab.cuda.context": SimpleNamespace(CudaContext=FakeRun), + "CelerisLab.lbm.block_stepper": SimpleNamespace(PeriodicBlockStepper=FakeRun), + "CelerisLab.lbm.refinement_stepper": SimpleNamespace(FixedPatchStepper=FakeRun)} + repo = Path(__file__).resolve().parents[3] + with tempfile.TemporaryDirectory() as tmp: + output = Path(tmp) / "projected_twin.json" + argv = ["run_case.py", "--repo", str(repo), "--re", "100", "--n", "128", "--depth", "2", + "--explosion", "projected_linear", "--flow", "shear_wave", + "--diagnostic", run_case.TWIN_SHEAR, "--output", str(output), + "--expected-revision", "f" * 40, "--run-gpu"] + 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=dict(revision="f" * 40)), \ + patch.object(run_case, "runtime_source", return_value={}), \ + patch.object(run_case, "runtime_identity", return_value={}), \ + patch.object(run_case, "build_capture", return_value=nullcontext()), \ + redirect_stdout(io.StringIO()): + run_case.main() + rec = json.loads(output.read_text()) + self.assertTrue(rec["completed"]) + self.assertEqual(rec["diagnostic"], run_case.TWIN_SHEAR) + self.assertEqual(rec["workload"]["active_cells_by_level"], self.twin.LEAVES) + self.assertEqual(rec["workload"]["active_collision_updates"], self.twin.WORK) + self.assertNotIn("final_snapshot", rec) + self.assertTrue(all(len(r["spurious_pressure_scaled_by_level"]) == 2 for r in rec["rows"])) + self.assertEqual(selectors, [("projected_linear", [list(self.twin.REFINEMENTS[0])])]) + self.assertEqual(sorted(p.name for p in Path(tmp).iterdir()), ["projected_twin.json"])