""" Phase D — Pending Changes View Tests Tests for GET /changes (structured change list) and GET /diff/{path} (inline diff). Git operations run against the real temp vault — no subprocess mocking needed here. """ import base64 import subprocess import textwrap from pathlib import Path import pytest pytestmark = pytest.mark.asyncio # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def encode_path(path: str) -> str: """URL-safe base64 encoding of a path, matching what the app uses.""" return base64.urlsafe_b64encode(path.encode()).decode() def git(vault: Path, *args) -> str: return subprocess.run( ["git", "-C", str(vault), *args], check=True, capture_output=True, text=True, ).stdout.strip() def commit_all(vault: Path, message: str): subprocess.run(["git", "-C", str(vault), "add", "-A"], check=True, capture_output=True) try: subprocess.run(["git", "-C", str(vault), "commit", "-m", message], check=True, capture_output=True) except subprocess.CalledProcessError: pass # --------------------------------------------------------------------------- # US-D1 — Changes endpoint structure # --------------------------------------------------------------------------- class TestChangesEndpoint: async def test_get_changes_returns_200(self, client): r = await client.get("/changes") assert r.status_code == 200 async def test_changes_returns_json(self, client): r = await client.get("/changes") assert "application/json" in r.headers.get("content-type", "") async def test_changes_returns_list(self, client): r = await client.get("/changes") data = r.json() assert isinstance(data, list) async def test_changes_empty_when_clean(self, client, populated_vault): r = await client.get("/changes") assert r.json() == [] async def test_each_change_has_required_fields(self, client, vault_with_pending): r = await client.get("/changes") items = r.json() assert len(items) > 0, "Expected pending changes" for item in items: assert "path" in item, f"Missing 'path' in item: {item}" assert "status" in item, f"Missing 'status' in item: {item}" assert "action" in item, f"Missing 'action' in item: {item}" async def test_status_values_are_valid(self, client, vault_with_pending): valid_statuses = {"modified", "added", "deleted", "renamed"} r = await client.get("/changes") for item in r.json(): assert item["status"] in valid_statuses, ( f"Invalid status '{item['status']}' — must be one of {valid_statuses}" ) # --------------------------------------------------------------------------- # US-D2 — Change categories # --------------------------------------------------------------------------- class TestChangeCategories: async def test_modified_file_shown_as_modified(self, client, populated_vault): # Edit an existing file on main cv = populated_vault / "Bewerbungen" / "CV.md" cv.write_text(cv.read_text() + "\n## Appendix\nNew content.\n") commit_all(populated_vault, "obsidian: edit CV") r = await client.get("/changes") items = r.json() cv_item = next((i for i in items if "CV.md" in i["path"]), None) assert cv_item is not None, "CV.md must appear in changes" assert cv_item["status"] == "modified" async def test_new_file_shown_as_added(self, client, populated_vault): new_file = populated_vault / "Projekte" / "NewDoc.md" new_file.parent.mkdir(exist_ok=True) new_file.write_text("# New Doc\nWritten in Obsidian.\n") commit_all(populated_vault, "obsidian: new doc") r = await client.get("/changes") items = r.json() new_item = next((i for i in items if "NewDoc.md" in i["path"]), None) assert new_item is not None, "NewDoc.md must appear in changes" assert new_item["status"] == "added" async def test_deleted_file_shown_as_deleted(self, client, populated_vault): cv = populated_vault / "Bewerbungen" / "CV.md" cv.unlink() commit_all(populated_vault, "obsidian: delete CV") r = await client.get("/changes") items = r.json() del_item = next((i for i in items if "CV.md" in i["path"]), None) assert del_item is not None, "Deleted CV.md must appear in changes" assert del_item["status"] == "deleted" async def test_renamed_file_shown_as_renamed(self, client, populated_vault): old_path = populated_vault / "Bewerbungen" / "CV.md" new_path = populated_vault / "Bewerbungen" / "Curriculum Vitae.md" old_path.rename(new_path) commit_all(populated_vault, "obsidian: rename CV") r = await client.get("/changes") items = r.json() # Either old or new name should appear with renamed status renamed = [i for i in items if i["status"] == "renamed"] assert len(renamed) >= 1, "Renamed file must appear with status=renamed" # Check both from_path and to_path since either may contain "CV" all_paths = " ".join( str(i.get("from_path", "")) + " " + str(i.get("to_path", i["path"])) for i in renamed ) assert "CV" in all_paths, "Renamed paths must reference the original filename" async def test_renamed_item_has_from_and_to_paths(self, client, populated_vault): old_path = populated_vault / "Bewerbungen" / "CV.md" new_path = populated_vault / "Bewerbungen" / "Resume.md" old_path.rename(new_path) commit_all(populated_vault, "obsidian: rename") r = await client.get("/changes") renamed = [i for i in r.json() if i["status"] == "renamed"] assert len(renamed) >= 1 item = renamed[0] assert "from" in item or "from_path" in item, "Rename must include source path" assert "to" in item or "to_path" in item, "Rename must include destination path" # --------------------------------------------------------------------------- # US-D3 — Diff preview # --------------------------------------------------------------------------- class TestDiffPreview: async def test_diff_endpoint_returns_200(self, client, populated_vault): # Edit a file to create a diff cv = populated_vault / "Bewerbungen" / "CV.md" original = cv.read_text() cv.write_text(original + "\n## New Section\n") commit_all(populated_vault, "edit for diff") encoded = encode_path("Bewerbungen/CV.md") r = await client.get(f"/diff/{encoded}") assert r.status_code == 200 async def test_diff_returns_html_fragment(self, client, populated_vault): cv = populated_vault / "Bewerbungen" / "CV.md" cv.write_text(cv.read_text() + "\nExtra line.\n") commit_all(populated_vault, "edit for diff") encoded = encode_path("Bewerbungen/CV.md") r = await client.get(f"/diff/{encoded}") assert "text/html" in r.headers.get("content-type", ""), ( "Diff endpoint must return HTML" ) async def test_diff_contains_two_columns(self, client, populated_vault): cv = populated_vault / "Bewerbungen" / "CV.md" cv.write_text(cv.read_text() + "\nAdded line.\n") commit_all(populated_vault, "edit for diff") encoded = encode_path("Bewerbungen/CV.md") r = await client.get(f"/diff/{encoded}") body = r.text.lower() # Two-column layout — check for table or grid structure assert "table" in body or "column" in body or "diff" in body, ( "Diff HTML must contain a two-column comparison layout" ) async def test_diff_for_unknown_file_returns_404(self, client, populated_vault): encoded = encode_path("DoesNotExist/ghost.md") r = await client.get(f"/diff/{encoded}") assert r.status_code == 404 async def test_diff_added_lines_have_distinct_marking(self, client, populated_vault): cv = populated_vault / "Bewerbungen" / "CV.md" cv.write_text(cv.read_text() + "\nThis line was added.\n") commit_all(populated_vault, "add line") encoded = encode_path("Bewerbungen/CV.md") r = await client.get(f"/diff/{encoded}") body = r.text # Added lines must be visually distinct (green class, + prefix, or ins tag) # difflib.HtmlDiff marks added lines with class="diff_add" assert any(marker in body for marker in ( 'class="diff_add"', "diff_add", 'class="add"', "", "diff-add", )), "Added lines must be visually marked in the diff" # --------------------------------------------------------------------------- # US-D4 — Deleted files skipped when allow_deletions=false # --------------------------------------------------------------------------- class TestDeletedFilesSkipped: async def test_deleted_file_action_is_skip_when_deletions_off( self, client, populated_vault, settings_file ): """With allow_deletions=false in settings, deleted files must show action=skip.""" import json settings = json.loads(settings_file.read_text()) settings["sync"]["allow_deletions"] = False settings_file.write_text(json.dumps(settings)) cv = populated_vault / "Bewerbungen" / "CV.md" cv.unlink() commit_all(populated_vault, "delete CV") r = await client.get("/changes") items = r.json() del_item = next((i for i in items if "CV.md" in i["path"]), None) assert del_item is not None assert del_item["action"] in ("skip", "skipped"), ( "Deleted file must have action=skip when deletions are disabled" ) async def test_deleted_file_action_is_delete_when_deletions_on( self, client, populated_vault, settings_file ): """With allow_deletions=true, deleted file action must be delete.""" import json settings = json.loads(settings_file.read_text()) settings["sync"]["allow_deletions"] = True settings_file.write_text(json.dumps(settings)) cv = populated_vault / "Bewerbungen" / "CV.md" cv.unlink() commit_all(populated_vault, "delete CV") r = await client.get("/changes") items = r.json() del_item = next((i for i in items if "CV.md" in i["path"]), None) assert del_item is not None assert del_item["action"] in ("delete", "archive"), ( "Deleted file must have action=delete when deletions are enabled" )