134 lines
4.5 KiB
Python
134 lines
4.5 KiB
Python
"""Unit tests for dashboard.handlers.git — _run_git helper + endpoint shapes."""
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
DASH = PROJECT_ROOT / "dashboard"
|
|
|
|
if str(DASH) not in sys.path:
|
|
sys.path.insert(0, str(DASH))
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def git_module():
|
|
from handlers import git as _g # type: ignore
|
|
return _g
|
|
|
|
|
|
@pytest.fixture
|
|
def handler(git_module):
|
|
"""A bare instance of GitHandlers with a captured send_json."""
|
|
class _Stub(git_module.GitHandlers):
|
|
def __init__(self):
|
|
self.captured = None
|
|
self.captured_code = None
|
|
|
|
def send_json(self, data, code=200):
|
|
self.captured = data
|
|
self.captured_code = code
|
|
|
|
return _Stub()
|
|
|
|
|
|
def test_run_git_is_a_subprocess_call(handler, git_module, tmp_path):
|
|
"""_run_git must use subprocess.run with the supplied cwd + timeout."""
|
|
with patch.object(git_module.subprocess, "run") as mock_run:
|
|
mock_run.return_value = subprocess.CompletedProcess(
|
|
args=["git", "status"], returncode=0, stdout="clean\n", stderr=""
|
|
)
|
|
result = handler._run_git(tmp_path, ["status"], timeout=3)
|
|
|
|
mock_run.assert_called_once()
|
|
args, kwargs = mock_run.call_args
|
|
assert args[0] == ["git", "status"]
|
|
assert kwargs["cwd"] == str(tmp_path)
|
|
assert kwargs["timeout"] == 3
|
|
assert kwargs["capture_output"] is True
|
|
assert kwargs["text"] is True
|
|
assert result.stdout == "clean\n"
|
|
|
|
|
|
def test_run_git_default_timeout_is_5(handler, git_module, tmp_path):
|
|
with patch.object(git_module.subprocess, "run") as mock_run:
|
|
mock_run.return_value = subprocess.CompletedProcess(
|
|
args=["git", "log"], returncode=0, stdout="", stderr=""
|
|
)
|
|
handler._run_git(tmp_path, ["log"])
|
|
_, kwargs = mock_run.call_args
|
|
assert kwargs["timeout"] == 5
|
|
|
|
|
|
def test_legacy_git_commit_handler_is_gone(git_module):
|
|
"""/api/git-commit was consolidated into /api/eco/git-commit."""
|
|
assert not hasattr(git_module.GitHandlers, "handle_git_commit")
|
|
|
|
|
|
def test_git_status_uses_echo_core_workspace(handler, git_module):
|
|
"""handle_git_status must target constants.GIT_WORKSPACE (echo-core), not clawd."""
|
|
captured_workspaces = []
|
|
|
|
def fake_run(_self, workspace, args, timeout=5):
|
|
captured_workspaces.append(workspace)
|
|
stdout_map = {
|
|
("branch", "--show-current"): "master\n",
|
|
("log", "-1", "--format=%h|%s|%cr"): "abc1234|test commit|1 hour ago\n",
|
|
("status", "--short"): "",
|
|
("diff", "--stat", "--cached"): "",
|
|
("diff", "--stat"): "",
|
|
}
|
|
return subprocess.CompletedProcess(
|
|
args=["git", *args],
|
|
returncode=0,
|
|
stdout=stdout_map.get(tuple(args), ""),
|
|
stderr="",
|
|
)
|
|
|
|
with patch.object(git_module.GitHandlers, "_run_git", fake_run):
|
|
handler.handle_git_status()
|
|
|
|
import constants # type: ignore
|
|
assert all(w == constants.GIT_WORKSPACE for w in captured_workspaces)
|
|
assert handler.captured is not None
|
|
assert handler.captured["branch"] == "master"
|
|
|
|
|
|
def test_git_status_parses_uncommitted_paths(handler, git_module):
|
|
def fake_run(_self, workspace, args, timeout=5):
|
|
if args == ["status", "--short"]:
|
|
return subprocess.CompletedProcess(
|
|
args=["git"] + args, returncode=0,
|
|
stdout=" M src/foo.py\n?? new_file.txt\n", stderr="",
|
|
)
|
|
if args[:1] == ["branch"]:
|
|
return subprocess.CompletedProcess(
|
|
args=["git"] + args, returncode=0, stdout="feat-x\n", stderr="",
|
|
)
|
|
if args[:1] == ["log"]:
|
|
return subprocess.CompletedProcess(
|
|
args=["git"] + args, returncode=0,
|
|
stdout="deadbee|msg|2 minutes ago\n", stderr="",
|
|
)
|
|
return subprocess.CompletedProcess(
|
|
args=["git"] + args, returncode=0, stdout="", stderr="",
|
|
)
|
|
|
|
with patch.object(git_module.GitHandlers, "_run_git", fake_run):
|
|
handler.handle_git_status()
|
|
|
|
data = handler.captured
|
|
assert data is not None
|
|
assert data["uncommittedCount"] == 2
|
|
assert data["clean"] is False
|
|
paths = [u["path"] for u in data["uncommittedParsed"]]
|
|
statuses = {u["status"] for u in data["uncommittedParsed"]}
|
|
assert "src/foo.py" in paths
|
|
assert "new_file.txt" in paths
|
|
assert "M" in statuses
|
|
assert "??" in statuses
|