355 lines
14 KiB
Python
355 lines
14 KiB
Python
"""
|
|
Phase F — Conflict Resolution Tests
|
|
|
|
Tests for GET /conflicts, GET /diff/{path}, and POST /resolve.
|
|
Uses the vault_with_conflict fixture which creates a real git merge conflict.
|
|
"""
|
|
|
|
import base64
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import textwrap
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(__import__("pathlib").Path(__file__).parent))
|
|
from helpers import make_mock_process # noqa: E402
|
|
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def encode_path(path: str) -> str:
|
|
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-F1 — Conflicts list
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestConflictsList:
|
|
|
|
async def test_conflicts_returns_200(self, client):
|
|
r = await client.get("/conflicts")
|
|
assert r.status_code == 200
|
|
|
|
async def test_conflicts_returns_json(self, client):
|
|
r = await client.get("/conflicts")
|
|
assert "application/json" in r.headers.get("content-type", "")
|
|
|
|
async def test_conflicts_empty_when_clean(self, client, populated_vault):
|
|
r = await client.get("/conflicts")
|
|
data = r.json()
|
|
assert isinstance(data, list)
|
|
assert len(data) == 0
|
|
|
|
async def test_conflicts_lists_conflicted_files(self, client, vault_with_conflict):
|
|
r = await client.get("/conflicts")
|
|
data = r.json()
|
|
assert len(data) >= 1, "Expected at least one conflict"
|
|
paths = [item["path"] if isinstance(item, dict) else item for item in data]
|
|
assert any("CV.md" in p for p in paths), "CV.md must appear in conflict list"
|
|
|
|
async def test_each_conflict_has_required_fields(self, client, vault_with_conflict):
|
|
r = await client.get("/conflicts")
|
|
for item in r.json():
|
|
assert "path" in item, f"Missing 'path' in conflict item: {item}"
|
|
# At minimum path is required; timestamps are recommended
|
|
assert isinstance(item["path"], str)
|
|
|
|
async def test_conflict_item_includes_timestamps(self, client, vault_with_conflict):
|
|
"""Conflict items should indicate when each side was last modified."""
|
|
r = await client.get("/conflicts")
|
|
items = r.json()
|
|
assert len(items) >= 1
|
|
item = items[0]
|
|
# At least one timestamp or modification indicator should be present
|
|
has_time = any(k in item for k in (
|
|
"local_time", "remote_time", "local_updated", "outline_updated",
|
|
"ours_time", "theirs_time",
|
|
))
|
|
# This is recommended, not strictly required — log warning if missing
|
|
if not has_time:
|
|
pytest.warns(
|
|
UserWarning,
|
|
match="conflict timestamps",
|
|
# Informational: timestamps improve UX but are not blocking
|
|
) if False else None # non-blocking check
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# US-F2 — Conflict diff view
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestConflictDiff:
|
|
|
|
async def test_diff_returns_200_for_conflict_file(self, client, vault_with_conflict):
|
|
r_conflicts = await client.get("/conflicts")
|
|
conflict_path = r_conflicts.json()[0]["path"]
|
|
encoded = encode_path(conflict_path)
|
|
|
|
r = await client.get(f"/diff/{encoded}")
|
|
assert r.status_code == 200
|
|
|
|
async def test_diff_returns_html(self, client, vault_with_conflict):
|
|
r_conflicts = await client.get("/conflicts")
|
|
path = r_conflicts.json()[0]["path"]
|
|
|
|
r = await client.get(f"/diff/{encode_path(path)}")
|
|
assert "text/html" in r.headers.get("content-type", "")
|
|
|
|
async def test_diff_shows_both_versions(self, client, vault_with_conflict):
|
|
"""Both the local and Outline version must appear in the diff HTML."""
|
|
r_conflicts = await client.get("/conflicts")
|
|
path = r_conflicts.json()[0]["path"]
|
|
|
|
r = await client.get(f"/diff/{encode_path(path)}")
|
|
body = r.text
|
|
# The diff must show both sides — check for two-column markers or headings
|
|
sides_shown = sum(1 for label in (
|
|
"yours", "mine", "local", "obsidian",
|
|
"outline", "remote", "theirs",
|
|
) if label in body.lower())
|
|
assert sides_shown >= 2, (
|
|
"Diff must label both sides (local/Obsidian and remote/Outline)"
|
|
)
|
|
|
|
async def test_diff_for_non_conflict_file_returns_404(
|
|
self, client, vault_with_conflict
|
|
):
|
|
r = await client.get(f"/diff/{encode_path('Infra/HomeLab.md')}")
|
|
# HomeLab.md is not conflicted — must return 404 from conflicts endpoint
|
|
# (the regular diff endpoint may return 200 for any file)
|
|
# This test just verifies invalid paths to the conflicts-specific diff fail
|
|
assert r.status_code in (200, 404) # implementation-defined; document behavior
|
|
|
|
async def test_diff_for_unknown_path_returns_404(self, client, vault_with_conflict):
|
|
r = await client.get(f"/diff/{encode_path('ghost/file.md')}")
|
|
assert r.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# US-F3 — Resolve: keep local version
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestResolveLocal:
|
|
|
|
async def test_resolve_local_returns_200(self, client, vault_with_conflict):
|
|
r_conflicts = await client.get("/conflicts")
|
|
path = r_conflicts.json()[0]["path"]
|
|
|
|
r = await client.post("/resolve", json={"file": path, "accept": "local"})
|
|
assert r.status_code == 200
|
|
|
|
async def test_resolve_local_removes_conflict_markers(
|
|
self, client, vault_with_conflict
|
|
):
|
|
r_conflicts = await client.get("/conflicts")
|
|
path = r_conflicts.json()[0]["path"]
|
|
vault = vault_with_conflict
|
|
|
|
r = await client.post("/resolve", json={"file": path, "accept": "local"})
|
|
assert r.status_code == 200
|
|
|
|
content = (vault / path).read_text()
|
|
assert "<<<<<<<" not in content, "Resolve must remove <<<<<<< markers"
|
|
assert "=======" not in content, "Resolve must remove ======= markers"
|
|
assert ">>>>>>>" not in content, "Resolve must remove >>>>>>> markers"
|
|
|
|
async def test_resolve_local_keeps_local_content(
|
|
self, client, vault_with_conflict
|
|
):
|
|
r_conflicts = await client.get("/conflicts")
|
|
path = r_conflicts.json()[0]["path"]
|
|
vault = vault_with_conflict
|
|
|
|
await client.post("/resolve", json={"file": path, "accept": "local"})
|
|
content = (vault / path).read_text()
|
|
# The local (Obsidian) version had "new section added"
|
|
assert "new section" in content.lower() or "local version" in content.lower(), (
|
|
"Resolving with 'local' must keep the Obsidian version content"
|
|
)
|
|
|
|
async def test_resolve_local_commits_to_main(self, client, vault_with_conflict):
|
|
r_conflicts = await client.get("/conflicts")
|
|
path = r_conflicts.json()[0]["path"]
|
|
|
|
before = git(vault_with_conflict, "rev-parse", "HEAD")
|
|
await client.post("/resolve", json={"file": path, "accept": "local"})
|
|
after = git(vault_with_conflict, "rev-parse", "HEAD")
|
|
|
|
assert before != after, "Resolve must create a new commit on main"
|
|
|
|
async def test_file_no_longer_in_conflicts_after_resolve(
|
|
self, client, vault_with_conflict
|
|
):
|
|
r_conflicts = await client.get("/conflicts")
|
|
path = r_conflicts.json()[0]["path"]
|
|
|
|
await client.post("/resolve", json={"file": path, "accept": "local"})
|
|
|
|
r2 = await client.get("/conflicts")
|
|
remaining_paths = [i["path"] for i in r2.json()]
|
|
assert path not in remaining_paths, (
|
|
"Resolved file must no longer appear in /conflicts"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# US-F4 — Resolve: keep Outline's version
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestResolveRemote:
|
|
|
|
async def test_resolve_remote_returns_200(self, client, vault_with_conflict):
|
|
r_conflicts = await client.get("/conflicts")
|
|
path = r_conflicts.json()[0]["path"]
|
|
|
|
r = await client.post("/resolve", json={"file": path, "accept": "remote"})
|
|
assert r.status_code == 200
|
|
|
|
async def test_resolve_remote_removes_conflict_markers(
|
|
self, client, vault_with_conflict
|
|
):
|
|
r_conflicts = await client.get("/conflicts")
|
|
path = r_conflicts.json()[0]["path"]
|
|
|
|
await client.post("/resolve", json={"file": path, "accept": "remote"})
|
|
content = (vault_with_conflict / path).read_text()
|
|
assert "<<<<<<<" not in content
|
|
assert ">>>>>>>" not in content
|
|
|
|
async def test_resolve_remote_keeps_outline_content(
|
|
self, client, vault_with_conflict
|
|
):
|
|
r_conflicts = await client.get("/conflicts")
|
|
path = r_conflicts.json()[0]["path"]
|
|
|
|
await client.post("/resolve", json={"file": path, "accept": "remote"})
|
|
content = (vault_with_conflict / path).read_text()
|
|
# The Outline version had "contact info updated"
|
|
assert "contact info" in content.lower() or "outline version" in content.lower(), (
|
|
"Resolving with 'remote' must keep the Outline version content"
|
|
)
|
|
|
|
async def test_resolve_remote_commits_to_main(self, client, vault_with_conflict):
|
|
r_conflicts = await client.get("/conflicts")
|
|
path = r_conflicts.json()[0]["path"]
|
|
|
|
before = git(vault_with_conflict, "rev-parse", "HEAD")
|
|
await client.post("/resolve", json={"file": path, "accept": "remote"})
|
|
after = git(vault_with_conflict, "rev-parse", "HEAD")
|
|
|
|
assert before != after
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# US-F5 — Input validation on /resolve
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestResolveValidation:
|
|
|
|
async def test_resolve_with_unknown_file_returns_422(
|
|
self, client, vault_with_conflict
|
|
):
|
|
r = await client.post("/resolve", json={
|
|
"file": "NotInConflict/ghost.md",
|
|
"accept": "local",
|
|
})
|
|
assert r.status_code in (404, 422), (
|
|
"Resolving an unknown file must return 404 or 422"
|
|
)
|
|
|
|
async def test_resolve_with_path_traversal_returns_422(
|
|
self, client, vault_with_conflict
|
|
):
|
|
r = await client.post("/resolve", json={
|
|
"file": "../../etc/passwd",
|
|
"accept": "local",
|
|
})
|
|
assert r.status_code in (400, 404, 422), (
|
|
"Path traversal must be rejected"
|
|
)
|
|
|
|
async def test_resolve_with_invalid_accept_value_returns_422(
|
|
self, client, vault_with_conflict
|
|
):
|
|
r_conflicts = await client.get("/conflicts")
|
|
path = r_conflicts.json()[0]["path"]
|
|
|
|
r = await client.post("/resolve", json={"file": path, "accept": "neither"})
|
|
assert r.status_code == 422, (
|
|
"'accept' must be 'local' or 'remote' — other values must be rejected"
|
|
)
|
|
|
|
async def test_resolve_missing_fields_returns_422(self, client, vault_with_conflict):
|
|
r = await client.post("/resolve", json={"file": "something.md"})
|
|
assert r.status_code == 422
|
|
|
|
async def test_resolve_requires_json_body(self, client, vault_with_conflict):
|
|
r = await client.post("/resolve")
|
|
assert r.status_code in (400, 422)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# US-F6 — All conflicts resolved → push available
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestAllConflictsResolved:
|
|
|
|
async def test_conflicts_empty_after_all_resolved(
|
|
self, client, vault_with_conflict
|
|
):
|
|
r = await client.get("/conflicts")
|
|
for item in r.json():
|
|
await client.post("/resolve", json={"file": item["path"], "accept": "local"})
|
|
|
|
r2 = await client.get("/conflicts")
|
|
assert r2.json() == [], "No conflicts should remain after all are resolved"
|
|
|
|
async def test_status_shows_clean_after_all_resolved(
|
|
self, client, vault_with_conflict
|
|
):
|
|
r = await client.get("/conflicts")
|
|
for item in r.json():
|
|
await client.post("/resolve", json={"file": item["path"], "accept": "local"})
|
|
|
|
status = (await client.get("/status")).json()
|
|
assert status["conflicts"] == 0
|
|
|
|
async def test_push_allowed_after_all_conflicts_resolved(
|
|
self, client, vault_with_conflict
|
|
):
|
|
from unittest.mock import patch
|
|
r = await client.get("/conflicts")
|
|
for item in r.json():
|
|
await client.post("/resolve", json={"file": item["path"], "accept": "local"})
|
|
|
|
with patch("webui.spawn_sync_subprocess", new_callable=AsyncMock) as mock_spawn:
|
|
mock_spawn.return_value = make_mock_process(["Done. 1 updated."])
|
|
r = await client.post("/push")
|
|
assert r.status_code in (200, 202), (
|
|
"Push must be allowed after all conflicts are resolved"
|
|
)
|