435 lines
24 KiB
Python
435 lines
24 KiB
Python
"""Host geometry, receipt-corruption tests and mocked scheduling; no CUDA/flow."""
|
|
|
|
from contextlib import nullcontext, redirect_stdout
|
|
from copy import deepcopy
|
|
from dataclasses import asdict, replace
|
|
import hashlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import numpy as np
|
|
|
|
import run_case
|
|
from assess_projected_pairs import (BUILD_SOURCES, CASES, NUMERICS, NUMERICAL_REVISION,
|
|
SOURCE_SHA256, assess)
|
|
from oracle import LIMITS, TIMES, U, rectangles
|
|
|
|
REPO = Path(__file__).resolve().parents[3]
|
|
OLD_REVISION = "519473a61c7b319183aa54e55b9f097499dde0bf"
|
|
MOCK_REVISION = "f" * 40
|
|
RUNTIME = dict(device=dict(name="host mock", pci_bus_id="mock", capability=[7, 0],
|
|
driver=12040, runtime=[12, 4, 0]), nvcc_path="/mock/nvcc",
|
|
nvcc_version="mock nvcc", python="mock Python", numpy=np.__version__, pycuda="mock")
|
|
|
|
|
|
def source():
|
|
return dict(revision=MOCK_REVISION, dirty=False, source_tree_sha256=SOURCE_SHA256,
|
|
harness_sha256={"run_case.py": "a" * 64},
|
|
build_capture_sha256=run_case.BUILD_CAPTURE_SHA256)
|
|
|
|
|
|
def build_metadata(size):
|
|
cfg = replace(run_case.config(size, 100), nx=10, ny=10)
|
|
cfg.validate()
|
|
return dict(source_sha256=BUILD_SOURCES[size], config=asdict(cfg), n_objects=0,
|
|
arch="sm_70", flags=["--use_fast_math"], ptx_sha256="b" * 64)
|
|
|
|
|
|
def records(early=False):
|
|
from CelerisLab.lbm.hierarchy import build_hierarchy
|
|
from CelerisLab.lbm.refinement import ACTIVE
|
|
|
|
out = []
|
|
for re, n, depth, scheme in CASES:
|
|
cfg = run_case.config(n, re)
|
|
refinements = [] if depth == 1 else rectangles(n, depth)
|
|
h = build_hierarchy(cfg, refinements)
|
|
main_steps, early_steps = run_case.sample_steps(n, early)
|
|
out.append(dict(
|
|
re_L=re, n=n, depth=depth, explosion=scheme, U=U, mach=float(U*np.sqrt(3)),
|
|
viscosity=cfg.viscosity, tau_by_level=[0.5+3*c.viscosity for c in h.configs],
|
|
refinements=refinements, numerics=NUMERICS.copy(), initialization=run_case.INITIALIZATION,
|
|
source=source(), numerical_revision=NUMERICAL_REVISION,
|
|
reference_harness=run_case.REFERENCE_HARNESS,
|
|
actual_build_sidecars=[build_metadata(n * 2**level) for level in range(depth)],
|
|
runtime_identity=deepcopy(RUNTIME), cuda_visible_devices="0",
|
|
runtime_source=run_case.runtime_source(REPO), completed=True,
|
|
workload=run_case.workload([k == ACTIVE for k in h.kinds], 16*n),
|
|
rows=[dict(step=s, tstar=s*U/n, **{k: 0.0 for k in LIMITS}) for s in main_steps],
|
|
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)))
|
|
return out
|
|
|
|
|
|
class ProjectedPairsTests(unittest.TestCase):
|
|
def test_exact_geometry_work_and_samples(self):
|
|
from CelerisLab.lbm.hierarchy import build_hierarchy, leaf_partition
|
|
|
|
data = records(True)
|
|
self.assertEqual(sum(r["workload"]["active_collision_updates"] for r in data), 589299712)
|
|
for r in data:
|
|
n, depth = r["n"], r["depth"]
|
|
h = build_hierarchy(run_case.config(n, 100), r["refinements"])
|
|
np.testing.assert_array_equal(leaf_partition(h), np.ones((n, n)))
|
|
main, early = run_case.sample_steps(n, True)
|
|
self.assertEqual(main, [0, 4*n, 8*n, 12*n, 16*n])
|
|
self.assertEqual(early, [k*(n//16) for k in range(65)])
|
|
np.testing.assert_allclose(np.array(early)*U/n, np.arange(65)*.001875, rtol=0, atol=3e-17)
|
|
if depth == 2:
|
|
self.assertEqual(r["refinements"], [[(16, 16, 112, 96)]])
|
|
self.assertTrue(assess(data)["passed"])
|
|
self.assertTrue(assess(records())["passed"])
|
|
|
|
def test_exact_old519_geometry_and_table_parity(self):
|
|
from CelerisLab.lbm.blocks import BlockLayout2D
|
|
from CelerisLab.lbm.hierarchy import build_hierarchy
|
|
from CelerisLab.lbm.refinement_tables import projected_explosion_tables
|
|
|
|
# Frozen from executing exact 519473a61c7b319183aa54e55b9f097499dde0bf
|
|
# geometry and table definitions, after checking unchanged config,
|
|
# descriptors and blocks. No historical Git object is needed at runtime.
|
|
# Historical source SHA256, refinement.py then refinement_tables.py:
|
|
# 08c2a0557276a98cbcaa459d9bb7a9735d8bbbf64a6114c57af2fda61481a687
|
|
# dabff948bb30496384d1a52c666b0edbc2085f9c61ed1bbc39f748502629113d
|
|
frozen = {
|
|
"covered": ("|b1", (128, 128), "14a1b5be54e3309bea825a00a653105951f0c805622b877a86dd453d65556df3"),
|
|
"fine_kind": ("|u1", (256, 256), "891ff72c3d60686c62b1f35ffe59d054ac23b8f9d59aeb5782bc941202794cc0"),
|
|
"ghost_fine": ("<u8", (2880,), "2d0311ac0feb1c0e1e2b6bcf806fe616b8217513de218407267dcd7652287579"),
|
|
"ghost_coarse": ("<u4", (2880,), "5a3b305a0f841c8d05a13f105fda02b43ef875ab1321f9bca579f2b8ae313233"),
|
|
"coalesce_coarse": ("<u8", (1060,), "a1d88c60f282153dcf2bb497c59d777968b6db773d061d8fea08163c98155967"),
|
|
"coalesce_fine": ("<u8", (1060, 4), "08914c5d17c1ec8e2bfd234f581e4b3283de26296ae7d80403385feed7b89532"),
|
|
"neighbors": ("<u8", (2880, 4), "cfef3a07103241e3dfff24fd65ab967afcf20520f699574b191044d47cf5d1f4"),
|
|
"offsets": ("<f4", (2880, 2), "c98ba658d5f2cc18605e029fbe1da448ded3fd2436edcaf8ea097fa46aeb1f2d"),
|
|
}
|
|
hierarchy = build_hierarchy(run_case.config(128, 100), rectangles(128, 2))
|
|
pair = hierarchy.pairs[0]
|
|
self.assertEqual((pair.coarse_shape, pair.fine_shape, pair.fine_origin),
|
|
((128, 128), (256, 256), (0.0, 0.0)))
|
|
self.assertEqual(len(np.unique(pair.ghost_coarse)), 720)
|
|
self.assertTrue((hierarchy.kinds[0].ravel()[pair.ghost_coarse] == 0).all())
|
|
layout = BlockLayout2D.build(128, 128, 8, 8, periodic=(True, True))
|
|
neighbors, offsets = projected_explosion_tables(pair, layout, hierarchy.kinds[0])
|
|
arrays = dict(neighbors=neighbors, offsets=offsets)
|
|
arrays.update({name: getattr(pair, name) for name in frozen if name not in arrays})
|
|
for name, array in arrays.items():
|
|
with self.subTest(array=name):
|
|
self.assertEqual((array.dtype.str, array.shape, hashlib.sha256(array.tobytes()).hexdigest()),
|
|
frozen[name])
|
|
|
|
def test_actual_build_capture_rejects_tampered_ptx_and_sidecar(self):
|
|
from CelerisLab.cuda import compiler_v2
|
|
from tests.validation.multigrid.check_projected_pairs import capture_builds
|
|
|
|
for tamper in (None, "ptx", "sidecar"):
|
|
with self.subTest(tamper=tamper), tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / "blocks.ptx"
|
|
path.write_bytes(b"host-only mock PTX")
|
|
metadata = dict(ptx_sha256=hashlib.sha256(path.read_bytes()).hexdigest())
|
|
if tamper == "ptx":
|
|
path.write_bytes(b"tampered")
|
|
if tamper == "sidecar":
|
|
metadata["ptx_sha256"] = "0" * 64
|
|
Path(str(path) + ".json").write_text(json.dumps(metadata))
|
|
report = {"actual_build_sidecars": []}
|
|
with patch.object(compiler_v2, "load_module", return_value="mock-module") as load:
|
|
with capture_builds(report, lambda: None):
|
|
if tamper:
|
|
with self.assertRaisesRegex(AssertionError, "PTX and build sidecar"):
|
|
compiler_v2.load_module(str(path))
|
|
load.assert_not_called()
|
|
self.assertEqual(report["actual_build_sidecars"], [])
|
|
else:
|
|
self.assertEqual(compiler_v2.load_module(str(path)), "mock-module")
|
|
self.assertEqual(report["actual_build_sidecars"], [metadata])
|
|
self.assertIs(compiler_v2.load_module, load)
|
|
|
|
def test_frozen_formula_and_finest_control(self):
|
|
data = records(True)
|
|
data[0]["rows"][2]["velocity_l2"] = .002
|
|
data[0]["rows"][0]["velocity_l2"] = .04 # retained absolute gate, excluded from worst
|
|
for r in data[2:]:
|
|
for row in r["rows"][1:]:
|
|
row.update(velocity_l2=.0039, pressure_l2=.02, energy_relative=.004)
|
|
r["early_pressure"]["rows"][-1]["pressure_l2"] = .02
|
|
r["early_pressure"]["sampled_max_pressure_l2"] = .02
|
|
data[1]["rows"][2]["velocity_l2"] = .00001
|
|
data[3]["early_pressure"]["rows"][7]["pressure_l2"] = .9
|
|
data[3]["early_pressure"]["sampled_max_pressure_l2"] = .9
|
|
result = assess(data)
|
|
self.assertTrue(result["passed"])
|
|
velocity = result["comparisons"][0]
|
|
self.assertEqual(velocity["baseline_budget"], .004)
|
|
self.assertAlmostEqual(velocity["amr_to_uniform256"], 390)
|
|
self.assertEqual(result["early_pressure"][-1]["sampled_max_pressure_l2"], .9)
|
|
data[3]["rows"][-1]["pressure_l2"] = .02001
|
|
result = assess(data)
|
|
self.assertFalse(result["passed"])
|
|
self.assertEqual(result["comparisons"][4]["amr"], .02001)
|
|
|
|
def test_scheme_verdicts_are_independent(self):
|
|
data = records()
|
|
data[2]["rows"][-1]["pressure_l2"] = .03
|
|
result = assess(data)
|
|
self.assertFalse(result["passed"])
|
|
self.assertFalse(result["scheme_results"]["constant"]["passed"])
|
|
self.assertTrue(result["scheme_results"]["projected_linear"]["passed"])
|
|
|
|
def test_corruptions_cannot_pass(self):
|
|
good = records(True)
|
|
corruptions = [
|
|
lambda r: r.update(completed=False),
|
|
lambda r: r["rows"].pop(),
|
|
lambda r: r["rows"][-1].update(pressure_l2=float("nan")),
|
|
lambda r: r["rows"][-1].update(energy_relative=float("inf")),
|
|
lambda r: r["rows"][-1].update(pressure_l2=1e308),
|
|
lambda r: r["rows"][-1].update(velocity_l2=-1),
|
|
lambda r: r["rows"][-1].update(step=2047),
|
|
lambda r: r["rows"][-1].update(tstar=.4801),
|
|
lambda r: r["source"].update(dirty=True),
|
|
lambda r: r["source"].update(source_tree_sha256="wrong"),
|
|
lambda r: r["source"].update(revision=OLD_REVISION),
|
|
lambda r: r["source"].pop("build_capture_sha256"),
|
|
lambda r: r.update(numerical_revision=OLD_REVISION),
|
|
lambda r: r.update(reference_harness="wrong"),
|
|
lambda r: r.pop("runtime_identity"),
|
|
lambda r: r["runtime_identity"].pop("nvcc_version"),
|
|
lambda r: r["runtime_identity"].update(device="wrong"),
|
|
lambda r: r["runtime_identity"]["device"].update(capability=[8, 0]),
|
|
lambda r: r.update(cuda_visible_devices="0,1"),
|
|
lambda r: r.pop("actual_build_sidecars"),
|
|
lambda r: r["actual_build_sidecars"].pop(),
|
|
lambda r: r["actual_build_sidecars"].reverse(),
|
|
lambda r: r["actual_build_sidecars"][0].update(source_sha256="c" * 64),
|
|
lambda r: r["actual_build_sidecars"][0].update(arch="sm_80"),
|
|
lambda r: r["actual_build_sidecars"][0].update(flags=[]),
|
|
lambda r: r["actual_build_sidecars"][0].update(ptx_sha256="not-a-hash"),
|
|
lambda r: r["actual_build_sidecars"][0]["config"].update(ddf_shifting=False),
|
|
lambda r: r["source"]["harness_sha256"].update(extra="mixed"),
|
|
lambda r: r.update(viscosity=.01),
|
|
lambda r: r["tau_by_level"].reverse(),
|
|
lambda r: r["numerics"].update(ddf_shifting=False),
|
|
lambda r: r.update(initialization="different equilibrium"),
|
|
lambda r: r.update(refinements=[[(16, 16, 112, 104)]]),
|
|
lambda r: r["workload"].update(active_collision_updates=0),
|
|
lambda r: r["runtime_source"].update(kernel_root="/wrong/installed/tree"),
|
|
lambda r: r["early_pressure"]["rows"].pop(),
|
|
lambda r: r["early_pressure"]["rows"][8].update(step=65),
|
|
lambda r: r["early_pressure"]["rows"][-1].update(pressure_l2=.005),
|
|
lambda r: r["early_pressure"]["rows"][8].update(pressure_l2=float("nan")),
|
|
lambda r: r["early_pressure"].update(sampled_max_pressure_l2=.5),
|
|
lambda r: r["early_pressure"].update(enabled=False),
|
|
]
|
|
for mutation in corruptions:
|
|
bad = deepcopy(good)
|
|
mutation(bad[-1])
|
|
with self.subTest(mutation=mutation):
|
|
result = assess(bad)
|
|
self.assertFalse(result["passed"])
|
|
json.dumps(result, allow_nan=False) # failure summary itself remains serializable
|
|
self.assertTrue(assess(good, expected_revision=MOCK_REVISION)["passed"])
|
|
self.assertFalse(assess(good, expected_revision=OLD_REVISION)["passed"])
|
|
old = deepcopy(good)
|
|
for record in old:
|
|
record["source"].update(revision=OLD_REVISION,
|
|
source_tree_sha256="766e173df81d2881d3871007f15d5e321f5c7205ae578b0f80b43d1e7fe3e220")
|
|
self.assertFalse(assess(old)["passed"])
|
|
self.assertFalse(assess(good[:-1])["passed"])
|
|
self.assertFalse(assess(good + [good[-1]])["passed"])
|
|
self.assertFalse(assess([])["passed"])
|
|
incomplete = deepcopy(good)
|
|
incomplete[-1]["completed"] = False
|
|
self.assertFalse(assess(incomplete)["comparisons"][3]["passed"])
|
|
|
|
def test_pinned_build_sources_match_host_generated_inputs(self):
|
|
from CelerisLab.cuda import compiler_v2
|
|
|
|
for size in (128, 256):
|
|
cfg = replace(run_case.config(size, 100), nx=10, ny=10)
|
|
cfg.validate()
|
|
root = compiler_v2.generate_config(cfg)
|
|
try:
|
|
actual = json.loads((Path(root) / ".celeris-build").read_text())
|
|
expected = build_metadata(size)
|
|
self.assertEqual(actual, {key: expected[key] for key in
|
|
("source_sha256", "config", "n_objects")})
|
|
finally:
|
|
compiler_v2.cleanup_build_dir(root)
|
|
|
|
def test_source_preflight_requires_clean_exact_revision_and_tree(self):
|
|
import oracle
|
|
from tests.validation.multigrid import check_projected_pairs as gate
|
|
|
|
for module in (run_case, oracle):
|
|
with self.subTest(imported_module=module.__name__), \
|
|
patch.object(module, "__file__", "/wrong/checkout/" + Path(module.__file__).name), \
|
|
patch.object(gate, "preflight") as preflight:
|
|
with self.assertRaisesRegex(RuntimeError, "Wrong runner or oracle checkout"):
|
|
run_case.verify_source(REPO, MOCK_REVISION)
|
|
preflight.assert_not_called()
|
|
for expected, status, diff, fingerprint in (
|
|
(MOCK_REVISION, "", 0, SOURCE_SHA256),
|
|
(OLD_REVISION, "", 0, SOURCE_SHA256),
|
|
(MOCK_REVISION, " M tracked.py", 0, SOURCE_SHA256),
|
|
(MOCK_REVISION, "", 1, SOURCE_SHA256),
|
|
(MOCK_REVISION, "", 0, "wrong")):
|
|
provenance = dict(source(), source_tree_sha256=fingerprint)
|
|
with self.subTest(expected=expected, status=status, diff=diff, fingerprint=fingerprint), \
|
|
patch.object(gate.subprocess, "check_output", side_effect=[MOCK_REVISION, status]), \
|
|
patch.object(gate.subprocess, "call", return_value=diff), \
|
|
patch.object(run_case, "provenance", return_value=provenance):
|
|
if expected != MOCK_REVISION or status or diff or fingerprint != SOURCE_SHA256:
|
|
with self.assertRaises((ValueError, RuntimeError)):
|
|
run_case.verify_source(REPO, expected)
|
|
else:
|
|
self.assertEqual(run_case.verify_source(REPO, expected), source())
|
|
|
|
def test_explicit_opt_in_selector_and_dry_run_without_cuda(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
output = Path(tmp) / "case.json"
|
|
base = ["run_case.py", "--repo", str(REPO), "--re", "100", "--n", "128",
|
|
"--depth", "1", "--output", str(output)]
|
|
pinned = base + ["--expected-revision", MOCK_REVISION]
|
|
for argv, device in ((base + ["--run-gpu"], "0"), (pinned, "0"),
|
|
(pinned + ["--run-gpu"], ""), (pinned + ["--run-gpu"], "0,1"),
|
|
(pinned + ["--run-gpu"], "-1"),
|
|
(pinned + ["--run-gpu", "--explosion", "projected"], "0"),
|
|
(pinned + ["--run-gpu", "--explosion", "projected_linear"], "0")):
|
|
with self.subTest(argv=argv, device=device), patch.object(sys, "argv", argv), \
|
|
patch.dict("os.environ", CUDA_VISIBLE_DEVICES=device), \
|
|
patch.object(run_case, "verify_source") as verify, patch("sys.stderr", io.StringIO()):
|
|
with self.assertRaises(SystemExit) as caught:
|
|
run_case.main()
|
|
self.assertEqual(caught.exception.code, 2)
|
|
verify.assert_not_called()
|
|
self.assertFalse(output.exists())
|
|
stdout = io.StringIO()
|
|
cuda_modules = {name: None for name in ("pycuda", "CelerisLab.cuda.context",
|
|
"CelerisLab.lbm.block_stepper", "CelerisLab.lbm.refinement_stepper")}
|
|
with patch.object(sys, "argv", pinned + ["--dry-run"]), \
|
|
patch.dict(sys.modules, cuda_modules), patch.dict("os.environ", CUDA_VISIBLE_DEVICES=""), \
|
|
patch.object(run_case, "verify_source", return_value=source()) as verify, \
|
|
patch.object(run_case, "runtime_identity", side_effect=AssertionError("CUDA reached")), \
|
|
redirect_stdout(stdout):
|
|
self.assertEqual(run_case.main(), 0)
|
|
verify.assert_called_once_with(REPO, MOCK_REVISION)
|
|
self.assertTrue(json.loads(stdout.getvalue())["dry_run"])
|
|
self.assertFalse(output.exists())
|
|
|
|
def test_runtime_rejects_installed_kernel_tree(self):
|
|
from CelerisLab.cuda import compiler_v2
|
|
with patch.object(compiler_v2, "_KERNEL_DIR", "/other/CelerisLab/lbm/kernels"):
|
|
with self.assertRaisesRegex(RuntimeError, "CUDA source/include root"):
|
|
run_case.runtime_source(REPO)
|
|
|
|
def test_existing_output_is_never_overwritten(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
output = Path(tmp) / "failed.json"
|
|
output.write_text('{"completed": false}\n')
|
|
argv = ["run_case.py", "--repo", str(REPO), "--re", "100", "--n", "128",
|
|
"--depth", "1", "--output", str(output),
|
|
"--expected-revision", MOCK_REVISION, "--run-gpu"]
|
|
with patch.object(sys, "argv", argv), patch("sys.stderr", io.StringIO()):
|
|
with self.assertRaises(SystemExit) as caught:
|
|
run_case.main()
|
|
self.assertEqual(caught.exception.code, 2)
|
|
self.assertEqual(output.read_text(), '{"completed": false}\n')
|
|
result = subprocess.run([sys.executable, str(Path(__file__).with_name("assess_projected_pairs.py")),
|
|
"unused.json", "--expected-revision", MOCK_REVISION,
|
|
"--output", str(output)], capture_output=True)
|
|
self.assertEqual(result.returncode, 2)
|
|
self.assertEqual(output.read_text(), '{"completed": false}\n')
|
|
|
|
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),
|
|
(True, 2, "resource"), (True, 2, "nonfinite"),
|
|
(True, 2, "probe_nonfinite")):
|
|
calls, selectors = [], []
|
|
|
|
class FakeRun:
|
|
def __init__(self, *args, **kwargs):
|
|
self.step_count = 0
|
|
if "coarse_boundary" in kwargs:
|
|
selectors.append(kwargs["explosion"])
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
pass
|
|
|
|
def initialize(self, *arrays):
|
|
pass
|
|
|
|
def step(self, count):
|
|
self.assertion = count >= 0
|
|
if not self.assertion:
|
|
raise AssertionError("backward step")
|
|
if fail == "resource" and self.step_count >= 16:
|
|
raise MemoryError("mock resource failure")
|
|
self.step_count += count
|
|
calls.append(self.step_count)
|
|
|
|
def download(self):
|
|
return None
|
|
|
|
modules = {
|
|
"CelerisLab.cuda.context": SimpleNamespace(CudaContext=FakeRun),
|
|
"CelerisLab.lbm.block_stepper": SimpleNamespace(PeriodicBlockStepper=FakeRun),
|
|
"CelerisLab.lbm.refinement_stepper": SimpleNamespace(FixedPatchStepper=FakeRun),
|
|
}
|
|
|
|
def metric(arrays, masks, step, n, nu, mass):
|
|
return dict(step=step, tstar=step*U/n, mass=16384.,
|
|
**{k: float("nan") if ((fail == "nonfinite" and step == 16)
|
|
or (fail == "probe_nonfinite" and step == 1))
|
|
and k == "energy_relative" else 0.0 for k in LIMITS})
|
|
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
output = Path(tmp) / "case.json"
|
|
argv = ["run_case.py", "--repo", str(REPO), "--re", "100", "--n", "128",
|
|
"--depth", str(depth), "--output", str(output),
|
|
"--expected-revision", MOCK_REVISION, "--run-gpu"]
|
|
if early:
|
|
argv += ["--early-pressure", "--explosion", "projected_linear"]
|
|
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, \
|
|
patch.object(run_case, "runtime_identity", return_value=deepcopy(RUNTIME)), \
|
|
patch.object(run_case, "build_capture", return_value=nullcontext()), \
|
|
patch.object(run_case, "metrics", side_effect=metric), redirect_stdout(io.StringIO()):
|
|
if fail:
|
|
with self.assertRaises(MemoryError if fail == "resource" else FloatingPointError):
|
|
run_case.main()
|
|
else:
|
|
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(receipt["source"], source())
|
|
self.assertEqual(receipt["runtime_identity"], RUNTIME)
|
|
self.assertEqual(receipt["completed"], not fail)
|
|
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)
|
|
else:
|
|
self.assertEqual(len(receipt["rows"]), 1)
|
|
self.assertEqual(len(receipt["early_pressure"]["rows"]),
|
|
{"resource": 3, "nonfinite": 2, "probe_nonfinite": 1}[fail])
|
|
if fail == "probe_nonfinite":
|
|
self.assertNotIn("ready_after_step", receipt)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|