weichen
d3d60f4a40
Why: waiting branches should keep their work while one candidate syncs and merges, and a conflict must not block unrelated healthy work. Evidence: SELF three Git scenarios and governance suite captured by make evidence; real CI and three-business-task acceptance remain NOT_RUN. Impact: local regression scenarios only; no automatic queue mutex, production merge or measured CI savings are claimed. Refs: U6; tests/test_merge_queue_trial.py; .task/feature-merge-queue-policy-draft.md
81 lines
3.6 KiB
Python
81 lines
3.6 KiB
Python
"""Real disposable Git scenarios; these do not claim CI or automatic queue locking."""
|
|
import unittest
|
|
|
|
import test_task as lifecycle
|
|
|
|
|
|
class QueueTrialTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.fixture = lifecycle.TaskLifecycleTests(methodName="runTest")
|
|
self.fixture.setUp()
|
|
self.addCleanup(self.fixture.doCleanups)
|
|
|
|
def create_task(self, branch):
|
|
f = self.fixture
|
|
wt = f.root / branch.replace("/", "-")
|
|
f.git(f.repo, "worktree", "add", "-b", branch, str(wt), "origin/main")
|
|
f.record_task_card(wt)
|
|
name = branch.split("/")[1]
|
|
(wt / name).write_text(name + "\n")
|
|
f.git(wt, "add", name)
|
|
f.git(wt, "commit", "-m", "Add independent fixture change")
|
|
return wt
|
|
|
|
def test_sequential_delivery_does_not_sync_waiting_branches(self):
|
|
f = self.fixture
|
|
branches = ["feature/a", "feature/b", "feature/c"]
|
|
worktrees = {branch: self.create_task(branch) for branch in branches}
|
|
waiting = {branch: f.git(wt, "rev-parse", "HEAD").stdout for branch, wt in worktrees.items()}
|
|
f.advance_remote()
|
|
for position, branch in enumerate(branches):
|
|
wt = worktrees[branch]
|
|
f.task(wt, "sync")
|
|
candidate = f.git(wt, "rev-parse", "HEAD").stdout
|
|
second = f.task(wt, "sync")
|
|
self.assertIn("changed=0", second.stdout)
|
|
self.assertEqual(f.git(wt, "rev-parse", "HEAD").stdout, candidate)
|
|
self.assertEqual(f.git(wt, "status", "--porcelain").stdout, "")
|
|
f.git(f.repo, "merge", "--no-ff", branch, "-m", "Merge fixture candidate")
|
|
f.git(f.repo, "push", "origin", "main")
|
|
for pending in branches[position + 1:]:
|
|
self.assertEqual(f.git(worktrees[pending], "rev-parse", "HEAD").stdout, waiting[pending])
|
|
for name in ("a", "b", "c"):
|
|
self.assertEqual((f.repo / name).read_text(), name + "\n")
|
|
|
|
def test_conflicted_task_does_not_block_an_independent_healthy_task(self):
|
|
f = self.fixture
|
|
conflict = self.create_task("feature/conflict")
|
|
healthy = self.create_task("feature/healthy")
|
|
(conflict / "tracked").write_text("task choice\n")
|
|
f.git(conflict, "commit", "-am", "Conflicting task edit")
|
|
(f.repo / "tracked").write_text("target choice\n")
|
|
f.git(f.repo, "commit", "-am", "Target edit")
|
|
f.git(f.repo, "push", "origin", "main")
|
|
failed = f.task(conflict, "sync", check=False)
|
|
self.assertNotEqual(failed.returncode, 0)
|
|
self.assertNotIn("[synced]", failed.stdout)
|
|
self.assertIn("UU tracked", f.git(conflict, "status", "--porcelain").stdout)
|
|
succeeded = f.task(healthy, "sync")
|
|
self.assertIn("[synced]", succeeded.stdout)
|
|
self.assertEqual((healthy / "tracked").read_text(), "target choice\n")
|
|
self.assertEqual((healthy / "healthy").read_text(), "healthy\n")
|
|
|
|
def test_external_target_advance_requires_a_new_candidate(self):
|
|
f = self.fixture
|
|
task = self.create_task("feature/a")
|
|
f.task(task, "sync")
|
|
original = f.git(task, "rev-parse", "HEAD").stdout
|
|
f.advance_remote()
|
|
target = f.git(f.repo, "rev-parse", "HEAD").stdout.strip()
|
|
result = f.task(task, "sync")
|
|
current = f.git(task, "rev-parse", "HEAD").stdout
|
|
self.assertNotEqual(current, original)
|
|
self.assertIn("changed=1", result.stdout)
|
|
f.git(task, "merge-base", "--is-ancestor", target, "HEAD")
|
|
self.assertEqual((task / "a").read_text(), "a\n")
|
|
self.assertEqual((task / "new-main").read_text(), "new main\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|