hermes-webui/tests/test_workspace_git.py

import json
import pathlib
import subprocess
import threading
import types
import uuid
import urllib.error
import urllib.parse
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from io import BytesIO

import pytest

from tests._pytest_port import BASE


ROOT = pathlib.Path(__file__).parent.parent


def _git(cwd, *args):
    result = subprocess.run(
        ["git", *args],
        cwd=str(cwd),
        shell=False,
        text=True,
        capture_output=True,
        timeout=20,
    )
    assert result.returncode == 0, result.stderr or result.stdout
    return result.stdout


def _init_repo(path):
    path.mkdir(parents=True, exist_ok=True)
    init = subprocess.run(
        ["git", "init", "-b", "master"],
        cwd=str(path),
        shell=False,
        text=True,
        capture_output=True,
        timeout=20,
    )
    if init.returncode != 0:
        _git(path, "init")
        _git(path, "checkout", "-B", "master")
    _git(path, "config", "user.email", "hermes-tests@example.invalid")
    _git(path, "config", "user.name", "Hermes Tests")
    return path


def _init_bare_repo(path):
    init = subprocess.run(
        ["git", "init", "--bare", "-b", "master", str(path)],
        shell=False,
        text=True,
        capture_output=True,
        timeout=20,
    )
    if init.returncode != 0:
        _git(path.parent, "init", "--bare", str(path))
        _git(path, "symbolic-ref", "HEAD", "refs/heads/master")
    return path


def _commit_all(path, message="initial"):
    _git(path, "add", ".")
    _git(path, "commit", "-m", message)


def _get(path):
    try:
        with urllib.request.urlopen(BASE + path, timeout=10) as r:
            return json.loads(r.read()), r.status
    except urllib.error.HTTPError as e:
        return json.loads(e.read()), e.code


def _post(path, body=None):
    data = json.dumps(body or {}).encode()
    req = urllib.request.Request(
        BASE + path,
        data=data,
        headers={"Content-Type": "application/json"},
    )
    try:
        with urllib.request.urlopen(req, timeout=10) as r:
            return json.loads(r.read()), r.status
    except urllib.error.HTTPError as e:
        return json.loads(e.read()), e.code


def _make_session(created_list, ws=None):
    body = {}
    if ws:
        body["workspace"] = str(ws)
    data, status = _post("/api/session/new", body)
    assert status == 200
    sid = data["session"]["session_id"]
    created_list.append(sid)
    return sid, pathlib.Path(data["session"]["workspace"])


class _CaptureHandler:
    def __init__(self):
        self.status = None
        self.headers = {}
        self.response_headers = []
        self.wfile = BytesIO()

    def send_response(self, status):
        self.status = status

    def send_header(self, key, value):
        self.response_headers.append((key, value))

    def end_headers(self):
        pass

    def payload(self):
        return json.loads(self.wfile.getvalue().decode("utf-8"))


def test_git_status_non_git_workspace(tmp_path):
    from api.workspace_git import git_status

    ws = tmp_path / "plain"
    ws.mkdir()
    assert git_status(ws) == {"is_git": False}


