mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-11-12 20:15:10 +00:00
272 lines
11 KiB
Python
272 lines
11 KiB
Python
"""Unit tests for remote git service-repo resolution. No network, no real git."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from unshackle.core import service_repo
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"value, expected",
|
|
[
|
|
("https://github.com/owner/repo", True),
|
|
("http://example.com/me/svc", True),
|
|
("git@gitlab.com:me/my-services.git", True),
|
|
("ssh://git@host/me/svc", True),
|
|
("me/svc.git", True),
|
|
("owner/repo", True),
|
|
("owner/repo@main", True),
|
|
("/abs/local/path", False),
|
|
("~/my-local-services", False),
|
|
(123, False),
|
|
],
|
|
)
|
|
def test_is_repo_spec(value, expected):
|
|
assert service_repo.is_repo_spec(value) is expected
|
|
|
|
|
|
def test_is_repo_spec_existing_path_wins(tmp_path):
|
|
# an owner/repo-looking value that exists locally is a path, not a repo
|
|
local = tmp_path / "owner" / "repo"
|
|
local.mkdir(parents=True)
|
|
assert service_repo.is_repo_spec(str(local)) is False
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"spec, url, branch, dest",
|
|
[
|
|
("owner/repo@main", "https://github.com/owner/repo", "main", "github.com__owner__repo"),
|
|
("owner/repo", "https://github.com/owner/repo", None, "github.com__owner__repo"),
|
|
("git@gitlab.com:me/my-services.git", "git@gitlab.com:me/my-services.git", None, "gitlab.com__me__my-services"),
|
|
("https://github.com/a/b", "https://github.com/a/b", None, "github.com__a__b"),
|
|
],
|
|
)
|
|
def test_parse_spec(spec, url, branch, dest):
|
|
assert service_repo.parse_spec(spec) == (url, branch, dest)
|
|
|
|
|
|
def test_slug_distinguishes_hosts():
|
|
# same owner/repo on different hosts must not share a clone dir
|
|
a = service_repo.parse_spec("https://github.com/owner/repo")[2]
|
|
b = service_repo.parse_spec("https://gitlab.com/owner/repo")[2]
|
|
assert a != b
|
|
|
|
|
|
def test_repos_base_uses_configured_local_dir(monkeypatch, tmp_path):
|
|
from unshackle.core.config import config
|
|
|
|
monkeypatch.setattr(config.directories, "services", ["owner/repo", str(tmp_path / "svc")], raising=False)
|
|
assert service_repo.repos_base() == (tmp_path / "svc" / "_repos")
|
|
|
|
|
|
def test_is_stale(tmp_path):
|
|
stamp = tmp_path / "stamp"
|
|
assert service_repo.is_stale(stamp, ttl=10) is True # missing → stale
|
|
stamp.write_text(str(time.time()))
|
|
assert service_repo.is_stale(stamp, ttl=10) is False # fresh
|
|
stamp.write_text(str(time.time() - 100))
|
|
assert service_repo.is_stale(stamp, ttl=10) is True # old
|
|
|
|
|
|
def test_clone_invokes_git(monkeypatch, tmp_path):
|
|
monkeypatch.setattr(service_repo.binaries, "Git", "git")
|
|
monkeypatch.setattr(service_repo, "repos_base", lambda: tmp_path)
|
|
|
|
calls = []
|
|
|
|
def fake_run(args, **kw):
|
|
calls.append(args)
|
|
(tmp_path / "github.com__owner__repo").mkdir(parents=True, exist_ok=True) # simulate clone output
|
|
return subprocess.CompletedProcess(args, 0, b"", b"")
|
|
|
|
monkeypatch.setattr(service_repo.subprocess, "run", fake_run)
|
|
|
|
dest = service_repo.resolve_service_repo("owner/repo@main")
|
|
assert dest == tmp_path / "github.com__owner__repo"
|
|
assert calls[0][:4] == ["git", "clone", "--depth", "1"]
|
|
assert "-b" in calls[0] and "main" in calls[0]
|
|
assert service_repo.stamp_for(dest).exists() # stamp written (beside the clone, not inside)
|
|
|
|
|
|
def test_clone_missing_git_returns_none(monkeypatch, tmp_path):
|
|
monkeypatch.setattr(service_repo.binaries, "Git", None)
|
|
monkeypatch.setattr(service_repo, "repos_base", lambda: tmp_path)
|
|
assert service_repo.resolve_service_repo("a/b") is None
|
|
|
|
|
|
def test_fresh_clone_skips_pull(monkeypatch, tmp_path):
|
|
dest = tmp_path / "github.com__a__b"
|
|
dest.mkdir(parents=True)
|
|
service_repo.stamp_for(dest).write_text(str(time.time()))
|
|
monkeypatch.setattr(service_repo, "repos_base", lambda: tmp_path)
|
|
monkeypatch.setattr(
|
|
service_repo.subprocess, "run", lambda *a, **k: pytest.fail("git should not run for a fresh clone")
|
|
)
|
|
assert service_repo.resolve_service_repo("a/b") == dest
|
|
|
|
|
|
def git_stub(porcelain=b"", ahead=b"0"):
|
|
"""Fake subprocess.run for git: dirty checks return given output, pull succeeds."""
|
|
|
|
def run(args, **kw):
|
|
sub = args[3] if len(args) > 3 else ""
|
|
if sub == "status":
|
|
return subprocess.CompletedProcess(args, 0, porcelain, b"")
|
|
if sub == "rev-list":
|
|
return subprocess.CompletedProcess(args, 0, ahead, b"")
|
|
return subprocess.CompletedProcess(args, 0, b"", b"") # pull
|
|
|
|
return run
|
|
|
|
|
|
def test_is_dirty_uncommitted(monkeypatch):
|
|
monkeypatch.setattr(service_repo.binaries, "Git", "git")
|
|
monkeypatch.setattr(service_repo.subprocess, "run", git_stub(porcelain=b" M EXAMPLE/__init__.py\n"))
|
|
assert service_repo.is_dirty(Path("/x")) is True
|
|
|
|
|
|
def test_is_dirty_commits_ahead(monkeypatch):
|
|
monkeypatch.setattr(service_repo.binaries, "Git", "git")
|
|
monkeypatch.setattr(service_repo.subprocess, "run", git_stub(porcelain=b"", ahead=b"2"))
|
|
assert service_repo.is_dirty(Path("/x")) is True
|
|
|
|
|
|
def test_is_dirty_clean(monkeypatch):
|
|
monkeypatch.setattr(service_repo.binaries, "Git", "git")
|
|
monkeypatch.setattr(service_repo.subprocess, "run", git_stub(porcelain=b"", ahead=b"0"))
|
|
assert service_repo.is_dirty(Path("/x")) is False
|
|
|
|
|
|
def test_is_dirty_ignores_untracked_files(monkeypatch):
|
|
# status must ask git to skip untracked files so importing services (__pycache__) isn't "dirty"
|
|
seen = []
|
|
monkeypatch.setattr(service_repo.binaries, "Git", "git")
|
|
|
|
def run(args, **kw):
|
|
seen.append(args)
|
|
out = b"0" if args[3] == "rev-list" else b""
|
|
return subprocess.CompletedProcess(args, 0, out, b"")
|
|
|
|
monkeypatch.setattr(service_repo.subprocess, "run", run)
|
|
service_repo.is_dirty(Path("/x"))
|
|
status_args = next(a for a in seen if a[3] == "status")
|
|
assert "--untracked-files=no" in status_args
|
|
|
|
|
|
def test_auto_refresh_refuses_dirty_clone(monkeypatch, tmp_path):
|
|
"""Auto path (force=False): a stale clone with local edits raises instead of clobbering."""
|
|
dest = tmp_path / "github.com__a__b"
|
|
dest.mkdir(parents=True)
|
|
service_repo.stamp_for(dest).write_text("0") # stale → refresh due
|
|
monkeypatch.setattr(service_repo, "repos_base", lambda: tmp_path)
|
|
monkeypatch.setattr(service_repo.binaries, "Git", "git")
|
|
monkeypatch.setattr(service_repo.subprocess, "run", git_stub(porcelain=b" M f\n"))
|
|
with pytest.raises(service_repo.DirtyServiceRepo) as ei:
|
|
service_repo.resolve_service_repo("a/b")
|
|
assert ei.value.path == dest
|
|
|
|
|
|
def test_manual_refresh_force_overwrites(monkeypatch, tmp_path):
|
|
"""Force path (services_repo_force): hard-resets a stale clone, never raises; leaves a fresh clone alone."""
|
|
dest = tmp_path / "github.com__a__b"
|
|
dest.mkdir(parents=True)
|
|
monkeypatch.setattr(service_repo, "repos_base", lambda: tmp_path)
|
|
monkeypatch.setattr(service_repo.binaries, "Git", "git")
|
|
calls = []
|
|
monkeypatch.setattr(
|
|
service_repo.subprocess,
|
|
"run",
|
|
lambda args, **kw: (calls.append(args[3]), subprocess.CompletedProcess(args, 0, b"", b""))[1],
|
|
)
|
|
assert service_repo.resolve_service_repo("a/b", force=True) == dest
|
|
assert "fetch" in calls and "reset" in calls # force-sync ran
|
|
assert "status" not in calls # no dirty gate on the force path
|
|
calls.clear()
|
|
service_repo.write_stamp(service_repo.stamp_for(dest)) # fresh stamp, inside the TTL
|
|
assert service_repo.resolve_service_repo("a/b", force=True) == dest
|
|
assert calls == [] # no network before the TTL expires
|
|
|
|
|
|
def test_changed_services_groups_by_top_dir(monkeypatch):
|
|
monkeypatch.setattr(service_repo.binaries, "Git", "git")
|
|
|
|
def run(args, **kw):
|
|
if "--name-status" in args:
|
|
out = b"A\tNEW/__init__.py\nM\tOLD/__init__.py\nM\tOLD/config.yaml\nD\tGONE/__init__.py\n"
|
|
else:
|
|
out = b""
|
|
return subprocess.CompletedProcess(args, 0, out, b"")
|
|
|
|
monkeypatch.setattr(service_repo.subprocess, "run", run)
|
|
lines = service_repo.changed_services(Path("/x"), "aaa", "bbb")
|
|
assert lines == ["-GONE", "+NEW", "~OLD"]
|
|
|
|
|
|
def test_local_dirty_services_lists_edited_dirs(monkeypatch):
|
|
monkeypatch.setattr(service_repo.binaries, "Git", "git")
|
|
monkeypatch.setattr(
|
|
service_repo.subprocess,
|
|
"run",
|
|
lambda *a, **k: subprocess.CompletedProcess(
|
|
a, 0, b"EXAMPLE/__init__.py\nEXAMPLE/config.yaml\nDEMO/__init__.py\n", b""
|
|
),
|
|
)
|
|
assert service_repo.local_dirty_services(Path("/x")) == ["DEMO", "EXAMPLE"]
|
|
|
|
|
|
def test_refresh_reports_discarded_local_edits(monkeypatch, tmp_path):
|
|
"""Force refresh with no upstream advance still reports the local edits it wiped."""
|
|
dest = tmp_path / "github.com__a__b"
|
|
dest.mkdir(parents=True)
|
|
monkeypatch.setattr(service_repo, "repos_base", lambda: tmp_path)
|
|
monkeypatch.setattr(service_repo.binaries, "Git", "git")
|
|
|
|
def run(args, **kw):
|
|
if "diff" in args and "--name-only" in args:
|
|
return subprocess.CompletedProcess(args, 0, b"EXAMPLE/__init__.py\n", b"") # local edit present
|
|
if "rev-parse" in args:
|
|
return subprocess.CompletedProcess(args, 0, b"samehead\n", b"") # upstream did not advance
|
|
return subprocess.CompletedProcess(args, 0, b"", b"") # fetch/reset/diff
|
|
|
|
monkeypatch.setattr(service_repo.subprocess, "run", run)
|
|
_, changes = service_repo.refresh_repo("a/b")
|
|
assert changes == ["!EXAMPLE"]
|
|
|
|
|
|
def test_changed_services_up_to_date(monkeypatch):
|
|
monkeypatch.setattr(service_repo.binaries, "Git", "git")
|
|
# same HEAD before/after → no diff computed at all
|
|
monkeypatch.setattr(service_repo.subprocess, "run", lambda *a, **k: pytest.fail("no git diff when HEAD unchanged"))
|
|
assert service_repo.changed_services(Path("/x"), "same", "same") == []
|
|
|
|
|
|
@pytest.mark.parametrize("order", [["a", "b"], ["b", "a"]])
|
|
def test_collision_first_source_wins(tmp_path, order):
|
|
"""Discovery dedupes by tag honoring list order: the FIRST source to define a tag wins,
|
|
whether it's a local dir or a repo clone; later duplicates are shadowed."""
|
|
dirs = {}
|
|
for name in ("a", "b"):
|
|
svc = tmp_path / name / "EXAMPLE"
|
|
svc.mkdir(parents=True)
|
|
(svc / "__init__.py").touch()
|
|
dirs[name] = tmp_path / name
|
|
|
|
seen: dict = {}
|
|
shadowed: list = []
|
|
for name in order: # services.py preserves config order
|
|
for path in dirs[name].glob("*/__init__.py"):
|
|
tag = path.parent.stem
|
|
if tag in seen:
|
|
shadowed.append((tag, path, seen[tag]))
|
|
else:
|
|
seen[tag] = path
|
|
|
|
winner, loser = order[0], order[1]
|
|
assert seen["EXAMPLE"] == dirs[winner] / "EXAMPLE" / "__init__.py"
|
|
assert shadowed and shadowed[0][1] == dirs[loser] / "EXAMPLE" / "__init__.py"
|