busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
18b9f2e94e2dcd6c547d32f6576ca753016a111c
ss-tools/backend/tests/services/git/test_git_status.py
busya 8c10632494 feat(semantic): curator-driven protocol hardening — decision memory + relation repair 
- Add @RATIONALE/@REJECTED to 103+ C4/C5 contracts across backend core, services, API routes, and frontend models
- Fix 109 unresolved @RELATION edges (Auth.*, SupersetClient.*, AgentChat.*, ADR cross-refs)
- Add 13 @ingroup tags for DSA/HCA attention grouping
- Repair 29 stale graph edges via index rebuild
- Update .kilo agent prompts and skills for GRACE-Poly v2.6 compliance
- Git integration: merge routes, branch lifecycle, remote providers, UX components
- 0 broken anchor pairs, index rebuilt with 0 parse warnings
2026-07-02 08:53:19 +03:00

515 lines
20 KiB
Python
Raw Blame History

# #region Test.GitService.Status [C:3] [TYPE Module] [SEMANTICS test,git,status,diff,history]
# @BRIEF Tests for GitServiceStatusMixin — parse porcelain, get_status, get_diff, get_commit_history.
# @RELATION BINDS_TO -> [GitServiceStatusMixin]
# @TEST_EDGE: empty_porcelain -> no output returns empty lists
# @TEST_EDGE: untracked_files -> ?? prefix parsed as untracked
# @TEST_EDGE: staged_and_modified -> XY prefix correctly split
# @TEST_EDGE: renamed_files -> -> arrow parsed as rename target
# @TEST_EDGE: git_status_failure -> exception caught, warning logged
# @TEST_EDGE: no_commits -> has_commits=false, branch name still returned
# @TEST_EDGE: divergent_branch -> ahead>0 AND behind>0 → DIVERGED
# @TEST_EDGE: ahead_remote -> ahead>0, behind=0 → AHEAD_REMOTE
# @TEST_EDGE: behind_remote -> ahead=0, behind>0 → BEHIND_REMOTE
# @TEST_EDGE: dirty_repo -> changes present → CHANGES
# @TEST_EDGE: clean_synced -> no changes, no ahead/behind → SYNCED
# @TEST_EDGE: diff_with_file -> file_path passed to git diff
# @TEST_EDGE: diff_staged -> --staged flag passed
# @TEST_EDGE: commit_history_empty -> no heads, no remotes returns []
# @TEST_EDGE: commit_history_exception -> exception logged, returns []
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "src"))
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from src.services.git._status import GitServiceStatusMixin
# ── Helper: create a testable instance of the mixin ──
class TestableGitStatus(GitServiceStatusMixin):
"""Concrete test class providing the minimum _locked and get_repo stubs.
_locked is a no-op context manager. get_repo returns a pre-set mock.
"""
def __init__(self, mock_repo=None):
self._mock_repo = mock_repo or MagicMock()
self._lock_called = False
async def get_repo(self, dashboard_id):
return self._mock_repo
def _locked(self, dashboard_id):
import contextlib
@contextlib.contextmanager
def _lock():
self._lock_called = True
yield
return _lock()
# ── _parse_status_porcelain ──
class TestParseStatusPorcelain:
"""_parse_status_porcelain — git status --porcelain parser."""
# #region test_porcelain_empty_output [C:2] [TYPE Function]
def test_porcelain_empty_output(self):
"""Empty output → all lists empty."""
svc = TestableGitStatus()
repo = MagicMock()
repo.git.status.return_value = ""
staged, modified, untracked = svc._parse_status_porcelain(repo)
assert staged == []
assert modified == []
assert untracked == []
# #endregion test_porcelain_empty_output
# #region test_porcelain_untracked_files [C:2] [TYPE Function]
def test_porcelain_untracked_files(self):
"""?? prefixed lines appear in untracked list."""
svc = TestableGitStatus()
repo = MagicMock()
repo.git.status.return_value = "?? new_file.txt\n?? untracked_dir/\n"
staged, modified, untracked = svc._parse_status_porcelain(repo)
assert untracked == ["new_file.txt", "untracked_dir/"]
assert staged == []
assert modified == []
# #endregion test_porcelain_untracked_files
# #region test_porcelain_staged [C:2] [TYPE Function]
def test_porcelain_staged(self):
"""XY where X!=space → staged."""
svc = TestableGitStatus()
repo = MagicMock()
repo.git.status.return_value = "M staged.txt\nA added.py\n"
staged, modified, untracked = svc._parse_status_porcelain(repo)
assert staged == ["staged.txt", "added.py"]
assert modified == []
# #endregion test_porcelain_staged
# #region test_porcelain_modified [C:2] [TYPE Function]
def test_porcelain_modified(self):
"""XY where Y!=space → modified."""
svc = TestableGitStatus()
repo = MagicMock()
repo.git.status.return_value = " M modified.rs\n"
staged, modified, untracked = svc._parse_status_porcelain(repo)
assert modified == ["modified.rs"]
assert staged == []
# #endregion test_porcelain_modified
# #region test_porcelain_staged_and_modified [C:2] [TYPE Function]
def test_porcelain_staged_and_modified(self):
"""MM → both staged and modified."""
svc = TestableGitStatus()
repo = MagicMock()
repo.git.status.return_value = "MM both.txt\n"
staged, modified, untracked = svc._parse_status_porcelain(repo)
assert "both.txt" in staged
assert "both.txt" in modified
# #endregion test_porcelain_staged_and_modified
# #region test_porcelain_renamed [C:2] [TYPE Function]
def test_porcelain_renamed(self):
"""R old -> new → staged shows new path."""
svc = TestableGitStatus()
repo = MagicMock()
repo.git.status.return_value = "R old_name.py -> new_name.py\n"
staged, modified, untracked = svc._parse_status_porcelain(repo)
assert "new_name.py" in staged
assert "old_name.py" not in str(staged)
# #endregion test_porcelain_renamed
# #region test_porcelain_exclude_intent_to_add [C:2] [TYPE Function]
def test_porcelain_exclude_intent_to_add(self):
"""!! lines (intent-to-add/assume-unchanged) are skipped."""
svc = TestableGitStatus()
repo = MagicMock()
repo.git.status.return_value = "!! ignored_pattern\n"
_, _, untracked = svc._parse_status_porcelain(repo)
assert untracked == []
# #endregion test_porcelain_exclude_intent_to_add
# #region test_porcelain_short_line [C:2] [TYPE Function]
def test_porcelain_short_line(self):
"""Line with <3 chars is skipped (e.g., empty/broken output)."""
svc = TestableGitStatus()
repo = MagicMock()
repo.git.status.return_value = "A\nBB\n M file.txt\n"
staged, modified, untracked = svc._parse_status_porcelain(repo)
# "A\n" and "BB\n" are skipped (< 3 chars). Only " M file.txt" is parsed.
assert "file.txt" in modified
assert len(staged) == 0
# #endregion test_porcelain_short_line
# #region test_porcelain_git_failure [C:2] [TYPE Function]
def test_porcelain_git_failure(self):
"""Exception from git.status returns empty lists, no crash."""
svc = TestableGitStatus()
repo = MagicMock()
repo.git.status.side_effect = Exception("git not available")
staged, modified, untracked = svc._parse_status_porcelain(repo)
assert staged == []
assert modified == []
assert untracked == []
# #endregion test_porcelain_git_failure
# ── get_status ──
class TestGetStatus:
"""get_status — full repository status computation."""
# #region test_get_status_no_commits [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_get_status_no_commits(self):
"""Repository with no commits: has_commits=false, branch returned."""
from unittest.mock import PropertyMock
repo = MagicMock()
head = MagicMock()
# Accessing repo.head.commit as a PROPERTY raises ValueError (like real GitPython)
type(head).commit = PropertyMock(side_effect=ValueError("no commits"))
repo.head = head
repo.active_branch.name = "main"
repo.active_branch.tracking_branch.return_value = None
svc = TestableGitStatus(repo)
result = await svc.get_status(42)
assert svc._lock_called is True
assert result["current_branch"] == "main"
assert result["has_upstream"] is False
assert result["last_commit_hash"] is None
assert result["last_commit_author"] is None
# #endregion test_get_status_no_commits
# #region test_get_status_diverged [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_get_status_diverged(self):
"""ahead>0 and behind>0 → DIVERGED."""
repo = MagicMock()
head = MagicMock()
head.commit = MagicMock()
repo.head = head
repo.active_branch.name = "feature"
tracking = MagicMock()
tracking.name = "origin/feature"
repo.active_branch.tracking_branch.return_value = tracking
# iter_commits for ahead: first call returns [1,2], second returns [3,4,5]
repo.iter_commits.side_effect = [
[MagicMock(), MagicMock()], # ahead: 2 commits
[MagicMock(), MagicMock(), MagicMock()], # behind: 3 commits
]
repo.git.status.return_value = ""
svc = TestableGitStatus(repo)
result = await svc.get_status(42)
assert result["sync_state"] == "DIVERGED"
assert result["ahead_count"] == 2
assert result["behind_count"] == 3
assert result["is_diverged"] is True
# #endregion test_get_status_diverged
# #region test_get_status_ahead [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_get_status_ahead(self):
"""ahead>0, behind=0 → AHEAD_REMOTE."""
repo = MagicMock()
head = MagicMock()
head.commit = MagicMock()
repo.head = head
repo.active_branch.name = "feature"
tracking = MagicMock()
repo.active_branch.tracking_branch.return_value = tracking
repo.iter_commits.side_effect = [
[MagicMock()], # ahead: 1 commit
[], # behind: 0
]
repo.git.status.return_value = ""
svc = TestableGitStatus(repo)
result = await svc.get_status(42)
assert result["sync_state"] == "AHEAD_REMOTE"
assert result["ahead_count"] == 1
assert result["behind_count"] == 0
# #endregion test_get_status_ahead
# #region test_get_status_behind [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_get_status_behind(self):
"""ahead=0, behind>0 → BEHIND_REMOTE."""
repo = MagicMock()
head = MagicMock()
head.commit = MagicMock(
hexsha="abcdef1234567890",
message="Update BI dashboard\n\nDetails",
committed_date=1710000000,
)
head.commit.author.name = "BI Analyst"
repo.head = head
repo.active_branch.name = "main"
tracking = MagicMock()
repo.active_branch.tracking_branch.return_value = tracking
repo.iter_commits.side_effect = [
[], # ahead: 0
[MagicMock()], # behind: 1
]
repo.git.status.return_value = ""
svc = TestableGitStatus(repo)
result = await svc.get_status(42)
assert result["sync_state"] == "BEHIND_REMOTE"
assert result["ahead_count"] == 0
assert result["behind_count"] == 1
# #endregion test_get_status_behind
# #region test_get_status_dirty [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_get_status_dirty(self):
"""Unstaged changes present → CHANGES."""
repo = MagicMock()
head = MagicMock()
head.commit = MagicMock()
repo.head = head
repo.active_branch.name = "main"
repo.active_branch.tracking_branch.return_value = None
repo.git.status.return_value = " M modified.txt\n?? new.txt\n"
svc = TestableGitStatus(repo)
result = await svc.get_status(42)
assert result["sync_state"] == "CHANGES"
assert result["is_dirty"] is True
assert "modified.txt" in result["modified_files"]
assert "new.txt" in result["untracked_files"]
# #endregion test_get_status_dirty
# #region test_get_status_synced [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_get_status_synced(self):
"""Clean, no divergence → SYNCED."""
repo = MagicMock()
head = MagicMock()
head.commit = MagicMock(
hexsha="abcdef1234567890",
message="Update BI dashboard\n\nDetails",
committed_date=1710000000,
)
head.commit.author.name = "BI Analyst"
repo.head = head
repo.active_branch.name = "main"
repo.active_branch.tracking_branch.return_value = None
repo.git.status.return_value = ""
svc = TestableGitStatus(repo)
result = await svc.get_status(42)
assert result["sync_state"] == "SYNCED"
assert result["is_dirty"] is False
assert result["last_commit_hash"] == "abcdef1234567890"
assert result["last_commit_message"] == "Update BI dashboard\n\nDetails"
assert result["last_commit_author"] == "BI Analyst"
assert result["last_commit_date"].startswith("2024-03-09T")
# #endregion test_get_status_synced
# #region test_get_status_tracking_exception [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_get_status_tracking_exception(self):
"""Exception in tracking_branch() → has_upstream=false."""
repo = MagicMock()
head = MagicMock()
head.commit = MagicMock()
repo.head = head
repo.active_branch.name = "main"
repo.active_branch.tracking_branch.side_effect = Exception("no remote")
repo.git.status.return_value = ""
svc = TestableGitStatus(repo)
result = await svc.get_status(42)
assert result["has_upstream"] is False
assert result["upstream_branch"] is None
# #endregion test_get_status_tracking_exception
# #region test_get_status_iter_commits_exception [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_get_status_iter_commits_exception(self):
"""Exception in iter_commits → ahead=0, behind=0."""
repo = MagicMock()
head = MagicMock()
head.commit = MagicMock()
repo.head = head
repo.active_branch.name = "main"
tracking = MagicMock()
repo.active_branch.tracking_branch.return_value = tracking
repo.iter_commits.side_effect = Exception("git error")
svc = TestableGitStatus(repo)
result = await svc.get_status(42)
assert result["ahead_count"] == 0
assert result["behind_count"] == 0
# #endregion test_get_status_iter_commits_exception
# ── rollback_commit ──
class TestRollbackCommit:
"""rollback_commit — creates a revert commit without rewriting history."""
# #region test_rollback_commit_reverts_target [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_rollback_commit_reverts_target(self):
repo = MagicMock()
repo.head.commit.hexsha = "rollback123"
svc = TestableGitStatus(repo)
result = await svc.rollback_commit(42, "abcdef1", "bad PROD deploy")
repo.commit.assert_called_once_with("abcdef1")
repo.git.revert.assert_called_once_with("--no-edit", "abcdef1")
assert result["status"] == "success"
assert result["reverted_commit"] == "abcdef1"
assert result["rollback_commit"] == "rollback123"
# #endregion test_rollback_commit_reverts_target
# ── get_diff ──
class TestGetDiff:
"""get_diff — diff generation."""
# #region test_get_diff_no_args [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_get_diff_no_args(self):
"""Default diff (no file, not staged)."""
repo = MagicMock()
repo.git.diff.return_value = "diff --git a/file.txt b/file.txt"
svc = TestableGitStatus(repo)
result = await svc.get_diff(42)
assert result == "diff --git a/file.txt b/file.txt"
repo.git.diff.assert_called_once_with()
# #endregion test_get_diff_no_args
# #region test_get_diff_staged [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_get_diff_staged(self):
"""--staged flag included."""
repo = MagicMock()
repo.git.diff.return_value = "staged diff"
svc = TestableGitStatus(repo)
result = await svc.get_diff(42, staged=True)
assert result == "staged diff"
repo.git.diff.assert_called_once_with("--staged")
# #endregion test_get_diff_staged
# #region test_get_diff_with_file [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_get_diff_with_file(self):
"""File path passed with -- separator."""
repo = MagicMock()
repo.git.diff.return_value = "file diff"
svc = TestableGitStatus(repo)
result = await svc.get_diff(42, file_path="src/main.py")
assert result == "file diff"
repo.git.diff.assert_called_once_with("--", "src/main.py")
# #endregion test_get_diff_with_file
# #region test_get_diff_staged_with_file [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_get_diff_staged_with_file(self):
"""Both staged and file_path passed."""
repo = MagicMock()
repo.git.diff.return_value = "staged file diff"
svc = TestableGitStatus(repo)
result = await svc.get_diff(42, file_path="test.py", staged=True)
assert result == "staged file diff"
repo.git.diff.assert_called_once_with("--staged", "--", "test.py")
# #endregion test_get_diff_staged_with_file
# ── get_commit_history ──
class TestGetCommitHistory:
"""get_commit_history — commit log retrieval."""
# #region test_commit_history_no_heads_no_remotes [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_commit_history_no_heads_no_remotes(self):
"""No heads and no remotes → returns []."""
repo = MagicMock()
repo.heads = []
repo.remotes = []
svc = TestableGitStatus(repo)
result = await svc.get_commit_history(42)
assert result == []
# #endregion test_commit_history_no_heads_no_remotes
# #region test_commit_history_returns_commits [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_commit_history_returns_commits(self):
"""Returns parsed commit dicts from iter_commits."""
import datetime
repo = MagicMock()
repo.heads = [MagicMock()]
repo.remotes = [MagicMock()]
commit1 = MagicMock()
commit1.hexsha = "abc123"
commit1.author.name = "Alice"
commit1.author.email = "alice@example.com"
commit1.committed_date = 1700000000
commit1.message = "First commit\n"
commit1.stats.files = {"file1.py": {}, "file2.py": {}}
repo.iter_commits.return_value = [commit1]
svc = TestableGitStatus(repo)
result = await svc.get_commit_history(42, limit=10)
assert len(result) == 1
assert result[0]["hash"] == "abc123"
assert result[0]["author"] == "Alice"
assert result[0]["email"] == "alice@example.com"
assert result[0]["message"] == "First commit"
assert "file1.py" in result[0]["files_changed"]
repo.iter_commits.assert_called_once_with(max_count=10)
# #endregion test_commit_history_returns_commits
# #region test_commit_history_exception [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_commit_history_exception(self):
"""Exception during iteration returns []."""
repo = MagicMock()
repo.heads = [MagicMock()]
repo.iter_commits.side_effect = Exception("git error")
svc = TestableGitStatus(repo)
result = await svc.get_commit_history(42)
assert result == []
# #endregion test_commit_history_exception
# #region test_commit_history_limit_default [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_commit_history_limit_default(self):
"""Default limit is 50."""
repo = MagicMock()
repo.heads = [MagicMock()]
repo.remotes = [MagicMock()]
repo.iter_commits.return_value = []
svc = TestableGitStatus(repo)
result = await svc.get_commit_history(42)
repo.iter_commits.assert_called_once_with(max_count=50)
assert result == []
# #endregion test_commit_history_limit_default
# #endregion Test.GitService.Status
Reference in New Issue View Git Blame Copy Permalink