76 lines
5.3 KiB
Python
76 lines
5.3 KiB
Python
import importlib.util
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
def run(cwd, *args): return subprocess.check_output(["git", *args], cwd=cwd, text=True).strip()
|
|
def mod():
|
|
p = Path(__file__).parents[1] / "active-runtime/app/aios_isolated_writer_api/integration_merge_writer.py"
|
|
s = importlib.util.spec_from_file_location("merge_under_test", p); m = importlib.util.module_from_spec(s); sys.modules[s.name] = m; s.loader.exec_module(m); return m
|
|
|
|
|
|
class IntegrationMergeWriterTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.m = mod(); self.temp = tempfile.TemporaryDirectory(); self.root = Path(self.temp.name); self.remote = self.root / "remote.git"; self.work = self.root / "work"
|
|
subprocess.run(["git", "init", "--bare", self.remote], check=True, capture_output=True); subprocess.run(["git", "init", self.work], check=True, capture_output=True)
|
|
for k, v in (("user.email", "t@example.invalid"), ("user.name", "T")): subprocess.run(["git", "config", k, v], cwd=self.work, check=True)
|
|
(self.work / "a").write_text("base"); subprocess.run(["git", "add", "."], cwd=self.work, check=True); subprocess.run(["git", "commit", "-m", "base"], cwd=self.work, check=True, capture_output=True)
|
|
subprocess.run(["git", "branch", "-M", "integration/x"], cwd=self.work, check=True); subprocess.run(["git", "remote", "add", "origin", self.remote], cwd=self.work, check=True); subprocess.run(["git", "push", "origin", "integration/x"], cwd=self.work, check=True, capture_output=True)
|
|
subprocess.run(["git", "checkout", "-b", "feature/x"], cwd=self.work, check=True, capture_output=True); (self.work / "b").write_text("source"); subprocess.run(["git", "add", "."], cwd=self.work, check=True); subprocess.run(["git", "commit", "-m", "source"], cwd=self.work, check=True, capture_output=True); self.source = run(self.work, "rev-parse", "HEAD"); subprocess.run(["git", "checkout", "integration/x"], cwd=self.work, check=True, capture_output=True)
|
|
|
|
def tearDown(self): self.temp.cleanup()
|
|
def test_no_ff_parent_order_and_scope_rejection(self):
|
|
remote_head = lambda branch: run(self.work, "ls-remote", "--heads", "origin", "refs/heads/" + branch).split()[0]
|
|
writer = self.m.IntegrationMergeWriter(self.work, "origin", remote_head)
|
|
result = writer.merge(self.m.MergeRequest("integration/x", "feature/x", self.source))
|
|
self.assertEqual("COMPLETED", result["state"])
|
|
self.assertEqual([result["pre_merge_oid"], self.source], run(self.work, "show", "-s", "--format=%P", "HEAD").split())
|
|
for integration, source, oid in (
|
|
("main", "feature/x", self.source),
|
|
("integration/x", "governance/approvals", self.source),
|
|
("integration/x", "refs/tags/v1", self.source),
|
|
("integration/x", "feature/x", "z" * 40),
|
|
):
|
|
with self.subTest(integration=integration, source=source), self.assertRaisesRegex(self.m.MergeError, "scope_rejected"):
|
|
writer.merge(self.m.MergeRequest(integration, source, oid))
|
|
|
|
def test_merge_conflict_aborts_and_restores_clean_worktree(self):
|
|
subprocess.run(["git", "checkout", "-b", "feature/conflict", "integration/x"], cwd=self.work, check=True, capture_output=True)
|
|
(self.work / "a").write_text("source conflict")
|
|
subprocess.run(["git", "commit", "-am", "source conflict"], cwd=self.work, check=True, capture_output=True)
|
|
source = run(self.work, "rev-parse", "HEAD")
|
|
subprocess.run(["git", "checkout", "integration/x"], cwd=self.work, check=True, capture_output=True)
|
|
(self.work / "a").write_text("target conflict")
|
|
subprocess.run(["git", "commit", "-am", "target conflict"], cwd=self.work, check=True, capture_output=True)
|
|
subprocess.run(["git", "push", "origin", "integration/x"], cwd=self.work, check=True, capture_output=True)
|
|
before = run(self.work, "rev-parse", "HEAD")
|
|
remote_head = lambda branch: run(self.work, "ls-remote", "--heads", "origin", "refs/heads/" + branch).split()[0]
|
|
writer = self.m.IntegrationMergeWriter(self.work, "origin", remote_head)
|
|
with self.assertRaisesRegex(self.m.MergeError, "integration_merge_conflict"):
|
|
writer.merge(self.m.MergeRequest("integration/x", "feature/conflict", source))
|
|
self.assertEqual(before, run(self.work, "rev-parse", "HEAD"))
|
|
self.assertEqual("", run(self.work, "status", "--porcelain=v1"))
|
|
|
|
def test_push_failure_restores_or_enters_reconciliation(self):
|
|
before = run(self.work, "rev-parse", "HEAD")
|
|
writer = self.m.IntegrationMergeWriter(self.work, str(self.root / "missing.git"), lambda branch: before)
|
|
restored = writer.merge(self.m.MergeRequest("integration/x", "feature/x", self.source))
|
|
self.assertEqual("PUSH_FAILED_RESTORED", restored["state"])
|
|
self.assertEqual(before, run(self.work, "rev-parse", "HEAD"))
|
|
|
|
calls = 0
|
|
def uncertain(branch):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 1: return before
|
|
raise TimeoutError()
|
|
writer = self.m.IntegrationMergeWriter(self.work, "origin", uncertain)
|
|
unknown = writer.merge(self.m.MergeRequest("integration/x", "feature/x", self.source))
|
|
self.assertEqual("RECONCILIATION_REQUIRED", unknown["state"])
|
|
|
|
|
|
if __name__ == "__main__": unittest.main()
|