test: select single-case resolution and compact refinement

tests/unit/test_refined_dfg.py

@@ -146,24 +146,26 @@ def test_full_requires_completed_pilot():
         r.verify_pilot(None, r.verify_source(r.DEFAULT_SOURCE, "MRT", "window"))
 
 
-def pilot_fixture(path, observation_mode="window", jet_scenario="none"):
+def pilot_fixture(path, observation_mode="window", jet_scenario="none", case_selection="both", uniform_diameter=80, placement="wide"):
     """Synthetic contract fixture only; never a physical result or run receipt."""
     manifest = (r.verify_source(r.DEFAULT_SOURCE, "MRT", "window", jet_scenario)
                 if observation_mode == "window" else
                 json.loads((r.HERE / "data/refined_dfg/source_manifest_mrt.json").read_text()))
+    selection = dict(case_selection=case_selection, uniform_diameter=uniform_diameter, placement=placement)
     (path / "plan.json").write_text(
         json.dumps(
             dict(
                 source_manifest=manifest,
+                **selection,
                 collision="MRT",
                 jet_scenario=jet_scenario,
                 observation_mode=observation_mode,
-                bounds_root_edge=list(r.BOUNDS),
+                bounds_root_edge=list(r.PLACEMENTS[placement]),
                 runner_sha256=r.runner_hash(),
             )
         )
     )
-    for case, diameter in (("uniform", 80), ("refined", 40)):
+    for case, diameter in r.selected_cases(case_selection, uniform_diameter):
         dest = path / case
         dest.mkdir()
         interval = round(0.02 * diameter / r.SPEED)
