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
95 changes: 95 additions & 0 deletions minigit/objects.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,18 @@
import hashlib
import zlib
from pathlib import Path
from typing import NamedTuple

from minigit.errors import ObjectCorruptError, ObjectNotFoundError


class TreeEntry(NamedTuple):
mode: str
type: str
hash: str
name: str


class ObjectStore:
root: Path
objects_dir: Path
Expand Down Expand Up @@ -83,6 +91,93 @@ def _object_path(self, hash: str) -> Path:
"""
return self.objects_dir / hash[:2] / hash[2:]

@staticmethod
def _validate_tree_entry(entry: TreeEntry) -> None:
"""
Helper Method
Validates a TreeEntry object for correct mode/type, hash, and name.
"""
if (entry.mode, entry.type) not in {
("100644", "blob"),
("100755", "blob"),
("40000", "tree"),
}:
raise ValueError
if len(entry.hash) != 40 or any(
character not in "0123456789abcdef" for character in entry.hash
):
raise ValueError
if entry.name in {"", ".", ".."} or any(
character in entry.name for character in "/\0\t\r\n"
):
raise ValueError

def write_tree(self, entries: list[TreeEntry]) -> str:
"""
Writes a list of TreeEntry objects to the object store in name-sorted order.
Returns the hash of the tree object.
Raises ValueError for incorrectly formatted entries.
"""

sorted_entries = sorted(entries, key=lambda entry: entry.name)

names = set()
for entry in sorted_entries:
self._validate_tree_entry(entry)
if entry.name in names:
raise ValueError
names.add(entry.name)

tree_data = b"".join(
f"{entry.mode} {entry.type} {entry.hash}\t{entry.name}\n".encode()
for entry in sorted_entries
)
return self.write_object(tree_data, "tree")

def read_tree(self, tree_hash: str) -> list[TreeEntry]:
"""
Reads a tree object from the object store and returns a list of TreeEntry objects.
Raises ObjectCorruptError if the object is not a tree or is incorrectly formatted.
"""

obj_type, tree_data = self.read_object(tree_hash)

if obj_type != "tree":
raise ObjectCorruptError(tree_hash)

entries = []

try:
text = tree_data.decode("utf-8")
if text and not text.endswith("\n"):
raise ValueError
lines = text.split("\n")[:-1]

names = set()

for line in lines:
if line.endswith("\r"):
raise ValueError

fields, separator, name = line.partition("\t")
if not separator:
raise ValueError

mode, entry_type, entry_hash = fields.split(" ")
entry = TreeEntry(mode, entry_type, entry_hash, name)
self._validate_tree_entry(entry)

if name in names:
raise ValueError

names.add(name)
entries.append(entry)

except (UnicodeDecodeError, ValueError):
raise ObjectCorruptError(tree_hash) from None

return entries


def run_hash_object(args) -> int:
"""
Expand Down
136 changes: 135 additions & 1 deletion tests/test_objects.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@

from minigit.cli import main
from minigit.errors import ObjectCorruptError, ObjectNotFoundError
from minigit.objects import ObjectStore
from minigit.objects import ObjectStore, TreeEntry


def test_round_trip(tmp_path: Path) -> None:
Expand Down Expand Up @@ -149,3 +149,137 @@ def test_duplicate_write_leaves_one_object_file(tmp_path):
store._object_path(obj_hash).parent,
store._object_path(obj_hash),
]


def test_writing_empty_tree(tmp_path):
"""
Tests that writing an empty tree returns a valid hash and can be read back.
"""
store = ObjectStore(tmp_path)

tree_hash = store.write_tree([])

assert store.read_object(tree_hash) == ("tree", b"")
assert store.read_tree(tree_hash) == []


def test_tree_round_trip_supports_nested_references_executable_and_spaces(tmp_path):
"""
Tests that writing a tree with nested references,
executable files, and spaces in names can be read back correctly
"""
store = ObjectStore(tmp_path)
blob_hash = "a" * 40
nested_hash = store.write_tree([TreeEntry("100644", "blob", blob_hash, "child")])
entries = [
TreeEntry("100755", "blob", blob_hash, "run script"),
TreeEntry("40000", "tree", nested_hash, "nested"),
]

