142 lines
4.3 KiB
Python
142 lines
4.3 KiB
Python
"""Tests for ProjectService."""
|
|
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
from flow.core.config import AppConfig, FlowContext
|
|
from flow.core.console import Console
|
|
from flow.core.errors import FlowError
|
|
from flow.core.platform import PlatformInfo
|
|
from flow.core.runtime import SystemRuntime
|
|
from flow.app.projects import ProjectService
|
|
|
|
|
|
def _make_ctx(projects_dir, monkeypatch):
|
|
monkeypatch.setattr("flow.actions.executor.paths.STATE_DIR", projects_dir / ".state")
|
|
return FlowContext(
|
|
config=AppConfig(projects_dir=str(projects_dir)),
|
|
manifest={},
|
|
platform=PlatformInfo(),
|
|
console=Console(color=False),
|
|
runtime=SystemRuntime(),
|
|
)
|
|
|
|
|
|
def _git(*args):
|
|
return subprocess.run(
|
|
["git", *[str(arg) for arg in args]],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
|
|
|
|
def _init_repo(path, commit=True):
|
|
"""Create a git repo with an initial commit."""
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
_git("init", path)
|
|
_git("-C", path, "config", "user.email", "test@test.com")
|
|
_git("-C", path, "config", "user.name", "Test")
|
|
if commit:
|
|
(path / "README.md").write_text("# test")
|
|
_git("-C", path, "add", ".")
|
|
_git("-C", path, "commit", "-m", "init")
|
|
|
|
|
|
def _add_upstream(repo):
|
|
remote = repo.parent / f"{repo.name}.git"
|
|
_git("init", "--bare", remote)
|
|
_git("-C", repo, "remote", "add", "origin", remote)
|
|
branch = _git("-C", repo, "branch", "--show-current").stdout.strip()
|
|
_git("-C", repo, "push", "-u", "origin", branch)
|
|
|
|
|
|
class TestProjectService:
|
|
def test_check_clean_repo(self, tmp_path, monkeypatch, capsys):
|
|
projects = tmp_path / "projects"
|
|
projects.mkdir()
|
|
repo = projects / "myrepo"
|
|
_init_repo(repo)
|
|
_add_upstream(repo)
|
|
|
|
ctx = _make_ctx(projects, monkeypatch)
|
|
svc = ProjectService(ctx)
|
|
svc.check(fetch=False)
|
|
|
|
output = capsys.readouterr().out
|
|
assert "myrepo" in output
|
|
assert "clean" in output
|
|
|
|
def test_check_uncommitted_changes(self, tmp_path, monkeypatch, capsys):
|
|
projects = tmp_path / "projects"
|
|
projects.mkdir()
|
|
repo = projects / "myrepo"
|
|
_init_repo(repo)
|
|
_add_upstream(repo)
|
|
(repo / "new_file.txt").write_text("changes")
|
|
|
|
ctx = _make_ctx(projects, monkeypatch)
|
|
svc = ProjectService(ctx)
|
|
svc.check(fetch=False)
|
|
|
|
output = capsys.readouterr().out
|
|
assert "uncommitted" in output
|
|
|
|
def test_check_no_git_repos(self, tmp_path, monkeypatch, capsys):
|
|
projects = tmp_path / "projects"
|
|
projects.mkdir()
|
|
(projects / "not-a-repo").mkdir()
|
|
|
|
ctx = _make_ctx(projects, monkeypatch)
|
|
svc = ProjectService(ctx)
|
|
svc.check(fetch=False)
|
|
|
|
output = capsys.readouterr().out
|
|
assert "No git" in output
|
|
|
|
def test_missing_projects_dir(self, tmp_path, monkeypatch, capsys):
|
|
ctx = _make_ctx(tmp_path / "nonexistent", monkeypatch)
|
|
svc = ProjectService(ctx)
|
|
svc.check(fetch=False)
|
|
assert "not found" in capsys.readouterr().out
|
|
|
|
def test_fetch_failure_raises(self, tmp_path, monkeypatch):
|
|
projects = tmp_path / "projects"
|
|
projects.mkdir()
|
|
repo = projects / "myrepo"
|
|
_init_repo(repo)
|
|
_git("-C", repo, "remote", "add", "origin", tmp_path / "missing-origin")
|
|
|
|
ctx = _make_ctx(projects, monkeypatch)
|
|
svc = ProjectService(ctx)
|
|
|
|
with pytest.raises(FlowError, match="Fetch remotes for myrepo"):
|
|
svc.check(fetch=True)
|
|
|
|
def test_status_failure_raises(self, tmp_path, monkeypatch):
|
|
projects = tmp_path / "projects"
|
|
projects.mkdir()
|
|
repo = projects / "myrepo"
|
|
_init_repo(repo)
|
|
(repo / ".git" / "HEAD").unlink()
|
|
|
|
ctx = _make_ctx(projects, monkeypatch)
|
|
svc = ProjectService(ctx)
|
|
|
|
with pytest.raises(FlowError, match="Check working tree for myrepo"):
|
|
svc.check(fetch=False)
|
|
|
|
def test_missing_upstream_raises_instead_of_clean(self, tmp_path, monkeypatch, capsys):
|
|
projects = tmp_path / "projects"
|
|
projects.mkdir()
|
|
_init_repo(projects / "myrepo")
|
|
|
|
ctx = _make_ctx(projects, monkeypatch)
|
|
svc = ProjectService(ctx)
|
|
|
|
with pytest.raises(FlowError, match="rev-list"):
|
|
svc.check(fetch=False)
|
|
|
|
assert "clean" not in capsys.readouterr().out
|