def test_git_status_handles_staged_unstaged_untracked_deleted_and_renamed(tmp_path):
    from api.workspace_git import git_status

    repo = _init_repo(tmp_path / "repo")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    (repo / "delete-me.txt").write_text("bye\n", encoding="utf-8")
    (repo / "old name.txt").write_text("move\n", encoding="utf-8")
    _commit_all(repo)

    (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8")
    (repo / "staged.txt").write_text("staged\n", encoding="utf-8")
    _git(repo, "add", "staged.txt")
    (repo / "delete-me.txt").unlink()
    _git(repo, "mv", "old name.txt", "new name.txt")
    (repo / "untracked space.txt").write_text("new\nfile\n", encoding="utf-8")

    status = git_status(repo)
    by_path = {item["path"]: item for item in status["files"]}

    assert status["is_git"] is True
    assert by_path["tracked.txt"]["unstaged"] is True
    assert by_path["staged.txt"]["staged"] is True
    assert by_path["delete-me.txt"]["status"] == "D"
    assert by_path["new name.txt"]["old_path"] == "old name.txt"
    assert by_path["untracked space.txt"]["untracked"] is True
    assert by_path["untracked space.txt"]["additions"] == 2
    assert status["totals"]["changed"] >= 5


def test_git_status_reports_ignored_files_without_counting_them_as_changes(tmp_path):
    from api.workspace_git import git_status

    repo = _init_repo(tmp_path / "repo")
    (repo / ".gitignore").write_text("*.log\nbuild/\n", encoding="utf-8")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)

    (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8")
    (repo / "debug.log").write_text("ignored log\n", encoding="utf-8")
    build = repo / "build"
    build.mkdir()
    (build / "artifact.txt").write_text("ignored artifact\n", encoding="utf-8")

    status = git_status(repo)
    by_path = {item["path"]: item for item in status["files"]}

    assert by_path["tracked.txt"]["unstaged"] is True
    assert by_path["debug.log"]["ignored"] is True
    assert by_path["debug.log"]["status"] == "Ignored"
    assert by_path["build/"]["ignored"] is True
    assert by_path["build/"]["staged"] is False
    assert by_path["build/"]["untracked"] is False
    assert status["totals"]["changed"] == 1
    assert status["totals"]["untracked"] == 0


def test_git_status_ignores_crlf_only_worktree_noise(tmp_path):
    from api.workspace_git import git_status

    repo = _init_repo(tmp_path / "repo")
    (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8", newline="\n")
    _commit_all(repo)

    (repo / "tracked.txt").write_text("one\r\ntwo\r\n", encoding="utf-8", newline="")

    raw = _git(repo, "status", "--porcelain", "--", "tracked.txt")
    assert raw.startswith(" M")

    status = git_status(repo)
    assert status["totals"]["changed"] == 0
    assert status["files"] == []
    assert status["noise_filtering"]["active"] is True
    assert status["noise_filtering"]["crlf_only"] == 1


def test_git_status_keeps_real_edit_with_crlf_endings(tmp_path):
    from api.workspace_git import git_status

    repo = _init_repo(tmp_path / "repo")
    (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8", newline="\n")
    _commit_all(repo)

    (repo / "tracked.txt").write_text("one\r\ntwo\r\nthree\r\n", encoding="utf-8", newline="")

    status = git_status(repo)
    by_path = {item["path"]: item for item in status["files"]}
    assert status["totals"]["changed"] == 1
    assert by_path["tracked.txt"]["unstaged"] is True
    assert by_path["tracked.txt"]["additions"] == 1
    assert by_path["tracked.txt"]["deletions"] == 0


def test_git_status_ignores_filemode_only_noise(tmp_path):
    from api.workspace_git import git_status

    repo = _init_repo(tmp_path / "repo")
    script = repo / "script.sh"
    script.write_text("#!/bin/sh\necho hi\n", encoding="utf-8")
    _commit_all(repo)

    _git(repo, "update-index", "--chmod=+x", "script.sh")

    raw = _git(repo, "status", "--porcelain", "--", "script.sh")
    assert "script.sh" in raw

    status = git_status(repo)
    assert status["totals"]["changed"] == 0
    assert status["files"] == []
    assert status["noise_filtering"]["active"] is True


def test_git_status_scopes_nested_workspace_to_that_directory(tmp_path):
    from api.workspace_git import git_status

    repo = _init_repo(tmp_path / "repo")
    nested = repo / "app"
    nested.mkdir()
    (nested / "inside.txt").write_text("inside\n", encoding="utf-8")
    (repo / "outside.txt").write_text("outside\n", encoding="utf-8")
    _commit_all(repo)

    (nested / "inside.txt").write_text("inside\nchanged\n", encoding="utf-8")
    (repo / "outside.txt").write_text("outside\nchanged\n", encoding="utf-8")

    status = git_status(nested)
    paths = {item["path"] for item in status["files"]}
    assert paths == {"inside.txt"}


def test_git_diff_generates_untracked_text_diff_and_blocks_escape(tmp_path):
    from api.workspace_git import GitWorkspaceError, git_diff

    repo = _init_repo(tmp_path / "repo")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)
    (repo / "new file.txt").write_text("hello\nworld\n", encoding="utf-8")

    diff = git_diff(repo, "new file.txt", "unstaged")
    assert diff["binary"] is False
    assert "+++ b/new file.txt" in diff["diff"]
    assert "+hello" in diff["diff"]

    with pytest.raises(GitWorkspaceError):
        git_diff(repo, "../outside.txt", "unstaged")


def test_git_status_reports_untracked_files_inside_directories(tmp_path):
    from api.workspace_git import git_discard, git_status

    repo = _init_repo(tmp_path / "repo")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)
    nested = repo / "newdir"
    nested.mkdir()
    (nested / "a.txt").write_text("hello\n", encoding="utf-8")

    status = git_status(repo)
    paths = {item["path"] for item in status["files"]}
    assert "newdir/a.txt" in paths
    assert "newdir/" not in paths

    git_discard(repo, ["newdir/a.txt"], delete_untracked=True)
    assert not (nested / "a.txt").exists()


def test_git_discard_untracked_file_tolerates_concurrent_missing_file(tmp_path, monkeypatch):
    import api.workspace_git as workspace_git

    repo = _init_repo(tmp_path / "repo")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)
    transient = repo / "transient.txt"
    transient.write_text("gone soon\n", encoding="utf-8")

    original_unlink_anchored = workspace_git.unlink_anchored
    raced = {"seen": False}

    def remove_before_unlink(root, target):
        if target == transient:
            raced["seen"] = True
            transient.unlink()
        return original_unlink_anchored(root, target)

    monkeypatch.setattr(workspace_git, "unlink_anchored", remove_before_unlink)

    status = workspace_git.git_discard(repo, ["transient.txt"], delete_untracked=True)

    assert raced["seen"] is True
    assert not transient.exists()
    assert status["totals"]["changed"] == 0


def test_git_status_reports_ignored_files_without_counting_them_as_changed(tmp_path):
    from api.workspace_git import git_status

    repo = _init_repo(tmp_path / "repo")
    (repo / ".gitignore").write_text("*.log\nbuild/\n", encoding="utf-8")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)

    (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8")
    (repo / "debug.log").write_text("ignored log\n", encoding="utf-8")
    build = repo / "build"
    build.mkdir()
    (build / "artifact.txt").write_text("ignored artifact\n", encoding="utf-8")

    status = git_status(repo)
    by_path = {item["path"]: item for item in status["files"]}

    assert by_path["tracked.txt"]["unstaged"] is True
    assert by_path["debug.log"]["ignored"] is True
    assert by_path["debug.log"]["status"] == "Ignored"
    assert by_path["debug.log"]["staged"] is False
    assert by_path["debug.log"]["unstaged"] is False
    assert by_path["debug.log"]["untracked"] is False
    assert any(item["ignored"] and item["path"].startswith("build") for item in status["files"])
    assert status["totals"]["changed"] == 1
    assert status["totals"]["untracked"] == 0


def test_git_diff_large_untracked_file_is_bounded(tmp_path):
    from api.workspace_git import DIFF_SIZE_LIMIT, git_diff, git_status

    repo = _init_repo(tmp_path / "repo")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)
    large = repo / "large.txt"
    large.write_text("x" * (DIFF_SIZE_LIMIT + 1), encoding="utf-8")

    status = git_status(repo)
    by_path = {item["path"]: item for item in status["files"]}
    assert by_path["large.txt"]["untracked"] is True
    assert by_path["large.txt"]["additions"] == 0

    diff = git_diff(repo, "large.txt", "unstaged")
    assert diff["too_large"] is True
    assert diff["diff"] == ""


