Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions diffgraph/git_snapshot.py
Original file line number Diff line number Diff line change
Expand Up @@ -643,6 +643,28 @@ def _working_tree_oid(
mode: Optional[str],
warnings: List[ResolutionWarning],
) -> Optional[str]:
"""Return the exact object ID for a worktree file, link, or gitlink."""

if mode == "160000":
output = _run(
["git", "rev-parse", "--verify", "HEAD^{commit}"],
os.path.join(root, path),
warnings,
"gitlink_head_failed",
path=path,
)
if output is None:
return None
oid = os.fsdecode(output).strip()
if not _is_hex_oid(oid):
warnings.append(ResolutionWarning(
"malformed_gitlink_head",
"Git returned an invalid gitlink commit object ID",
path,
))
return None
return oid

result = _working_tree_blob(root, path, mode, warnings)
return result[1] if result is not None else None

Expand Down
50 changes: 43 additions & 7 deletions diffgraph/structural.py
Original file line number Diff line number Diff line change
Expand Up @@ -116,6 +116,21 @@ def _binary_sides(old: Optional[bytes], new: Optional[bytes]) -> Tuple[str, ...]
return tuple(sides)


def _non_regular_modes(entry: SnapshotEntry) -> Tuple[str, ...]:
"""Return Git modes whose contents must not be treated as source text.

Symlink targets and gitlinks are represented by object-like values in a
raw Git diff, but neither is a source-file snapshot. Parsing a symlink
target such as ``replacement.py`` as Python can otherwise yield a
misleading successful analysis with empty topology.
"""
return tuple(sorted({
mode
for mode in (entry.old_mode, entry.new_mode)
if mode is not None and mode not in ("100644", "100755")
}))


def _parser():
parser = getattr(_PARSER_STATE, "python_parser", None)
if parser is not None:
Expand Down Expand Up @@ -512,13 +527,20 @@ def analyze_local_diff(
path = entry.new_path or entry.old_path
if path is None:
raise RuntimeError("snapshot entry has neither an old nor a new path")
# Gitlink (mode 160000) OIDs identify commits, not blobs, so content
# reads would fail or be misleading. Skip reads for gitlink sides and
# let _non_regular_modes() report the opaque entry cleanly; continue
# reading symlink sides normally.
try:
old = _blob(root, entry.old_oid)
new = (
_blob(root, entry.new_oid)
if staged or is_commit_range
else _worktree_bytes(root, entry)
old = (
None if entry.old_mode == "160000" else _blob(root, entry.old_oid)
)
if entry.new_mode == "160000":
new = None
elif staged or is_commit_range:
new = _blob(root, entry.new_oid)
else:
new = _worktree_bytes(root, entry)
except (OSError, GitSnapshotError) as error:
old = new = None
warnings.append(_warning("PARTIAL_ANALYSIS", path, "snapshot read failed: {}".format(error)))
Expand All @@ -529,15 +551,20 @@ def analyze_local_diff(
new is None and entry.new_oid is not None
)
binary_sides = () if snapshot_missing else _binary_sides(old, new)
non_regular_modes = _non_regular_modes(entry)
Comment thread
coderabbitai[bot] marked this conversation as resolved.
lines_added, lines_removed = (
(None, None) if snapshot_missing or binary_sides else _line_counts(old, new)
(None, None)
if snapshot_missing or binary_sides or non_regular_modes
else _line_counts(old, new)
)
file_entry = {
"id": "file::" + path, "path": path,
"old_path": entry.old_path if entry.status in ("R", "C") else None,
"language": (
"python"
if not binary_sides and Path(path).suffix.lower() == ".py"
if not binary_sides
and not non_regular_modes
and Path(path).suffix.lower() == ".py"
else None
),
"change_kind": _change_kind(entry.status, entry.old_oid, entry.new_oid),
Expand All @@ -557,6 +584,15 @@ def analyze_local_diff(
),
))
continue
if non_regular_modes:
skipped += 1
warnings.append(_warning(
"PARTIAL_ANALYSIS",
path,
"Non-regular Git mode {} detected; structural parsing and line "
"counts were skipped.".format(", ".join(non_regular_modes)),
))
continue
if file_entry["language"] != "python":
skipped += 1
warnings.append(_warning("UNSUPPORTED_LANGUAGE", path, "Deterministic extraction currently supports Python (.py) only."))
Expand Down
134 changes: 127 additions & 7 deletions tests/test_structural.py
Original file line number Diff line number Diff line change
Expand Up @@ -214,6 +214,7 @@ def test_binary_python_snapshot_preserves_identity_without_parsing(tmp_path, sta
],
ids=["binary-to-text", "binary-to-binary"],
)

