250 lines
11 KiB
Python
250 lines
11 KiB
Python
# #region Test.Git.Sync.Edge [C:3] [TYPE Module] [SEMANTICS test, git, sync, push, pull, coverage]
|
|
# @BRIEF Edge case tests for GitServiceSyncMixin — push/pull paths.
|
|
# @RELATION BINDS_TO -> [GitServiceSyncMixin]
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "src"))
|
|
|
|
import contextlib
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from git.exc import GitCommandError
|
|
|
|
|
|
class TestPushChanges:
|
|
@pytest.mark.asyncio
|
|
async def test_no_branches(self):
|
|
from src.services.git._sync import GitServiceSyncMixin
|
|
mixin = GitServiceSyncMixin()
|
|
mixin._locked = lambda x: contextlib.nullcontext()
|
|
repo = MagicMock()
|
|
repo.heads = []
|
|
with patch.object(mixin, 'get_repo', new_callable=AsyncMock, create=True, return_value=repo):
|
|
await mixin.push_changes(1) # should not raise
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_origin_remote(self):
|
|
from src.services.git._sync import GitServiceSyncMixin
|
|
mixin = GitServiceSyncMixin()
|
|
mixin._locked = lambda x: contextlib.nullcontext()
|
|
repo = MagicMock()
|
|
repo.heads = [MagicMock()]
|
|
repo.remote.side_effect = ValueError("no origin")
|
|
with patch.object(mixin, 'get_repo', new_callable=AsyncMock, create=True, return_value=repo):
|
|
with pytest.raises(HTTPException, match="origin"):
|
|
await mixin.push_changes(1)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_tracking_branch_sets_upstream(self):
|
|
from src.services.git._sync import GitServiceSyncMixin
|
|
mixin = GitServiceSyncMixin()
|
|
mixin._locked = lambda x: contextlib.nullcontext()
|
|
mixin._align_origin_host_with_config = MagicMock(return_value=None)
|
|
repo = MagicMock()
|
|
repo.heads = [MagicMock()]
|
|
origin = MagicMock()
|
|
repo.remote.return_value = origin
|
|
current_branch = MagicMock()
|
|
current_branch.name = "feature"
|
|
current_branch.tracking_branch.return_value = None
|
|
repo.active_branch = current_branch
|
|
|
|
with patch.object(mixin, 'get_repo', new_callable=AsyncMock, create=True, return_value=repo):
|
|
with patch("src.services.git._sync.SessionLocal") as mock_sl_cls:
|
|
mock_db = MagicMock()
|
|
mock_sl_cls.return_value = mock_db
|
|
mock_db_repo = MagicMock()
|
|
mock_db_repo.remote_url = "https://git.com/org/repo.git"
|
|
mock_db_repo.config_id = 1
|
|
mock_db.query.return_value.filter.return_value.first.return_value = mock_db_repo
|
|
mock_db_config = MagicMock()
|
|
mock_db_config.url = "https://git.com"
|
|
mock_db.query.return_value.filter.return_value.first.return_value = mock_db_config
|
|
|
|
await mixin.push_changes(1)
|
|
repo.git.push.assert_called_with("--set-upstream", "origin", "feature:feature")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tracking_branch_push(self):
|
|
from src.services.git._sync import GitServiceSyncMixin
|
|
mixin = GitServiceSyncMixin()
|
|
mixin._locked = lambda x: contextlib.nullcontext()
|
|
mixin._align_origin_host_with_config = MagicMock(return_value=None)
|
|
repo = MagicMock()
|
|
repo.heads = [MagicMock()]
|
|
origin = MagicMock()
|
|
repo.remote.return_value = origin
|
|
current_branch = MagicMock()
|
|
current_branch.name = "main"
|
|
current_branch.tracking_branch.return_value = "origin/main"
|
|
repo.active_branch = current_branch
|
|
|
|
with patch.object(mixin, 'get_repo', new_callable=AsyncMock, create=True, return_value=repo):
|
|
with patch("src.services.git._sync.SessionLocal") as mock_sl_cls:
|
|
mock_db = MagicMock()
|
|
mock_sl_cls.return_value = mock_db
|
|
mock_db.query.return_value.filter.return_value.first.return_value = None # no db_repo
|
|
|
|
await mixin.push_changes(1)
|
|
origin.push.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_push_git_command_error_nonfastforward(self):
|
|
from src.services.git._sync import GitServiceSyncMixin
|
|
mixin = GitServiceSyncMixin()
|
|
mixin._locked = lambda x: contextlib.nullcontext()
|
|
mixin._align_origin_host_with_config = MagicMock(return_value=None)
|
|
repo = MagicMock()
|
|
repo.heads = [MagicMock()]
|
|
origin = MagicMock()
|
|
repo.remote.return_value = origin
|
|
current_branch = MagicMock()
|
|
current_branch.name = "main"
|
|
current_branch.tracking_branch.return_value = None
|
|
repo.active_branch = current_branch
|
|
repo.git.push.side_effect = GitCommandError("push", "non-fast-forward")
|
|
|
|
with patch.object(mixin, 'get_repo', new_callable=AsyncMock, create=True, return_value=repo):
|
|
with patch("src.services.git._sync.SessionLocal") as mock_sl_cls:
|
|
mock_db = MagicMock()
|
|
mock_sl_cls.return_value = mock_db
|
|
mock_db.query.return_value.filter.return_value.first.return_value = None
|
|
with pytest.raises(HTTPException, match="conflict|rejected|409"):
|
|
await mixin.push_changes(1)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_push_generic_exception(self):
|
|
from src.services.git._sync import GitServiceSyncMixin
|
|
mixin = GitServiceSyncMixin()
|
|
mixin._locked = lambda x: contextlib.nullcontext()
|
|
mixin._align_origin_host_with_config = MagicMock(return_value=None)
|
|
repo = MagicMock()
|
|
repo.heads = [MagicMock()]
|
|
origin = MagicMock()
|
|
repo.remote.return_value = origin
|
|
current_branch = MagicMock()
|
|
current_branch.name = "main"
|
|
current_branch.tracking_branch.return_value = None
|
|
repo.active_branch = current_branch
|
|
repo.git.push.side_effect = GitCommandError("push", "some error")
|
|
|
|
with patch.object(mixin, 'get_repo', new_callable=AsyncMock, create=True, return_value=repo):
|
|
with patch("src.services.git._sync.SessionLocal") as mock_sl_cls:
|
|
mock_db = MagicMock()
|
|
mock_sl_cls.return_value = mock_db
|
|
mock_db.query.return_value.filter.return_value.first.return_value = None
|
|
with pytest.raises(HTTPException, match="500"):
|
|
await mixin.push_changes(1)
|
|
|
|
|
|
class TestPullChanges:
|
|
@pytest.mark.asyncio
|
|
async def test_unfinished_merge_detected(self):
|
|
from src.services.git._sync import GitServiceSyncMixin
|
|
mixin = GitServiceSyncMixin()
|
|
mixin._locked = lambda x: contextlib.nullcontext()
|
|
repo = MagicMock()
|
|
repo.git_dir = "/tmp/repo/.git"
|
|
repo.working_tree_dir = "/tmp/repo"
|
|
repo.active_branch.name = "main"
|
|
|
|
with patch.object(mixin, 'get_repo', new_callable=AsyncMock, create=True, return_value=repo), \
|
|
patch("os.path.exists", return_value=True), \
|
|
patch.object(mixin, '_build_unfinished_merge_payload', return_value={
|
|
"repository_path": "/tmp/repo",
|
|
"git_dir": "/tmp/repo/.git",
|
|
"current_branch": "main",
|
|
"merge_head": "abc123",
|
|
"merge_message_preview": "Merge branch",
|
|
"conflicts_count": 2,
|
|
}, create=True):
|
|
with pytest.raises(HTTPException, match="409"):
|
|
await mixin.pull_changes(1)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remote_branch_not_found(self):
|
|
from src.services.git._sync import GitServiceSyncMixin
|
|
mixin = GitServiceSyncMixin()
|
|
mixin._locked = lambda x: contextlib.nullcontext()
|
|
repo = MagicMock()
|
|
repo.git_dir = "/tmp/repo/.git"
|
|
repo.working_tree_dir = "/tmp/repo"
|
|
current_branch = MagicMock()
|
|
current_branch.name = "feature"
|
|
repo.active_branch = current_branch
|
|
origin = MagicMock()
|
|
repo.remote.return_value = origin
|
|
repo.refs = []
|
|
|
|
with patch.object(mixin, 'get_repo', new_callable=AsyncMock, create=True, return_value=repo), \
|
|
patch("os.path.exists", return_value=False):
|
|
with pytest.raises(HTTPException, match="409"):
|
|
await mixin.pull_changes(1)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_conflict_during_pull(self):
|
|
from src.services.git._sync import GitServiceSyncMixin
|
|
mixin = GitServiceSyncMixin()
|
|
mixin._locked = lambda x: contextlib.nullcontext()
|
|
repo = MagicMock()
|
|
repo.git_dir = "/tmp/repo/.git"
|
|
repo.working_tree_dir = "/tmp/repo"
|
|
current_branch = MagicMock()
|
|
current_branch.name = "main"
|
|
repo.active_branch = current_branch
|
|
origin = MagicMock()
|
|
repo.remote.return_value = origin
|
|
# Simulate remote branch existence
|
|
ref = MagicMock()
|
|
ref.name = "origin/main"
|
|
repo.refs = [ref]
|
|
repo.git.pull.side_effect = GitCommandError("pull", "conflict")
|
|
|
|
with patch.object(mixin, 'get_repo', new_callable=AsyncMock, create=True, return_value=repo), \
|
|
patch("os.path.exists", return_value=False):
|
|
with pytest.raises(HTTPException, match="409"):
|
|
await mixin.pull_changes(1)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_origin_not_found(self):
|
|
from src.services.git._sync import GitServiceSyncMixin
|
|
mixin = GitServiceSyncMixin()
|
|
mixin._locked = lambda x: contextlib.nullcontext()
|
|
repo = MagicMock()
|
|
repo.git_dir = "/tmp/repo/.git"
|
|
repo.working_tree_dir = "/tmp/repo"
|
|
repo.active_branch.name = "main"
|
|
repo.remote.side_effect = ValueError
|
|
|
|
with patch.object(mixin, 'get_repo', new_callable=AsyncMock, create=True, return_value=repo), \
|
|
patch("os.path.exists", return_value=False):
|
|
with pytest.raises(HTTPException, match="origin"):
|
|
await mixin.pull_changes(1)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generic_pull_error(self):
|
|
from src.services.git._sync import GitServiceSyncMixin
|
|
mixin = GitServiceSyncMixin()
|
|
mixin._locked = lambda x: contextlib.nullcontext()
|
|
repo = MagicMock()
|
|
repo.git_dir = "/tmp/repo/.git"
|
|
repo.working_tree_dir = "/tmp/repo"
|
|
current_branch = MagicMock()
|
|
current_branch.name = "main"
|
|
repo.active_branch = current_branch
|
|
origin = MagicMock()
|
|
repo.remote.return_value = origin
|
|
ref = MagicMock()
|
|
ref.name = "origin/main"
|
|
repo.refs = [ref]
|
|
repo.git.pull.side_effect = Exception("unknown error")
|
|
|
|
with patch.object(mixin, 'get_repo', new_callable=AsyncMock, create=True, return_value=repo), \
|
|
patch("os.path.exists", return_value=False):
|
|
with pytest.raises(HTTPException, match="500"):
|
|
await mixin.pull_changes(1)
|
|
# #endregion Test.Git.Sync.Edge
|