tests/test_incremental_mtime_collision.py
Rama v8 · réplica de lectura del repositorio original.
"""An incremental run must not skip a file whose mtime did not move.
``detect_incremental`` treats "mtime unchanged" as proof the content is
unchanged. That holds only while the filesystem can separate the two writes: an
edit keeping the file the same length and landing inside one timestamp tick
moves neither size nor mtime, so the file is classified unchanged and never
re-extracted, while the graph keeps serving the old content.
The stat-only fastpath is the point of the gate and must survive: a settled
corpus with a manifest from an earlier run has to cost zero content hashes.
"""
import os
import tempfile
import time
from pathlib import Path
import pytest
from graphify import detect as det
@pytest.fixture()
def corpus(tmp_path):
src = tmp_path / "src"
src.mkdir()
for i in range(3):
(src / f"f{i}.md").write_text(f"# Doc {i}\n\nBody {i}.\n", encoding="utf-8")
# manifest lives outside the scanned tree so it is never part of the corpus
return src, str(tmp_path / "manifest.json")
def _queued(result):
return [f for flist in result["new_files"].values() for f in flist]
def test_same_size_rewrite_in_one_tick_is_requeued(corpus):
"""The bug: a rewrite the filesystem cannot distinguish by stat alone.
The mtime is pinned back to the manifest's recorded value so the collision
is reproduced deterministically rather than depending on how fast the
manifest write happened to be.
"""
import json
src, manifest = corpus
det.save_manifest(det.detect(src)["files"], manifest, root=src, kind="semantic")
# The gate only distrusts an mtime stamped within a tick of the row's `seen`.
# How long save_manifest took is machine-dependent, so stamp the same tick
# explicitly instead of relying on a fast disk.
rows = json.loads(Path(manifest).read_text(encoding="utf-8"))
for row in rows.values():
row["seen"] = row["mtime"]
Path(manifest).write_text(json.dumps(rows), encoding="utf-8")
target = src / "f1.md"
stat_before = target.stat()
target.write_text("# Doc X\n\nBody Y.\n", encoding="utf-8")
# same length, and stat now reports exactly what the manifest recorded
os.utime(target, ns=(stat_before.st_atime_ns, stat_before.st_mtime_ns))
assert target.stat().st_size == stat_before.st_size
assert target.stat().st_mtime_ns == stat_before.st_mtime_ns
queued = _queued(det.detect_incremental(src, manifest, kind="semantic"))
assert len(queued) == 1, "a rewritten file must be re-queued"
assert queued[0].endswith("f1.md")
def test_a_row_stamped_just_before_the_mtime_still_hashes(corpus):
"""`seen` from time.time() can land a few hundred ns BEFORE the mtime (#5036).
Both clocks describe the same write, so the old `seen - mtime < 0 -> trust
stat` shortcut skipped the hash on a row that could still hide a rewrite.
The delta is only meaningful as a magnitude: anything inside the tick is
unprovable in either direction.
"""
import json
src, manifest = corpus
det.save_manifest(det.detect(src)["files"], manifest, root=src, kind="semantic")
rows = json.loads(Path(manifest).read_text(encoding="utf-8"))
for row in rows.values():
row["seen"] = row["mtime"] - 0.000001 # ~1 us before the stamp
Path(manifest).write_text(json.dumps(rows), encoding="utf-8")
target = src / "f1.md"
stat_before = target.stat()
target.write_text("# Doc 2\n\nA substantially longer body than before.\n", encoding="utf-8")
os.utime(target, ns=(stat_before.st_atime_ns, stat_before.st_mtime_ns))
assert target.stat().st_mtime_ns == stat_before.st_mtime_ns
queued = _queued(det.detect_incremental(src, manifest, kind="semantic"))
assert len(queued) == 1 and queued[0].endswith("f1.md")
def test_a_row_stamped_well_before_the_mtime_keeps_the_fastpath(corpus, monkeypatch):
"""Control for the fix: a window-old row is from an earlier run, not a tick."""
import json
src, manifest = corpus
det.save_manifest(det.detect(src)["files"], manifest, root=src, kind="semantic")
rows = json.loads(Path(manifest).read_text(encoding="utf-8"))
for row in rows.values():
row["seen"] = row["mtime"] - 3600 # an hour before; a later write moved mtime
Path(manifest).write_text(json.dumps(rows), encoding="utf-8")
calls = {"n": 0}
real = det._md5_file
def counting(p):
calls["n"] += 1
return real(p)
monkeypatch.setattr(det, "_md5_file", counting)
result = det.detect_incremental(src, manifest, kind="semantic")
assert _queued(result) == []
assert calls["n"] == 0, "a row from an earlier run must not be re-hashed"
def test_a_settled_corpus_costs_no_content_hashes(corpus, monkeypatch):
"""The fastpath must survive: an untouched corpus does zero MD5 work."""
src, manifest = corpus
det.save_manifest(det.detect(src)["files"], manifest, root=src, kind="semantic")
old = time.time() - 3600
for f in src.glob("*.md"):
os.utime(f, (old, old))
det.save_manifest(det.detect(src)["files"], manifest, root=src, kind="semantic")
calls = {"n": 0}
real = det._md5_file
def counting(p):
calls["n"] += 1
return real(p)
monkeypatch.setattr(det, "_md5_file", counting)
result = det.detect_incremental(src, manifest, kind="semantic")
assert calls["n"] == 0, "an unchanged corpus must not be re-hashed"
assert _queued(result) == []
def test_a_genuinely_edited_file_is_still_requeued(corpus):
"""Control: the ordinary size-change path is untouched."""
src, manifest = corpus
det.save_manifest(det.detect(src)["files"], manifest, root=src, kind="semantic")
(src / "f2.md").write_text(
"# Doc 2\n\nA substantially longer body than before.\n", encoding="utf-8")
queued = _queued(det.detect_incremental(src, manifest, kind="semantic"))
assert len(queued) == 1 and queued[0].endswith("f2.md")
def test_an_untouched_file_is_not_requeued_after_a_neighbour_changes(corpus):
"""Only the edited file moves; its neighbours keep the fastpath."""
src, manifest = corpus
det.save_manifest(det.detect(src)["files"], manifest, root=src, kind="semantic")
old = time.time() - 3600
for f in src.glob("*.md"):
os.utime(f, (old, old))
det.save_manifest(det.detect(src)["files"], manifest, root=src, kind="semantic")
(src / "f0.md").write_text("# Doc 0\n\nA different and longer body.\n",
encoding="utf-8")
queued = _queued(det.detect_incremental(src, manifest, kind="semantic"))
assert len(queued) == 1 and queued[0].endswith("f0.md")
def test_a_legacy_manifest_row_without_seen_still_works(corpus):
"""Rows predating the `seen` field come from an earlier run: trusted."""
import json
src, manifest = corpus
det.save_manifest(det.detect(src)["files"], manifest, root=src, kind="semantic")
data = json.loads(Path(manifest).read_text(encoding="utf-8"))
for row in data.values():
if isinstance(row, dict):
row.pop("seen", None)
Path(manifest).write_text(json.dumps(data), encoding="utf-8")
assert _queued(det.detect_incremental(src, manifest, kind="semantic")) == []