Skip to content
Open
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
110 changes: 101 additions & 9 deletions minigit/index.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -9,9 +9,12 @@
Build the `WorkingTree` class here, per the interface contract.
"""

import os
from dataclasses import dataclass
from typing import NamedTuple

from minigit.errors import MiniGitError


class IndexEntry(NamedTuple):
mode: str
Expand All@@ -37,31 +40,90 @@ def __init__(self, repo_path=".", store=None):
store = ObjectStore(repo_path)

self.store = store
self._entries: list[IndexEntry] = []

def read_index(self) -> list[IndexEntry]:
return sorted(self._entries, key=lambda e: e.path)
if not os.path.exists(self.index_path):
return []

entries = []
with open(self.index_path, encoding="utf-8") as f:
for line in f:
line = line.rstrip("\n")
if not line:
continue
mode, hash_, path = line.split(" ", 2)
entries.append(IndexEntry(mode, hash_, path))

return sorted(entries, key=lambda e: e.path)

def write_index(self, entries) -> None:
self._entries = sorted(entries, key=lambda e: e.path)
entries = sorted(entries, key=lambda e: e.path)
lines = [f"{e.mode} {e.hash} {e.path}" for e in entries]
content = "\n".join(lines)
if content:
content += "\n"

os.makedirs(os.path.dirname(self.index_path), exist_ok=True)
with open(self.index_path, "w", encoding="utf-8") as f:
f.write(content)

def stage_file(self, path) -> None:
with open(f"{self.root}/{path}", "rb") as f:
full_path = os.path.join(self.root, path)

if not os.path.isfile(full_path):
raise MiniGitError(f"No such file: {path}")

rel_path = os.path.relpath(full_path, self.root)
rel_path = rel_path.replace(os.sep, "/")

with open(full_path, "rb") as f:
data = f.read()

blob_hash = self.store.write_object(data, "blob")

self._entries = [e for e in self._entries if e.path != path]
if os.access(full_path, os.X_OK):
mode = "100755"
else:
mode = "100644"

self._entries.append(IndexEntry("100644", blob_hash, path))
self._entries.sort(key=lambda e: e.path)
entries = [e for e in self.read_index() if e.path != path]
entries.append(IndexEntry(mode, blob_hash, path))
self.write_index(entries)

def build_tree_from_index(self) -> str:
return self.store.write_object(b"", "tree")

def diff_working_tree_vs(self, tree_hash) -> DiffResult:
return DiffResult([], [], [])

def _working_status(self) -> DiffResult:
result = DiffResult([], [], [])
entries = self.read_index()
known_paths = {e.path for e in entries}

# check staged entries against disk
for entry in entries:
full_path = f"{self.root}/{entry.path}"
if not os.path.isfile(full_path):
result.deleted.append(entry.path)
else:
with open(full_path, "rb") as f:
data = f.read()
current_hash = self.store.write_object(data, "blob")
if current_hash != entry.hash:
result.modified.append(entry.path)

# walk the working tree for untracked files
for dirpath, dirnames, filenames in os.walk(self.root):
dirnames[:] = [d for d in dirnames if d != ".minigit"]
for filename in filenames:
full_path = os.path.join(dirpath, filename)
rel_path = os.path.relpath(full_path, self.root)
if rel_path not in known_paths:
result.added.append(rel_path)

return result

def checkout(self, tree_hash) -> None:
pass

Expand All@@ -74,8 +136,38 @@ def cmd_add(args) -> int:

def cmd_status(args) -> int:
wt = WorkingTree()
for entry in wt.read_index():
print(entry.path)
result = wt._working_status()
entries = wt.read_index()

# build each category once, upfront
changed_paths = result.modified + result.deleted
staged = [e.path for e in entries if e.path not in changed_paths]
not_staged = sorted(changed_paths)
untracked = sorted(result.added)

printed_anything = False

if staged:
print("staged:")
for path in staged:
print(f" {path}")
printed_anything = True

if not_staged:
print("not staged:")
for path in not_staged:
print(f" {path}")
printed_anything = True

if untracked:
print("untracked:")
for path in untracked:
print(f" {path}")
printed_anything = True

if not printed_anything:
print("clean")

return 0


Expand Down
103 changes: 71 additions & 32 deletions tests/test_index.py
Original file line numberDiff line numberDiff line change
@@ -1,68 +1,107 @@
"""Tests for minigit/index.py - WorkingTree, IndexEntry, DiffResult."""

from minigit.index import WorkingTree
from minigit.objects import ObjectStore


class FakeObjectStore:
"""Stand-in for the real ObjectStore - keeps everything in memory."""
def test_index_works_new_workingtree(tmp_path):
(tmp_path / "hello.txt").write_text("hi")
store = ObjectStore(str(tmp_path))

def __init__(self):
self.objects = {}
wt1 = WorkingTree(repo_path=str(tmp_path), store=store)
wt1.stage_file("hello.txt")

def write_object(self, data, obj_type):
fake_hash = f"hash{len(data)}"
self.objects[fake_hash] = data
return fake_hash
# a brand new WorkingTree, simulating a separate CLI invocation
wt2 = WorkingTree(repo_path=str(tmp_path), store=store)
entries = wt2.read_index()


def test_staging_adds_one_entry(tmp_path):
file = tmp_path / "hello.txt"
file.write_text("hi")

wt = WorkingTree(repo_path=str(tmp_path), store=FakeObjectStore())
wt.stage_file("hello.txt")

entries = wt.read_index()
assert len(entries) == 1
assert entries[0].path == "hello.txt"


def test_staging_twice_replaces_not_duplicates(tmp_path):
file = tmp_path / "hello.txt"
file.write_text("hi")
def test_staging_twice(tmp_path):
(tmp_path / "hello.txt").write_text("hi")
store = ObjectStore(str(tmp_path))
wt = WorkingTree(repo_path=str(tmp_path), store=store)

wt = WorkingTree(repo_path=str(tmp_path), store=FakeObjectStore())
wt.stage_file("hello.txt")
wt.stage_file("hello.txt")

entries = wt.read_index()
assert len(entries) == 1


def test_read_index_sorted(tmp_path):
def test_index_file(tmp_path):
(tmp_path / "z.txt").write_text("z")
(tmp_path / "a.txt").write_text("a")
store = ObjectStore(str(tmp_path))
wt = WorkingTree(repo_path=str(tmp_path), store=store)

wt = WorkingTree(repo_path=str(tmp_path), store=FakeObjectStore())
wt.stage_file("z.txt")

index_path = tmp_path / ".minigit" / "index"
lines = index_path.read_text(encoding="utf-8").splitlines()
assert len(lines) == 1

(tmp_path / "a.txt").write_text("a")
wt.stage_file("a.txt")

lines = index_path.read_text(encoding="utf-8").splitlines()
assert len(lines) == 2
assert lines[0].endswith("a.txt")
assert lines[1].endswith("z.txt")


def test_path_with_space(tmp_path):
(tmp_path / "my notes.txt").write_text("hello")
store = ObjectStore(str(tmp_path))
wt = WorkingTree(repo_path=str(tmp_path), store=store)

wt.stage_file("my notes.txt")

entries = wt.read_index()
assert entries[0].path == "a.txt"
assert entries[1].path == "z.txt"
assert len(entries) == 1
assert entries[0].path == "my notes.txt"


def test_editing_after_staging_does_not_change_entry(tmp_path):
def test_edit_after_staging(tmp_path):
file = tmp_path / "hello.txt"
file.write_text("original content")
file.write_text("original")
store = ObjectStore(str(tmp_path))
wt = WorkingTree(repo_path=str(tmp_path), store=store)

wt = WorkingTree(repo_path=str(tmp_path), store=FakeObjectStore())
wt.stage_file("hello.txt")

hash_before = wt.read_index()[0].hash

file.write_text("changed content!!!")
file.write_text("changed!!!")

hash_after = wt.read_index()[0].hash
assert hash_before == hash_after # snapshot rule still holds

status = wt._working_status()
assert "hello.txt" in status.modified


def test_untracked(tmp_path):
(tmp_path / "hello.txt").write_text("hi")
(tmp_path / "extra.txt").write_text("not staged at all")
store = ObjectStore(str(tmp_path))
wt = WorkingTree(repo_path=str(tmp_path), store=store)

wt.stage_file("hello.txt")

status = wt._working_status()
assert "extra.txt" in status.added
assert "hello.txt" not in status.added


def test_deleted(tmp_path):
file = tmp_path / "hello.txt"
file.write_text("hi")
store = ObjectStore(str(tmp_path))
wt = WorkingTree(repo_path=str(tmp_path), store=store)

wt.stage_file("hello.txt")
file.unlink()

assert hash_before == hash_after
status = wt._working_status()
assert "hello.txt" in status.deleted
Loading