def test_git_stage_unstage_discard_and_commit(tmp_path):
    from api.workspace_git import git_commit, git_discard, git_stage, git_status, git_unstage

    repo = _init_repo(tmp_path / "repo")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)

    (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8")
    staged = git_stage(repo, ["tracked.txt"])
    assert staged["totals"]["staged"] == 1

    unstaged = git_unstage(repo, ["tracked.txt"])
    assert unstaged["totals"]["staged"] == 0
    assert unstaged["totals"]["unstaged"] == 1

    git_discard(repo, ["tracked.txt"])
    assert git_status(repo)["totals"]["changed"] == 0

    (repo / "tracked.txt").write_text("one\nthree\n", encoding="utf-8")
    git_stage(repo, ["tracked.txt"])
    committed = git_commit(repo, "Update tracked file")
    assert committed["ok"] is True
    assert committed["commit"]
    assert committed["status"]["totals"]["changed"] == 0


def test_git_commit_selected_ignores_unrelated_real_index(tmp_path):
    from api.workspace_git import git_commit_selected, git_status

    repo = _init_repo(tmp_path / "repo")
    (repo / "selected.txt").write_text("one\n", encoding="utf-8")
    (repo / "staged.txt").write_text("alpha\n", encoding="utf-8")
    _commit_all(repo)

    (repo / "selected.txt").write_text("one\ntwo\n", encoding="utf-8")
    (repo / "staged.txt").write_text("alpha\nbeta\n", encoding="utf-8")
    _git(repo, "add", "staged.txt")

    committed = git_commit_selected(repo, "Commit selected only", ["selected.txt"])
    assert committed["ok"] is True
    assert committed["paths"] == ["selected.txt"]
    assert _git(repo, "show", "--name-only", "--format=", "HEAD").splitlines() == ["selected.txt"]

    by_path = {item["path"]: item for item in git_status(repo)["files"]}
    assert "selected.txt" not in by_path
    assert by_path["staged.txt"]["staged"] is True


def test_git_commit_selected_supports_initial_commit(tmp_path):
    from api.workspace_git import git_commit_selected, git_status

    repo = _init_repo(tmp_path / "repo")
    (repo / "first.txt").write_text("first\n", encoding="utf-8")

    committed = git_commit_selected(repo, "Initial selected commit", ["first.txt"])
    assert committed["ok"] is True
    assert _git(repo, "show", "--name-only", "--format=", "HEAD").splitlines() == ["first.txt"]
    assert git_status(repo)["totals"]["changed"] == 0


def test_git_commit_selected_preserves_rename_semantics(tmp_path):
    from api.workspace_git import git_commit_selected, git_status

    repo = _init_repo(tmp_path / "repo")
    (repo / "old.txt").write_text("old\n", encoding="utf-8")
    _commit_all(repo)

    _git(repo, "mv", "old.txt", "new.txt")

    committed = git_commit_selected(repo, "Rename selected file", ["new.txt"])
    assert committed["ok"] is True
    assert _git(repo, "ls-tree", "--name-only", "HEAD").splitlines() == ["new.txt"]
    assert "old.txt" not in _git(repo, "status", "--porcelain=v2")
    assert git_status(repo)["totals"]["changed"] == 0


def test_git_commit_selected_handles_untracked_and_mixed_paths(tmp_path):
    from api.workspace_git import git_commit_selected

    repo = _init_repo(tmp_path / "repo")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)

    (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8")
    (repo / "new.txt").write_text("new\n", encoding="utf-8")

    committed = git_commit_selected(repo, "Commit mixed selected files", ["tracked.txt", "new.txt"])
    assert committed["ok"] is True
    assert set(_git(repo, "show", "--name-only", "--format=", "HEAD").splitlines()) == {
        "tracked.txt",
        "new.txt",
    }


def test_git_commit_selected_respects_nested_workspace_scope(tmp_path):
    from api.workspace_git import GitWorkspaceError, git_commit_selected

    repo = _init_repo(tmp_path / "repo")
    nested = repo / "app"
    nested.mkdir()
    (nested / "inside.txt").write_text("inside\n", encoding="utf-8")
    (repo / "outside.txt").write_text("outside\n", encoding="utf-8")
    _commit_all(repo)

    (nested / "inside.txt").write_text("inside\nchanged\n", encoding="utf-8")
    (repo / "outside.txt").write_text("outside\nchanged\n", encoding="utf-8")

    committed = git_commit_selected(nested, "Nested selected commit", ["inside.txt"])
    assert committed["paths"] == ["inside.txt"]
    assert _git(repo, "show", "--name-only", "--format=", "HEAD").splitlines() == ["app/inside.txt"]

    with pytest.raises(GitWorkspaceError) as outside:
        git_commit_selected(nested, "Outside", ["../outside.txt"])
    assert outside.value.code == "path_outside_workspace"


def test_git_commit_selected_rejects_conflicts_and_path_traversal(tmp_path):
    from api.workspace_git import GitWorkspaceError, git_commit_selected

    repo = _init_repo(tmp_path / "repo")
    (repo / "conflict.txt").write_text("base\n", encoding="utf-8")
    _commit_all(repo)
    _git(repo, "checkout", "-b", "side")
    (repo / "conflict.txt").write_text("side\n", encoding="utf-8")
    _commit_all(repo, "side")
    _git(repo, "checkout", "master")
    (repo / "conflict.txt").write_text("main\n", encoding="utf-8")
    _commit_all(repo, "main")
    subprocess.run(["git", "merge", "side"], cwd=repo, shell=False, text=True, capture_output=True, timeout=20)

    with pytest.raises(GitWorkspaceError) as conflict:
        git_commit_selected(repo, "Nope", ["conflict.txt"])
    assert conflict.value.code == "conflict"

    with pytest.raises(GitWorkspaceError) as traversal:
        git_commit_selected(repo, "Nope", ["../outside.txt"])
    assert traversal.value.code == "path_outside_workspace"


def test_selected_commit_message_prompt_uses_selected_diff(tmp_path):
    from api.workspace_git import selected_commit_message_prompt

    repo = _init_repo(tmp_path / "repo")
    (repo / "selected.txt").write_text("one\n", encoding="utf-8")
    (repo / "other.txt").write_text("alpha\n", encoding="utf-8")
    _commit_all(repo)
    (repo / "selected.txt").write_text("one\ntwo\n", encoding="utf-8")
    (repo / "other.txt").write_text("alpha\nbeta\n", encoding="utf-8")

    prompt = selected_commit_message_prompt(repo, ["selected.txt"])
    assert "selected.txt" in prompt["user_prompt"]
    assert "+two" in prompt["user_prompt"]
    assert "other.txt" not in prompt["user_prompt"]
    assert "beta" not in prompt["user_prompt"]


def test_staged_commit_message_prompt_uses_only_staged_diff(tmp_path):
    from api.workspace_git import (
        GitWorkspaceError,
        clean_generated_commit_message,
        staged_commit_message_prompt,
    )

    repo = _init_repo(tmp_path / "repo")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)

    (repo / "tracked.txt").write_text("one\nstaged\n", encoding="utf-8")
    _git(repo, "add", "tracked.txt")
    (repo / "tracked.txt").write_text("one\nstaged\nunstaged\n", encoding="utf-8")

    prompt = staged_commit_message_prompt(repo)
    assert prompt["truncated"] is False
    assert "tracked.txt" in prompt["user_prompt"]
    assert "+staged" in prompt["user_prompt"]
    assert "unstaged" not in prompt["user_prompt"]
    assert "Never mention AI, Cursor, Zed, agents" in prompt["system_prompt"]

    _git(repo, "restore", "--staged", "tracked.txt")
    with pytest.raises(GitWorkspaceError):
        staged_commit_message_prompt(repo)

    assert clean_generated_commit_message("```text\nSubject\n\n- Body\n```") == "Subject\n\n- Body"


