"""Host-only checks for the opt-in CUDA gate; never imports or compiles CUDA.""" from copy import deepcopy import importlib.util from pathlib import Path import unittest from unittest.mock import patch import numpy as np from CelerisLab.config import LBMConfig from CelerisLab.lbm.blocks import BlockLayout2D from CelerisLab.lbm.hierarchy import build_hierarchy from CelerisLab.lbm.refinement import ACTIVE, OUTER_GHOST, FixedPatch2D, FixedPatchSet2D from CelerisLab.lbm.refinement_tables import projected_explosion_tables from tests.unit.test_hierarchy import CFG, FOUR from tests.unit.test_projected_explosion import decode, reconstruct, totals PATH = Path(__file__).resolve().parents[1] / 'validation/multigrid/check_projected_pairs.py' spec = importlib.util.spec_from_file_location('projected_pair_gate', PATH) gate = importlib.util.module_from_spec(spec) spec.loader.exec_module(gate) class ProjectedGateOracle(unittest.TestCase): def check_pair(self, pair, layout, kind): ids, offsets = projected_explosion_tables(pair, layout, kind) neighbors = decode(layout, ids) coarse = np.random.default_rng(731).normal(size=(9, *pair.coarse_shape)) legal = np.zeros(pair.coarse_shape, bool) legal.ravel()[pair.ghost_coarse] = True coarse[:, ~legal] = np.nan expected, _ = reconstruct(coarse, pair, neighbors, offsets) # The coordinate oracle must not accidentally share production tables. with patch('CelerisLab.lbm.refinement_tables.projected_explosion_tables', side_effect=AssertionError('oracle used production stencil')): actual = gate.independent_explosion(coarse, pair, layout, kind) np.testing.assert_allclose(actual, expected, rtol=0, atol=2e-15) center = coarse.reshape(9, -1)[:, pair.ghost_coarse] self.assertTrue(np.isfinite(actual).all()) self.assertGreater(np.max(abs(actual[0] - center[0])), .01) self.assertGreater(np.max(abs(actual[1:] - center[1:])), .01) return coarse, neighbors, offsets, actual def test_scalar_union_nested_sparse_and_unequal_strides(self): def fine_layout(pair, block): return (pair.fine_layout(block) if isinstance(pair, FixedPatchSet2D) else BlockLayout2D.build(*pair.fine_shape, *block, periodic=(False,False))) small = LBMConfig(nx=16, ny=16, viscosity=.02) fixtures = [(small, [(4,4,12,12), (8,8,16,16), (8,8,16,16)]), (CFG, FOUR), (small, [[(4,4,5,7), (4,5,7,7)]])] for cfg, refs in fixtures: hierarchy = build_hierarchy(cfg, refs) for index, pair in enumerate(hierarchy.pairs): block, child_block = ((4,4), (4,8)) if index % 2 == 0 else ((4,8), (4,4)) layout = (BlockLayout2D.build(cfg.ny, cfg.nx, *block) if index == 0 else fine_layout(hierarchy.pairs[index-1], block)) with self.subTest(refs=refs, pair=index): self.assertNotEqual(np.prod(layout.local_shape), np.prod(fine_layout(pair, child_block).local_shape)) if index and isinstance(pair, FixedPatchSet2D): self.assertTrue((~layout.owned).any()) self.check_pair(pair, layout, hierarchy.kinds[index]) def test_axis_periodicity_and_mean_preserving_mutants(self): pair = FixedPatch2D.from_config(LBMConfig(nx=16,ny=16,viscosity=.1), (2,2,14,14)) kind = np.where(pair.covered, OUTER_GHOST, ACTIVE) outputs = [] for periodic in ((False,False), (True,False), (False,True), (True,True)): layout = BlockLayout2D.build(16,16,4,4,periodic=periodic) coarse, neighbors, offsets, actual = self.check_pair(pair, layout, kind) outputs.append(actual) donors, first, inverse = np.unique(pair.ghost_coarse, return_index=True, return_inverse=True) wrong = neighbors[first[np.roll(np.arange(len(donors)), 1)]][inverse] for bad_neighbors, bad_offsets in ((neighbors, -offsets), (wrong, offsets)): mutant, _ = reconstruct(coarse, pair, bad_neighbors, bad_offsets) self.assertGreater(np.max(abs(mutant - actual)), .01) for donor in donors: rows = pair.ghost_coarse == donor self.assertEqual(rows.sum(), 4) np.testing.assert_allclose(mutant[:,rows].mean(axis=1), coarse.reshape(9,-1)[:,donor], rtol=0, atol=1e-15) bad_kind = kind.copy(); bad_kind.ravel()[pair.ghost_coarse[0]] = OUTER_GHOST with self.assertRaisesRegex(AssertionError, 'illegal center donor'): gate.independent_explosion(coarse, pair, layout, bad_kind) for i, j in ((0,1), (0,2), (1,3), (2,3)): self.assertGreater(np.max(abs(outputs[i] - outputs[j])), .01) def test_concave_replay_preserves_existing_fp64_budget(self): cfg, refs, blocks = gate.cases()['concave'] hierarchy = build_hierarchy(cfg, refs); pair = hierarchy.pairs[0] layout = BlockLayout2D.build(cfg.ny, cfg.nx, *blocks[0]) rng = np.random.default_rng(17) values = tuple(gate.W[:,None,None] * (1 + rng.normal(0,.002,(9,*shape))) for shape in (pair.coarse_shape, pair.fine_shape)) start = totals(*values, pair) values[0][:,pair.covered] = np.nan for _ in range(8): values = gate.replay_two_hops(values, pair, layout, hierarchy.kinds[0]) np.testing.assert_allclose(totals(*values, pair), start, rtol=0, atol=4e-12) self.assertEqual(gate.REPLAY_ATOL, 3e-6) # CUDA replay bound is separate. class ProjectedGateReceipts(unittest.TestCase): def test_digest_exact_finite_bits_and_nan_masks(self): values = np.array([0., -0., np.nan, 1.], dtype=np.float32) self.assertEqual(gate.digest(values), gate.digest(values.copy())) payload = values.copy(); payload.view(np.uint32)[2] = 0x7fc00001 self.assertEqual(gate.digest(values), gate.digest(payload)) variants = [np.array([-0., -0., np.nan, 1.], np.float32), np.array([0., np.nan, -0., 1.], np.float32), np.array([0., -0., np.nan, np.nextafter(np.float32(1), np.float32(2))], np.float32), values.reshape(2,2), values.astype(np.float64)] for changed in variants: self.assertNotEqual(gate.digest(values), gate.digest(changed)) with self.assertRaisesRegex(AssertionError, 'infinite'): gate.digest(np.array([np.inf])) def test_snapshot_negative_controls_and_exact_dev_options(self): snapshot = [dict(step=0, levels=[dict(raw='raw',public='public',active=4)], mass_momentum=[1.,0.,0.],observed_level_hops=[0,0])] gate.compare_snapshots(snapshot, deepcopy(snapshot)) diagnostic = deepcopy(snapshot); diagnostic[0]['mass_momentum'][0] = 2. gate.compare_snapshots(snapshot, diagnostic) # Diagnostics are not bitwise gate fields. mutants = [[]] for key, value in (('raw','changed'), ('public','changed'), ('active',3)): changed = deepcopy(snapshot); changed[0]['levels'][0][key] = value mutants.append(changed) changed = deepcopy(snapshot); changed[0]['step'] = 1; mutants.append(changed) changed = deepcopy(snapshot); changed[0]['observed_level_hops'] = [0,1]; mutants.append(changed) for changed in mutants: with self.assertRaises(AssertionError): gate.compare_snapshots(snapshot, changed) self.assertEqual(gate.default_options('omitted'), {}) self.assertEqual(gate.default_options('constant'), {'explosion':'constant'}) self.assertEqual(gate.default_options('projected_linear'), {'explosion':'projected_linear'}) def test_baseline_receipt_rejects_provenance_and_contract_mutants(self): keys = [f'{name}:shifted={shifted}' for name in gate.cases() for shifted in (False,True)] good = dict(completed=True,dry_run=False,mode='baseline',production_revision=gate.BASE, source=dict(revision=gate.BASE,dirty=False),driver_sha256='driver', stops=list(gate.STOPS),operator_tolerance=[6e-8,2e-7], rest_tolerance=[3e-7,1e-6],replay_tolerance=[3e-6,0],default_snapshots=dict.fromkeys(keys, []), actual_build_sidecars=[{'ptx_sha256':'captured'}]) self.assertIs(gate.baseline_receipt(good, 'driver', keys), good) mutants = dict(completed=False,dry_run=True,mode='candidate',production_revision='wrong', driver_sha256='wrong',stops=[0,1,2],operator_tolerance=[1e-6,2e-7], rest_tolerance=[3e-7,1e-5],replay_tolerance=[3e-6,1e-7],actual_build_sidecars=[],default_snapshots={}) for field, value in mutants.items(): bad = deepcopy(good); bad[field] = value with self.subTest(field=field), self.assertRaises(ValueError): gate.baseline_receipt(bad, 'driver', keys) for changes in (dict(revision='wrong'), dict(dirty=True), dict(dirty=None)): bad = deepcopy(good); bad['source'].update(changes) with self.assertRaises(ValueError): gate.baseline_receipt(bad, 'driver', keys) bad = deepcopy(good); bad['default_snapshots']['unexpected'] = [] with self.assertRaises(ValueError): gate.baseline_receipt(bad, 'driver', keys) if __name__ == '__main__': unittest.main()