1"""Exercise the mirror script against local Git repositories and a stub GitHub API."""
  2import json
  3import os
  4from pathlib import Path
  5import subprocess
  6import tempfile
  7import unittest
  8
  9SCRIPT = Path(__file__).resolve().parents[1] / "mirror-github.sh"
 10
 11
 12class MirrorGitHubTests(unittest.TestCase):
 13    def setUp(self):
 14        self.temp = tempfile.TemporaryDirectory()
 15        self.addCleanup(self.temp.cleanup)
 16        self.root = Path(self.temp.name)
 17        self.sources = self.root / "sources"
 18        self.sources.mkdir()
 19        self.destination = self.root / "served repos"
 20        self.fixtures = self.root / "pages"
 21        self.fixtures.mkdir()
 22        tools = self.root / "tools"
 23        tools.mkdir()
 24        # Stub only HTTP discovery; every clone/fetch uses real Git with local URL rewriting.
 25        curl = tools / "curl"
 26        curl.write_text('''#!/usr/bin/env python3
 27import json, os, sys
 28from pathlib import Path
 29from urllib.parse import parse_qs, urlparse
 30if os.environ.get("TEST_API_FAILURE"):
 31    sys.exit(22)
 32args = sys.argv[1:]
 33assert not any("test-token" in arg for arg in args)
 34headers = sys.stdin.read()
 35if os.environ.get("GITHUB_TOKEN"):
 36    assert "Authorization: Bearer test-token" in headers
 37url = urlparse(args[-1])
 38assert url.path == "/users/testuser/repos"
 39query = parse_qs(url.query)
 40assert query["per_page"] == ["100"]
 41page = query["page"][0]
 42with open(os.environ["TEST_API_CALLS"], "a") as calls:
 43    calls.write(page + "\\n")
 44Path(args[args.index("--output") + 1]).write_bytes(
 45    (Path(os.environ["TEST_API_PAGES"]) / (page + ".json")).read_bytes())
 46''')
 47        curl.chmod(0o755)
 48        self.env = os.environ.copy()
 49        # Isolate Git configuration, identities, credentials, and URL rewrites from the host.
 50        self.env.update({
 51            "PATH": str(tools) + os.pathsep + self.env["PATH"],
 52            "GIT_CONFIG_NOSYSTEM": "1",
 53            "GIT_CONFIG_GLOBAL": "/dev/null",
 54            "GIT_CONFIG_COUNT": "1",
 55            "GIT_CONFIG_KEY_0": f"url.{self.sources.as_uri()}/.insteadOf",
 56            "GIT_CONFIG_VALUE_0": "https://github.com/testuser/",
 57            "GIT_AUTHOR_NAME": "Mirror Test",
 58            "GIT_AUTHOR_EMAIL": "mirror@example.com",
 59            "GIT_COMMITTER_NAME": "Mirror Test",
 60            "GIT_COMMITTER_EMAIL": "mirror@example.com",
 61            "TEST_API_PAGES": str(self.fixtures),
 62            "TEST_API_CALLS": str(self.root / "calls"),
 63            "GITHUB_TOKEN": "test-token",
 64        })
 65        self.env.pop("TEST_API_FAILURE", None)
 66
 67    def git(self, directory, *args):
 68        result = subprocess.run(["git", "-C", str(directory), *args], env=self.env,
 69                                text=True, capture_output=True, check=True)
 70        return result.stdout.strip()
 71
 72    def source(self, name, empty=False):
 73        path = self.sources / (name + ".git")
 74        path.mkdir()
 75        self.git(path, "init", "--initial-branch=main")
 76        if not empty:
 77            (path / "README.md").write_text("# Example\n")
 78            self.git(path, "add", "README.md")
 79            self.git(path, "commit", "-m", "Initial source")
 80        return path
 81
 82    def page(self, repos, number=1):
 83        (self.fixtures / f"{number}.json").write_text(json.dumps(repos))
 84
 85    def record(self, name, **overrides):
 86        return {"name": name, "private": False, "default_branch": "main",
 87                "description": "A description\twith tabs\nand newlines.", **overrides}
 88
 89    def run_script(self, success=True):
 90        result = subprocess.run([str(SCRIPT), "testuser", str(self.destination)],
 91                                env=self.env, text=True, capture_output=True, timeout=30)
 92        if success:
 93            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
 94        else:
 95            self.assertNotEqual(result.returncode, 0)
 96        return result
 97
 98    def test_clone_refresh_and_prune(self):
 99        source = self.source("project")