def test_git_fetch_pull_and_push_with_upstream(tmp_path):
    from api.workspace_git import git_fetch, git_pull, git_push, git_status

    remote = _init_bare_repo(tmp_path / "remote.git")

    origin = _init_repo(tmp_path / "origin")
    (origin / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(origin)
    _git(origin, "remote", "add", "origin", str(remote))
    _git(origin, "push", "-u", "origin", "HEAD")
    _git(remote, "symbolic-ref", "HEAD", "refs/heads/master")

    clone = tmp_path / "clone"
    _git(tmp_path, "clone", str(remote), str(clone))
    _git(clone, "config", "user.email", "hermes-tests@example.invalid")
    _git(clone, "config", "user.name", "Hermes Tests")

    (origin / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8")
    _commit_all(origin, "Remote update")
    _git(origin, "push")

    fetched = git_fetch(clone)
    assert fetched["status"]["behind"] == 1

    pulled = git_pull(clone)
    assert pulled["status"]["behind"] == 0
    assert (clone / "tracked.txt").read_text(encoding="utf-8") == "one\ntwo\n"

    (clone / "tracked.txt").write_text("one\ntwo\nthree\n", encoding="utf-8")
    _git(clone, "add", "tracked.txt")
    _git(clone, "commit", "-m", "Local update")
    assert git_status(clone)["ahead"] == 1

    pushed = git_push(clone)
    assert pushed["status"]["ahead"] == 0


def test_git_branches_lists_local_remote_and_upstream(tmp_path):
    from api.workspace_git import git_branches

    remote = _init_bare_repo(tmp_path / "remote.git")
    origin = _init_repo(tmp_path / "origin")
    (origin / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(origin)
    _git(origin, "branch", "-M", "main")
    _git(origin, "remote", "add", "origin", str(remote))
    _git(origin, "push", "-u", "origin", "main")
    _git(remote, "symbolic-ref", "HEAD", "refs/heads/main")

    clone = tmp_path / "clone"
    _git(tmp_path, "clone", str(remote), str(clone))
    branches = git_branches(clone)
    assert branches["current"] == "main"
    assert branches["detached"] is False
    assert any(item["name"] == "main" and item["upstream"] == "origin/main" for item in branches["local"])
    main = next(item for item in branches["local"] if item["name"] == "main")
    assert "updated_relative" in main and "author" in main and "subject" in main
    assert any(item["name"] == "origin/main" for item in branches["remote"])
    assert not any(item["name"] == "origin" for item in branches["remote"])


def test_git_checkout_local_new_remote_dirty_and_invalid_refs(tmp_path):
    from api.workspace_git import GitWorkspaceError, git_branches, git_checkout

    remote = _init_bare_repo(tmp_path / "remote.git")
    origin = _init_repo(tmp_path / "origin")
    (origin / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(origin)
    _git(origin, "branch", "-M", "main")
    _git(origin, "remote", "add", "origin", str(remote))
    _git(origin, "push", "-u", "origin", "main")
    _git(remote, "symbolic-ref", "HEAD", "refs/heads/main")
    _git(origin, "checkout", "-b", "remote-feature")
    (origin / "remote.txt").write_text("remote\n", encoding="utf-8")
    _commit_all(origin, "remote feature")
    _git(origin, "push", "-u", "origin", "remote-feature")

    clone = tmp_path / "clone"
    _git(tmp_path, "clone", str(remote), str(clone))
    _git(clone, "config", "user.email", "hermes-tests@example.invalid")
    _git(clone, "config", "user.name", "Hermes Tests")

    created = git_checkout(clone, "main", "new", new_branch="local-work")
    assert created["current_branch"] == "local-work"
    assert git_branches(clone)["current"] == "local-work"

    switched = git_checkout(clone, "main", "local")
    assert switched["current_branch"] == "main"

    tracked = git_checkout(clone, "origin/remote-feature", "remote", new_branch="remote-feature", track=True)
    assert tracked["current_branch"] == "remote-feature"
    assert git_branches(clone)["upstream"] == "origin/remote-feature"

    (clone / "tracked.txt").write_text("dirty\n", encoding="utf-8")
    with pytest.raises(GitWorkspaceError) as dirty:
        git_checkout(clone, "main", "local")
    assert dirty.value.code == "dirty_worktree"
    _git(clone, "restore", "tracked.txt")

    with pytest.raises(GitWorkspaceError) as invalid:
        git_checkout(clone, "does-not-exist", "local")
    assert invalid.value.code in {"invalid_ref", "git_failed"}


def test_git_checkout_detached_requires_explicit_mode(tmp_path):
    from api.workspace_git import git_branches, git_checkout

    repo = _init_repo(tmp_path / "repo")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)
    sha = _git(repo, "rev-parse", "--short", "HEAD").strip()

    result = git_checkout(repo, sha, "detached")
    assert result["ok"] is True
    branches = git_branches(repo)
    assert branches["detached"] is True
    assert branches["current"] == sha


def test_git_stash_and_checkout_is_explicit(tmp_path):
    from api.workspace_git import git_stash_and_checkout, git_status

    repo = _init_repo(tmp_path / "repo")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)
    _git(repo, "checkout", "-b", "target")
    _git(repo, "checkout", "master")
    (repo / "tracked.txt").write_text("dirty\n", encoding="utf-8")

    result = git_stash_and_checkout(repo, "target", "local")
    assert result["ok"] is True
    assert result["stashed"] is True
    assert result["stash_name"].startswith("hermes-webui branch switch")
    assert result["current_branch"] == "target"
    assert git_status(repo)["totals"]["changed"] == 0
    assert "hermes-webui branch switch to target" in _git(repo, "stash", "list")


def test_git_stash_and_checkout_restores_branch_changes_when_returning(tmp_path):
    from api.workspace_git import git_stash_and_checkout, git_status

    repo = _init_repo(tmp_path / "repo")
    _git(repo, "branch", "-M", "main")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)
    _git(repo, "checkout", "-b", "feature")
    _git(repo, "checkout", "main")

    (repo / "tracked.txt").write_text("main dirty\n", encoding="utf-8")
    (repo / "main-only.txt").write_text("untracked on main\n", encoding="utf-8")

    to_feature = git_stash_and_checkout(repo, "feature", "local")
    assert to_feature["ok"] is True
    assert to_feature["stashed"] is True
    assert to_feature["current_branch"] == "feature"
    assert git_status(repo)["totals"]["changed"] == 0
    assert not (repo / "main-only.txt").exists()

    (repo / "feature-only.txt").write_text("untracked on feature\n", encoding="utf-8")
    to_main = git_stash_and_checkout(repo, "main", "local")

    assert to_main["ok"] is True
    assert to_main["stashed"] is True
    assert to_main["current_branch"] == "main"
    assert to_main["restored_stash"]["branch"] == "main"
    assert (repo / "tracked.txt").read_text(encoding="utf-8") == "main dirty\n"
    assert (repo / "main-only.txt").read_text(encoding="utf-8") == "untracked on main\n"
    assert not (repo / "feature-only.txt").exists()
    stash_list = _git(repo, "stash", "list")
    assert "On main: hermes-webui branch switch" not in stash_list
    assert "On feature: hermes-webui branch switch" in stash_list


def test_git_stash_and_checkout_reports_restore_conflicts_without_dropping_stash(tmp_path):
    from api.workspace_git import git_stash_and_checkout

    repo = _init_repo(tmp_path / "repo")
    _git(repo, "branch", "-M", "main")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)
    _git(repo, "checkout", "-b", "feature")
    _git(repo, "checkout", "main")
    (repo / "tracked.txt").write_text("main dirty\n", encoding="utf-8")

    git_stash_and_checkout(repo, "feature", "local")
    _git(repo, "checkout", "main")
    (repo / "tracked.txt").write_text("main changed while parked\n", encoding="utf-8")
    _commit_all(repo, "advance main")
    _git(repo, "checkout", "feature")

    result = git_stash_and_checkout(repo, "main", "local")

    assert result["ok"] is True
    assert result["current_branch"] == "main"
    assert result["restore_failed"] is True
    assert result["restore_stash"]["branch"] == "main"
    assert "On main: hermes-webui branch switch" in _git(repo, "stash", "list")


