test: make recovery fixtures native-page-size aware

tests/unit/test_kvcr.py

@@ -170,8 +170,9 @@ def test_a_guarded_startup_that_fails_gives_back_everything_it_took(
 ) -> None:
     """Refused before the claim, or unwound after it: core closed, pool returned."""
     events: list[str] = []
+    block_size = mmap.PAGESIZE // 2 if stage == "g3-invalid" else 1024
     hold = _fake_hold(
-        local_dram=LocalDramOptions([("", 1234, 8192)]),
+        local_dram=LocalDramOptions([("", 1234, 2 * block_size)]),
         _attachment=_UNSERVED_POOL,
         _control_listener_fd=None,
         release=lambda **_kwargs: events.append("hold.release"),
@@ -237,9 +238,7 @@ def is_quiescent(self) -> bool:
         KVCR(
             KVCRConfig(
                 nixl_agent_name="target",
-                pool_layouts=[
-                    ("", mmap.PAGESIZE // 2 if stage == "g3-invalid" else 1024)
-                ],
+                pool_layouts=[("", block_size)],
                 nixl_listen_port=1,
             ),
             KVCRBindings(Mock(), Mock(), Mock(), framework_control=control),

tests/unit/test_recovery_journal.py

@@ -259,11 +259,13 @@ def test_invalidation_stops_every_role_until_an_owner_reset(
 
 
 @contextmanager
-def _attached(tmp_path: Path) -> Iterator[KVCRPoolAttachment]:
+def _attached(
+    tmp_path: Path, pool_size_bytes: int = 8192 + 4096
+) -> Iterator[KVCRPoolAttachment]:
     """An owned pool, mapped, the way a Guard or a claimant holds one."""
     owner = _KVCRPoolOwner.allocate(
         pool_id="pool_0",
-        pool_size_bytes=8192 + 4096,
+        pool_size_bytes=pool_size_bytes,
         journal_bytes=8192,
         pool_dir=tmp_path,
     )
@@ -285,12 +287,17 @@ def _write_slot(pool: KVCRPoolAttachment, terms: bytes, key: bytes, slot: int) -
     write_recovery_snapshot(pool, terms, frames)
 
 
+@pytest.mark.parametrize(
+    "pool_size_bytes",
+    [8192 + 4096, 4 * mmap.ALLOCATIONGRANULARITY, 4 * mmap.ALLOCATIONGRANULARITY + 1],
+    ids=["small-pool", "page-aligned-pool", "unaligned-pool"],
+)
 def test_a_handback_region_lives_and_dies_inside_the_pool_file(
-    tmp_path: Path, caplog
+    tmp_path: Path, caplog, pool_size_bytes: int
 ) -> None:
     """Replayed whole under its own terms, discardable when torn, gone once released."""
     caplog.set_level(logging.DEBUG, logger="kvcr.recovery_journal")
-    with _attached(tmp_path) as pool:
+    with _attached(tmp_path, pool_size_bytes) as pool:
         path = Path(pool._spec.path)
         pools = (
             _PoolDescriptor("pool0", 2048, 1024, pool._spec.journal_bytes),
@@ -307,10 +314,10 @@ def test_a_handback_region_lives_and_dies_inside_the_pool_file(
         }
         write_recovery_snapshot(pool, terms, _recovery_frames(records))
         written = [r for r in caplog.records if "snapshot_written " in r.message]
-        assert written[-1].args == (
-            pool._spec.pool_id,
-            path.stat().st_size - pool._spec.mapping_bytes,
-        )
+        with pool.mapped_snapshot() as snapshot:
+            assert snapshot is not None
+            # The snapshot's byte count excludes alignment padding after the pool.
+            assert written[-1].args == (pool._spec.pool_id, len(snapshot))
         # Inside the pool file, so it has no name of its own to be found under.
         assert set(tmp_path.iterdir()) == {path}
         assert path.stat().st_size > pool._spec.mapping_bytes