smriti/backend/tests/integration/test_worktrees.py

"""Integration tests for the V5 worktree API.

These tests use a real temporary git repository on disk. That is the
important safety boundary for this feature: V1 is about filesystem/index
isolation, so the API must exercise real `git worktree` behavior.
"""
from __future__ import annotations

import subprocess
import uuid
from pathlib import Path
from types import SimpleNamespace

from fastapi import HTTPException

from app.api.routes import worktrees
from app.db.models import WorkTree


def _git(repo: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
    result = subprocess.run(
        ["git", *args],
        cwd=repo,
        capture_output=True,
        text=True,
        check=False,
    )
    if check:
        assert result.returncode == 0, result.stderr
    return result


def _create_repo(client, name="Worktree Test Repo", project_root: str | None = None):
    payload = {"name": name}
    if project_root is not None:
        payload["project_root"] = project_root
    r = client.post("/api/v2/repos", json=payload)
    assert r.status_code == 201, r.text
    return r.json()["id"]


def _create_session(client, repo_id, title="worktree test"):
    r = client.post(
        f"/api/v4/chat/spaces/{repo_id}/sessions",
        json={"title": title, "provider": "openrouter", "model": "mock"},
    )
    assert r.status_code == 201, r.text
    return r.json()["id"]


def _commit_with_root(client, repo_id, session_id, project_root: Path):
    payload = {
        "repo_id": repo_id,
        "session_id": session_id,
        "message": "base",
        "summary": "base",
        "project_root": str(project_root),
    }
    r = client.post("/api/v4/chat/commit", json=payload)
    assert r.status_code == 201, r.text
    return r.json()


def _create_project_with_root(client, git_repo: Path):
    repo_id = _create_repo(client)
    session_id = _create_session(client, repo_id)
    _commit_with_root(client, repo_id, session_id, git_repo)
    return repo_id


def _create_worktree(client, space_id: str, **kwargs):
    payload = {
        "space_id": space_id,
        "agent": kwargs.pop("agent", "codex-local"),
        **kwargs,
    }
    return client.post("/api/v5/worktrees", json=payload)


def test_create_worktree_inserts_row_creates_directory_and_branch(
    client,
    tmp_path,
    monkeypatch,
):
    git_repo = _init_git_repo(tmp_path)
    space_id = _create_project_with_root(client, git_repo)
    monkeypatch.setattr(worktrees.Path, "home", lambda: tmp_path / "home")

    r = _create_worktree(client, space_id, agent="Codex Local")

    assert r.status_code == 201, r.text
    data = r.json()
    worktree_path = Path(data["path"])
    assert worktree_path.exists()
    assert data["agent"] == "Codex Local"
    assert data["branch_name"].startswith("smriti/codex-local/")
    assert data["base_commit_sha"] == _git(git_repo, "rev-parse", "HEAD").stdout.strip()
    assert data["status"] == "active"
    assert _git(
        git_repo,
        "show-ref",
        "--verify",
        f"refs/heads/{data['branch_name']}",
    ).returncode == 0


def test_create_worktree_honors_explicit_branch_base_and_path(client, tmp_path):
    git_repo = _init_git_repo(tmp_path)
    space_id = _create_project_with_root(client, git_repo)
    base_sha = _git(git_repo, "rev-parse", "HEAD").stdout.strip()
    target = tmp_path / "custom-worktree"

    r = _create_worktree(
        client,
        space_id,
        branch_name="feature/custom-worktree",
        base_commit_sha=base_sha,
        base_path=str(target),
    )

    assert r.status_code == 201, r.text
    data = r.json()
    assert data["branch_name"] == "feature/custom-worktree"
    assert data["base_commit_sha"] == base_sha
    assert data["path"] == str(target.resolve())
    assert target.exists()


def test_create_worktree_existing_branch_rejected_without_row_or_directory(
    client,
    tmp_path,
):
    git_repo = _init_git_repo(tmp_path)
    space_id = _create_project_with_root(client, git_repo)
    _git(git_repo, "branch", "existing-worktree-branch")
    target = tmp_path / "should-not-exist"

    r = _create_worktree(
        client,
        space_id,
        branch_name="existing-worktree-branch",
        base_path=str(target),
    )

    assert r.status_code == 409
    assert "Branch already exists" in r.json()["detail"]
    assert not target.exists()
    assert client.get(f"/api/v5/worktrees?space_id={space_id}").json() == []


def test_create_worktree_uses_canonical_project_root(client, tmp_path):
    canonical_repo = _init_git_repo(tmp_path, name="canonical-project")
    fallback_repo = _init_git_repo(tmp_path, name="fallback-project")
    repo_id = _create_repo(
        client,
        name="Canonical Root Repo",
        project_root=str(canonical_repo),
    )
    session_id = _create_session(client, repo_id)
    _commit_with_root(client, repo_id, session_id, fallback_repo)
    target = tmp_path / "canonical-worktree"

    r = _create_worktree(
        client,
        repo_id,
        branch_name="feature/canonical-root",
        base_path=str(target),
    )

    assert r.status_code == 201, r.text
    assert target.exists()
    assert (
        _git(canonical_repo, "show-ref", "--verify", "refs/heads/feature/canonical-root")
        .returncode
        == 0
    )
    assert (
        _git(
            fallback_repo,
            "show-ref",
            "--verify",
            "refs/heads/feature/canonical-root",
            check=False,
        ).returncode
        != 0
    )


def test_create_worktree_lazy_backfills_project_root_from_commit(client, tmp_path):
    git_repo = _init_git_repo(tmp_path)
    repo_id = _create_project_with_root(client, git_repo)
    assert client.get(f"/api/v2/repos/{repo_id}").json()["project_root"] is None

    r = _create_worktree(
        client,
        repo_id,
        branch_name="feature/lazy-backfill",
        base_path=str(tmp_path / "lazy-backfill-worktree"),
    )

    assert r.status_code == 201, r.text
    assert client.get(f"/api/v2/repos/{repo_id}").json()["project_root"] == str(
        git_repo.resolve()
    )


def test_list_show_and_close_clean_worktree(client, tmp_path):
    git_repo = _init_git_repo(tmp_path)
    space_id = _create_project_with_root(client, git_repo)
    target = tmp_path / "list-show-close"
    created = _create_worktree(client, space_id, base_path=str(target)).json()

    list_r = client.get(f"/api/v5/worktrees?space_id={space_id}")
    assert list_r.status_code == 200
    assert [w["id"] for w in list_r.json()] == [created["id"]]

    show_r = client.get(f"/api/v5/worktrees/{created['id']}")
    assert show_r.status_code == 200
    assert show_r.json()["path"] == str(target.resolve())

    close_r = client.delete(f"/api/v5/worktrees/{created['id']}")
    assert close_r.status_code == 200, close_r.text
    closed = close_r.json()
    assert closed["status"] == "closed"
    assert closed["closed_at"] is not None
    assert not target.exists()

    active_r = client.get(f"/api/v5/worktrees?space_id={space_id}")
    assert active_r.status_code == 200
    assert active_r.json() == []

    all_r = client.get(f"/api/v5/worktrees?space_id={space_id}&include_closed=true")
    assert len(all_r.json()) == 1


def test_short_id_resolution_unique_prefix(client, tmp_path):
    git_repo = _init_git_repo(tmp_path)
    space_id = _create_project_with_root(client, git_repo)
    created = _create_worktree(
        client,
        space_id,
        base_path=str(tmp_path / "short-id-show"),
    ).json()

    r = client.get(f"/api/v5/worktrees/{created['id'][:8]}")

    assert r.status_code == 200, r.text
    assert r.json()["id"] == created["id"]


def test_short_id_resolution_full_uuid_still_works(client, tmp_path):
    git_repo = _init_git_repo(tmp_path)
    space_id = _create_project_with_root(client, git_repo)
    created = _create_worktree(
        client,
        space_id,
        base_path=str(tmp_path / "full-id-show"),
    ).json()

    r = client.get(f"/api/v5/worktrees/{created['id']}")

    assert r.status_code == 200, r.text
    assert r.json()["id"] == created["id"]


def test_short_id_resolution_no_match_returns_404(client):
    r = client.get("/api/v5/worktrees/deadbeef")

    assert r.status_code == 404
    assert "No worktree matches prefix" in r.json()["detail"]


def test_short_id_resolution_ambiguous_returns_422(client, db_session):
    repo_id = _create_repo(client)
    first = WorkTree(
        id=uuid.UUID("abcd1111-0000-4000-8000-000000000001"),
        repo_id=uuid.UUID(repo_id),
        agent="codex-local",
        path="/tmp/wt-one",
        branch_name="feature/one",
        status="active",
    )
    second = WorkTree(
        id=uuid.UUID("abcd2222-0000-4000-8000-000000000002"),
        repo_id=uuid.UUID(repo_id),
        agent="claude-code",
        path="/tmp/wt-two",
        branch_name="feature/two",
        status="active",
    )
    db_session.add_all([first, second])
    db_session.commit()

    r = client.get("/api/v5/worktrees/abcd")

    assert r.status_code == 422
    assert "ambiguous" in r.json()["detail"]
    assert "abcd1111" in r.json()["detail"]
    assert "abcd2222" in r.json()["detail"]


def test_short_id_resolution_too_short_returns_400(client):
    r = client.get("/api/v5/worktrees/abc")

    assert r.status_code == 400
    assert "at least 4 characters" in r.json()["detail"]


def test_short_id_resolution_close_path(client, tmp_path):
    git_repo = _init_git_repo(tmp_path)
    space_id = _create_project_with_root(client, git_repo)
    target = tmp_path / "short-id-close"
    created = _create_worktree(client, space_id, base_path=str(target)).json()

    r = client.delete(f"/api/v5/worktrees/{created['id'][:8]}")

    assert r.status_code == 200, r.text
    assert r.json()["id"] == created["id"]
    assert r.json()["status"] == "closed"
    assert not target.exists()


def test_list_includes_probe_data_for_active_worktrees(client, tmp_path, monkeypatch):
    git_repo = _init_git_repo(tmp_path)
    space_id = _create_project_with_root(client, git_repo)
    target = tmp_path / "probe-worktree"
    created = _create_worktree(client, space_id, base_path=str(target)).json()

    def fake_probe(worktree_id, path, branch):
        assert worktree_id == created["id"]
        assert path == str(target.resolve())
        assert branch == created["branch_name"]
        return {
            "id": worktree_id,
            "path": path,
            "branch": branch,
            "dirty_files": 3,
            "dirty_paths": ["cli/main.py", "backend/app/api/routes/worktrees.py"],
            "ahead": 1,
            "behind": 0,
            "last_commit_sha": "abc1234",
            "last_commit_relative": "5 minutes ago",
        }

    monkeypatch.setattr(worktrees, "_probe_worktree", fake_probe)

    r = client.get(f"/api/v5/worktrees?space_id={space_id}")

    assert r.status_code == 200, r.text
    data = r.json()
    assert data[0]["id"] == created["id"]
    assert data[0]["probe"] == {
        "dirty_files": 3,
        "dirty_paths": ["cli/main.py", "backend/app/api/routes/worktrees.py"],
        "ahead": 1,
        "behind": 0,
        "last_commit_sha": "abc1234",
        "last_commit_relative": "5 minutes ago",
    }


def test_list_probe_null_for_closed_worktrees(client, tmp_path, monkeypatch):
    git_repo = _init_git_repo(tmp_path)
    space_id = _create_project_with_root(client, git_repo)
    created = _create_worktree(
        client,
        space_id,
        base_path=str(tmp_path / "closed-probe"),
    ).json()
    close_r = client.delete(f"/api/v5/worktrees/{created['id']}")
    assert close_r.status_code == 200, close_r.text

    def fail_probe(worktree_id, path, branch):
        raise AssertionError("closed worktrees should not be probed")

    monkeypatch.setattr(worktrees, "_probe_worktree", fail_probe)

    r = client.get(f"/api/v5/worktrees?space_id={space_id}&include_closed=true")

    assert r.status_code == 200, r.text
    assert r.json()[0]["id"] == created["id"]
    assert r.json()[0]["probe"] is None


def test_show_nonexistent_worktree_returns_404(client):
    r = client.get(f"/api/v5/worktrees/{uuid.uuid4()}")
    assert r.status_code == 404


def test_close_dirty_worktree_requires_force(client, tmp_path):
    git_repo = _init_git_repo(tmp_path)
    space_id = _create_project_with_root(client, git_repo)
    target = tmp_path / "dirty-worktree"
    created = _create_worktree(client, space_id, base_path=str(target)).json()
    (target / "dirty.txt").write_text("uncommitted\n")

    r = client.delete(f"/api/v5/worktrees/{created['id']}")

    assert r.status_code == 409
    assert "uncommitted changes" in r.json()["detail"]
    assert target.exists()

    force_r = client.delete(f"/api/v5/worktrees/{created['id']}?force=true")
    assert force_r.status_code == 200, force_r.text
    assert force_r.json()["status"] == "closed"
    assert not target.exists()


def test_close_already_closed_returns_409(client, tmp_path):
    git_repo = _init_git_repo(tmp_path)
    space_id = _create_project_with_root(client, git_repo)
    created = _create_worktree(
        client,
        space_id,
        base_path=str(tmp_path / "already-closed"),
    ).json()
    first = client.delete(f"/api/v5/worktrees/{created['id']}")
    assert first.status_code == 200

    second = client.delete(f"/api/v5/worktrees/{created['id']}")

    assert second.status_code == 409
    assert "already closed" in second.json()["detail"]


def test_create_worktree_requires_project_root(client, tmp_path):
    space_id = _create_repo(client)
    target = tmp_path / "no-root-worktree"

    r = _create_worktree(client, space_id, base_path=str(target))

    assert r.status_code == 400
    assert "project_root" in r.json()["detail"]
    assert "set-project-root" in r.json()["detail"]
    assert not target.exists()


def test_git_infrastructure_error_does_not_write_row_or_leave_target(
    client,
    tmp_path,
    monkeypatch,
):
    git_repo = _init_git_repo(tmp_path)
    space_id = _create_project_with_root(client, git_repo)
    target = tmp_path / "git-missing"
    head_sha = _git(git_repo, "rev-parse", "HEAD").stdout.strip()

    def fake_run_git(args, *, cwd=None, timeout=30.0):
        if args[0] == "show-ref":
            return SimpleNamespace(returncode=1, stdout="", stderr="")
        if args[0] == "rev-parse":
            return SimpleNamespace(returncode=0, stdout=f"{head_sha}\n", stderr="")
        if args[:2] == ["worktree", "add"]:
            raise HTTPException(status_code=500, detail="git executable not found")
        raise AssertionError(f"unexpected git call: {args}")

    monkeypatch.setattr(worktrees, "_run_git", fake_run_git)

    r = _create_worktree(client, space_id, base_path=str(target))

    assert r.status_code == 500
    assert "git executable not found" in r.json()["detail"]
    assert not target.exists()
    assert client.get(f"/api/v5/worktrees?space_id={space_id}").json() == []


def _init_git_repo(tmp_path: Path, name: str = "project") -> Path:
    repo = tmp_path / name
    repo.mkdir()
    _git(repo, "init")
    _git(repo, "config", "user.email", "test@example.com")
    _git(repo, "config", "user.name", "Test User")
    (repo / "README.md").write_text("hello\n")
    _git(repo, "add", "README.md")
    _git(repo, "commit", "-m", "initial")
    return repo