def test_git_stash_checkout_validates_before_stashing(tmp_path):
    from api.workspace_git import GitWorkspaceError, git_stash_and_checkout

    repo = _init_repo(tmp_path / "repo")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)
    (repo / "tracked.txt").write_text("dirty\n", encoding="utf-8")

    with pytest.raises(GitWorkspaceError) as invalid:
        git_stash_and_checkout(repo, "missing-branch", "local")

    assert invalid.value.code == "invalid_ref"
    assert "M tracked.txt" in _git(repo, "status", "--porcelain")
    assert _git(repo, "stash", "list") == ""


def test_git_routes_status_diff_stage_unstage_discard_commit(cleanup_test_sessions):
    sid, base_ws = _make_session(cleanup_test_sessions)
    repo = base_ws / f"git-route-{uuid.uuid4().hex[:8]}"
    _init_repo(repo)
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)

    _post("/api/session/update", {"session_id": sid, "workspace": str(repo), "model": "openai/gpt-5.4-mini"})
    (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8")

    status, code = _get(f"/api/git/status?session_id={sid}")
    assert code == 200
    assert status["git"]["totals"]["unstaged"] == 1

    diff, code = _get(
        f"/api/git/diff?session_id={sid}&path={urllib.parse.quote('tracked.txt')}&kind=unstaged"
    )
    assert code == 200
    assert "+two" in diff["diff"]["diff"]

    staged, code = _post("/api/git/stage", {"session_id": sid, "paths": ["tracked.txt"]})
    assert code == 200 and staged["git"]["totals"]["staged"] == 1

    unstaged, code = _post("/api/git/unstage", {"session_id": sid, "paths": ["tracked.txt"]})
    assert code == 200 and unstaged["git"]["totals"]["unstaged"] == 1

    discarded, code = _post("/api/git/discard", {"session_id": sid, "paths": ["tracked.txt"]})
    assert code == 200 and discarded["git"]["totals"]["changed"] == 0

    (repo / "tracked.txt").write_text("one\nthree\n", encoding="utf-8")
    _post("/api/git/stage", {"session_id": sid, "paths": ["tracked.txt"]})
    committed, code = _post("/api/git/commit", {"session_id": sid, "message": "Route commit"})
    assert code == 200
    assert committed["ok"] is True
    assert committed["status"]["totals"]["changed"] == 0


def test_git_routes_branches_and_checkout(cleanup_test_sessions):
    sid, base_ws = _make_session(cleanup_test_sessions)
    repo = base_ws / f"git-branch-route-{uuid.uuid4().hex[:8]}"
    _init_repo(repo)
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)
    _git(repo, "branch", "-M", "main")
    _git(repo, "checkout", "-b", "feature")
    _git(repo, "checkout", "main")

    _post("/api/session/update", {"session_id": sid, "workspace": str(repo), "model": "openai/gpt-5.4-mini"})
    branches, code = _get(f"/api/git/branches?session_id={sid}")
    assert code == 200
    assert branches["branches"]["current"] == "main"
    assert any(item["name"] == "feature" for item in branches["branches"]["local"])

    checked, code = _post(
        "/api/git/checkout",
        {"session_id": sid, "ref": "feature", "mode": "local", "dirty_mode": "block"},
    )
    assert code == 200
    assert checked["ok"] is True
    assert checked["current_branch"] == "feature"
    assert checked["git"]["branch"] == "feature"


