test: add one projected x-gap-width shear contrast

tests/validation/taylor_green/test_shear_wave.py

@@ -1066,11 +1066,13 @@ class ProbeTraceTests(unittest.TestCase):
         from types import SimpleNamespace
         from contextlib import nullcontext
         arrays = self.exact_arrays(masks, n, 0.0)
-        counts = []
+        counts, 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
@@ -1106,7 +1108,11 @@ class ProbeTraceTests(unittest.TestCase):
                  patch.object(run_case, "build_capture", return_value=nullcontext()), \
                  redirect_stdout(io.StringIO()):
                 run_case.main()
-            return json.loads(output.read_text()), counts, sorted(p.name for p in Path(tmp).iterdir())
+            rec = json.loads(output.read_text())
+            expected = ([(scheme, [[tuple(r) for r in level] for level in rec["refinements"]])]
+                        if depth > 1 else [])
+            self.assertEqual(selectors, expected)
+            return rec, counts, sorted(p.name for p in Path(tmp).iterdir())
 
     def test_mocked_probe_traces(self):
         sp = self.probes
@@ -1162,3 +1168,124 @@ class ProbeTraceTests(unittest.TestCase):
         self.assertTrue(rec["completed"])
         self.assertEqual(rec["rows"], baseline["rows"])
         self.assertIn("exceeds", rec["probe_trace"]["error"])