tree_hash = store.write_tree(entries)

assert store.read_tree(tree_hash) == [entries[1], entries[0]]


def test_tree_hash_is_deterministic_across_input_order(tmp_path):
"""
Tests that writing a tree with the same entries in different orders produces the same hash.
"""
store = ObjectStore(tmp_path)
entries = [
TreeEntry("100644", "blob", "a" * 40, "z.txt"),
TreeEntry("100644", "blob", "b" * 40, "a.txt"),
]

assert store.write_tree(entries) == store.write_tree(list(reversed(entries)))


@pytest.mark.parametrize(
"entry",
[
TreeEntry("100600", "blob", "a" * 40, "file"),
TreeEntry("100644", "commit", "a" * 40, "file"),
TreeEntry("100644", "blob", "a" * 39, "file"),
TreeEntry("100644", "blob", "g" * 40, "file"),
TreeEntry("100644", "blob", "a" * 40, ""),
TreeEntry("100644", "blob", "a" * 40, "dir/file"),
TreeEntry("100644", "blob", "a" * 40, "."),
TreeEntry("100644", "blob", "a" * 40, ".."),
TreeEntry("100644", "blob", "a" * 40, "has\t tab"),
TreeEntry("100644", "blob", "a" * 40, "has\nnewline"),
],
)
def test_write_tree_rejects_malformed_entries(tmp_path, entry):
"""
Tests that writing a tree with incorrectly formatted entries raises a ValueError.
"""
with pytest.raises(ValueError):
ObjectStore(tmp_path).write_tree([entry])


@pytest.mark.parametrize(
"tree_data",
[
b"100644 blob aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa no-tab\n",
b"100644 blob aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\t\n",
b"100644 blob aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\tdup\n"
b"100644 blob bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb\tdup\n",
b"100644 blob aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\tfile\n",
b"100644 blob aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\tfile",
b"100644 blob aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\tfile\r\n",
b"\xff",
],
)
def test_read_tree_rejects_malformed_data(tmp_path, tree_data):
"""
Tests that reading a tree with incorrectly formatted data raises an ObjectCorruptError
"""
store = ObjectStore(tmp_path)
tree_hash = store.write_object(tree_data, "tree")

with pytest.raises(ObjectCorruptError):
store.read_tree(tree_hash)


def test_read_tree_rejects_non_tree_object(tmp_path):
"""
Tests that reading a tree from a non-tree object raises an ObjectCorruptError
"""
store = ObjectStore(tmp_path)
blob_hash = store.write_object(b"content", "blob")

with pytest.raises(ObjectCorruptError):
store.read_tree(blob_hash)


def test_read_tree_missing_object(tmp_path):
"""
Tests that reading a non-existent tree raises an ObjectNotFoundError.
"""
with pytest.raises(ObjectNotFoundError):
ObjectStore(tmp_path).read_tree("a" * 40)


def test_tree_round_trip_with_new_store(tmp_path):
"""
Tests that writing a tree and reading it back with a new ObjectStore instance works correctly.
"""
entries = [TreeEntry("100644", "blob", "a" * 40, "file")]
first_store = ObjectStore(tmp_path)
tree_hash = first_store.write_tree(entries)

second_store = ObjectStore(tmp_path)

assert second_store.read_tree(tree_hash) == entries


@pytest.mark.parametrize("name", ["vertical\vtab", "form\ffeed", "unicode\u2028separator"])
def test_tree_round_trip_preserves_non_delimiter_characters(tmp_path, name):
store = ObjectStore(tmp_path)
entries = [TreeEntry("100644", "blob", "a" * 40, name)]
assert store.read_tree(store.write_tree(entries)) == entries


def test_tree_rejects_null_in_filename(tmp_path):
with pytest.raises(ValueError):
ObjectStore(tmp_path).write_tree([TreeEntry("100644", "blob", "a" * 40, "bad\0name")])
Loading