@@ -190,6 +192,7 @@ def pilot_fixture(path, observation_mode="window", jet_scenario="none"):
             json.dumps(
                 dict(
                     case=case,
+                    **selection,
                     collision="MRT",
                     jet_scenario=jet_scenario,
                     observation_mode=observation_mode,
@@ -208,6 +211,7 @@ def pilot_fixture(path, observation_mode="window", jet_scenario="none"):
             json.dumps(
                 dict(
                     case=case,
+                    **selection,
                     collision="MRT",
                     jet_scenario=jet_scenario,
                     observation_mode=observation_mode,
@@ -526,6 +530,7 @@ def test_cv_snapshot_common_units_and_window_endpoints():
     for refined, shape, origin, diameter, impulse in (
         (True, (240, 568), (38., 22.), 40, [4., -8.]),
         (False, (330, 1761), (.25, .5), 80, [16., -32.]),
+        (False, (658, 3521), (.375, .75), 160, [64., -128.]),
     ):
         rho, ux, uy, fluid = np.ones(shape), np.full(shape, .01), np.zeros(shape), np.ones(shape, bool)
         begin, n = round(290*diameter/r.SPEED), round(.02*diameter/r.SPEED)
@@ -538,7 +543,8 @@ def test_cv_snapshot_common_units_and_window_endpoints():
         assert r.cv_selected(n, diameter, "pilot")
         assert sum(r.cv_selected(step, diameter, "full") for step in range(begin-n, end+n+1, n)) == 501
     for key in ("time_root", "interval_start_root", "interval_dt_root", "interval_impulse_root", "inventory"):
-        np.testing.assert_allclose(records[0][key], records[1][key], rtol=0, atol=1e-10)
+        for record in records[1:]:
+            np.testing.assert_allclose(records[0][key], record[key], rtol=0, atol=1e-10)
     assert records[0]["interval_dt_root"] == 80 and np.isclose(records[0]["tstar"], 290.02)
     np.testing.assert_array_equal(records[0]["interval_impulse_root"], [4, -8])
 
@@ -564,6 +570,57 @@ def test_cv_pilot_mode_and_helper_provenance():
         assert r.verify_pilot(path, manifest, "MRT", "window", "fixed", True) == 3606
 
 
+def test_compact_geometry_and_cv_stencils():
+    from CelerisLab.lbm.refinement import ACTIVE
+    from CelerisLab.lbm.descriptors import SOLID
+
+    _, pair, _, host = r.host_geometry(r.DEFAULT_SOURCE, h, "MRT", "fixed", "compact")
+    assert pair.fine_shape == (168, 168) and tuple(pair.fine_origin) == (46, 46)
+    fluid = (pair.fine_kind == ACTIVE) & ((host.flags.reshape(pair.fine_shape) & SOLID) == 0)
+    rho, ux, uy = np.ones(pair.fine_shape), np.full(pair.fine_shape, .01), np.zeros(pair.fine_shape)
+    record = r.cv_snapshot(rho, ux, uy, fluid, pair.fine_origin, True, 1160000, 80, 40, [1, 2], "compact")
+    np.testing.assert_allclose(record["outward_flux"], 0, atol=1e-12)
+    stencil = np.zeros(pair.fine_shape, bool)
+    for rect in r.cv_contours(True, 40, "compact"):
+        x0, y0, x1, y1 = np.rint((rect - np.tile(pair.fine_origin, 2))/.5).astype(int)
+        stencil[y0-2:y1+2, x0-2:x0+2] = stencil[y0-2:y1+2, x1-2:x1+2] = True
+        stencil[y0-2:y0+2, x0-2:x1+2] = stencil[y1-2:y1+2, x0-2:x1+2] = True
+    assert fluid[stencil].all() and not stencil.ravel()[host.transfer_fine_cells].any()
+
+
+def test_d160_config_and_single_case_receipts():
+    from tempfile import TemporaryDirectory
+    from unittest.mock import patch
+
+    cfg, _, p = r.config(h, 160, "MRT", "fixed")
+    assert (p["nx"], p["ny"], p["center"], p["radius"]) == (3521, 658, [320, 320.5], 80)
+    assert p["viscosity"] == .016 and p["mean_speed"] == .01 and not cfg["method"]["ddf_shifting"]
+    assert r.uniform_origin(160) == (.375, .75)
+    np.testing.assert_array_equal(r.cv_contours(False, 160), r.CV_CONTOURS + [.125, 0, .125, 0])
+    for scenario, expected_exit in (("fixed", 0), ("zero", 2)):
+        with TemporaryDirectory() as tmp:
+            path = Path(tmp)
+            manifest = pilot_fixture(path, jet_scenario=scenario, case_selection="uniform", uniform_diameter=160)
+            assert r.verify_pilot(path, manifest, "MRT", "window", scenario, False, "uniform", 160) == 1803
+            for selection, diameter, placement in (("both", 160, "wide"), ("uniform", 80, "wide"), ("uniform", 160, "compact")):
+                with unittest.TestCase().assertRaises(ValueError):
+                    r.verify_pilot(path, manifest, "MRT", "window", scenario, False, selection, diameter, placement)
+            # Synthetic dispatch/result check: no CUDA imports or execution.
+            output = path / "full"
+            argv = ["runner", "--collision", "MRT", "--exchange-window", "--jet-scenario", scenario,
+                    "--case", "uniform", "--uniform-diameter", "160", "--mode", "full",
+                    "--pilot-results", str(path), "--output", str(output), "--max-wall-seconds", "10", "--gpu-gates-approved"]
+            result = dict(case="uniform", healthy=True, periodic=scenario == "zero", jet_diagnostic_windows=[],
+                          status="driven_diagnostics_complete" if scenario == "fixed" else "periodic",
+                          reference_comparison=dict(reference_agreement=False))
+            with patch.object(sys, "argv", argv), patch.dict(r.os.environ, CUDA_VISIBLE_DEVICES="mock-no-gpu"), \
+                    patch.object(r, "run_case", return_value=result) as run:
+                assert r.main() == expected_exit and run.call_count == 1
+            comparison = json.loads((output / "comparison.json").read_text())
+            assert comparison["matched_case_pass"] is False and "refined_vs_uniform" not in comparison
+            assert comparison["status"].startswith("single_case_")
+
+
 if __name__ == "__main__":
     suite = unittest.TestSuite(
         unittest.FunctionTestCase(fn)