mirror of
https://github.com/boykopovar/AnyPS5.git
synced 2026-10-22 00:22:57 -05:00
199 lines
9.2 KiB
Python
199 lines
9.2 KiB
Python
from pathlib import Path
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
import check_main_sync
|
|
|
|
TOOL = Path(__file__).resolve().parents[1] / "check_main_sync.py"
|
|
IDENTITY = {
|
|
"GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@example.com",
|
|
"GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@example.com",
|
|
}
|
|
|
|
|
|
class Repo:
|
|
def __init__(self, path):
|
|
self.path = path
|
|
self.git("init", "-q", "-b", "main")
|
|
self.write("shared.txt", "one\ntwo\nthree\n")
|
|
self.write("other.txt", "x\n")
|
|
self.commit("initial")
|
|
|
|
def git(self, *args):
|
|
subprocess.run(["git", "-C", str(self.path), *args], check=True, capture_output=True, env={**os.environ, **IDENTITY})
|
|
|
|
def write(self, name, text):
|
|
(self.path / name).write_text(text, encoding="utf-8")
|
|
|
|
def commit(self, message):
|
|
self.git("add", "-A")
|
|
self.git("commit", "-q", "-m", message)
|
|
|
|
def merge_pull(self, number, branch):
|
|
self.git("merge", "--no-ff", "-q", "-m", f"Merge pull request #{number} from someone/{branch}", branch)
|
|
|
|
|
|
def run_tool(path, *args):
|
|
return subprocess.run(
|
|
[sys.executable, str(TOOL), "--repo", str(path), "--base", "main", *args],
|
|
capture_output=True,
|
|
encoding="utf-8",
|
|
env={**os.environ, "PYTHONIOENCODING": "utf-8"},
|
|
)
|
|
|
|
|
|
class CheckMainSyncTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.directory = tempfile.TemporaryDirectory()
|
|
self.repo = Repo(Path(self.directory.name))
|
|
|
|
def tearDown(self):
|
|
self.directory.cleanup()
|
|
|
|
def branch(self, name):
|
|
self.repo.git("checkout", "-q", "-b", name)
|
|
|
|
def test_branch_level_with_main_is_clean_and_not_behind(self):
|
|
self.branch("feature")
|
|
self.repo.write("other.txt", "y\n")
|
|
self.repo.commit("feature change")
|
|
data = check_main_sync.report(self.repo.path, "main", "feature")
|
|
self.assertEqual((data["behind"], data["ahead"], data["conflicts"], data["shared"]), (0, 1, [], []))
|
|
self.assertEqual(run_tool(self.repo.path, "--head", "feature", "--strict").returncode, 0)
|
|
|
|
def test_behind_main_is_reported_and_only_fails_when_strict(self):
|
|
self.branch("feature")
|
|
self.repo.write("other.txt", "y\n")
|
|
self.repo.commit("feature change")
|
|
self.repo.git("checkout", "-q", "main")
|
|
self.repo.write("extra.txt", "new\n")
|
|
self.repo.commit("main moves")
|
|
relaxed = run_tool(self.repo.path, "--head", "feature")
|
|
strict = run_tool(self.repo.path, "--head", "feature", "--strict")
|
|
self.assertEqual(relaxed.returncode, 0, relaxed.stderr)
|
|
self.assertIn("1 ahead and 1 behind main", relaxed.stdout)
|
|
self.assertIn("merge: clean", relaxed.stdout)
|
|
self.assertEqual(strict.returncode, 1)
|
|
|
|
def test_conflict_names_the_file_and_the_merged_pull_request(self):
|
|
self.branch("feature")
|
|
self.repo.write("shared.txt", "one\nmine\nthree\n")
|
|
self.repo.commit("feature edit")
|
|
self.repo.git("checkout", "-q", "main")
|
|
self.repo.git("checkout", "-q", "-b", "theirs")
|
|
self.repo.write("shared.txt", "one\ntheirs\nthree\n")
|
|
self.repo.commit("their edit")
|
|
self.repo.git("checkout", "-q", "main")
|
|
self.repo.merge_pull(4242, "theirs")
|
|
result = run_tool(self.repo.path, "--head", "feature")
|
|
self.assertEqual(result.returncode, 1, result.stderr)
|
|
self.assertIn("merge: CONFLICT in 1 file(s)", result.stdout)
|
|
self.assertIn(" shared.txt", result.stdout)
|
|
self.assertIn("merged pull requests that touched them: #4242", result.stdout)
|
|
|
|
def test_files_changed_on_both_sides_are_listed_when_the_merge_is_clean(self):
|
|
self.branch("feature")
|
|
self.repo.write("shared.txt", "zero\none\ntwo\nthree\n")
|
|
self.repo.commit("feature edit")
|
|
self.repo.git("checkout", "-q", "main")
|
|
self.repo.git("checkout", "-q", "-b", "theirs")
|
|
self.repo.write("shared.txt", "one\ntwo\nthree\nfour\n")
|
|
self.repo.commit("their edit")
|
|
self.repo.git("checkout", "-q", "main")
|
|
self.repo.merge_pull(77, "theirs")
|
|
result = run_tool(self.repo.path, "--head", "feature")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("merge: clean", result.stdout)
|
|
self.assertIn("1 file(s) changed on both sides", result.stdout)
|
|
self.assertIn("#77", result.stdout)
|
|
|
|
def test_paths_with_spaces_and_unicode_are_reported_verbatim(self):
|
|
name = "sp ace \u00e9.txt"
|
|
self.repo.write(name, "one\ntwo\n")
|
|
self.repo.commit("add")
|
|
self.branch("feature")
|
|
self.repo.write(name, "one\nmine\n")
|
|
self.repo.commit("feature edit")
|
|
self.repo.git("checkout", "-q", "main")
|
|
self.repo.git("checkout", "-q", "-b", "theirs")
|
|
self.repo.write(name, "one\ntheirs\n")
|
|
self.repo.commit("their edit")
|
|
self.repo.git("checkout", "-q", "main")
|
|
self.repo.merge_pull(9, "theirs")
|
|
result = run_tool(self.repo.path, "--head", "feature")
|
|
self.assertEqual(result.returncode, 1, result.stderr)
|
|
self.assertIn(f" {name}\n", result.stdout.replace("\r", ""))
|
|
self.assertIn("#9", result.stdout)
|
|
|
|
def test_unrelated_histories_are_an_error(self):
|
|
self.repo.git("checkout", "-q", "--orphan", "island")
|
|
self.repo.write("lone.txt", "x\n")
|
|
self.repo.commit("island")
|
|
result = run_tool(self.repo.path, "--head", "island")
|
|
self.assertEqual(result.returncode, 2)
|
|
self.assertIn("no common ancestor", result.stderr)
|
|
|
|
def check_unicode_output(self, conflict):
|
|
name = "sp ace \u00e9 \u6e2c \U0001f680.txt"
|
|
self.repo.write(name, "one\ntwo\nthree\n")
|
|
self.repo.commit("add unicode path")
|
|
self.branch("feature")
|
|
self.repo.write(name, "one\nmine\nthree\n" if conflict else "zero\none\ntwo\nthree\n")
|
|
self.repo.commit("feature edit")
|
|
self.repo.git("checkout", "-q", "main")
|
|
self.repo.git("checkout", "-q", "-b", "theirs")
|
|
self.repo.write(name, "one\ntheirs\nthree\n" if conflict else "one\ntwo\nthree\nfour\n")
|
|
self.repo.commit("their edit")
|
|
self.repo.git("checkout", "-q", "main")
|
|
self.repo.merge_pull(9, "theirs")
|
|
expected = {"ascii": r"sp ace \xe9 \u6e2c \U0001f680.txt",
|
|
"cp1252": "sp ace \u00e9 " + r"\u6e2c \U0001f680.txt",
|
|
"utf-8": name}
|
|
for encoding, shown in expected.items():
|
|
for strict in ((False,) if conflict else (False, True)):
|
|
with self.subTest(encoding=encoding, strict=strict):
|
|
result = subprocess.run([sys.executable, str(TOOL), "--repo", str(self.repo.path),
|
|
"--base", "main", "--head", "feature", *(["--strict"] if strict else [])],
|
|
capture_output=True, text=True, encoding=encoding,
|
|
env={**os.environ, "PYTHONIOENCODING": encoding + ":strict"})
|
|
self.assertEqual(result.returncode, 1 if conflict or strict else 0, result.stderr)
|
|
self.assertIn("merge: CONFLICT" if conflict else "merge: clean", result.stdout)
|
|
self.assertIn(f" {shown}\n", result.stdout.replace("\r", ""))
|
|
self.assertIn("merged pull requests that touched them: #9", result.stdout)
|
|
self.assertNotIn("Traceback", result.stderr)
|
|
|
|
def test_clean_unicode_report_preserves_exit_status_and_output_encoding(self):
|
|
self.check_unicode_output(False)
|
|
|
|
def test_conflicting_unicode_report_preserves_exit_status_and_output_encoding(self):
|
|
self.check_unicode_output(True)
|
|
|
|
def test_missing_base_is_an_error_not_a_pass(self):
|
|
result = subprocess.run([sys.executable, str(TOOL), "--repo", self.directory.name, "--base", "upstream/main"], capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 2)
|
|
self.assertIn("no base ref found", result.stderr)
|
|
|
|
def test_explicit_base_is_fetched_before_it_is_resolved(self):
|
|
with tempfile.TemporaryDirectory() as other:
|
|
subprocess.run(["git", "clone", "-q", self.directory.name, other], check=True, capture_output=True)
|
|
self.repo.git("remote", "add", "upstream", other)
|
|
result = subprocess.run([sys.executable, str(TOOL), "--repo", self.directory.name, "--base", "upstream/main", "--fetch"], capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("merge: clean", result.stdout)
|
|
|
|
def test_default_base_is_fetched_before_it_is_resolved(self):
|
|
with tempfile.TemporaryDirectory() as other:
|
|
subprocess.run(["git", "clone", "-q", self.directory.name, other], check=True, capture_output=True)
|
|
self.repo.git("remote", "add", "upstream", other)
|
|
result = subprocess.run([sys.executable, str(TOOL), "--repo", self.directory.name, "--fetch"], capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("merge: clean", result.stdout)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|