149 lines
6.4 KiB
Python
149 lines
6.4 KiB
Python
"""Tests for the recipe refresh that keeps the offline abra calls working on current recipes."""
|
|
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from git import Repo
|
|
from git.exc import GitCommandError
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
import alakazam
|
|
from alakazam import fetch_recipes, sync_repo
|
|
|
|
INSTANCE_CONFIGS = {"a.org": {"authentik": {}, "nextcloud": {}}}
|
|
|
|
|
|
def commit(repo, name, content="x"):
|
|
"""Adds a file and commits it, returning the new commit."""
|
|
path = Path(repo.working_dir) / name
|
|
path.write_text(content)
|
|
repo.index.add([name])
|
|
return repo.index.commit(f"add {name}")
|
|
|
|
|
|
@pytest.fixture
|
|
def remote_and_clone(tmp_path):
|
|
"""A bare-ish upstream repository plus a clone of it, as abra would leave one behind."""
|
|
upstream = Repo.init(tmp_path / "upstream", initial_branch="main")
|
|
upstream.config_writer().set_value("user", "name", "test").release()
|
|
upstream.config_writer().set_value("user", "email", "test@example.org").release()
|
|
commit(upstream, "compose.yml")
|
|
clone = Repo.clone_from(str(tmp_path / "upstream"), str(tmp_path / "clone"))
|
|
clone.config_writer().set_value("user", "name", "test").release()
|
|
clone.config_writer().set_value("user", "email", "test@example.org").release()
|
|
return upstream, clone
|
|
|
|
|
|
class TestSyncRepo:
|
|
def test_fast_forwards_to_the_remote(self, remote_and_clone):
|
|
upstream, clone = remote_and_clone
|
|
commit(upstream, "abra.sh")
|
|
sync_repo(Path(clone.working_dir))
|
|
assert (Path(clone.working_dir) / "abra.sh").exists()
|
|
|
|
def test_fetches_tags(self, remote_and_clone):
|
|
"""A pinned recipe version is a tag, so abra needs them even without a fast-forward."""
|
|
upstream, clone = remote_and_clone
|
|
upstream.create_tag("1.2.3")
|
|
sync_repo(Path(clone.working_dir))
|
|
assert "1.2.3" in [tag.name for tag in clone.tags]
|
|
|
|
def test_writes_the_fetch_head_marker(self, remote_and_clone):
|
|
_, clone = remote_and_clone
|
|
marker = Path(clone.working_dir) / ".git" / "FETCH_HEAD"
|
|
marker.unlink(missing_ok=True)
|
|
sync_repo(Path(clone.working_dir))
|
|
assert marker.exists()
|
|
|
|
def test_a_checked_out_version_is_left_alone(self, remote_and_clone):
|
|
"""abra checks a pinned version out itself, moving it here would fight with that."""
|
|
upstream, clone = remote_and_clone
|
|
upstream.create_tag("1.2.3")
|
|
sync_repo(Path(clone.working_dir))
|
|
clone.git.checkout("1.2.3")
|
|
commit(upstream, "abra.sh")
|
|
sync_repo(Path(clone.working_dir))
|
|
assert clone.head.is_detached
|
|
assert not (Path(clone.working_dir) / "abra.sh").exists()
|
|
|
|
def test_a_dirty_repository_is_left_alone(self, remote_and_clone, caplog):
|
|
upstream, clone = remote_and_clone
|
|
(Path(clone.working_dir) / "compose.yml").write_text("local change")
|
|
commit(upstream, "abra.sh")
|
|
sync_repo(Path(clone.working_dir))
|
|
assert (Path(clone.working_dir) / "compose.yml").read_text() == "local change"
|
|
assert "uncommitted changes" in caplog.text
|
|
|
|
def test_a_repository_without_a_remote_is_skipped(self, tmp_path):
|
|
repo = Repo.init(tmp_path / "solo", initial_branch="main")
|
|
repo.config_writer().set_value("user", "name", "test").release()
|
|
repo.config_writer().set_value("user", "email", "test@example.org").release()
|
|
commit(repo, "compose.yml")
|
|
sync_repo(tmp_path / "solo")
|
|
|
|
def test_an_unreachable_remote_raises(self, tmp_path, remote_and_clone):
|
|
"""fetch_recipes has to catch this, a broken forge must not end the run."""
|
|
_, clone = remote_and_clone
|
|
clone.remotes[0].set_url(str(tmp_path / "gone"))
|
|
with pytest.raises(GitCommandError):
|
|
sync_repo(Path(clone.working_dir))
|
|
|
|
|
|
class TestFetchRecipes:
|
|
@pytest.fixture
|
|
def abra_dir(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(alakazam, "ABRA_DIR", tmp_path)
|
|
synced = []
|
|
monkeypatch.setattr(alakazam, "sync_repo", synced.append)
|
|
for name in ("catalogue", "recipes/authentik", "recipes/nextcloud"):
|
|
(tmp_path / name / ".git").mkdir(parents=True)
|
|
return synced
|
|
|
|
def test_syncs_the_catalogue_and_every_used_recipe(self, tmp_path, abra_dir):
|
|
"""The catalogue feeds 'abra recipe versions', leaving it out freezes version lookups."""
|
|
fetch_recipes(INSTANCE_CONFIGS)
|
|
assert [p.name for p in abra_dir] == ["catalogue", "authentik", "nextcloud"]
|
|
|
|
def test_a_recently_fetched_repository_is_skipped(self, tmp_path, abra_dir):
|
|
(tmp_path / "recipes/authentik/.git/FETCH_HEAD").touch()
|
|
fetch_recipes(INSTANCE_CONFIGS)
|
|
assert "authentik" not in [p.name for p in abra_dir]
|
|
|
|
def test_a_stale_marker_does_not_skip(self, tmp_path, abra_dir):
|
|
marker = tmp_path / "recipes/authentik/.git/FETCH_HEAD"
|
|
marker.touch()
|
|
os.utime(marker, (0, 0))
|
|
fetch_recipes(INSTANCE_CONFIGS)
|
|
assert "authentik" in [p.name for p in abra_dir]
|
|
|
|
def test_the_marker_is_written_even_without_a_remote(self, tmp_path, abra_dir):
|
|
"""A fetch writes FETCH_HEAD itself, this is the fallback for the paths that do not fetch."""
|
|
fetch_recipes(INSTANCE_CONFIGS)
|
|
assert (tmp_path / "recipes/authentik/.git/FETCH_HEAD").exists()
|
|
|
|
def test_a_failing_repository_does_not_stop_the_others(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(alakazam, "ABRA_DIR", tmp_path)
|
|
synced = []
|
|
|
|
def failing(path):
|
|
synced.append(path)
|
|
if path.name == "authentik":
|
|
raise GitCommandError("git fetch", 128, b"could not resolve host")
|
|
|
|
monkeypatch.setattr(alakazam, "sync_repo", failing)
|
|
for name in ("catalogue", "recipes/authentik", "recipes/nextcloud"):
|
|
(tmp_path / name / ".git").mkdir(parents=True)
|
|
fetch_recipes(INSTANCE_CONFIGS)
|
|
assert [p.name for p in synced] == ["catalogue", "authentik", "nextcloud"]
|
|
|
|
def test_a_recipe_that_was_never_cloned_is_tolerated(self, tmp_path, monkeypatch):
|
|
"""abra clones a missing recipe itself, EnsureExists runs before its offline check."""
|
|
monkeypatch.setattr(alakazam, "ABRA_DIR", tmp_path)
|
|
monkeypatch.setattr(alakazam, "sync_repo", alakazam.sync_repo)
|
|
(tmp_path / "catalogue" / ".git").mkdir(parents=True)
|
|
fetch_recipes(INSTANCE_CONFIGS)
|
|
assert not (tmp_path / "recipes/authentik").exists()
|