"""CPU-only union geometry and production host-control-flow tests. The AST loader omits CUDA imports and supplies explicit mocks; it executes unchanged production definitions, never CUDA kernels or a numerical emulator. """ import ast from contextlib import nullcontext from dataclasses import replace from pathlib import Path import types import unittest from unittest.mock import Mock, patch import numpy as np from CelerisLab.config import LBMConfig from CelerisLab.lbm import refinement as r def host_module(): path=Path(r.__file__).with_name('refinement_stepper.py') tree=ast.parse(path.read_text()) tree.body=[node for node in tree.body if not ( isinstance(node,ast.Import) and any(a.name=='pycuda.driver' for a in node.names) or isinstance(node,ast.ImportFrom) and node.module in ( 'cuda','curved_links'))] module=types.ModuleType('CelerisLab.lbm._test_host_refinement') module.__package__='CelerisLab.lbm' from CelerisLab.body.jets import surface_link_velocity module.__dict__.update(cuda=Mock(),compiler=Mock(), surface_link_velocity=surface_link_velocity) module.cuda.mem_get_info.return_value=(1<<40,1<<40) # free-memory preflight passes exec(compile(tree,str(path),'exec'),module.__dict__) return module host=host_module() CFG = LBMConfig(nx=48, ny=32, viscosity=.1) PARENTS = ((4,4,20,28), (24,4,44,28)) # exact gap=4 coarse cells CHILDREN = ((16,16,32,40), (56,16,80,40)) # parent-allocated L1 cells class Gate(unittest.TestCase): def setUp(self): self.parent = r.FixedPatchSet2D.from_config(CFG, PARENTS) def test_one_element_equivalence(self): cfg = LBMConfig(nx=16,ny=16,viscosity=.1) local = r.FixedPatch2D.from_config(cfg,(4,4,12,12)) global_parent = r.FixedPatchSet2D.from_config(cfg,[(4,4,12,12)]) a, ak = r.nest_patch(cfg,local,(8,8,16,16)) b, bk = r.nest_patch_set(cfg,global_parent,[(12,12,20,20)]) local_set, local_kind = r.nest_patch_set(cfg,local,[(8,8,16,16)]) np.testing.assert_array_equal(ak,local_kind) np.testing.assert_array_equal(a.fine_kind,local_set.patches[0].fine_kind) np.testing.assert_array_equal(a.ghost_coarse,local_set.ghost_coarse) np.testing.assert_array_equal(a.coalesce_coarse,local_set.coalesce_coarse) np.testing.assert_array_equal(ak, bk[4:28,4:28]) # L2 node-index offsets in global L2 grid: local parent origin=2 L0. ay,ax=np.nonzero(a.fine_kind == r.ACTIVE) by,bx=np.nonzero(b.fine_kind == r.ACTIVE) np.testing.assert_array_equal(np.stack([ay+20,ax+20],-1),np.stack([by,bx],-1)) np.testing.assert_array_equal(a.fine_kind,b.fine_kind[20:44,20:44]) # Compare complete per-region maps after undoing set addressing. bp=b.patches[0] np.testing.assert_array_equal(a.coalesce_fine,bp.coalesce_fine) aq,ac=np.divmod(a.coalesce_coarse,24*24) bq,bc=np.divmod(bp.coalesce_coarse,32*32) np.testing.assert_array_equal(aq,bq) np.testing.assert_array_equal(np.stack(np.divmod(ac,24),-1)+4,np.stack(np.divmod(bc,32),-1)) ay,ax=np.divmod(a.ghost_coarse,24) by,bx=np.divmod(bp.ghost_coarse,32) np.testing.assert_array_equal(np.stack([ay+4,ax+4],-1),np.stack([by,bx],-1)) def test_twin_tile_chains_and_order(self): children=((16,16,32,40),(56,16,72,40)) child,kind=r.nest_patch_set(CFG,self.parent,children) other,ok=r.nest_patch_set(CFG,r.FixedPatchSet2D.from_config(CFG,PARENTS[::-1]),children[::-1]) for name in ('fine_kind','ghost_fine','ghost_coarse','coalesce_coarse','coalesce_fine'): np.testing.assert_array_equal(getattr(child,name),getattr(other,name)) np.testing.assert_array_equal(kind,ok) for p in child.patches: self.assertEqual(p.fine_shape,child.patches[0].fine_shape) np.testing.assert_array_equal(child.patches[0].fine_kind,child.patches[1].fine_kind) self.assertEqual(child.tau_coarse,self.parent.tau_fine) self.assertAlmostEqual(child.tau_fine,1.7) def test_asymmetric_no_child_region(self): child,kind=r.nest_patch_set(CFG,self.parent,[CHILDREN[0]]) np.testing.assert_array_equal(kind[:,48:],self.parent.fine_kind[:,48:]) self.assertEqual(int(child.covered.sum()),16*24) def test_two_children_one_parent(self): child,kind=r.nest_patch_set(CFG,self.parent,[(16,16,32,24),(16,28,32,40)]) self.assertEqual(len(child.patches),2) self.assertEqual(int(child.covered.sum()),16*20) self.assertEqual(int((kind==r.ACTIVE).sum()),int((self.parent.fine_kind==r.ACTIVE).sum())-16*20) def test_holes_donors_receivers_footprint_and_invalid_pull(self): child,_=r.nest_patch_set(CFG,self.parent,CHILDREN) cases=[('inherited ACTIVE', lambda: r.nest_patch_set(CFG,self.parent,[(40,16,48,40)])), ('donors/receivers', lambda: r.nest_patch_set(CFG,self.parent,[(8,16,32,40)]))] for text,call in cases: with self.subTest(text=text),self.assertRaisesRegex(ValueError,text):call() # Synthetic upstream dependency inside child coverage isolates the final guard. upstream=self.parent.coalesce_fine.copy() upstream[0,:]=20*self.parent.fine_shape[1]+20 # q=0, deep inherited ACTIVE with self.assertRaisesRegex(ValueError,'restriction footprint'): r._nested_kind(replace(self.parent,coalesce_fine=upstream),child) hole=int(np.flatnonzero(~self.parent.fine_layout((4,4)).owned.ravel())[0]) donors=child.ghost_coarse.copy();donors[0]=hole with self.assertRaisesRegex(ValueError,'donors/receivers'): r._nested_kind(self.parent,replace(child,ghost_coarse=donors)) receivers=child.coalesce_coarse.copy();receivers[0]=hole with self.assertRaisesRegex(ValueError,'donors/receivers'): r._nested_kind(self.parent,replace(child,coalesce_coarse=receivers)) # Remove all supplying rows; covered-to-uncovered pulls must be rejected. with self.assertRaisesRegex(ValueError,'surviving parent ACTIVE pull'): r._nested_kind(self.parent,replace(child,coalesce_coarse=child.coalesce_coarse[:0],coalesce_fine=child.coalesce_fine[:0])) def test_existing_l2_body_jet_preparation_on_union(self): from CelerisLab.body.geometry.circle import CircleGeometry from CelerisLab.body.objects import SimObject from CelerisLab.body.jets import Jet, Tube, Parallel, jet_link_shapes child,_=r.nest_patch_set(CFG,self.parent,CHILDREN) centers=((12.,14.),(34.,14.)) objects=[SimObject(99,geometry=CircleGeometry(*c,.8),center=c,radius=.8) for c in centers] jets=[Jet(Tube((c[0]-1.5,c[1]),0.,3.,.3),Parallel(0.),amplitude=.002) for c in centers] middle=replace(CFG,nx=self.parent.fine_shape[1],ny=self.parent.fine_shape[0],viscosity=.2) boundary=r.prepare_fine_boundary(middle,child,objects,jets=jets, parent_origin=self.parent.fine_origin,parent_spacing=.5) ids=boundary.curved_host[3] np.testing.assert_array_equal(np.unique(ids),[0,1]) self.assertEqual(int((ids==0).sum()),int((ids==1).sum())) np.testing.assert_allclose(boundary.curved_hits[ids==1]-boundary.curved_hits[ids==0], np.broadcast_to([22.,0.],boundary.curved_hits[ids==0].shape),atol=1e-14,rtol=0) _,jid=jet_link_shapes(jets,boundary.curved_hits) for i in (0,1): self.assertTrue((ids[jid==i]==i).all()) self.assertTrue((jid==i).any()) def test_sparse_parent_tables_preserve_inherited_kinds(self): child,kind=r.nest_patch_set(CFG,self.parent,CHILDREN) layout=self.parent.fine_layout((4,4));order=np.argsort(child.coalesce_coarse) donors,targets,bkind=host.blocked_coarse_tables(child,layout,order,composed_kind=kind) self.assertTrue((donors < layout.hole).all()) self.assertTrue((targets < np.prod(layout.blocked_shape)).all()) np.testing.assert_array_equal(bkind[layout.owner[layout.owned]],kind[layout.owned]) inherited=self.parent.fine_kind != r.ACTIVE np.testing.assert_array_equal(kind[inherited],self.parent.fine_kind[inherited]) self.assertTrue((kind[child.covered] == r.OUTER_GHOST).all()) l2=child.fine_layout((4,4)) ghosts,children,ckind=host.blocked_fine_tables(child,l2,order) self.assertTrue((ghosts20: raise RuntimeError('table upload failed') module.cuda.memcpy_htod.side_effect=upload objects,jets=bodies_and_jets() if bodies else ((),()) try: run=module.FixedPatchStepper(CFG,context=context, **levels(children=children,objects=((),(),objects),jets=((),(),jets))) except RuntimeError: for allocation in allocations: allocation.free.assert_called_once_with() raise return run,module,allocations class ConstructorAndSchedule(unittest.TestCase): def test_invalid_inputs_fail_before_context_or_allocations(self): context=Mock() b=(4,4) cases=[dict(children=(8,8,16,16)),dict(children=[]), dict(children=[(16,16,32)]), dict(children=[(16.5,16,32,40)]), dict(children=[(16,16,32,24),(16,27,32,55)]), dict(children=[(40,16,48,40)]), dict(children=[(8,16,32,40)]), dict(blocks=(b,b,None)),dict(blocks=(b,None,b)),dict(blocks=(None,b,b)), dict(blocks=(b,b,(5,4))),dict(blocks=(b,(5,4),b)), dict(blocks=(b,b,(4,))),dict(children=(16,16,32,40))] for changes in cases: with self.subTest(changes=changes),self.assertRaises((ValueError,TypeError)): host.FixedPatchStepper(CFG,context=context,**levels(**changes)) with self.assertRaisesRegex(ValueError,'margin'): host.FixedPatchStepper(CFG,context=context,**levels(parents=[(1,4,20,28),(23,4,44,28)])) objects,jets=bodies_and_jets() with self.assertRaisesRegex(ValueError,'only on the finest level'): host.FixedPatchStepper(CFG,context=context,**levels(objects=((),objects,()))) from CelerisLab.body.jets import Jet,Tube,Parallel spanning=Jet(Tube((10.5,14.),0.,25.,.3),Parallel(0.),amplitude=0.) with self.assertRaisesRegex(ValueError,'one jet'): host.FixedPatchStepper(CFG,context=context, **levels(objects=((),(),objects),jets=((),(),[spanning]))) context.activate.assert_not_called() host.cuda.mem_alloc.assert_not_called() def test_aggregate_schedule_independent_of_region_count(self): for children,bodies in ((CHILDREN,True),(CHILDREN[:1],False), (((16,16,32,24),(16,28,32,40)),False)): run,_,allocations=mock_backend(children=children,bodies=bodies) self.assertEqual(len(run._levels),3) c,f,g=run._levels self.assertEqual(len({id(level.current) for level in run._levels} | {id(level.spare) for level in run._levels}),6) events=[] initial_c=c.current def record(name): return lambda source,*a,**kw: events.append((name,source)) run._pairs[0].explode.side_effect=record('explode01') run._pairs[1].explode.side_effect=record('explode12') for i,level in enumerate(run._levels): level.kernel.side_effect=record(f'step{i}') if bodies: g.boundary.kernel.side_effect=record('bfl2') run._pairs[1].coalesce.side_effect=lambda src,dst,*a,**kw: events.append(('restrict21',dst)) run._pairs[0].coalesce.side_effect=lambda src,dst,*a,**kw: events.append(('restrict10',dst)) run._status='ready' expected=[('explode01',initial_c)] for middle in (f.current,f.spare): expected.append(('explode12',middle)) for leaf in (g.current,g.spare): expected.append(('halo2',leaf)) if bodies: expected.append(('bfl2',leaf)) expected.append(('step2',leaf)) expected.extend([('halo1',middle),('restrict21',middle),('step1',middle)]) expected.extend([('halo0',initial_c),('restrict10',initial_c),('step0',initial_c)]) def gather(level,buffer,**kw): events.append((f'halo{next(i for i,l in enumerate(run._levels) if l is level)}',buffer)) with patch.object(run,'_gather_halo',side_effect=gather): run.step() self.assertEqual(events,expected) self.assertEqual([l.kernel.call_count for l in run._levels],[1,2,4]) # A stale-feed implementation must fail the source-identity guard. feeds=[] advance=run._advance def checked_advance(index, buffer, **kw): if index == 2: # L1 -> L2 feed: must be the L1 hop source self.assertIs(buffer,(f.current,f.spare)[len(feeds)]) feeds.append(buffer) advance(index,buffer,**kw) with patch.object(run,'_advance',side_effect=lambda index,buffer,**kw: checked_advance(index,f.current if index == 2 else buffer,**kw)): with self.assertRaises(AssertionError): run.step() # A per-region shared-parent advance must fail the 1:2:4 guard. run._status='ready' counts=[0,0,0] def count(index, multiple=1): def step(*a,**kw): counts[index]+=multiple return step c.kernel.side_effect=count(0) f.kernel.side_effect=count(1,2) # mutant: twice per shared L1 hop g.kernel.side_effect=count(2) run.step() with self.assertRaises(AssertionError): self.assertEqual(counts,[1,2,4]) run.close();run.close() for allocation in allocations: allocation.free.assert_called_once_with() def test_partial_construction_frees_every_allocation(self): for options in (dict(fail_compile=True),dict(fail_upload=True)): with self.subTest(options=options),self.assertRaisesRegex(RuntimeError,'failed'): mock_backend(**options) class RemovedLegacyAPI(unittest.TestCase): """The pre-level-list constructor/initialize forms are gone and fail loudly.""" def test_removed_constructor_forms_raise_type_error(self): import inspect context = Mock() with self.assertRaises(TypeError): # positional bounds host.FixedPatchStepper(CFG, PARENTS, context=context) removed = dict(bounds=PARENTS, fine_block=(4, 4), base_block=(4, 4), fine_objects=(), jets=(), child=(8, 8, 16, 16), child_block=(4, 4), child_objects=(), child_jets=(), child_rectangles_in_parent=CHILDREN) for name, value in removed.items(): with self.subTest(keyword=name), self.assertRaises(TypeError): host.FixedPatchStepper(CFG, context=context, **levels(), **{name: value}) context.activate.assert_not_called() self.assertEqual(list(inspect.signature(host.FixedPatchStepper.initialize).parameters), ["self", "level_g"]) # no coarse_g/fine_g/child_g if __name__=='__main__': unittest.main()