def test_git_routes_selected_commit_and_structured_error(cleanup_test_sessions):
    sid, base_ws = _make_session(cleanup_test_sessions)
    repo = base_ws / f"git-selected-route-{uuid.uuid4().hex[:8]}"
    _init_repo(repo)
    (repo / "selected.txt").write_text("one\n", encoding="utf-8")
    (repo / "other.txt").write_text("alpha\n", encoding="utf-8")
    _commit_all(repo)

    _post("/api/session/update", {"session_id": sid, "workspace": str(repo), "model": "openai/gpt-5.4-mini"})
    (repo / "selected.txt").write_text("one\ntwo\n", encoding="utf-8")
    (repo / "other.txt").write_text("alpha\nbeta\n", encoding="utf-8")
    _git(repo, "add", "other.txt")

    bad, code = _post("/api/git/commit-selected", {"session_id": sid, "message": "Bad", "paths": ["../x"]})
    assert code == 400
    assert bad["code"] == "path_outside_workspace"

    committed, code = _post(
        "/api/git/commit-selected",
        {"session_id": sid, "message": "Selected route commit", "paths": ["selected.txt"]},
    )
    assert code == 200
    assert committed["ok"] is True
    assert committed["paths"] == ["selected.txt"]
    assert _git(repo, "show", "--name-only", "--format=", "HEAD").splitlines() == ["selected.txt"]


def test_git_discard_untracked_delete_uses_anchored_unlink_after_validation_race(tmp_path, monkeypatch):
    import os
    import shutil

    import api.workspace_git as workspace_git
    from api.workspace import safe_resolve_ws as real_safe_resolve_ws

    repo = _init_repo(tmp_path / "repo")
    (repo / "tracked.txt").write_text("tracked\n", encoding="utf-8")
    _commit_all(repo)

    (repo / "d").mkdir()
    (repo / "d" / "f").write_text("workspace untracked\n", encoding="utf-8")
    outside = tmp_path / "outside"
    outside.mkdir()
    victim = outside / "f"
    victim.write_text("outside victim\n", encoding="utf-8")

    state = {"calls": 0, "swapped": False}

    def racing_safe_resolve(root, requested):
        target = real_safe_resolve_ws(root, requested)
        if requested == "d/f":
            state["calls"] += 1
        # git_discard validates once for the Git pathspec and once immediately
        # before deletion. Race the second validation-to-use window.
        if requested == "d/f" and state["calls"] == 2 and not state["swapped"]:
            shutil.rmtree(repo / "d")
            os.symlink(outside, repo / "d")
            state["swapped"] = True
        return target

    monkeypatch.setattr(workspace_git, "safe_resolve_ws", racing_safe_resolve)

    with pytest.raises(ValueError, match="Path traversal blocked"):
        workspace_git.git_discard(repo, ["d/f"], delete_untracked=True)

    assert state["swapped"] is True
    assert victim.exists()
    assert victim.read_text(encoding="utf-8") == "outside victim\n"


