Files
flow/tests/test_sync.py
2026-02-25 15:32:23 +02:00

89 lines
2.7 KiB
Python

"""Tests for flow.commands.sync."""
from types import SimpleNamespace
import pytest
from flow.commands import sync
def _git_clean_repo(_repo, *cmd, capture=True):
_ = capture
if cmd == ("rev-parse", "--abbrev-ref", "HEAD"):
return SimpleNamespace(returncode=0, stdout="main\n")
if cmd == ("diff", "--quiet"):
return SimpleNamespace(returncode=0, stdout="")
if cmd == ("diff", "--cached", "--quiet"):
return SimpleNamespace(returncode=0, stdout="")
if cmd == ("ls-files", "--others", "--exclude-standard"):
return SimpleNamespace(returncode=0, stdout="")
if cmd == ("rev-parse", "--abbrev-ref", "main@{u}"):
return SimpleNamespace(returncode=0, stdout="origin/main\n")
if cmd == ("rev-list", "--oneline", "main@{u}..main"):
return SimpleNamespace(returncode=0, stdout="")
if cmd == ("for-each-ref", "--format=%(refname:short)", "refs/heads"):
return SimpleNamespace(returncode=0, stdout="main\n")
raise AssertionError(f"Unexpected git command: {cmd!r}")
@pytest.mark.parametrize("git_style", ["dir", "file"])
def test_check_repo_detects_git_dir_and_worktree_file(tmp_path, monkeypatch, git_style):
repo = tmp_path / "repo"
repo.mkdir()
if git_style == "dir":
(repo / ".git").mkdir()
else:
(repo / ".git").write_text("gitdir: /tmp/worktrees/repo\n", encoding="utf-8")
monkeypatch.setattr(sync, "_git", _git_clean_repo)
name, issues = sync._check_repo(str(repo), do_fetch=False)
assert name == "repo"
assert issues == []
class _ConsoleCapture:
def __init__(self):
self.info_messages = []
self.error_messages = []
self.success_messages = []
def info(self, message):
self.info_messages.append(message)
def error(self, message):
self.error_messages.append(message)
def success(self, message):
self.success_messages.append(message)
def test_run_fetch_includes_worktree_style_repo(tmp_path, monkeypatch):
projects = tmp_path / "projects"
projects.mkdir()
worktree_repo = projects / "worktree"
worktree_repo.mkdir()
(worktree_repo / ".git").write_text("gitdir: /tmp/worktrees/worktree\n", encoding="utf-8")
(projects / "non_git").mkdir()
calls = []
def _git_fetch(repo, *cmd, capture=True):
_ = capture
calls.append((repo, cmd))
return SimpleNamespace(returncode=0, stdout="")
monkeypatch.setattr(sync, "_git", _git_fetch)
console = _ConsoleCapture()
ctx = SimpleNamespace(config=SimpleNamespace(projects_dir=str(projects)), console=console)
sync.run_fetch(ctx, SimpleNamespace())
assert calls == [(str(worktree_repo), ("fetch", "--all", "--quiet"))]
assert console.success_messages == ["All remotes fetched."]