Files
ss-tools/backend/tests/services/git/test_git_branch_phase13.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

621 lines
26 KiB
Python

# #region Test.Git.Branch.Phase13 [C:3] [TYPE Module] [SEMANTICS test,git,merge,delete,branch,validate,classify]
# @BRIEF Tests for Phase 13 branch features: merge_branch, delete_branch, branch validation, classification.
# @RELATION BINDS_TO -> [GitServiceMergeMixin]
# @RELATION BINDS_TO -> [GitServiceBranchMixin]
# @TEST_EDGE: merge_success -> status=success, commit_hash populated
# @TEST_EDGE: merge_conflict -> status=conflicts, merge aborted, conflicts list returned
# @TEST_EDGE: merge_already_up_to_date -> status=success, no error
# @TEST_EDGE: merge_auto_delete -> source_deleted=true after successful merge
# @TEST_EDGE: merge_same_branch -> HTTPException 400
# @TEST_EDGE: merge_missing_source -> HTTPException 404
# @TEST_EDGE: merge_missing_target -> HTTPException 404
# @TEST_EDGE: delete_protected -> HTTPException 403 for dev/preprod/prod
# @TEST_EDGE: delete_force -> bypasses protection with force=True
# @TEST_EDGE: delete_active_branch -> switches to fallback before deleting
# @TEST_EDGE: delete_remote -> pushes :branch to origin
# @TEST_EDGE: validate_valid -> returns True for valid names
# @TEST_EDGE: validate_dotdot -> HTTPException 400 for ".."
# @TEST_EDGE: validate_leading_slash -> HTTPException 400 for "/feature"
# @TEST_EDGE: classify_environment -> returns "environment" for dev/preprod/prod
# @TEST_EDGE: classify_feature -> returns "feature" for feature/*
# @TEST_EDGE: classify_hotfix -> returns "hotfix" for hotfix/*
# @TEST_EDGE: classify_legacy -> returns "legacy" for main/master
import contextlib
import os
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "src"))
import pytest
from unittest.mock import ANY, AsyncMock, MagicMock, patch
from fastapi import HTTPException
from git.exc import GitCommandError
from src.services.git._branch import GitServiceBranchMixin
from src.services.git._merge import GitServiceMergeMixin
# ── Testable classes ─────────────────────────────────────────────
class TestableGitMerge(GitServiceMergeMixin):
"""Testable merge mixin with mock _locked and _get_unmerged_file_paths."""
def __init__(self, mock_repo=None):
self._mock_repo = mock_repo or MagicMock()
def _locked(self, dashboard_id):
@contextlib.contextmanager
def _lock():
yield
return _lock()
def _get_unmerged_file_paths(self, repo):
return []
class TestableGitBranch(GitServiceBranchMixin):
"""Testable branch mixin with mock _locked."""
def __init__(self, mock_repo=None):
self._mock_repo = mock_repo or MagicMock()
def _locked(self, dashboard_id):
@contextlib.contextmanager
def _lock():
yield
return _lock()
# ═══════════════════════════════════════════════════════════════════
# merge_branch tests
# ═══════════════════════════════════════════════════════════════════
class TestMergeBranch:
"""merge_branch — merge source into target with conflict detection."""
# #region test_merge_success [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_merge_success(self):
"""Successful merge returns status=success with commit_hash."""
repo = MagicMock()
repo.active_branch.name = "dev"
head_feature = MagicMock()
head_feature.name = "feature/test"
head_dev = MagicMock()
head_dev.name = "dev"
repo.heads = [head_feature, head_dev]
repo.head.commit.hexsha = "abc123def"
origin = MagicMock()
repo.remote.return_value = origin
svc = TestableGitMerge(repo)
svc.get_repo = AsyncMock(return_value=repo)
result = await svc.merge_branch(1, "feature/test", "dev", message="merge feature")
assert result["status"] == "success"
assert result["commit_hash"] == "abc123def"
assert result["conflicts"] == []
assert result["source_deleted"] is False
origin.fetch.assert_called_once()
repo.git.merge.assert_called_once()
origin.push.assert_called_once()
repo.git.checkout.assert_called() # restored original branch
# #endregion test_merge_success
# #region test_merge_conflict [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_merge_conflict(self):
"""Merge conflict → status=conflicts, merge aborted, conflict files returned."""
repo = MagicMock()
repo.active_branch.name = "dev"
head_feature = MagicMock()
head_feature.name = "feature/test"
head_dev = MagicMock()
head_dev.name = "dev"
repo.heads = [head_feature, head_dev]
origin = MagicMock()
repo.remote.return_value = origin
# Simulate Git "CONFLICT" during merge
repo.git.merge.side_effect = GitCommandError("merge", "CONFLICT in file.yaml")
svc = TestableGitMerge(repo)
svc.get_repo = AsyncMock(return_value=repo)
# Override to return specific conflict files
svc._get_unmerged_file_paths = MagicMock(return_value=["file.yaml", "other.yaml"])
result = await svc.merge_branch(1, "feature/test", "dev")
assert result["status"] == "conflicts"
assert result["commit_hash"] is None
assert result["conflicts"] == ["file.yaml", "other.yaml"]
assert "Merge conflict" in result["error_message"]
assert result["source_deleted"] is False
# merge was called and then --abort was called
repo.git.merge.assert_any_call("feature/test", "-m", ANY)
repo.git.merge.assert_any_call("--abort")
origin.push.assert_not_called() # no push on conflict
# #endregion test_merge_conflict
# #region test_merge_already_up_to_date [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_merge_already_up_to_date(self):
"""Already up-to-date → status=success, no error."""
repo = MagicMock()
repo.active_branch.name = "dev"
head_feature = MagicMock()
head_feature.name = "feature/test"
head_dev = MagicMock()
head_dev.name = "dev"
repo.heads = [head_feature, head_dev]
repo.head.commit.hexsha = "already123"
origin = MagicMock()
repo.remote.return_value = origin
repo.git.merge.side_effect = GitCommandError("merge", "Already up to date.")
svc = TestableGitMerge(repo)
svc.get_repo = AsyncMock(return_value=repo)
result = await svc.merge_branch(1, "feature/test", "dev")
assert result["status"] == "success"
assert result["commit_hash"] == "already123"
assert result["error_message"] is None
# Remote push should still be attempted after detecting "already up to date"
origin.push.assert_called_once()
# #endregion test_merge_already_up_to_date
# #region test_merge_auto_delete [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_merge_auto_delete(self):
"""Auto-delete source branch after successful merge."""
repo = MagicMock()
repo.active_branch.name = "dev"
head_feature = MagicMock()
head_feature.name = "feature/test"
head_dev = MagicMock()
head_dev.name = "dev"
repo.heads = [head_feature, head_dev]
repo.head.commit.hexsha = "merged123"
origin = MagicMock()
repo.remote.return_value = origin
svc = TestableGitMerge(repo)
svc.get_repo = AsyncMock(return_value=repo)
with patch.object(svc, 'delete_branch', new_callable=AsyncMock, create=True) as mock_delete:
result = await svc.merge_branch(1, "feature/test", "dev", auto_delete_source=True)
assert result["status"] == "success"
assert result["source_deleted"] is True
mock_delete.assert_called_once_with(1, "feature/test", force=True)
origin.push.assert_called() # target branch pushed
# #endregion test_merge_auto_delete
# #region test_merge_same_branch [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_merge_same_branch(self):
"""Same source and target → HTTPException 400."""
repo = MagicMock()
svc = TestableGitMerge(repo)
svc.get_repo = AsyncMock(return_value=repo)
with pytest.raises(HTTPException) as exc_info:
await svc.merge_branch(1, "dev", "dev")
assert exc_info.value.status_code == 400
assert "must be different" in str(exc_info.value.detail)
# #endregion test_merge_same_branch
# #region test_merge_source_not_found [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_merge_source_not_found(self):
"""Missing source branch → HTTPException 404."""
repo = MagicMock()
repo.active_branch.name = "dev"
head_dev = MagicMock()
head_dev.name = "dev"
repo.heads = [head_dev]
svc = TestableGitMerge(repo)
svc.get_repo = AsyncMock(return_value=repo)
with pytest.raises(HTTPException) as exc_info:
await svc.merge_branch(1, "feature/nonexistent", "dev")
assert exc_info.value.status_code == 404
# #endregion test_merge_source_not_found
# #region test_merge_target_not_found [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_merge_target_not_found(self):
"""Missing target branch → HTTPException 404."""
repo = MagicMock()
repo.active_branch.name = "feature/test"
head_feature = MagicMock()
head_feature.name = "feature/test"
repo.heads = [head_feature]
svc = TestableGitMerge(repo)
svc.get_repo = AsyncMock(return_value=repo)
with pytest.raises(HTTPException) as exc_info:
await svc.merge_branch(1, "feature/test", "nonexistent")
assert exc_info.value.status_code == 404
# #endregion test_merge_target_not_found
# #region test_merge_empty_source [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_merge_empty_source(self):
"""Empty source branch → HTTPException 400."""
repo = MagicMock()
svc = TestableGitMerge(repo)
svc.get_repo = AsyncMock(return_value=repo)
with pytest.raises(HTTPException) as exc_info:
await svc.merge_branch(1, "", "dev")
assert exc_info.value.status_code == 400
# #endregion test_merge_empty_source
# #region test_merge_restores_original_branch [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_merge_restores_original_branch(self):
"""Original branch restored in finally even after merge error."""
repo = MagicMock()
repo.active_branch.name = "hotfix/bug"
head_hotfix = MagicMock()
head_hotfix.name = "hotfix/bug"
head_dev = MagicMock()
head_dev.name = "dev"
repo.heads = [head_hotfix, head_dev]
origin = MagicMock()
repo.remote.return_value = origin
# Simulate unexpected error after checkout
def merge_side_effect(*args, **kwargs):
raise RuntimeError("disk full")
repo.git.merge.side_effect = merge_side_effect
svc = TestableGitMerge(repo)
svc.get_repo = AsyncMock(return_value=repo)
with pytest.raises(HTTPException) as exc_info:
await svc.merge_branch(1, "hotfix/bug", "dev")
assert exc_info.value.status_code == 500
# Verify original branch was restored (last checkout call should be to original branch)
# Note: calls include checkout(target), then checkout(original) in finally
assert repo.git.checkout.called
# #endregion test_merge_restores_original_branch
# ═══════════════════════════════════════════════════════════════════
# delete_branch tests
# ═══════════════════════════════════════════════════════════════════
class TestDeleteBranch:
"""delete_branch — delete with environment branch protection."""
# #region test_delete_success [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_delete_success(self):
"""Successful delete removes local and remote branch."""
repo = MagicMock()
repo.active_branch.name = "dev"
head_feature = MagicMock()
head_feature.name = "feature/to-delete"
head_dev = MagicMock()
head_dev.name = "dev"
repo.heads = [head_feature, head_dev]
origin = MagicMock()
repo.remote.return_value = origin
svc = TestableGitBranch(repo)
svc.get_repo = AsyncMock(return_value=repo)
result = await svc.delete_branch(1, "feature/to-delete")
assert result["status"] == "deleted"
assert result["branch_name"] == "feature/to-delete"
repo.delete_head.assert_called_once_with("feature/to-delete")
origin.push.assert_called_once_with(refspec=":feature/to-delete")
# #endregion test_delete_success
# #region test_delete_protected_dev [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_delete_protected_dev(self):
"""Protected 'dev' → HTTPException 403."""
repo = MagicMock()
head_dev = MagicMock(); head_dev.name = "dev"
head_prod = MagicMock(); head_prod.name = "prod"
repo.heads = [head_dev, head_prod]
svc = TestableGitBranch(repo)
svc.get_repo = AsyncMock(return_value=repo)
with pytest.raises(HTTPException) as exc_info:
await svc.delete_branch(1, "dev")
assert exc_info.value.status_code == 403
repo.delete_head.assert_not_called()
# #endregion test_delete_protected_dev
# #region test_delete_protected_prod [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_delete_protected_prod(self):
"""Protected 'prod' → HTTPException 403."""
repo = MagicMock()
head_dev = MagicMock(); head_dev.name = "dev"
head_prod = MagicMock(); head_prod.name = "prod"
repo.heads = [head_dev, head_prod]
svc = TestableGitBranch(repo)
svc.get_repo = AsyncMock(return_value=repo)
with pytest.raises(HTTPException) as exc_info:
await svc.delete_branch(1, "prod")
assert exc_info.value.status_code == 403
@pytest.mark.asyncio
async def test_delete_protected_preprod(self):
"""Protected 'preprod' → HTTPException 403."""
repo = MagicMock()
head_dev = MagicMock(); head_dev.name = "dev"
head_pre = MagicMock(); head_pre.name = "preprod"
repo.heads = [head_dev, head_pre]
svc = TestableGitBranch(repo)
svc.get_repo = AsyncMock(return_value=repo)
with pytest.raises(HTTPException) as exc_info:
await svc.delete_branch(1, "preprod")
assert exc_info.value.status_code == 403
# #region test_delete_force_bypasses_protection [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_delete_force_bypasses_protection(self):
"""Force=True bypasses environment branch protection."""
repo = MagicMock()
repo.active_branch.name = "prod"
head_prod = MagicMock(); head_prod.name = "prod"
head_dev = MagicMock(); head_dev.name = "dev"
repo.heads = [head_prod, head_dev]
origin = MagicMock()
repo.remote.return_value = origin
svc = TestableGitBranch(repo)
svc.get_repo = AsyncMock(return_value=repo)
result = await svc.delete_branch(1, "prod", force=True)
assert result["status"] == "deleted"
repo.delete_head.assert_called_once()
# #endregion test_delete_force_bypasses_protection
# #region test_delete_active_branch [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_delete_active_branch(self):
"""Active branch → switches to fallback before delete."""
repo = MagicMock()
repo.active_branch.name = "feature/old"
head_feature = MagicMock(); head_feature.name = "feature/old"
head_dev = MagicMock(); head_dev.name = "dev"
repo.heads = [head_feature, head_dev]
origin = MagicMock()
repo.remote.return_value = origin
svc = TestableGitBranch(repo)
svc.get_repo = AsyncMock(return_value=repo)
result = await svc.delete_branch(1, "feature/old")
assert result["status"] == "deleted"
repo.git.checkout.assert_called_with("dev")
repo.delete_head.assert_called_once_with("feature/old")
# #endregion test_delete_active_branch
# #region test_delete_not_found [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_delete_not_found(self):
"""Missing branch → HTTPException 404."""
repo = MagicMock()
head_dev = MagicMock(); head_dev.name = "dev"
repo.heads = [head_dev]
svc = TestableGitBranch(repo)
svc.get_repo = AsyncMock(return_value=repo)
with pytest.raises(HTTPException) as exc_info:
await svc.delete_branch(1, "nonexistent")
assert exc_info.value.status_code == 404
# #endregion test_delete_not_found
# #region test_delete_no_origin_does_not_fail [C:2] [TYPE Function]
@pytest.mark.asyncio
async def test_delete_no_origin_does_not_fail(self):
"""No origin remote → local delete succeeds, remote skipped gracefully."""
repo = MagicMock()
repo.active_branch.name = "dev"
head_feature = MagicMock(); head_feature.name = "feature/test"
head_dev = MagicMock(); head_dev.name = "dev"
repo.heads = [head_feature, head_dev]
repo.remote.side_effect = ValueError("no origin")
svc = TestableGitBranch(repo)
svc.get_repo = AsyncMock(return_value=repo)
result = await svc.delete_branch(1, "feature/test")
assert result["status"] == "deleted"
repo.delete_head.assert_called_once()
# #endregion test_delete_no_origin_does_not_fail
# ═══════════════════════════════════════════════════════════════════
# validate_branch_name tests
# ═══════════════════════════════════════════════════════════════════
class TestValidateBranchName:
"""validate_branch_name — regex validation for branch names."""
# #region test_valid_simple [C:2] [TYPE Function]
def test_valid_simple(self):
"""Valid simple name returns True."""
assert GitServiceBranchMixin.validate_branch_name("feature") is True
# #endregion test_valid_simple
# #region test_valid_slash [C:2] [TYPE Function]
def test_valid_slash(self):
"""Valid name with slashes."""
assert GitServiceBranchMixin.validate_branch_name("feature/my-new-feature") is True
# #endregion test_valid_slash
# #region test_valid_hyphen_underscore [C:2] [TYPE Function]
def test_valid_hyphen_underscore(self):
"""Valid name with hyphens and underscores."""
assert GitServiceBranchMixin.validate_branch_name("hotfix/bug-123_v2") is True
# #endregion test_valid_hyphen_underscore
# #region test_invalid_dotdot [C:2] [TYPE Function]
def test_invalid_dotdot(self):
"""Name with '..' → HTTPException 400."""
with pytest.raises(HTTPException) as exc_info:
GitServiceBranchMixin.validate_branch_name("feature/../escape")
assert exc_info.value.status_code == 400
assert ".." in str(exc_info.value.detail)
# #endregion test_invalid_dotdot
# #region test_invalid_leading_slash [C:2] [TYPE Function]
def test_invalid_leading_slash(self):
"""Leading '/' → HTTPException 400."""
with pytest.raises(HTTPException) as exc_info:
GitServiceBranchMixin.validate_branch_name("/feature/test")
assert exc_info.value.status_code == 400
# #endregion test_invalid_leading_slash
# #region test_invalid_trailing_slash [C:2] [TYPE Function]
def test_invalid_trailing_slash(self):
"""Trailing '/' → HTTPException 400."""
with pytest.raises(HTTPException) as exc_info:
GitServiceBranchMixin.validate_branch_name("feature/test/")
assert exc_info.value.status_code == 400
# #endregion test_invalid_trailing_slash
# #region test_invalid_leading_hyphen [C:2] [TYPE Function]
def test_invalid_leading_hyphen(self):
"""Leading '-' → HTTPException 400."""
with pytest.raises(HTTPException) as exc_info:
GitServiceBranchMixin.validate_branch_name("-feature")
assert exc_info.value.status_code == 400
# #endregion test_invalid_leading_hyphen
# #region test_invalid_empty [C:2] [TYPE Function]
def test_invalid_empty(self):
"""Empty string → HTTPException 400."""
with pytest.raises(HTTPException) as exc_info:
GitServiceBranchMixin.validate_branch_name("")
assert exc_info.value.status_code == 400
# #endregion test_invalid_empty
# #region test_invalid_spaces [C:2] [TYPE Function]
def test_invalid_spaces(self):
"""Spaces in name → HTTPException 400."""
with pytest.raises(HTTPException) as exc_info:
GitServiceBranchMixin.validate_branch_name("feature with spaces")
assert exc_info.value.status_code == 400
# #endregion test_invalid_spaces
# #region test_invalid_too_long [C:2] [TYPE Function]
def test_invalid_too_long(self):
"""Name > 255 chars → HTTPException 400."""
long_name = "a" * 256
with pytest.raises(HTTPException) as exc_info:
GitServiceBranchMixin.validate_branch_name(long_name)
assert exc_info.value.status_code == 400
# #endregion test_invalid_too_long
# ═══════════════════════════════════════════════════════════════════
# _classify_branch_type tests
# ═══════════════════════════════════════════════════════════════════
class TestClassifyBranchType:
"""_classify_branch_type — branch name to semantic type."""
# #region test_classify_dev [C:2] [TYPE Function]
def test_classify_dev(self):
"""'dev''environment'."""
assert GitServiceBranchMixin._classify_branch_type("dev") == "environment"
# #endregion test_classify_dev
# #region test_classify_preprod [C:2] [TYPE Function]
def test_classify_preprod(self):
"""'preprod''environment'."""
assert GitServiceBranchMixin._classify_branch_type("preprod") == "environment"
# #endregion test_classify_preprod
# #region test_classify_prod [C:2] [TYPE Function]
def test_classify_prod(self):
"""'prod''environment'."""
assert GitServiceBranchMixin._classify_branch_type("prod") == "environment"
# #endregion test_classify_prod
# #region test_classify_feature [C:2] [TYPE Function]
def test_classify_feature(self):
"""'feature/auth''feature'."""
assert GitServiceBranchMixin._classify_branch_type("feature/auth") == "feature"
# #endregion test_classify_feature
# #region test_classify_hotfix [C:2] [TYPE Function]
def test_classify_hotfix(self):
"""'hotfix/bug-123''hotfix'."""
assert GitServiceBranchMixin._classify_branch_type("hotfix/bug-123") == "hotfix"
# #endregion test_classify_hotfix
# #region test_classify_bugfix [C:2] [TYPE Function]
def test_classify_bugfix(self):
"""'bugfix/minor-fix''bugfix'."""
assert GitServiceBranchMixin._classify_branch_type("bugfix/minor-fix") == "bugfix"
# #endregion test_classify_bugfix
# #region test_classify_legacy_main [C:2] [TYPE Function]
def test_classify_legacy_main(self):
"""'main''legacy'."""
assert GitServiceBranchMixin._classify_branch_type("main") == "legacy"
# #endregion test_classify_legacy_main
# #region test_classify_legacy_master [C:2] [TYPE Function]
def test_classify_legacy_master(self):
"""'master''legacy'."""
assert GitServiceBranchMixin._classify_branch_type("master") == "legacy"
# #endregion test_classify_legacy_master
# #region test_classify_remote_ref [C:2] [TYPE Function]
def test_classify_remote_ref(self):
"""Remote refs flagged as 'remote_ref'."""
assert GitServiceBranchMixin._classify_branch_type("origin/dev", True) == "remote_ref"
assert GitServiceBranchMixin._classify_branch_type("origin/feature/test") == "remote_ref"
# #endregion test_classify_remote_ref
# #region test_classify_other [C:2] [TYPE Function]
def test_classify_other(self):
"""Unclassified name → 'other'."""
assert GitServiceBranchMixin._classify_branch_type("random-branch") == "other"
# #endregion test_classify_other
# ═══════════════════════════════════════════════════════════════════
# get_branch_protection_rules tests
# ═══════════════════════════════════════════════════════════════════
class TestGetBranchProtectionRules:
"""get_branch_protection_rules — returns protection config for env branches."""
# #region test_returns_rules_for_all_env_branches [C:2] [TYPE Function]
def test_returns_rules_for_all_env_branches(self):
"""Returns protection rules for dev, preprod, and prod."""
rules = GitServiceBranchMixin.get_branch_protection_rules()
branch_names = {r["branch_name"] for r in rules}
assert branch_names == {"dev", "preprod", "prod"}
for rule in rules:
assert rule["allow_direct_delete"] is False
assert rule["allow_force_push"] is False
# #endregion test_returns_rules_for_all_env_branches
# #endregion Test.Git.Branch.Phase13