def test_git_status_ignores_repo_local_fsmonitor_command(tmp_path):
    import os
    import sys

    if os.name == "nt":
        pytest.skip("executable fsmonitor helper setup is POSIX-only")

    from api.workspace_git import git_status

    repo = _init_repo(tmp_path / "repo")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)
    marker = tmp_path / "fsmonitor-ran"
    helper = tmp_path / "fsmonitor_helper.py"
    helper.write_text(
        "#!/usr/bin/env python3\n"
        "import pathlib, sys\n"
        "pathlib.Path(sys.argv[1]).write_text('fsmonitor executed', encoding='utf-8')\n"
        "print('')\n",
        encoding="utf-8",
    )
    helper.chmod(0o755)
    _git(repo, "config", "core.fsmonitor", f"{sys.executable} {helper} {marker}")

    status = git_status(repo)

    assert status["is_git"] is True
    assert not marker.exists()


def test_git_fetch_blocks_repo_local_ext_transport_execution(tmp_path):
    import os
    import sys

    if os.name == "nt":
        pytest.skip("ext transport helper setup is POSIX-only")

    from api.workspace_git import GitWorkspaceError, git_fetch

    repo = _init_repo(tmp_path / "repo")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)
    marker = tmp_path / "ext-transport-ran"
    helper = tmp_path / "ext_helper.py"
    helper.write_text(
        "#!/usr/bin/env python3\n"
        "import pathlib, sys\n"
        "pathlib.Path(sys.argv[1]).write_text('ext executed: ' + ' '.join(sys.argv[2:]), encoding='utf-8')\n"
        "sys.exit(1)\n",
        encoding="utf-8",
    )
    helper.chmod(0o755)
    _git(repo, "config", "protocol.ext.allow", "always")
    _git(repo, "remote", "add", "origin", f"ext::{sys.executable} {helper} {marker} %S foo")

    with pytest.raises(GitWorkspaceError) as exc:
        git_fetch(repo)

    assert exc.value.code == "git_failed"
    assert not marker.exists()


