validate-pr: accept context-only cherry-picks

.github/scripts/test_validate_pr.py

@@ -0,0 +1,327 @@
+#!/usr/bin/env python3
+"""Black-box regression tests for validate-pr cherry-pick matching."""
+
+import os
+import pathlib
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+
+
+VALIDATE_PR = pathlib.Path(__file__).with_name("validate-pr")
+SUBJECT = "fixture: insert payload"
+UPSTREAM_SOB = "Signed-off-by: Upstream Author <upstream@example.com>"
+LOCAL_SOB = "Signed-off-by: Local Author <local@example.com>"
+
+
+class ValidatePrPatchIdTest(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.repo = pathlib.Path(self._tmp.name) / "repo"
+        self.repo.mkdir()
+        self.real_git = shutil.which("git")
+        self.assertIsNotNone(self.real_git)
+
+        self._git("init")
+        self._set_identity("Fixture Author", "fixture@example.com")
+        self._write_fixture([
+            "first-before",
+            "anchor",
+            "first-after",
+            "spacer-1",
+            "spacer-2",
+            "spacer-3",
+            "spacer-4",
+            "second-before",
+            "anchor",
+            "second-after",
+        ])
+        self.base = self._commit("fixture: base", LOCAL_SOB)
+
+    def _relocate_repo(self, name):
+        relocated = self.repo.parent / name
+        self.repo.rename(relocated)
+        self.repo = relocated
+
+    def _git(self, *args, input_text=None):
+        result = subprocess.run(
+            ["git", *args],
+            cwd=self.repo,
+            input=input_text,
+            text=True,
+            capture_output=True,
+        )
+        if result.returncode:
+            self.fail(
+                "git {} failed ({}):\n{}{}".format(
+                    " ".join(args), result.returncode,
+                    result.stdout, result.stderr))
+        return result.stdout.strip()
+
+    def _set_identity(self, name, email):
+        self._git("config", "user.name", name)
+        self._git("config", "user.email", email)
+
+    def _write_fixture(self, lines):
+        (self.repo / "fixture.txt").write_text("\n".join(lines) + "\n")
+
+    def _read_fixture(self):
+        return (self.repo / "fixture.txt").read_text().splitlines()
+
+    def _use_identical_occurrence_fixture(self):
+        block = [
+            "same-before-4",
+            "same-before-3",
+            "same-before-2",
+            "same-before-1",
+            "anchor",
+            "same-after-1",
+            "same-after-2",
+            "same-after-3",
+            "same-after-4",
+        ]
+        separator = ["separator-{}".format(i) for i in range(1, 8)]
+        self._write_fixture(["prefix", *block, *separator, *block, "suffix"])
+        self.base = self._commit("fixture: duplicate context", LOCAL_SOB)
+
+    def _commit(self, subject, body):
+        self._git("add", "fixture.txt")
+        self._git("commit", "-F", "-",
+                  input_text="{}\n\n{}\n".format(subject, body))
+        return self._git("rev-parse", "HEAD")
+
+    def _patch_id(self, commit):
+        patch = self._git("show", commit)
+        output = self._git("patch-id", "--stable", input_text=patch)
+        return output.split()[0]
+
+    def _build_case(self, change, context_parent=True):
+        self._git("checkout", "-b", "upstream-topic", self.base)
+        self._set_identity("Upstream Author", "upstream@example.com")
+        lines = self._read_fixture()
+        lines.insert(lines.index("anchor") + 1, "payload")
+        self._write_fixture(lines)
+        upstream = self._commit(SUBJECT, UPSTREAM_SOB)
+        self._git("update-ref", "refs/remotes/upstream/linux", upstream)
+
+        self._git("checkout", "-b", "local", self.base)
+        self._set_identity("Local Author", "local@example.com")
+        if context_parent:
+            lines = self._read_fixture()
+            lines[0] = "local-first-before"
+            self._write_fixture(lines)
+            self._commit("fixture: adjust local context", LOCAL_SOB)
+        parent = self._git("rev-parse", "HEAD")
+
+        if change == "exact":
+            self._git("cherry-pick", "-x", "--signoff", upstream)
+        else:
+            lines = self._read_fixture()
+            anchors = [i for i, line in enumerate(lines) if line == "anchor"]
+            if change == "context":
+                lines.insert(anchors[0] + 1, "payload")
+            elif change == "mutated":
+                lines.insert(anchors[0] + 1, "payload-mutated")
+            elif change == "wrong-occurrence":
+                lines.insert(anchors[1] + 1, "payload")
+            else:
+                self.fail("unknown fixture change: {}".format(change))
+            self._write_fixture(lines)
+            self._commit(
+                SUBJECT,
+                "{}\n\n(cherry picked from commit {})\n{}".format(
+                    UPSTREAM_SOB, upstream, LOCAL_SOB))
+
+        local = self._git("rev-parse", "HEAD")
+        return parent, local
+
+    def _fake_git_env(self, mode):
+        fake_bin = self.repo / "fake-bin"
+        fake_bin.mkdir(exist_ok=True)
+        wrapper = fake_bin / "git"
+        wrapper.write_text("""#!/usr/bin/env python3
+import os
+import subprocess
+import sys
+
+real_git = os.environ["VALIDATE_PR_REAL_GIT"]
+mode = os.environ["VALIDATE_PR_FAKE_GIT_MODE"]
+args = sys.argv[1:]
+
+if mode == "show-failure" and args and args[0] == "show":
+    result = subprocess.run([real_git, *args], capture_output=True)
+    sys.stdout.buffer.write(result.stdout)
+    sys.stderr.buffer.write(result.stderr)
+    sys.exit(1)
+
+if (args[:2] == ["patch-id", "--stable"] and
+        mode in ("patch-id-malformed", "patch-id-extra-line")):
+    sys.stdin.buffer.read()
+    if mode == "patch-id-malformed":
+        sys.stdout.write("not-a-patch-id not-a-commit-id\\n")
+        sys.exit(0)
+    if mode == "patch-id-extra-line":
+        sys.stdout.write("{} {}\\n{} {}\\n".format(
+            "a" * 40, "b" * 40, "c" * 40, "d" * 40))
+        sys.exit(0)
+
+if (mode == "rev-parse-invalid-utf8" and
+        args[:3] == ["rev-parse", "--git-path", "objects"]):
+    sys.stdout.buffer.write(b"\\xff\\n")
+    sys.exit(0)
+
+if mode == "merge-tree-failure" and args and args[0] == "merge-tree":
+    sys.stderr.write("synthetic merge-tree failure\\n")
+    sys.exit(2)
+
+if mode == "merge-tree-extra-line" and args and args[0] == "merge-tree":
+    result = subprocess.run([real_git, *args], capture_output=True)
+    sys.stdout.buffer.write(result.stdout)
+    if result.stdout and not result.stdout.endswith(b"\\n"):
+        sys.stdout.buffer.write(b"\\n")
+    sys.stdout.buffer.write(b"unexpected-extra-line\\n")
+    sys.stderr.buffer.write(result.stderr)
+    sys.exit(result.returncode)
+
+os.execv(real_git, [real_git, *args])
+""")
+        wrapper.chmod(0o755)
+        env = os.environ.copy()
+        env["PATH"] = str(fake_bin) + os.pathsep + env.get("PATH", "")
+        env["VALIDATE_PR_REAL_GIT"] = self.real_git
+        env["VALIDATE_PR_FAKE_GIT_MODE"] = mode
+        return env
+
+    def _validate(self, parent, local, git_mode=None):
+        return subprocess.run(
+            [sys.executable, str(VALIDATE_PR),
+             "{}..{}".format(parent, local), "upstream", "linux",
+             "--no-update"],
+            cwd=self.repo,
+            env=self._fake_git_env(git_mode) if git_mode else None,
+            text=True,
+            capture_output=True,
+        )
+
+    @staticmethod
+    def _output(result):
+        return "stdout:\n{}\nstderr:\n{}".format(
+            result.stdout, result.stderr)
+
+    def _patch_id_status(self, result, local):
+        marker = "│ {} │".format(local[:12])
+        for line in result.stdout.splitlines():
+            if marker in line:
+                cells = line.split("│")
+                self.assertGreaterEqual(len(cells), 6, self._output(result))
+                return cells[3].strip()
+        self.fail("no digest row for {}:\n{}".format(
+            local[:12], self._output(result)))
+
+    def test_accepts_context_only_replay(self):
+        parent, local = self._build_case("context")
+
+        result = self._validate(parent, local)
+
+        self.assertEqual(result.returncode, 0, self._output(result))
+        self.assertEqual(self._patch_id_status(result, local), "context")
+
+    def test_accepts_context_only_replay_with_colon_in_object_path(self):
+        self._relocate_repo("repo:colon")
+        parent, local = self._build_case("context")
+
+        result = self._validate(parent, local)
+
+        self.assertEqual(result.returncode, 0, self._output(result))
+        self.assertEqual(self._patch_id_status(result, local), "context")
+
+    def test_accepts_exact_cherry_pick(self):
+        parent, local = self._build_case("exact", context_parent=False)
+
+        result = self._validate(parent, local)
+
+        self.assertEqual(result.returncode, 0, self._output(result))
+        self.assertEqual(self._patch_id_status(result, local), "match")
+
+    def test_rejects_git_show_failure_with_patch_output(self):
+        parent, local = self._build_case("exact", context_parent=False)
+
+        result = self._validate(parent, local, "show-failure")
+
+        self.assertEqual(result.returncode, 1, self._output(result))
+        self.assertIn("patch-ID mismatch with upstream", result.stdout)
+
+    def test_rejects_malformed_patch_id_output(self):
+        parent, local = self._build_case("exact", context_parent=False)
+
+        result = self._validate(parent, local, "patch-id-malformed")
+
+        self.assertEqual(result.returncode, 1, self._output(result))
+        self.assertIn("patch-ID mismatch with upstream", result.stdout)
+
+    def test_rejects_merge_tree_output_with_extra_line(self):
+        parent, local = self._build_case("context")
+
+        result = self._validate(parent, local, "merge-tree-extra-line")
+
+        self.assertEqual(result.returncode, 1, self._output(result))
+        self.assertIn("patch-ID mismatch with upstream", result.stdout)
+
+    def test_reports_replay_environment_failure(self):
+        parent, local = self._build_case("exact", context_parent=False)
+
+        result = self._validate(parent, local, "rev-parse-invalid-utf8")
+
+        self.assertEqual(result.returncode, 1, self._output(result))
+        self.assertIn("unable to verify patch replay", result.stderr)
+
+    def test_reports_merge_tree_failure(self):
+        parent, local = self._build_case("exact", context_parent=False)
+
+        result = self._validate(parent, local, "merge-tree-failure")
+
+        self.assertEqual(result.returncode, 1, self._output(result))
+        self.assertIn("synthetic merge-tree failure", result.stderr)
+
+    def test_rejects_multiple_patch_id_output_lines(self):
+        parent, local = self._build_case("exact", context_parent=False)
+
+        result = self._validate(parent, local, "patch-id-extra-line")
+
+        self.assertEqual(result.returncode, 1, self._output(result))
+        self.assertIn("patch-ID mismatch with upstream", result.stdout)
+
+    def test_rejects_mutated_payload(self):
+        parent, local = self._build_case("mutated")
+
+        result = self._validate(parent, local)
+
+        self.assertEqual(result.returncode, 1, self._output(result))
+        self.assertIn("patch-ID mismatch with upstream", result.stdout)
+
+    def test_rejects_same_change_at_different_occurrence(self):
+        parent, local = self._build_case("wrong-occurrence")
+
+        result = self._validate(parent, local)
+
+        self.assertEqual(result.returncode, 1, self._output(result))
+        self.assertIn("patch-ID mismatch with upstream", result.stdout)
+
+    def test_rejects_equal_patch_id_at_different_occurrence(self):
+        self._use_identical_occurrence_fixture()
+        parent, local = self._build_case(
+            "wrong-occurrence", context_parent=False)
+        upstream = self._git("rev-parse", "refs/remotes/upstream/linux")
+
+        self.assertEqual(self._patch_id(local), self._patch_id(upstream))
+        result = self._validate(parent, local)
+
+        self.assertEqual(result.returncode, 1, self._output(result))
+        self.assertIn("patch-ID mismatch with upstream", result.stdout)
+
+
+if __name__ == "__main__":
+    unittest.main()