def test_pre_change_binary_snapshots_skip_all_source_analysis(
tmp_path, old_content, new_content, binary_sides
):
Expand Down Expand Up @@ -469,6 +470,119 @@ def test_cli_terminal_all_disables_review_item_cap(tmp_path, monkeypatch):
assert "item_10" in result.output
assert "more" not in result.output

def test_symlink_type_change_is_an_opaque_snapshot_not_python_source(tmp_path):
"""A symlink target ending in .py must not create false topology."""
root = repo(tmp_path)
write(root, "module.py", "def previous():\n return 1\n")
commit(root)

(root / "module.py").unlink()
os.symlink("replacement.py", root / "module.py")
git(root, "add", "-A")

artifact = analyze_local_diff(str(root), staged=True)

assert_valid(artifact)
file_entry = artifact["files"][0]
provenance = json.loads(file_entry["evidence"][0]["detail"])
assert provenance["old_mode"] == "100644"
assert provenance["new_mode"] == "120000"
assert file_entry["language"] is None
assert file_entry["lines_added"] is None
assert file_entry["lines_removed"] is None
assert artifact["symbols"] == []
assert artifact["relationships"] == []
assert artifact["metadata"]["files_skipped"] == 1
assert artifact["metadata"]["warnings"] == [{
"code": "PARTIAL_ANALYSIS",
"file": "module.py",
"detail": "Non-regular Git mode 120000 detected; structural parsing and line counts were skipped.",
}]


def assert_opaque_gitlink(artifact, expected_old_oid, expected_new_oid):
"""Assert that a gitlink keeps commit provenance without source claims."""
assert_valid(artifact)
file_entry = next(item for item in artifact["files"] if item["path"] == "vendor/tool")
provenance = json.loads(file_entry["evidence"][0]["detail"])
assert provenance["old_mode"] in (None, "160000")
assert provenance["new_mode"] in (None, "160000")
assert provenance["old_oid"] == expected_old_oid
assert provenance["new_oid"] == expected_new_oid
assert file_entry["language"] is None
assert file_entry["lines_added"] is None
assert file_entry["lines_removed"] is None
assert artifact["symbols"] == []
assert artifact["relationships"] == []
assert artifact["metadata"]["files_skipped"] == 1
assert artifact["metadata"]["warnings"] == [{
"code": "PARTIAL_ANALYSIS",
"file": "vendor/tool",
"detail": "Non-regular Git mode 160000 detected; structural parsing and line counts were skipped.",
}]


def test_staged_gitlink_is_an_opaque_commit_snapshot(tmp_path):
"""A staged gitlink addition must preserve its commit ID without a blob read."""
root = repo(tmp_path)
write(root, "tracked.txt", "baseline\n")
commit(root)
gitlink_oid = git(root, "rev-parse", "HEAD")
git(root, "update-index", "--add", "--cacheinfo", "160000,{},vendor/tool".format(gitlink_oid))

artifact = analyze_local_diff(str(root), staged=True)

assert_opaque_gitlink(artifact, None, gitlink_oid)