def test_git_fetch_blocks_repo_local_credential_helper_execution(tmp_path):
    import os
    import sys

    if os.name == "nt":
        pytest.skip("executable credential helper setup is POSIX-only")

    from api.workspace_git import GitWorkspaceError, git_fetch

    class AuthRequiredHandler(BaseHTTPRequestHandler):
        def do_GET(self):
            self.send_response(401)
            self.send_header("WWW-Authenticate", 'Basic realm="hermes-test"')
            self.end_headers()

        def log_message(self, format, *args):
            del format, args

    repo = _init_repo(tmp_path / "repo")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)
    marker = tmp_path / "credential-helper-ran"
    helper = tmp_path / "credential_helper.py"
    helper.write_text(
        "#!/usr/bin/env python3\n"
        "import pathlib, sys\n"
        "pathlib.Path(sys.argv[1]).write_text('credential helper executed', encoding='utf-8')\n",
        encoding="utf-8",
    )
    helper.chmod(0o755)

    server = ThreadingHTTPServer(("127.0.0.1", 0), AuthRequiredHandler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        url = f"http://127.0.0.1:{server.server_port}/repo.git"
        _git(repo, "remote", "add", "origin", url)
        _git(repo, "config", "credential.helper", f"!{sys.executable} {helper} {marker}")

        with pytest.raises(GitWorkspaceError) as exc:
            git_fetch(repo)
    finally:
        server.shutdown()
        thread.join(timeout=5)
        server.server_close()

    assert exc.value.code == "auth_failed"
    assert not marker.exists()


def test_git_fetch_blocks_repo_local_askpass_execution(tmp_path):
    import os
    import sys

    if os.name == "nt":
        pytest.skip("executable askpass helper setup is POSIX-only")

    from api.workspace_git import GitWorkspaceError, git_fetch

    class AuthRequiredHandler(BaseHTTPRequestHandler):
        def do_GET(self):
            self.send_response(401)
            self.send_header("WWW-Authenticate", 'Basic realm="hermes-test"')
            self.end_headers()

        def log_message(self, format, *args):
            del format, args

    repo = _init_repo(tmp_path / "repo")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)
    marker = tmp_path / "askpass-ran"
    helper = tmp_path / "askpass_helper.py"
    helper.write_text(
        "#!/usr/bin/env python3\n"
        "import pathlib, sys\n"
        "pathlib.Path(sys.argv[1]).write_text('askpass executed', encoding='utf-8')\n"
        "print('pw')\n",
        encoding="utf-8",
    )
    helper.chmod(0o755)

    server = ThreadingHTTPServer(("127.0.0.1", 0), AuthRequiredHandler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        url = f"http://127.0.0.1:{server.server_port}/repo.git"
        _git(repo, "remote", "add", "origin", url)
        _git(repo, "config", "core.askPass", f"{sys.executable} {helper} {marker}")

        with pytest.raises(GitWorkspaceError) as exc:
            git_fetch(repo)
    finally:
        server.shutdown()
        thread.join(timeout=5)
        server.server_close()

    assert exc.value.code == "auth_failed"
    assert not marker.exists()


def test_git_env_scrub_removes_redirecting_vars_and_preserves_temp_index(monkeypatch):
    from api.workspace_git import _clean_git_env

    monkeypatch.setenv("GIT_DIR", "/tmp/evil-git-dir")
    monkeypatch.setenv("GIT_WORK_TREE", "/tmp/evil-work-tree")
    monkeypatch.setenv("GIT_CONFIG_GLOBAL", "/tmp/evil-config")
    monkeypatch.setenv("GIT_CONFIG_SYSTEM", "/tmp/evil-system-config")
    monkeypatch.setenv("GIT_CONFIG_COUNT", "1")
    monkeypatch.setenv("GIT_CONFIG_KEY_0", "core.sshCommand")
    monkeypatch.setenv("GIT_CONFIG_VALUE_0", "ssh -i /tmp/evil-key")
    monkeypatch.setenv("GIT_CONFIG_PARAMETERS", "'core.sshCommand=ssh -i /tmp/evil-key'")
    monkeypatch.setenv("GIT_ASKPASS", "/tmp/evil-askpass")
    monkeypatch.setenv("SSH_ASKPASS", "/tmp/evil-ssh-askpass")
    monkeypatch.setenv("GIT_SSH", "/tmp/evil-ssh")
    monkeypatch.setenv("GIT_SSH_COMMAND", "ssh -i /tmp/evil-key")
    monkeypatch.setenv("GIT_TERMINAL_PROMPT", "1")

    env = _clean_git_env({"GIT_INDEX_FILE": "/tmp/hermes-index"})

    assert "GIT_DIR" not in env
    assert "GIT_WORK_TREE" not in env
    assert "GIT_CONFIG_GLOBAL" not in env
    assert "GIT_CONFIG_SYSTEM" not in env
    assert "GIT_CONFIG_COUNT" not in env
    assert "GIT_CONFIG_KEY_0" not in env
    assert "GIT_CONFIG_VALUE_0" not in env
    assert "GIT_CONFIG_PARAMETERS" not in env
    assert "GIT_ASKPASS" not in env
    assert "SSH_ASKPASS" not in env
    assert "GIT_SSH" not in env
    assert "GIT_SSH_COMMAND" not in env
    assert env["GIT_TERMINAL_PROMPT"] == "0"
    assert env["GIT_INDEX_FILE"] == "/tmp/hermes-index"


def test_git_error_classifier_identifies_non_fast_forward_push():
    from api.workspace_git import _classify_git_error

    assert _classify_git_error("Updates were rejected", ["push"]) == "non_fast_forward"
    assert _classify_git_error("non-fast-forward", ["push"]) == "non_fast_forward"
    assert _classify_git_error("fetch first", ["push"]) == "non_fast_forward"


def test_git_commit_hook_failure_returns_hook_failed_code(tmp_path):
    from api.workspace_git import GitWorkspaceError, git_commit, git_stage

    repo = _init_repo(tmp_path / "repo")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)
    hook = repo / ".git" / "hooks" / "pre-commit"
    hook.write_text("#!/bin/sh\necho hook blocked >&2\nexit 1\n", encoding="utf-8")
    hook.chmod(0o755)

    (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8")
    git_stage(repo, ["tracked.txt"])

    with pytest.raises(GitWorkspaceError) as exc:
        git_commit(repo, "Hook should fail")
    assert exc.value.code == "hook_failed"


def test_destructive_workspace_git_flag_defaults_off_and_accepts_truthy(monkeypatch):
    from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, workspace_git_destructive_enabled

    monkeypatch.delenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, raising=False)
    assert workspace_git_destructive_enabled() is False

    monkeypatch.setenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, "1")
    assert workspace_git_destructive_enabled() is True

    monkeypatch.setenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, "true")
    assert workspace_git_destructive_enabled() is True


def test_git_active_stream_lock_detection(monkeypatch):
    from api import routes
    from api.config import STREAMS, STREAMS_LOCK

    session = types.SimpleNamespace(active_stream_id="stream-git-lock-test")
    with STREAMS_LOCK:
        STREAMS[session.active_stream_id] = object()
    try:
        assert routes._git_locked_by_active_stream(session) is True
    finally:
        with STREAMS_LOCK:
            STREAMS.pop(session.active_stream_id, None)

    assert routes._git_locked_by_active_stream(session) is False


def test_git_commit_route_rejects_active_stream(monkeypatch, tmp_path):
    from api import routes
    from api.config import STREAMS, STREAMS_LOCK
    from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV

    # Enable destructive ops for this in-process test — conftest.py sets the env
    # var on the test_server subprocess env block, but this test calls
    # _handle_git_commit() directly in the pytest process, which inherits
    # the default-OFF setting. Without this monkeypatch, the destructive-mode
    # gate fires first (403) before the active-stream check (409) can run.
    monkeypatch.setenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, "1")

    repo = _init_repo(tmp_path / "repo")
    (repo / "tracked.txt").write_text("one\n", encoding="utf-8")
    _commit_all(repo)
    _git(repo, "add", "tracked.txt")
    session = types.SimpleNamespace(
        session_id="sid-active-git",
        workspace=str(repo),
        active_stream_id="stream-active-git",
    )

    monkeypatch.setattr(routes, "get_session", lambda sid: session)
    handler = _CaptureHandler()
    with STREAMS_LOCK:
        STREAMS[session.active_stream_id] = object()
    try:
        assert routes._handle_git_commit(
            handler,
            {"session_id": session.session_id, "message": "Should be blocked"},
        ) is True
    finally:
        with STREAMS_LOCK:
            STREAMS.pop(session.active_stream_id, None)

    assert handler.status == 409
    payload = handler.payload()
    assert payload["code"] == "active_stream"
    assert "active" in payload["error"].lower()