mirror of
https://github.com/boykopovar/AnyPS5.git
synced 2026-10-18 23:07:07 -05:00
327 lines
16 KiB
Python
327 lines
16 KiB
Python
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
import pr_overlap
|
|
|
|
|
|
class OpenPullRequestsTests(unittest.TestCase):
|
|
def test_collects_pull_requests_beyond_the_previous_limit(self):
|
|
pull_requests = [
|
|
{"number": number, "user": {"login": f"author{number}"},
|
|
"head": {"sha": f"sha{number}"}, "body": None}
|
|
for number in range(1, 202)
|
|
]
|
|
pull_requests[-1]["body"] = "Depends on: #12\n- #34\n"
|
|
|
|
def response(*args):
|
|
if args[0] == "api":
|
|
return json.dumps(pull_requests)
|
|
limit = int(args[args.index("-L") + 1])
|
|
return json.dumps([
|
|
{"number": pr["number"], "author": pr["user"], "headRefOid": pr["head"]["sha"],
|
|
"body": pr["body"]} for pr in pull_requests[:limit]
|
|
])
|
|
|
|
with patch.object(pr_overlap, "REPO", "owner/project", create=True), \
|
|
patch.object(pr_overlap, "gh", side_effect=response):
|
|
found = pr_overlap.open_prs("main")
|
|
self.assertEqual(set(found), set(range(1, 202)))
|
|
self.assertEqual(found[201], {
|
|
"ref": "refs/pr/201", "sha": "sha201", "author": "author201", "depends": {12, 34},
|
|
})
|
|
self.assertEqual(found[1]["depends"], set())
|
|
|
|
def test_base_branch_is_encoded_in_the_paginated_request(self):
|
|
with patch.object(pr_overlap, "REPO", "owner/project", create=True), \
|
|
patch.object(pr_overlap, "gh", return_value="[]") as gh:
|
|
self.assertEqual(pr_overlap.open_prs("topic/one&two"), {})
|
|
gh.assert_called_once_with(
|
|
"api", "--paginate",
|
|
"repos/owner/project/pulls?state=open&base=topic%2Fone%26two&per_page=100",
|
|
)
|
|
|
|
def test_single_page_preserves_author_head_and_dependencies(self):
|
|
pull_requests = [{"number": 7, "user": {"login": "contributor"}, "head": {"sha": "head-sha"},
|
|
"body": "Depends on: none\nUnrelated issue #42"}]
|
|
with patch.object(pr_overlap, "REPO", "owner/project", create=True), \
|
|
patch.object(pr_overlap, "gh", return_value=json.dumps(pull_requests)):
|
|
found = pr_overlap.open_prs("main")
|
|
self.assertEqual(found, {
|
|
7: {"ref": "refs/pr/7", "sha": "head-sha", "author": "contributor", "depends": set()},
|
|
})
|
|
|
|
def test_collects_concatenated_json_pages_from_paginated_gh_output(self):
|
|
pull_requests = [
|
|
{"number": number, "user": {"login": f"author{number}"},
|
|
"head": {"sha": f"sha{number}"}, "body": None}
|
|
for number in range(1, 4)
|
|
]
|
|
pages = [pull_requests[:2], pull_requests[2:]]
|
|
output = "\n".join(json.dumps(page) for page in pages)
|
|
|
|
with patch.object(pr_overlap, "REPO", "owner/project", create=True), \
|
|
patch.object(pr_overlap, "gh", return_value=output):
|
|
found = pr_overlap.open_prs("main")
|
|
|
|
self.assertEqual(set(found), {1, 2, 3})
|
|
self.assertEqual(found[3]["sha"], "sha3")
|
|
|
|
def test_rejects_empty_paginated_output(self):
|
|
with patch.object(pr_overlap, "REPO", "owner/project", create=True), \
|
|
patch.object(pr_overlap, "gh", return_value=" \n"):
|
|
with self.assertRaisesRegex(ValueError, "returned no JSON pages"):
|
|
pr_overlap.open_prs("main")
|
|
|
|
|
|
class TopInsertTests(unittest.TestCase):
|
|
def test_top_of_file_insertions_clash_on_line_one(self):
|
|
pr = {"tree": "t", "files": {"f.cpp"}, "added": set()}
|
|
with patch.object(pr_overlap, "git", return_value="@@ -0,0 +1 @@\n+x\n"):
|
|
self.assertEqual(pr_overlap.hunks(pr, "f.cpp"), [(1, 1)])
|
|
self.assertEqual(pr_overlap.clashes(pr, pr, "f.cpp"), [(1, 1)])
|
|
|
|
def test_top_of_file_span_links_to_line_one(self):
|
|
self.assertEqual(pr_overlap.spans({"f.cpp": [(1, 1)]}, "https://x/blob/sha"),
|
|
"`f.cpp` [L1](https://x/blob/sha/f.cpp#L1)")
|
|
|
|
def test_ordinary_hunk_is_unchanged(self):
|
|
pr = {"tree": "t"}
|
|
with patch.object(pr_overlap, "git", return_value="@@ -3,2 +3,2 @@\n-a\n-b\n+c\n+d\n"):
|
|
self.assertEqual(pr_overlap.hunks(pr, "f.cpp"), [(3, 4)])
|
|
|
|
|
|
class FetchTests(unittest.TestCase):
|
|
def test_fetches_each_head_commit_the_api_reported(self):
|
|
prs = {
|
|
7: {"ref": "refs/pr/7", "sha": "a" * 40, "author": "one", "depends": set()},
|
|
9: {"ref": "refs/pr/9", "sha": "b" * 40, "author": "two", "depends": set()},
|
|
}
|
|
with patch.object(pr_overlap, "REPO", "owner/project", create=True), \
|
|
patch.object(pr_overlap, "git", return_value="") as git:
|
|
pr_overlap.fetch("main", prs)
|
|
git.assert_called_once_with(
|
|
"fetch", "--no-tags", "-q", "https://github.com/owner/project.git",
|
|
"+refs/heads/main:refs/pr/base", f"+{'a' * 40}:refs/pr/7", f"+{'b' * 40}:refs/pr/9",
|
|
)
|
|
|
|
def test_batches_large_sets_of_pull_request_heads(self):
|
|
prs = {
|
|
number: {
|
|
"ref": f"refs/pr/{number}", "sha": f"{number:040x}",
|
|
"author": "contributor", "depends": set(),
|
|
}
|
|
for number in range(250)
|
|
}
|
|
with patch.object(pr_overlap, "REPO", "owner/project", create=True), \
|
|
patch.object(pr_overlap, "git", return_value="") as git:
|
|
pr_overlap.fetch("main", prs)
|
|
self.assertEqual(git.call_count, 3)
|
|
self.assertEqual([len(call.args) - 4 for call in git.call_args_list], [100, 100, 51])
|
|
self.assertEqual(git.call_args_list[0].args[4], "+refs/heads/main:refs/pr/base")
|
|
self.assertEqual(git.call_args_list[-1].args[-1], "+00000000000000000000000000000000000000f9:refs/pr/249")
|
|
|
|
|
|
class ConflictTests(unittest.TestCase):
|
|
def test_uses_the_shared_base_parent_without_newer_merge_tree_option(self):
|
|
first = {"tree": "first-tree", "files": {"shared.cpp"}, "added": set()}
|
|
second = {"tree": "second-tree", "files": {"shared.cpp"}, "added": set()}
|
|
with patch.object(pr_overlap, "merge", return_value=("merged-tree", set())) as merge:
|
|
self.assertEqual(pr_overlap.conflicts(first, second), {})
|
|
merge.assert_called_once_with("first-tree", "second-tree")
|
|
|
|
def real_pull_requests(self, main_changes, older_changes, current_changes):
|
|
temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(temporary.cleanup)
|
|
root = Path(temporary.name).resolve()
|
|
self.assertTrue(root.is_relative_to(Path(tempfile.gettempdir()).resolve()))
|
|
self.addCleanup(os.chdir, os.getcwd())
|
|
os.chdir(root)
|
|
|
|
def git(*args):
|
|
return subprocess.run(
|
|
["git", "-c", "user.name=Overlap Test", "-c", "user.email=test@example.com", *args],
|
|
capture_output=True, text=True, encoding="utf-8", check=True).stdout.strip()
|
|
|
|
git("init", "-q")
|
|
git("config", "commit.gpgsign", "false")
|
|
for name in ("shared.cpp", "other.cpp", "unrelated.cpp"):
|
|
(root / name).write_text("int value = 0;\n", encoding="utf-8")
|
|
git("add", ".")
|
|
git("commit", "-qm", "test: create base")
|
|
base = git("rev-parse", "HEAD")
|
|
|
|
def branch(name, parent, changes):
|
|
git("checkout", "-qb", name, parent)
|
|
for path, value in changes.items():
|
|
(root / path).write_text(f"int value = {value};\n", encoding="utf-8")
|
|
git("add", ".")
|
|
git("commit", "-qm", "test: change values", "--allow-empty")
|
|
return git("rev-parse", "HEAD")
|
|
|
|
older = branch("older", base, older_changes)
|
|
main = branch("updated-base", base, main_changes)
|
|
current = branch("current", main, current_changes)
|
|
git("update-ref", "refs/pr/base", main)
|
|
prs = {1: {"ref": current, "author": "current-author", "depends": set()},
|
|
2: {"ref": older, "author": "older-author", "depends": set()}}
|
|
for pr in prs.values():
|
|
pr_overlap.scan(pr)
|
|
return prs
|
|
|
|
def test_reports_shared_files_when_older_pr_conflicts_with_main(self):
|
|
prs = self.real_pull_requests({"shared.cpp": 2}, {"shared.cpp": 1}, {"shared.cpp": 3})
|
|
self.assertIsNone(prs[2]["tree"])
|
|
self.assertIsNotNone(prs[1]["tree"])
|
|
found = pr_overlap.overlaps(prs, 1, 2)
|
|
self.assertIsNotNone(found)
|
|
self.assertEqual(found["uncompared"], {"shared.cpp"})
|
|
report = pr_overlap.report({2: found}, "https://example.invalid/blob/base")
|
|
self.assertIn("could not be completed", report)
|
|
self.assertIn("`shared.cpp`", report)
|
|
self.assertNotIn("Merging one will break the other", report)
|
|
|
|
def test_conflict_in_another_file_does_not_hide_shared_changes(self):
|
|
prs = self.real_pull_requests({"other.cpp": 2}, {"other.cpp": 1, "shared.cpp": 1}, {"shared.cpp": 3})
|
|
self.assertIsNone(prs[2]["tree"])
|
|
found = pr_overlap.overlaps(prs, 1, 2)
|
|
self.assertIsNotNone(found)
|
|
self.assertEqual(found["uncompared"], {"shared.cpp"})
|
|
|
|
def test_unmergeable_pr_with_no_shared_files_is_not_reported(self):
|
|
prs = self.real_pull_requests({"other.cpp": 2}, {"other.cpp": 1}, {"unrelated.cpp": 3})
|
|
self.assertIsNone(prs[2]["tree"])
|
|
self.assertIsNone(pr_overlap.overlaps(prs, 1, 2))
|
|
|
|
def test_mergeable_branches_still_report_confirmed_line_conflicts(self):
|
|
prs = self.real_pull_requests({}, {"shared.cpp": 1}, {"shared.cpp": 2})
|
|
self.assertTrue(all(pr["tree"] for pr in prs.values()))
|
|
found = pr_overlap.overlaps(prs, 1, 2)
|
|
self.assertEqual(found["conflicts"], {"shared.cpp": [(1, 1)]})
|
|
self.assertNotIn("uncompared", found)
|
|
|
|
def test_mixed_report_keeps_confirmed_and_uncompared_sections(self):
|
|
rows = {2: {"added": set(), "conflicts": {"other.cpp": [(1, 1)]}},
|
|
3: {"added": set(), "uncompared": {"shared.cpp"}}}
|
|
report = pr_overlap.report(rows, "https://example.invalid/blob/base")
|
|
self.assertIn("[L1]", report)
|
|
self.assertIn("could not be completed", report)
|
|
self.assertIn("`shared.cpp`", report)
|
|
self.assertTrue(report.startswith(pr_overlap.MARKER))
|
|
|
|
def test_report_links_pull_requests_without_cross_references(self):
|
|
rows = {2: {"added": set(), "conflicts": {"other.cpp": [(1, 1)]}},
|
|
3: {"added": set(), "uncompared": {"shared.cpp"}}}
|
|
report = pr_overlap.report(rows, "https://github.com/owner/project/blob/base")
|
|
self.assertIsNone(re.search(r"#\d|https://github\.com/owner/project/(pull|issues)/", report))
|
|
self.assertIn("[PR 2](https://www.github.com/owner/project/pull/2)", report)
|
|
self.assertIn("[PR 3](https://www.github.com/owner/project/pull/3)", report)
|
|
|
|
|
|
class UnicodePathTests(unittest.TestCase):
|
|
def test_scans_and_reports_conflicts_with_git_path_quoting_enabled(self):
|
|
previous = os.getcwd()
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
|
|
def git(*args):
|
|
return subprocess.run(
|
|
["git", "-c", "user.name=Overlap Test", "-c", "user.email=test@example.com",
|
|
"-c", "commit.gpgsign=false", *args], cwd=root,
|
|
capture_output=True, text=True, encoding="utf-8", check=True).stdout.strip()
|
|
|
|
filename = "core/libs/prx/libSceExample/caf\u00e9.cpp"
|
|
source = root / filename
|
|
source.parent.mkdir(parents=True)
|
|
source.write_text("int value = 0;\n", encoding="utf-8")
|
|
git("init", "-q")
|
|
git("config", "core.quotePath", "true")
|
|
git("add", ".")
|
|
git("commit", "-qm", "test: create base")
|
|
base = git("rev-parse", "HEAD")
|
|
git("update-ref", "refs/pr/base", base)
|
|
for branch, value in (("left", 1), ("right", 2)):
|
|
git("checkout", "-qb", branch, base)
|
|
source.write_text(f"int value = {value};\n", encoding="utf-8")
|
|
git("add", ".")
|
|
git("commit", "-qm", "fix: change value")
|
|
try:
|
|
os.chdir(root)
|
|
left, right = {"ref": "left"}, {"ref": "right"}
|
|
pr_overlap.scan(left)
|
|
pr_overlap.scan(right)
|
|
self.assertEqual(left["files"], {filename})
|
|
self.assertEqual(right["files"], {filename})
|
|
self.assertEqual(pr_overlap.conflicts(left, right), {filename: [(1, 1)]})
|
|
finally:
|
|
os.chdir(previous)
|
|
|
|
|
|
class CommentStateTests(unittest.TestCase):
|
|
def setUp(self):
|
|
if os.name == "nt" or not all(shutil.which(tool) for tool in ("bash", "jq", "sed", "sha256sum")):
|
|
self.skipTest("the workflow step runs in bash with jq, sed and sha256sum on Linux")
|
|
lines = (Path(__file__).resolve().parents[2] / ".github/workflows/pr-overlap.yml").read_text(encoding="utf-8").splitlines()
|
|
start = next(index for index, line in enumerate(lines) if "tools/pr_overlap.py" in line)
|
|
indent = len(lines[start]) - len(lines[start].lstrip())
|
|
while start > 0 and len(lines[start - 1]) - len(lines[start - 1].lstrip()) == indent:
|
|
start -= 1
|
|
step = []
|
|
for line in lines[start:]:
|
|
if line.strip() and len(line) - len(line.lstrip()) < indent:
|
|
break
|
|
step.append(line[indent:])
|
|
self.step = "\n".join(step).replace("${{ github.event.repository.default_branch }}", "main")
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temp.name)
|
|
(self.root / "overlap").mkdir()
|
|
|
|
def tearDown(self):
|
|
self.temp.cleanup()
|
|
|
|
def report(self, commit, lines):
|
|
found = {7: {"conflicts": {"core/a.cpp": [lines]}, "added": set()}}
|
|
return pr_overlap.report(found, f"https://github.com/owner/project/blob/{commit}")
|
|
|
|
def run_step(self, report, comment):
|
|
(self.root / "overlap" / "42.md").write_text(report, encoding="utf-8", newline="\n")
|
|
(self.root / "comment").write_text(json.dumps([5, comment]) + "\n" if comment else "", encoding="utf-8", newline="\n")
|
|
calls = self.root / "calls"
|
|
calls.write_text("", encoding="utf-8")
|
|
script = "\n".join([
|
|
"set -eo pipefail",
|
|
"python3() { :; }",
|
|
'gh() { printf \'%s\\n\' "$*" >> "$RUNNER_TEMP/calls"; case "$*" in "api repos/"*) cat "$RUNNER_TEMP/comment";; esac; }',
|
|
self.step,
|
|
])
|
|
(self.root / "step.sh").write_text(script, encoding="utf-8", newline="\n")
|
|
environment = dict(os.environ, RUNNER_TEMP=str(self.root), GH_REPO="owner/project")
|
|
result = subprocess.run(["bash", "step.sh"], cwd=self.root, env=environment, capture_output=True, text=True, timeout=20)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
return [call.split(" ")[:3] for call in calls.read_text(encoding="utf-8").splitlines()]
|
|
|
|
def test_new_main_commit_alone_makes_no_requests(self):
|
|
first = self.report("a" * 40, (3, 4))
|
|
self.assertEqual(self.run_step(first, None), [["api", "repos/owner/project/issues/42/comments", "--paginate"], ["pr", "comment", "42"]])
|
|
self.assertEqual(self.run_step(self.report("b" * 40, (3, 4)), first), [])
|
|
|
|
def test_lost_state_reads_the_comment_without_rewriting_it(self):
|
|
(self.root / "overlap-state.json").write_text("{}", encoding="utf-8")
|
|
self.assertEqual(self.run_step(self.report("b" * 40, (3, 4)), self.report("a" * 40, (3, 4))), [["api", "repos/owner/project/issues/42/comments", "--paginate"]])
|
|
|
|
def test_changed_lines_still_update_the_comment(self):
|
|
first = self.report("a" * 40, (3, 4))
|
|
self.run_step(first, None)
|
|
self.assertEqual(self.run_step(self.report("b" * 40, (5, 6)), first), [["api", "repos/owner/project/issues/42/comments", "--paginate"], ["api", "-X", "PATCH"]])
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|