+
+
+class WideXDiagnosticTests(unittest.TestCase):
+    """One width contrast; historical qualification matrices remain unchanged."""
+
+    def setUp(self):
+        import shear_nested
+        import shear_probes
+        import shear_wide_x
+        from CelerisLab.lbm.hierarchy import build_hierarchy
+        self.wide, self.nested, self.probes = shear_wide_x, shear_nested, shear_probes
+        self.h = build_hierarchy(run_case.config(128, 100), self.wide.REFINEMENTS)
+
+    def test_geometry_counts_and_legal_pair_tables(self):
+        from CelerisLab.lbm.blocks import BlockLayout2D
+        from CelerisLab.lbm.hierarchy import build_hierarchy, leaf_partition
+        from CelerisLab.lbm.refinement import ACTIVE
+        from CelerisLab.lbm.refinement_tables import projected_explosion_tables
+        wide, h = self.wide, self.h
+        self.assertEqual(wide.REFINEMENTS[0], self.nested.REFINEMENTS[0])
+        self.assertEqual(wide.REFINEMENTS[1], [(56, 44, 96, 212), (160, 60, 200, 228)])
+        for outer, inner in zip(wide.REFINEMENTS[0], wide.REFINEMENTS[1]):
+            x0, y0, x1, y1 = (v / 2 for v in inner)
+            self.assertEqual((x0 - outer[0], outer[2] - x1), (12, 12))
+            self.assertEqual((y0 - outer[1], outer[3] - y1), (6, 6))
+        # Independent physical rectangle masks at each native spacing.
+        masks = []
+        for level, kind in enumerate(h.kinds):
+            x, y = centers(kind.shape, level)
+            outer = ((x >= 16) & (x < 60) & (y >= 16) & (y < 112)
+                     | (x >= 68) & (x < 112) & (y >= 24) & (y < 120))
+            inner = ((x >= 28) & (x < 48) & (y >= 22) & (y < 106)
+                     | (x >= 80) & (x < 100) & (y >= 30) & (y < 114))
+            masks.append(~outer if level == 0 else outer & ~inner if level == 1 else inner)
+            np.testing.assert_array_equal(kind == ACTIVE, masks[-1])
+        self.assertEqual([int(m.sum()) for m in masks], [7936, 20352, 53760])
+        np.testing.assert_array_equal(leaf_partition(h), np.ones((128, 128)))
+        self.assertEqual(run_case.workload(masks, 2048)["active_collision_updates"], 540016640)
+        self.assertEqual(wide.LEAVES, [7936, 20352, 53760])
+        self.assertEqual(wide.HOPS, [2048, 4096, 8192])
+        self.assertEqual(wide.WORK, 540016640)
+        np.testing.assert_allclose([c.viscosity for c in h.configs], wide.NU, rtol=0, atol=0)
+        np.testing.assert_allclose([1 / c.omega for c in h.configs], wide.TAU, rtol=1e-15)
+        old = build_hierarchy(run_case.config(128, 100), self.nested.REFINEMENTS)
+        for name in ("ghost_coarse", "ghost_fine", "coalesce_coarse", "coalesce_fine"):
+            np.testing.assert_array_equal(getattr(h.pairs[0], name), getattr(old.pairs[0], name))
+        root = BlockLayout2D.build(128, 128, 8, 8, periodic=(True, True))
+        for level, pair in enumerate(h.pairs):
+            layout = root if level == 0 else h.pairs[level - 1].fine_layout((8, 8))
+            neighbors, offsets = projected_explosion_tables(pair, layout, h.kinds[level])
+            legal = np.unique(pair.ghost_coarse)
+            legal = legal[(h.kinds[level].ravel()[legal] == ACTIVE) & ~pair.covered.ravel()[legal]]
+            self.assertTrue(np.isin(neighbors, layout.owner.ravel()[legal]).all())
+            self.assertEqual(neighbors.shape, (len(pair.ghost_fine), 4))
+            self.assertEqual(len(np.unique(pair.ghost_fine)), len(pair.ghost_fine))
+            self.assertTrue((np.abs(offsets) == .25).all())
+            if level == 0:
+                expected = projected_explosion_tables(old.pairs[0], root, old.kinds[0])
+                for a, b in zip((neighbors, offsets), expected):
+                    np.testing.assert_array_equal(a, b)
+
+    def test_fixed_probes_and_complete_runner_schedule(self):
+        from CelerisLab.lbm.refinement import ACTIVE
+        masks = [k == ACTIVE for k in self.h.kinds]
+        probes = self.probes.select(masks)
+        old = self.probes.select(self.nested.leaf_masks())
+        self.assertEqual([p["level"] for p in probes[:12]], [0] * 3 + [1] * 9)
+        self.assertLess(max(p["center"][0] for p in probes[:12]), 28)
+        self.assertEqual(probes[:8], old[:8])  # identical-center old L0/L1 probes
+        self.assertEqual(probes[12:], old[12:])  # unchanged y-normal transect
+        helper = ProbeTraceTests()
+        helper.setUp()
+        rec, counts, files = helper.run_mocked(128, 3, run_case.WIDE_X_SHEAR,
+                                               "projected_linear", masks)
+        self.assertTrue(rec["completed"])
+        self.assertEqual(rec["refinements"], [[list(r) for r in level] for level in self.wide.REFINEMENTS])
+        self.assertEqual(rec["workload"]["active_cells_by_level"], self.wide.LEAVES)
+        self.assertEqual(rec["workload"]["active_collision_updates"], self.wide.WORK)
+        self.assertEqual([r["step"] for r in rec["rows"]], [0, 512, 1024, 1536, 2048])
+        self.assertEqual(rec["probe_trace"]["steps"], list(self.probes.STEPS))
+        self.assertEqual(rec["probe_trace"]["probes"], probes)
+        self.assertEqual(counts[-1], 2048)
+        self.assertEqual(files, ["probe.json"])
+        self.assertNotIn("final_snapshot", rec)
+
+    def test_one_case_admission_and_pin(self):
+        self.assertEqual(run_case.SHEAR_WIDE_X_SHA256,
+                         hashlib.sha256((HERE / "shear_wide_x.py").read_bytes()).hexdigest())
+        repo = HERE.parents[2]
+        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",
+                    "--flow", "shear_wave", "--diagnostic", run_case.WIDE_X_SHEAR,
+                    "--n", "128", "--depth", "3", "--explosion", "projected_linear", "--probe-trace"]
+            for extra in ([], ["--n", "512"], ["--depth", "2"], ["--explosion", "constant"],
+                          ["--re", "1000"], ["--flow", "taylor_green"], ["--final-snapshot"],
+                          ["--early-pressure"]):
+                with self.subTest(extra=extra), patch.object(sys, "argv", base + extra), \
+                     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 not extra:
+                        self.assertEqual(run_case.main(), 0)
+                        verify.assert_called_once_with(repo, "f" * 40, "shear_wide_x", probes=True)
+                    else:
+                        with self.assertRaises(SystemExit):
+                            run_case.main()
+                        verify.assert_not_called()
+            with patch.object(sys, "argv", base[:-1]), patch("sys.stderr", io.StringIO()):
+                with self.assertRaises(SystemExit):
+                    run_case.main()
+
+    def test_historical_matrix_rejects_width_case(self):
+        helper = NestedPresetTests()
+        helper.setUp()
+        records = helper.nested_set()
+        self.assertTrue(helper.mod.assess_nested(records, "r")["passed"])
+        records[-1].update(diagnostic=run_case.WIDE_X_SHEAR,
+                           refinements=self.wide.REFINEMENTS)
+        self.assertEqual(helper.mod.assess_nested(records, "r")["status"], "INVALID_SET")
+        self.assertNotIn(run_case.WIDE_X_SHEAR, helper.mod.PRESETS)