"""Standalone CPU manufactured checks for the oblique shear wave. Run from this directory with PYTHONPATH=/src (same as test_oracle.py). """ import hashlib import io import json import math import sys import tempfile from contextlib import redirect_stdout from unittest.mock import patch import unittest from pathlib import Path import numpy as np import oracle import run_case import shear_wave as sw from oracle import CX, CY, U, TIMES, centers, rectangles N = 32 NU = U * N / sw.RE HERE = Path(__file__).resolve().parent ORACLE_SHA256_EEC4677 = "83e4d7425d45538278e9fb181af450a36358bd9a5ed6be57ff611dd9ae5eb35b" def composite_masks(n): """Level-wide union masks for rectangles(n, 2): root complement + fine box.""" (x0, y0, x1, y1), = rectangles(n, 2)[0] root = np.ones((n, n), dtype=bool) root[y0:y1, x0:x1] = False fine = np.zeros((2 * n, 2 * n), dtype=bool) fine[2 * y0:2 * y1, 2 * x0:2 * x1] = True return [root, fine] def populations(masks, field, t=0.0): """Equilibrium G of an arbitrary (ux, uy, rho) field on every level.""" out = [] for level, m in enumerate(masks): x, y = centers(m.shape, level) ux, uy, rho = field(x, y, t) out.append(sw.equilibrium(rho, ux, uy)) return out def exact_field(phase=0.0, rate_factor=1.0, mode=sw.MODE, longitudinal=0.0, ripple=0.0, offset=0.0): k = 2 * np.pi / N * np.array(mode, dtype=float) e = np.array([-mode[1], mode[0]], dtype=float) / math.hypot(*mode) rate = rate_factor * sw.decay_rate(N, NU) kh = k / np.linalg.norm(k) def field(x, y, t): theta = k[0] * x + k[1] * y + phase a = U * np.exp(-rate * t) ux = a * e[0] * np.sin(theta) + longitudinal * kh[0] * np.sin(theta) uy = a * e[1] * np.sin(theta) + longitudinal * kh[1] * np.sin(theta) rho = 1 + offset + ripple * np.cos(theta) return ux, uy, rho return field class ConventionTests(unittest.TestCase): def test_mode_angle_and_polarization(self): k = sw.wavevector(N) self.assertAlmostEqual(sw.ANGLE_DEG, 26.565051177077994) self.assertAlmostEqual(math.degrees(math.atan2(k[1], k[0])), sw.ANGLE_DEG) self.assertAlmostEqual(float(sw.E @ k), 0.0) self.assertAlmostEqual(float(sw.E @ sw.E), 1.0) self.assertAlmostEqual(sw.eps(128), 5 * (2 * math.pi / 128) ** 2) def test_navier_stokes_and_vanishing_advection(self): # Independent central differences: du/dt = nu lap u, (u.grad)u = 0, div u = 0. x = np.array([2.7, 8.1, 13.3]) y = np.array([3.2, 7.4, 21.6]) t, h = 10.0, 1e-3 f = lambda x, y, t: np.array(sw.exact(x, y, t, N, NU)[:2]) z = f(x, y, t) dt = (f(x, y, t + h) - f(x, y, t - h)) / (2 * h) dx = (f(x + h, y, t) - f(x - h, y, t)) / (2 * h) dy = (f(x, y + h, t) - f(x, y - h, t)) / (2 * h) lap = (f(x + h, y, t) + f(x - h, y, t) + f(x, y + h, t) + f(x, y - h, t) - 4 * z) / h**2 np.testing.assert_allclose(dt, NU * lap, atol=1e-9) # Analytic identities: div u = A cos(theta) (E.k) and (u.grad)u = A sin(theta) (E.k) u/A. k = sw.wavevector(N) self.assertLess(abs(float(sw.E @ k)), 1e-17) # Central differences do not cancel exactly: each derivative carries # -(k_j h)^2/6 relative truncation, so div_h = -A cos(theta) h^2/6 sum_j E_j k_j^3 # + O(h^4) (nonzero for an oblique k). Bound it with that leading term plus roundoff. a = U * math.exp(-sw.decay_rate(N, NU) * t) trunc = a * h * h / 6 * float(np.abs(sw.E * k**3).sum()) * 1.01 # Argument rounding: eps*max|x|/h*U*|k| ~ 2.2e-16*21.6/1e-3*0.03*0.44 ~ 6e-14. roundoff = 1e-13 self.assertLess(np.max(np.abs(dx[0] + dy[1])), trunc + roundoff) # Discrete advection: same truncation scaled by |u| <= a. self.assertLess(np.max(np.abs(z[0] * dx + z[1] * dy)), a * trunc + roundoff) # Halving h reduces the divergence truncation about fourfold. hh = h / 2 div2 = ((f(x + hh, y, t) - f(x - hh, y, t))[0] + (f(x, y + hh, t) - f(x, y - hh, t))[1]) / (2 * hh) np.testing.assert_allclose(div2 * 4, dx[0] + dy[1], rtol=1e-2, atol=6 * roundoff) self.assertTrue(np.all(sw.exact(x, y, t, N, NU)[2] == 0)) class InitialTests(unittest.TestCase): def test_moments_on_each_level(self): for level, m in enumerate(composite_masks(N)): g = sw.initial(m.shape, level, N, NU) x, y = centers(m.shape, level) ux, uy, _ = sw.exact(x, y, 0.0, N, NU) rho = g.sum(axis=0) np.testing.assert_allclose(rho, 1.0, atol=1e-14) np.testing.assert_allclose(np.tensordot(CX, g, 1) / rho, ux, atol=1e-15) np.testing.assert_allclose(np.tensordot(CY, g, 1) / rho, uy, atol=1e-15) def test_equilibrium_matches_frozen_oracle_formula(self): # Same equilibrium as oracle.initial when fed the TG field. shape = (8, 8) x, y = centers(shape, 1) ux, uy, p = oracle.exact(x, y, 0.0, 4.0, 0.01) np.testing.assert_array_equal( sw.equilibrium(1 + 3 * p, ux, uy), oracle.initial(shape, 1, 4.0, 0.01) ) class MetricTests(unittest.TestCase): def setUp(self): self.masks = composite_masks(N) def sample(self, field, t): return sw.metrics(populations(self.masks, field, t), self.masks, t, N, NU, None) def test_exact_field_on_composite_partition(self): for t in (0.0, 100.0, 512.0): r = self.sample(exact_field(), t) self.assertLess(r["velocity_l2"], 1e-12) self.assertLess(r["residual_l2"], 1e-12) self.assertLess(r["longitudinal_l2"], 1e-12) self.assertLess(r["phase_drift_abs"], 1e-12) self.assertAlmostEqual(r["amplitude_ratio"], 1.0, places=12) self.assertLess(abs(r["amplitude_error"]), 1e-12) self.assertLess(r["spurious_pressure_scaled"], 1e-10) self.assertEqual(len(r["spurious_pressure_scaled_by_level"]), 2) self.assertEqual(r["leaf_cells"], int(sum(m.sum() for m in self.masks))) def test_phase_control(self): r = self.sample(exact_field(phase=0.01), 0.0) self.assertAlmostEqual(r["phase_drift"], 0.01, delta=1e-9) self.assertLess(abs(r["amplitude_error"]), 1e-12) # R = hypot(a, b) is phase-free self.assertLess(r["residual_l2"], 1e-12) # a pure phase shift stays modal def test_rate_control_and_fit(self): for factor, expected in ((1.0, 0.0), (1.01, 0.01)): field = exact_field(rate_factor=factor) rows = [] for tstar in TIMES: step = round(tstar * N / U) rows.append(dict(self.sample(field, step), step=step)) fit = sw.fit_rate(rows, N, NU) self.assertAlmostEqual(fit["rate_error"], expected, delta=1e-9) np.testing.assert_allclose(fit["interval_rate_errors"], expected, atol=1e-9) self.assertLess(fit["log_fit_residual_max"], 1e-9) self.assertAlmostEqual(fit["exact_rate_tstar"], 5 * (2 * math.pi) ** 2 / sw.RE) self.assertAlmostEqual(fit["exact_rate_tstar"], 1.97392088, places=8) def test_fit_ignores_startup_sample(self): rows = [dict(tstar=0.0, step=0, a_hat=2 * U, b_hat=0.0)] rows += [dict(tstar=s, step=round(s * N / U), b_hat=0.0, a_hat=U * math.exp(-sw.decay_rate(N, NU) * round(s * N / U))) for s in TIMES[1:]] self.assertLess(sw.fit_rate(rows, N, NU)["rate_error"], 1e-12) rows[-1]["a_hat"] = float("nan") with self.assertRaises(ValueError): sw.fit_rate(rows, N, NU) def test_wrong_mode_control(self): # Swapping (2,1) for (1,2), i.e. 26.6 vs 63.4 deg, is not silently modal. # Composite quadrature need not give an exactly zero projection. r = self.sample(exact_field(mode=(1, 2)), 0.0) self.assertLess(abs(r["amplitude_ratio"]), 0.05) self.assertGreater(r["residual_l2"], 0.95) def test_longitudinal_control(self): r = self.sample(exact_field(longitudinal=1e-3), 0.0) self.assertAlmostEqual(r["longitudinal_l2"], 1e-3 / U, delta=1e-9) # The full-vector residual and velocity error see it too. self.assertAlmostEqual(r["residual_l2"], 1e-3 / U, delta=1e-9) self.assertAlmostEqual(r["velocity_l2"], 1e-3 / U, delta=1e-9) self.assertLess(r["phase_drift_abs"], 1e-12) self.assertAlmostEqual(r["amplitude_ratio"], 1.0, places=12) def test_pressure_gauge_and_ripple(self): r = self.sample(exact_field(offset=1e-4), 0.0) self.assertLess(r["spurious_pressure_scaled"], 1e-10) # one global gauge ripple = 3e-5 r = self.sample(exact_field(ripple=ripple), 0.0) # p = ripple/3 cos(theta); rms over a whole number of periods = ripple/(3 sqrt 2). # Composite midpoint quadrature is not exactly periodic: 3% tolerance. expected = ripple / (3 * math.sqrt(2)) / (U * U / 4) self.assertAlmostEqual(r["spurious_pressure_scaled"], expected, delta=0.03 * expected) self.assertAlmostEqual(r["spurious_pressure_rms"], expected * U * U / 4, delta=0.03 * expected * U * U / 4) def test_volume_weighting_and_double_count(self): bad = [np.ones_like(m) for m in self.masks] with self.assertRaises(ValueError): sw.metrics(populations(bad, exact_field()), bad, 0.0, N, NU) # Unweighted (count-based) averaging would change the fit; weights matter. r = self.sample(exact_field(), 0.0) self.assertEqual(r["leaf_cells"], N * N - self.masks[1].sum() // 4 + self.masks[1].sum()) def record(n, depth, rows, explosion="constant", completed=True, source="s"): for row in rows: row["spurious_pressure_scaled_by_level"] = [row["spurious_pressure_scaled"]] * depth return dict(n=n, depth=depth, viscosity=U * n / sw.RE, rows=rows, completed=completed, explosion=explosion, source=source) def rows_for(n, rate_error=0.0, residual=1e-4, pressure=1e-3, phase=0.0): rate = (1 + rate_error) * sw.decay_rate(n, U * n / sw.RE) out = [] for tstar in TIMES: step = round(tstar * n / U) amp = U * math.exp(-rate * step) out.append(dict(tstar=step * U / n, step=step, a_hat=amp, b_hat=0.0, amplitude_ratio=1.0, modal_amplitude=amp, amplitude_exact=amp, phase_drift=phase, mass=float(n * n), rho_min=0.999, rho_max=1.001, max_speed=amp, longitudinal_l2=0.0, spurious_pressure_max_scaled=pressure, amplitude_error=0.0, residual_l2=residual, phase_drift_abs=phase, spurious_pressure_scaled=pressure, spurious_pressure_rms=pressure * U * U / 4, velocity_l2=residual, mass_relative=0.0)) return out class AssessTests(unittest.TestCase): def test_budget_form_is_max_not_sum(self): base = record(128, 1, rows_for(128, rate_error=1e-3, pressure=0.015)) fine = record(256, 1, rows_for(256, rate_error=2e-4)) # rate budget = max(2*1e-3, 0.002) = 0.002; 0.0025 would pass a sum form. refined = record(128, 2, rows_for(128, rate_error=0.0025)) report = sw.assess(base, fine, [refined]) self.assertEqual(report["status"], "FAIL") self.assertAlmostEqual(report["refined"]["constant"]["budget"]["rate_error"], 0.002) # pressure budget = max(2*0.015, 0.02) = 0.03 (control-dominated). self.assertAlmostEqual( report["refined"]["constant"]["budget"]["spurious_pressure_scaled"], 0.03) self.assertIn("finest", report["refined"]["constant"]["control_ratios"]) ok = record(128, 2, rows_for(128, rate_error=0.0015), explosion="projected_linear") self.assertEqual(sw.assess(base, fine, [ok])["status"], "PASS") def test_control_outcomes_are_distinct(self): fine = record(256, 1, rows_for(256)) refined = record(128, 2, rows_for(128)) failing = record(128, 1, rows_for(128, residual=0.5)) # complete, > eps(128) report = sw.assess(failing, fine, [refined]) self.assertEqual(report["status"], "CONTROL_SCREEN_FAIL") own = report["refined"]["constant"] self.assertTrue(own["own_screens_pass"]) # refined own screens kept self.assertFalse(own["controlled_pass"]) # but no controlled comparison passes good = record(128, 1, rows_for(128)) self.assertEqual(sw.assess(None, fine, [refined])["status"], "INVALID_CONTROL") incomplete = record(128, 1, rows_for(128), completed=False) self.assertEqual(sw.assess(incomplete, fine, [refined])["status"], "INVALID_CONTROL") mixed = record(256, 1, rows_for(256), source="other") self.assertEqual(sw.assess(good, mixed, [refined])["status"], "INVALID_CONTROL") def test_invalid_rows_are_rejected_before_screens(self): base = record(128, 1, rows_for(128)) fine = record(256, 1, rows_for(256)) for mutate in ( lambda rows: rows[2].update(rho_min=float("nan")), lambda rows: rows[2].update(spurious_pressure_scaled_by_level=[0.0, float("nan")]), lambda rows: rows[2].update(step=rows[1]["step"], tstar=rows[1]["tstar"]), lambda rows: rows[2].update(residual_l2=-1e-6), lambda rows: rows[2].pop("longitudinal_l2"), ): refined = record(128, 2, rows_for(128)) mutate(refined["rows"]) self.assertEqual(sw.assess(base, fine, [refined])["status"], "INVALID_RUN") # Legitimate signs stay allowed: negative amplitude and phase errors. refined = record(128, 2, rows_for(128)) for row in refined["rows"]: row.update(amplitude_error=-1e-3, phase_drift=-1e-4, phase_drift_abs=1e-4) report = sw.assess(base, fine, [refined]) self.assertEqual(report["status"], "PASS") self.assertEqual(sw.assess(base, fine, [])["status"], "INVALID_CONTROL") with self.assertRaises(ValueError): sw.assess(base, fine, [record(128, 2, rows_for(128)), record(128, 2, rows_for(128))]) def test_modal_extinction_is_a_failure_not_malformed(self): def extinct(rows): for row in rows[1:]: row.update(a_hat=0.0, b_hat=0.0, modal_amplitude=0.0, amplitude_ratio=0.0, amplitude_error=-1.0) return rows base = record(128, 1, rows_for(128)) fine = record(256, 1, rows_for(256)) refined = record(128, 2, extinct(rows_for(128))) report = sw.assess(base, fine, [refined]) self.assertEqual(report["status"], "FAIL") own = report["refined"]["constant"] self.assertTrue(own["valid"]) self.assertFalse(own["rate_fit_valid"]) self.assertIsNone(own["rate_error"]) self.assertIsNone(own["control_ratios"]["root"]["rate_error"]) self.assertFalse(own["own_screens_pass"] or own["controlled_pass"] or own["budget_pass"]) self.assertEqual(own["residual_l2"], 1e-4) # other finite metrics retained json.loads(json.dumps(report, allow_nan=False)) control = sw.assess(record(128, 1, extinct(rows_for(128))), fine, [record(128, 2, rows_for(128))]) self.assertEqual(control["status"], "CONTROL_SCREEN_FAIL") self.assertIsNone(control["refined"]["constant"]["budget"]["rate_error"]) self.assertFalse(control["refined"]["constant"]["controlled_pass"]) json.loads(json.dumps(control, allow_nan=False)) def test_wrong_shapes(self): base = record(128, 1, rows_for(128)) fine = record(256, 1, rows_for(256)) refined = record(128, 2, rows_for(128)) with self.assertRaises(ValueError): sw.assess(base, record(128, 1, rows_for(128)), [refined]) with self.assertRaises(ValueError): sw.assess(record(128, 1, rows_for(128)), fine, [record(256, 2, rows_for(256))]) def test_incomplete_record_fails(self): base = record(128, 1, rows_for(128)) fine = record(256, 1, rows_for(256)) refined = record(128, 2, rows_for(128), completed=False) self.assertEqual(sw.assess(base, fine, [refined])["status"], "FAIL") class FakeRun: def __init__(self, arrays): self.arrays = arrays def download(self): return self.arrays if len(self.arrays) > 1 else self.arrays[0] class RunnerSampleTests(unittest.TestCase): def test_refined_shear_sample_with_list_diagnostics(self): masks = composite_masks(N) arrays = [sw.initial(m.shape, level, N, NU) for level, m in enumerate(masks)] row = run_case.sample(FakeRun(arrays), masks, 0, N, NU, None, sw.metrics, 2) self.assertIsInstance(row["spurious_pressure_scaled_by_level"], list) self.assertTrue(run_case.finite_row(row)) bad = [a.copy() for a in arrays] bad[1][0, masks[1]] = np.nan with self.assertRaises(FloatingPointError): run_case.sample(FakeRun(bad), masks, 0, N, NU, None, sw.metrics, 2, "READY") self.assertFalse(run_case.finite_row(dict(a=1.0, b=[0.0, float("inf")]))) def test_taylor_green_sample_unchanged(self): masks = composite_masks(N) arrays = [oracle.initial(m.shape, level, N, NU) for level, m in enumerate(masks)] row = run_case.sample(FakeRun(arrays), masks, 0, N, NU, None, oracle.metrics, 2) self.assertEqual(row, oracle.metrics(arrays, masks, 0, N, NU, None)) uniform = [oracle.initial((N, N), 0, N, NU)] row = run_case.sample(FakeRun(uniform), [np.ones((N, N), bool)], 0, N, NU, None, oracle.metrics, 1) self.assertEqual(row["leaf_cells"], N * N) def receipt(n, depth, explosion="constant", diagnostic=None): """Synthetic receipt satisfying every contract field of the fixed four-case set.""" import assess_projected_pairs as tg steps = run_case.sample_steps(n)[0] rows = rows_for(n) for row, step in zip(rows, steps): row.update(step=step, tstar=step * U / n, spurious_pressure_scaled_by_level=[row["spurious_pressure_scaled"]] * depth) cells = [n * n] if depth == 1 else [17 * n * n // 32, 15 * n * n // 8] hops = [16 * n * 2**level for level in range(depth)] builds = [] for level in range(depth): size = n * 2**level builds.append(dict(source_sha256=tg.BUILD_SOURCES[size], arch="sm_70", flags=["--use_fast_math"], n_objects=0, ptx_sha256="a" * 64, config=dict(nx=10, ny=10, nz=1, lattice_model="D2Q9", collision="SRT", store_precision="FP32", streaming="double_buffer", ddf_shifting=True, les_enabled=False, velocity=0.0, viscosity=U * size / 100, rho=1.0, omega=1 / (.5 + 3 * U * size / 100), threads_per_block=256))) checkout = "/x/CelerisLab" return dict( n=n, depth=depth, re_L=100, explosion=explosion, U=U, mach=U * math.sqrt(3), viscosity=U * n / 100, tau_by_level=[0.5 + 3 * U * n / 100 * 2**l for l in range(depth)], source=dict(revision="r", dirty=False, source_tree_sha256=run_case.SOURCE_SHA256, harness_sha256={"x.py": "h"}, build_capture_sha256=run_case.BUILD_CAPTURE_SHA256, shear_wave_sha256=run_case.SHEAR_WAVE_SHA256), runtime_identity=dict(device=dict(name="V100", pci_bus_id="0", capability=[7, 0], driver=1, runtime=[12, 0]), nvcc_path="/nvcc", nvcc_version="v", python="3", numpy="1", pycuda="2"), runtime_source=dict(checkout=checkout, module=checkout + "/src/CelerisLab/__init__.py", compiler=checkout + "/src/CelerisLab/cuda/compiler_v2.py", kernel_root=checkout + "/src/CelerisLab/lbm/kernels"), cuda_visible_devices="0", numerical_revision=run_case.NUMERICAL_REVISION, reference_harness=run_case.REFERENCE_HARNESS, numerics=dict(tg.NUMERICS), initialization=sw.INITIALIZATION, limits=dict(sw.LIMITS), early_pressure=dict(enabled=False, rows=[], sampled_max_pressure_l2=None), refinements=[] if depth == 1 else rectangles(n, depth), workload=dict(active_cells_by_level=cells, level_hops=hops, root_intervals=16 * n, active_collision_updates_by_level=[c * h for c, h in zip(cells, hops)], active_collision_updates=sum(c * h for c, h in zip(cells, hops))), actual_build_sidecars=builds, rows=rows, completed=True, flow="shear_wave", mode=list(sw.MODE), polarization=sw.E.tolist(), angle_deg=sw.ANGLE_DEG, phase=sw.PHI, eps=sw.eps(n), **({"diagnostic": diagnostic} if diagnostic else {})) class ContractTests(unittest.TestCase): def setUp(self): import assess_shear_wave self.assess = assess_shear_wave.assess self.records = [receipt(128, 1), receipt(256, 1), receipt(128, 2), receipt(128, 2, "projected_linear")] def test_complete_contract_passes(self): result = self.assess(self.records, "r") self.assertTrue(result["contract_valid"], [k for k, v in result["checks"].items() if not v]) self.assertEqual(result["status"], "PASS") def test_eps_metadata_is_portable_but_bounded(self): import assess_shear_wave self.assertTrue(assess_shear_wave.close_ulps(0.012047856934923534, 0.012047856934923532)) exact = sw.eps(128) for value, ok in ((math.nextafter(exact, 1.0), True), (math.nextafter(math.nextafter(exact, 0.0), 0.0), True), (exact * (1 + 1e-12), False), (exact * 2, False), (str(exact), False), (float("nan"), False), (None, False)): records = [dict(r) for r in self.records] records[2]["eps"] = value status = self.assess(records, "r")["status"] self.assertEqual(status == "PASS", ok, (value, status)) def test_reproduced_false_passes_are_rejected(self): a = self.assess self.assertEqual(a(self.records[:2], "r")["status"], "INVALID_SET") # empty refined dup = [*self.records[:3], receipt(128, 2)] self.assertEqual(a(dup, "r")["status"], "INVALID_SET") # duplicate constant-only for mutate in ( lambda r: r.pop("source"), lambda r: r.update(flow="taylor_green"), lambda r: r.update(mode=[1, 2]), lambda r: r.update(re_L=1000), lambda r: r.update(refinements=[[(0, 0, 8, 8)]]), lambda r: r["actual_build_sidecars"][0].update(arch="sm_80"), lambda r: r["rows"][0].update(step=1, tstar=U / 128), # wrong initial time lambda r: r["source"].update(shear_wave_sha256="0" * 64), ): records = [dict(r, rows=[dict(x) for x in r["rows"]], source=dict(r["source"]), actual_build_sidecars=[dict(b) for b in r["actual_build_sidecars"]]) for r in self.records] mutate(records[2]) self.assertEqual(a(records, "r")["status"], "INVALID_SET") mixed = [dict(r) for r in self.records] mixed[1] = dict(mixed[1], source=dict(mixed[1]["source"], revision="other")) self.assertEqual(a(mixed, "r")["status"], "INVALID_SET") self.assertEqual(a(self.records, "wrong-revision")["status"], "INVALID_SET") def scaled_set(rate_error=0.0): d = run_case.SCALED_SHEAR out = [receipt(256, 1, diagnostic=d), receipt(512, 1, diagnostic=d), receipt(256, 2, diagnostic=d), receipt(256, 2, "projected_linear", diagnostic=d)] if rate_error: out[3]["rows"] = rows_for(256, rate_error=rate_error) for row, step in zip(out[3]["rows"], run_case.sample_steps(256)[0]): row.update(step=step, tstar=step * U / 256, spurious_pressure_scaled_by_level=[row["spurious_pressure_scaled"]] * 2) return out class ScaledContractTests(unittest.TestCase): def setUp(self): import assess_shear_wave self.mod = assess_shear_wave def test_scaled_contract_and_signed_excess(self): result = self.mod.assess(scaled_set(1.5e-3), "r", scaled=True) self.assertTrue(result["contract_valid"], [k for k, v in result["checks"].items() if not v]) self.assertEqual([c[1] for c in result["cases"]], [256, 512, 256, 256]) projected = result["signed"]["projected_linear"] self.assertAlmostEqual(projected["signed_rate_bias"], 1.5e-3, delta=1e-9) self.assertAlmostEqual(projected["excess_over_uniform_root"]["signed_rate_bias"], 1.5e-3, delta=1e-9) self.assertIn("excess_over_uniform_fine", projected) def test_scaled_and_default_sets_do_not_mix(self): n128 = [receipt(128, 1), receipt(256, 1), receipt(128, 2), receipt(128, 2, "projected_linear")] self.assertEqual(self.mod.assess(n128, "r", scaled=True)["status"], "INVALID_SET") self.assertEqual(self.mod.assess(scaled_set(), "r")["status"], "INVALID_SET") unlabeled = scaled_set() unlabeled[0].pop("diagnostic") self.assertEqual(self.mod.assess(unlabeled, "r", scaled=True)["status"], "INVALID_SET") tg_label = scaled_set() tg_label[2]["diagnostic"] = run_case.SCALED_DIAGNOSTIC self.assertEqual(self.mod.assess(tg_label, "r", scaled=True)["status"], "INVALID_SET") def test_resolution_trend_and_reference_pin(self): reference = [receipt(128, 1), receipt(256, 1), receipt(128, 2), receipt(128, 2, "projected_linear")] new = self.mod.assess(scaled_set(1.5e-3), "r", scaled=True) trend = self.mod.resolution_trend(new, reference, "r") self.assertTrue(trend["valid"]) change = trend["changes"]["projected_linear"]["signed_rate_bias"] self.assertAlmostEqual(change["signed_change"], 1.5e-3, delta=1e-9) self.assertIn("excess_over_uniform_fine.residual_l2", trend["changes"]["constant"]) with tempfile.TemporaryDirectory() as tmp: paths = [] for i, r in enumerate(reference): paths.append(Path(tmp) / f"{i}.json") paths[-1].write_text(json.dumps(r)) with self.assertRaises(ValueError): self.mod.load_reference(paths) def test_scaled_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"] shear = ["--flow", "shear_wave"] scaled = ["--diagnostic", run_case.SCALED_SHEAR] for extra, valid in ( (["--n", "512", "--depth", "1"] + shear + scaled, True), (["--n", "256", "--depth", "2", "--explosion", "projected_linear"] + shear + scaled, True), (["--n", "512", "--depth", "1"] + shear, False), (["--n", "128", "--depth", "2"] + shear + scaled, False), (["--n", "512", "--depth", "1"] + scaled, False), # needs shear flow (["--n", "512", "--depth", "1", "--early-pressure", "--diagnostic", run_case.SCALED_DIAGNOSTIC] + shear, False), # mutually exclusive (["--n", "256", "--depth", "2", "--early-pressure"] + shear + scaled, False)): argv = base + extra with self.subTest(argv=argv), patch.object(sys, "argv", argv), \ patch.object(run_case, "verify_source", return_value={}), \ 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) else: with self.assertRaises(SystemExit): run_case.main() class DefaultPathTests(unittest.TestCase): def test_frozen_oracle_is_byte_identical(self): digest = hashlib.sha256((HERE / "oracle.py").read_bytes()).hexdigest() self.assertEqual(digest, ORACLE_SHA256_EEC4677) def test_default_flow_is_frozen_taylor_green(self): init, met, limits, text = run_case.flow_functions() self.assertIs(init, oracle.initial) self.assertIs(met, oracle.metrics) self.assertIs(limits, oracle.LIMITS) self.assertEqual(text, run_case.INITIALIZATION) self.assertEqual(run_case.flow_functions("taylor_green"), (init, met, limits, text)) shear = run_case.flow_functions("shear_wave") self.assertIs(shear[0], sw.initial) self.assertIs(shear[1], sw.metrics) with self.assertRaises(ValueError): run_case.flow_functions("tg") if __name__ == "__main__": unittest.main()