100        initial_hash = self.git(source, "rev-parse", "HEAD")
101        self.git(source, "branch", "obsolete")
102        self.git(source, "tag", "v1")
103        self.page([self.record("project", fork=True, archived=True)])
104        self.run_script()
105        mirror = self.destination / "project.git"
106        self.assertEqual(self.git(mirror, "rev-parse", "HEAD"), self.git(source, "rev-parse", "HEAD"))
107        self.assertEqual(self.git(mirror, "rev-parse", "--is-bare-repository"), "true")
108        self.assertEqual((mirror / "description").read_text(), self.record("project")["description"] + "\n")
109        self.git(source, "branch", "-m", "trunk")
110        self.git(source, "branch", "-D", "obsolete")
111        self.git(source, "tag", "-d", "v1")
112        (source / "README.md").write_text("# Changed\n")
113        self.git(source, "commit", "-am", "Upstream update")
114        self.page([self.record("project", default_branch="trunk", description=None)])
115        self.run_script()
116        self.assertEqual(self.git(mirror, "symbolic-ref", "HEAD"), "refs/heads/trunk")
117        self.assertEqual(self.git(mirror, "rev-parse", "HEAD"), self.git(source, "rev-parse", "HEAD"))
118        self.assertEqual(self.git(mirror, "for-each-ref", "--format=%(refname)"), "refs/heads/trunk")
119        self.assertEqual((mirror / "description").read_text(), "\n")
120        self.git(source, "update-ref", "refs/heads/trunk", initial_hash)
121        self.run_script()
122        self.assertEqual(self.git(mirror, "rev-parse", "HEAD"), initial_hash)
123
124    def test_pagination_and_public_only_filter(self):
125        self.source("last-page")
126        # A full first page must not stop pagination even if every entry is filtered out.
127        self.page([self.record(f"private-{i}", private=True) for i in range(100)])
128        self.page([self.record("last-page")], number=2)
129        self.run_script()
130        self.assertEqual((self.root / "calls").read_text(), "1\n2\n")
131        self.assertEqual(sorted(p.name for p in self.destination.glob("*.git")), ["last-page.git"])
132
133    def test_preserves_unrelated_destination_and_continues(self):
134        self.source("project")
135        self.source("other")
136        existing = self.destination / "project.git"
137        existing.mkdir(parents=True)
138        (existing / "important.txt").write_text("Keep this file")
139        self.page([self.record("project"), self.record("other")])
140        result = self.run_script(success=False)
141        self.assertIn("not a matching GitHub mirror", result.stderr)
142        self.assertEqual((existing / "important.txt").read_text(), "Keep this file")
143        self.assertTrue((self.destination / "other.git" / "HEAD").is_file())
144
145    def test_refuses_existing_bare_clones_and_foreign_mirrors(self):
146        self.destination.mkdir()
147        for name, mode in (("bare", "--bare"), ("foreign", "--mirror")):
148            source = self.source(name)
149            target = self.destination / (name + ".git")
150            self.git(self.root, "clone", mode, str(source), str(target))
151        self.page([self.record("bare"), self.record("foreign")])
152        self.run_script(success=False)
153        for name in ("bare", "foreign"):
154            target = self.destination / (name + ".git")
155            self.assertEqual(self.git(target, "config", "--get", "remote.origin.url"),
156                             str(self.sources / (name + ".git")))
157            self.assertEqual(self.git(target, "rev-parse", "HEAD"),
158                             self.git(self.sources / (name + ".git"), "rev-parse", "HEAD"))
159
160    def test_refuses_symlink_destination(self):
161        source = self.source("project")
162        self.destination.mkdir()
163        (self.destination / "project.git").symlink_to(source, target_is_directory=True)
164        self.page([self.record("project")])
165        self.run_script(success=False)
166        self.assertFalse((source / "description").exists())
167
168    def test_empty_repository_and_failed_clone(self):
169        self.source("empty", empty=True)
170        self.source("working")
171        self.page([self.record("missing"), self.record("empty"), self.record("working")])
172        result = self.run_script(success=False)
173        self.assertIn("Mirrored 2 repositories", result.stdout)
174        self.assertFalse((self.destination / "missing.git").exists())
175        self.assertTrue((self.destination / "working.git" / "HEAD").is_file())
176        self.assertEqual([p for p in self.destination.glob(".github-mirror.*") if p.is_dir()], [])
177
178    def test_api_failure_does_not_touch_repositories(self):
179        self.env["TEST_API_FAILURE"] = "1"
180        result = self.run_script(success=False)
181        self.assertIn("GitHub API request failed", result.stderr)
182        self.assertEqual(list(self.destination.glob("*.git")), [])
183        self.assertEqual([p for p in self.destination.glob(".github-mirror.*") if p.is_dir()], [])
184
185    def test_destination_lock(self):
186        self.destination.mkdir()
187        self.page([])
188        import fcntl
189        with (self.destination / ".github-mirror.lock").open("w") as lock:
190            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
191            result = self.run_script(success=False)
192            self.assertIn("already running", result.stderr)
193        self.assertFalse((self.root / "calls").exists())
194
195    def test_rejects_invalid_username(self):
196        result = subprocess.run([str(SCRIPT), "../unsafe", str(self.destination)],
197                                env=self.env, text=True, capture_output=True)
198        self.assertNotEqual(result.returncode, 0)
199        self.assertFalse(self.destination.exists())
200
201
202if __name__ == "__main__":
203    unittest.main()