def test_unstaged_gitlink_uses_checked_out_submodule_commit(tmp_path):
"""An unstaged gitlink change must resolve the checked-out submodule HEAD."""
child = tmp_path / "child"
child.mkdir()
git(child, "init")
git(child, "config", "user.name", "Structural Tests")
git(child, "config", "user.email", "structural@example.test")
write(child, "value.txt", "one\n")
commit(child)

root = repo(tmp_path)
git(root, "-c", "protocol.file.allow=always", "submodule", "add", str(child), "vendor/tool")
commit(root)
old_oid = git(root / "vendor/tool", "rev-parse", "HEAD")
git(root / "vendor/tool", "config", "user.name", "Structural Tests")
git(root / "vendor/tool", "config", "user.email", "structural@example.test")
write(root / "vendor/tool", "value.txt", "two\n")
commit(root / "vendor/tool")
new_oid = git(root / "vendor/tool", "rev-parse", "HEAD")

artifact = analyze_local_diff(str(root))

assert_opaque_gitlink(artifact, old_oid, new_oid)


def test_commit_range_gitlink_is_an_opaque_commit_snapshot(tmp_path):
"""A commit-range gitlink change must preserve both commit IDs without reads."""
root = repo(tmp_path)
write(root, "tracked.txt", "first\n")
commit(root)
first_oid = git(root, "rev-parse", "HEAD")
write(root, "tracked.txt", "second\n")
commit(root)
second_oid = git(root, "rev-parse", "HEAD")

git(root, "update-index", "--add", "--cacheinfo", "160000,{},vendor/tool".format(first_oid))
git(root, "commit", "-m", "add gitlink")
base_oid = git(root, "rev-parse", "HEAD")
git(root, "update-index", "--cacheinfo", "160000,{},vendor/tool".format(second_oid))
git(root, "commit", "-m", "update gitlink")
head_oid = git(root, "rev-parse", "HEAD")

artifact = analyze_local_diff(
str(root), base_ref=base_oid, head_ref=head_oid
)

assert_opaque_gitlink(artifact, first_oid, second_oid)


@pytest.mark.parametrize("pathspec", ["--compact", "--all"])
def test_cli_terminal_preserves_flag_like_pathspec_after_separator(
Expand Down Expand Up @@ -815,7 +929,7 @@ def test_aliased_import_uses_name_field_and_reports_alias_edit_as_modified(tmp_p
assert imported["change_kind"] == "modified"


def test_worktree_symlink_uses_exact_link_bytes_without_partial_warning(tmp_path):
def test_worktree_symlink_preserves_exact_link_bytes_without_source_analysis(tmp_path):
root = repo(tmp_path)
os.symlink("original.py", root / "link.py")
commit(root)
Expand All @@ -824,12 +938,18 @@ def test_worktree_symlink_uses_exact_link_bytes_without_partial_warning(tmp_path

artifact = analyze_local_diff(str(root))
assert_valid(artifact)
assert artifact["metadata"]["files_analyzed"] == 1
assert not any(
warning["code"] == "PARTIAL_ANALYSIS"
for warning in artifact["metadata"]["warnings"]
)
provenance = json.loads(artifact["files"][0]["evidence"][0]["detail"])
assert artifact["metadata"]["files_analyzed"] == 0
assert artifact["metadata"]["files_skipped"] == 1
file_entry = artifact["files"][0]
assert file_entry["language"] is None
assert file_entry["lines_added"] is None
assert file_entry["lines_removed"] is None
assert artifact["metadata"]["warnings"] == [{
"code": "PARTIAL_ANALYSIS",
"file": "link.py",
"detail": "Non-regular Git mode 120000 detected; structural parsing and line counts were skipped.",
}]
provenance = json.loads(file_entry["evidence"][0]["detail"])
assert provenance["new_oid"] == git(
root, "hash-object", "--stdin", input_bytes=b"replacement.py"
)
Expand Down
Loading