test(annotations): assert the rewrite outcome, not the platform

Native Windows CI detected the same-size, mtime-restored rewrite on
3.12 and missed it on 3.13 in the same run: the path stat's change time
(Python 3.12+ fast path) only moves when the clock ticks between the
pre-stat and the rewrite, and the tick wait was skipped on Windows. The
wait now runs everywhere with fresh probe names (Windows tunnels the
creation time of a reused name), and the test asserts the outcome: a
detected change leaves no temp file; an undetected one is accepted on
Windows only and must produce a digest of the copied bytes; POSIX must
detect it. The injected-fstat test keeps the detection path covered on
every platform.
This commit is contained in:
Vito Sansevero
2026-10-22 06:13:57 -07:00
parent 7a47f936da
commit 21ba73ff39
+22 -12
View File
@@ -4,7 +4,6 @@ import hashlib
import os
import stat
import struct
import sys
import tempfile
import threading
import time
@@ -101,13 +100,23 @@ class CaptureStoreTestCase(unittest.TestCase):
self.skipTest(f"symlinks unavailable here: {error}")
def wait_for_change_time_after(self, ns: int) -> None:
"""Block until the filesystem clock passes *ns* (coarse ticks on Linux)."""
probe = self.parent / "tick"
"""Block until a new file's ``st_ctime_ns`` passes *ns*.
File timestamps tick coarsely (Linux jiffies, Windows ~1-16 ms). Each
probe is a fresh name: Windows reports the creation time (or, on
Python 3.12+, the change time) and tunnels the creation time of a
reused name for 15 s.
"""
deadline = time.monotonic() + WAIT
while time.monotonic() < deadline:
for index in range(1_000_000):
probe = self.parent / f"tick-{index}"
probe.write_bytes(b"x")
if os.stat(probe).st_ctime_ns > ns:
advanced = os.stat(probe).st_ctime_ns > ns
probe.unlink()
if advanced:
return
if time.monotonic() >= deadline:
break
time.sleep(0.001)
self.fail("the filesystem clock did not advance") # pragma: no cover
@@ -280,8 +289,7 @@ class CaptureStoreTestCase(unittest.TestCase):
# so there the torn copy completes and its digest describes the copy.
source.write_bytes(source_bytes)
before = os.stat(source)
if os.name == "posix": # Windows st_ctime is the creation time
self.wait_for_change_time_after(before.st_ctime_ns)
self.wait_for_change_time_after(before.st_ctime_ns)
rewritten = bytes(b ^ 0xFF for b in source_bytes)
def rewriting(store_, stream, size):
@@ -293,9 +301,12 @@ class CaptureStoreTestCase(unittest.TestCase):
os.utime(source, ns=(before.st_atime_ns, before.st_mtime_ns))
return data
# Detection depends on whether fstat exposes a change time: POSIX
# always, Windows only on Python 3.13+. Where it is missing the torn
# copy completes and its digest must describe the copied bytes.
# Only a change time can tell: POSIX always has one; Windows exposes
# it from stat on Python 3.12+ (where the OS supports the fast path)
# and never from fstat, so detection there is best effort. The test
# asserts the outcome, not the platform: a detected change leaves no
# temp file; an undetected one (Windows only) completes with a digest
# that describes the copied bytes.
with patch.object(CaptureStore, "_read", rewriting):
try:
cap = store.capture(self.root, "a.png", epoch=EPOCH, scope=SCOPE)
@@ -303,13 +314,12 @@ class CaptureStoreTestCase(unittest.TestCase):
detected = True
else:
detected = False
self.assertEqual(os.name, "nt", "only Windows may miss the rewrite")
copied = Path(cap.path).read_bytes()
self.assertEqual(cap.sha256, hashlib.sha256(copied).hexdigest())
store.release(cap.token)
if os.name == "posix":
self.assertTrue(detected)
else:
self.assertEqual(detected, sys.version_info >= (3, 13))
self.assert_empty(store)
self.assertEqual(os.stat(source).st_mtime_ns, before.st_mtime_ns)