209 lines
8.2 KiB
Python
209 lines
8.2 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
import os
|
|
import shutil
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
import uuid
|
|
from subprocess import CompletedProcess
|
|
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
|
|
|
|
from git_repo import ( # noqa: E402
|
|
DIVERGED,
|
|
LOCAL_AHEAD,
|
|
REMOTE_AHEAD,
|
|
SYNCED,
|
|
UNAVAILABLE,
|
|
GitRepository,
|
|
GitRepositoryError,
|
|
_is_authentication_failure,
|
|
_display_url,
|
|
)
|
|
|
|
|
|
class GitRepositoryTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
cache_root = Path(__file__).resolve().parents[1] / ".cache"
|
|
cache_root.mkdir(exist_ok=True)
|
|
self.base = cache_root / f"git-repo-test-{uuid.uuid4().hex}"
|
|
self.base.mkdir()
|
|
self.remote = self.base / "labyricorn-site.git"
|
|
self.seed = self.base / "seed"
|
|
self.work = self.base / "work"
|
|
self.other = self.base / "other"
|
|
|
|
self.git(self.base, "init", "--bare", "--initial-branch=main", str(self.remote))
|
|
self.git(self.base, "init", "--initial-branch=main", str(self.seed))
|
|
self.configure_identity(self.seed)
|
|
(self.seed / "README.md").write_text("initial\n", encoding="utf-8")
|
|
self.git(self.seed, "add", "README.md")
|
|
self.git(self.seed, "commit", "-m", "Initial")
|
|
self.git(self.seed, "remote", "add", "origin", str(self.remote))
|
|
self.git(self.seed, "push", "-u", "origin", "main")
|
|
|
|
self.git(self.base, "clone", str(self.remote), str(self.work))
|
|
self.git(self.base, "clone", str(self.remote), str(self.other))
|
|
self.configure_identity(self.work)
|
|
self.configure_identity(self.other)
|
|
self.repository = GitRepository.locate(self.work / "README.md")
|
|
|
|
def tearDown(self) -> None:
|
|
shutil.rmtree(self.base, onexc=self.make_writable)
|
|
|
|
@staticmethod
|
|
def make_writable(function: object, path: str, _error: BaseException) -> None:
|
|
os.chmod(path, stat.S_IWRITE)
|
|
function(path)
|
|
|
|
@staticmethod
|
|
def git(cwd: Path, *args: str) -> subprocess.CompletedProcess[str]:
|
|
result = subprocess.run(
|
|
["git", *args],
|
|
cwd=cwd,
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
)
|
|
if result.returncode != 0:
|
|
raise AssertionError(result.stderr.strip() or result.stdout.strip())
|
|
return result
|
|
|
|
def configure_identity(self, repository: Path) -> None:
|
|
self.git(repository, "config", "user.name", "Launcher Test")
|
|
self.git(repository, "config", "user.email", "[email protected]")
|
|
|
|
def commit(self, repository: Path, filename: str, content: str) -> None:
|
|
(repository / filename).write_text(content, encoding="utf-8")
|
|
self.git(repository, "add", filename)
|
|
self.git(repository, "commit", "-m", f"Update {filename}")
|
|
|
|
def push_other_commit(self) -> None:
|
|
self.commit(self.other, "remote.txt", "remote\n")
|
|
self.git(self.other, "push", "origin", "main")
|
|
|
|
def test_synchronized_and_dirty_states(self) -> None:
|
|
state = self.repository.refresh()
|
|
self.assertEqual(SYNCED, state.sync_state)
|
|
self.assertEqual("labyricorn-site", state.repository_name)
|
|
self.assertEqual("main", state.branch)
|
|
self.assertEqual("origin/main", state.upstream)
|
|
self.assertEqual(0, state.dirty_count)
|
|
|
|
(self.work / "local.txt").write_text("uncommitted\n", encoding="utf-8")
|
|
dirty = self.repository.refresh()
|
|
self.assertEqual(SYNCED, dirty.sync_state)
|
|
self.assertEqual(1, dirty.dirty_count)
|
|
|
|
def test_local_ahead_and_push(self) -> None:
|
|
self.commit(self.work, "local.txt", "local\n")
|
|
state = self.repository.refresh()
|
|
self.assertEqual(LOCAL_AHEAD, state.sync_state)
|
|
self.assertEqual(1, state.ahead)
|
|
self.assertTrue(state.can_push)
|
|
|
|
result = self.repository.push(state)
|
|
self.assertEqual(SYNCED, result.sync_state)
|
|
|
|
def test_dirty_worktree_publish_commits_and_pushes(self) -> None:
|
|
(self.work / "local.txt").write_text("publish me\n", encoding="utf-8")
|
|
state = self.repository.refresh()
|
|
changes = self.repository.publishing_changes()
|
|
self.assertTrue(state.can_publish)
|
|
self.assertIn("local.txt", changes)
|
|
|
|
result = self.repository.publish(state, "Publish local change", changes)
|
|
self.assertEqual(SYNCED, result.sync_state)
|
|
self.assertEqual("Publish local change", self.git(self.work, "log", "-1", "--format=%s").stdout.strip())
|
|
self.assertTrue((self.other / "local.txt").exists() is False)
|
|
self.git(self.other, "pull", "--ff-only")
|
|
self.assertEqual("publish me\n", (self.other / "local.txt").read_text(encoding="utf-8"))
|
|
|
|
def test_publish_stops_if_changes_differ_from_confirmation(self) -> None:
|
|
(self.work / "first.txt").write_text("first\n", encoding="utf-8")
|
|
state = self.repository.refresh()
|
|
changes = self.repository.publishing_changes()
|
|
(self.work / "second.txt").write_text("second\n", encoding="utf-8")
|
|
|
|
with self.assertRaisesRegex(GitRepositoryError, "changed after confirmation"):
|
|
self.repository.publish(state, "Should not publish", changes)
|
|
self.assertEqual("Initial", self.git(self.work, "log", "-1", "--format=%s").stdout.strip())
|
|
|
|
def test_remote_ahead_and_fast_forward_pull(self) -> None:
|
|
self.push_other_commit()
|
|
state = self.repository.refresh()
|
|
self.assertEqual(REMOTE_AHEAD, state.sync_state)
|
|
self.assertEqual(1, state.behind)
|
|
self.assertTrue(state.can_pull)
|
|
|
|
result = self.repository.pull(state)
|
|
self.assertEqual(SYNCED, result.sync_state)
|
|
self.assertTrue((self.work / "remote.txt").is_file())
|
|
|
|
def test_dirty_worktree_blocks_pull(self) -> None:
|
|
self.push_other_commit()
|
|
(self.work / "local.txt").write_text("uncommitted\n", encoding="utf-8")
|
|
state = self.repository.refresh()
|
|
self.assertEqual(REMOTE_AHEAD, state.sync_state)
|
|
self.assertFalse(state.can_pull)
|
|
with self.assertRaisesRegex(GitRepositoryError, "uncommitted changes"):
|
|
self.repository.pull(state)
|
|
|
|
def test_diverged_history_has_no_sync_action(self) -> None:
|
|
self.commit(self.work, "local.txt", "local\n")
|
|
self.push_other_commit()
|
|
state = self.repository.refresh()
|
|
self.assertEqual(DIVERGED, state.sync_state)
|
|
self.assertEqual(1, state.ahead)
|
|
self.assertEqual(1, state.behind)
|
|
self.assertFalse(state.can_pull)
|
|
self.assertFalse(state.can_push)
|
|
with self.assertRaisesRegex(GitRepositoryError, "diverged"):
|
|
self.repository.push(state)
|
|
|
|
def test_missing_origin_and_upstream_are_reported(self) -> None:
|
|
self.git(self.work, "remote", "remove", "origin")
|
|
missing_origin = self.repository.refresh()
|
|
self.assertEqual(UNAVAILABLE, missing_origin.sync_state)
|
|
self.assertIn("No usable origin", missing_origin.problem or "")
|
|
|
|
self.git(self.work, "remote", "add", "origin", str(self.remote))
|
|
missing_upstream = self.repository.refresh()
|
|
self.assertEqual(UNAVAILABLE, missing_upstream.sync_state)
|
|
self.assertIn("no usable upstream", missing_upstream.problem or "")
|
|
|
|
def test_http_credentials_are_not_displayed(self) -> None:
|
|
self.assertEqual(
|
|
"https://git.example.test/owner/repository.git",
|
|
_display_url("https://user:[email protected]/owner/repository.git"),
|
|
)
|
|
|
|
def test_authentication_failure_is_detected(self) -> None:
|
|
result = CompletedProcess(
|
|
["git", "push"],
|
|
128,
|
|
"",
|
|
"remote: Failed to authenticate user\nfatal: Authentication failed",
|
|
)
|
|
self.assertTrue(_is_authentication_failure(result))
|
|
|
|
def test_clear_https_credential_targets_only_origin_host(self) -> None:
|
|
self.git(self.work, "config", "credential.helper", "")
|
|
self.git(
|
|
self.work,
|
|
"remote",
|
|
"set-url",
|
|
"origin",
|
|
"https://git.example.test/team/repository.git",
|
|
)
|
|
self.assertEqual("git.example.test", self.repository.clear_https_credential())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|