|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""Unit tests for merge_live.py (apply-repo-settings export deep-merge). |
| 3 | +
|
| 4 | +Run: python3 merge_live_test.py (requires ruamel.yaml) |
| 5 | +""" |
| 6 | +import json |
| 7 | +import os |
| 8 | +import subprocess |
| 9 | +import sys |
| 10 | +import tempfile |
| 11 | +import unittest |
| 12 | + |
| 13 | +HERE = os.path.dirname(os.path.abspath(__file__)) |
| 14 | +sys.path.insert(0, HERE) |
| 15 | + |
| 16 | +from merge_live import deep_merge # noqa: E402 |
| 17 | + |
| 18 | +MERGE = os.path.join(HERE, "merge_live.py") |
| 19 | + |
| 20 | + |
| 21 | +def merge_file(yaml_text, source_obj, write=True): |
| 22 | + """Run merge_live.py against yaml_text with source_obj; return (changed, new_text).""" |
| 23 | + with tempfile.NamedTemporaryFile("w", suffix=".yml", delete=False) as fh: |
| 24 | + fh.write(yaml_text) |
| 25 | + path = fh.name |
| 26 | + try: |
| 27 | + args = [sys.executable, MERGE, "--file", path] |
| 28 | + if write: |
| 29 | + args.append("--write") |
| 30 | + out = subprocess.run( |
| 31 | + args, input=json.dumps(source_obj), capture_output=True, text=True, check=True |
| 32 | + ).stdout |
| 33 | + changed = "changed=true" in out |
| 34 | + with open(path, encoding="utf-8") as fh: |
| 35 | + return changed, fh.read() |
| 36 | + finally: |
| 37 | + os.unlink(path) |
| 38 | + |
| 39 | + |
| 40 | +class DeepMergeLogic(unittest.TestCase): |
| 41 | + def test_scalar_target_wins(self): |
| 42 | + # file (target) value is kept; live (source) never overwrites it. |
| 43 | + out = deep_merge({"has_issues": False}, {"has_issues": True}) |
| 44 | + self.assertEqual(out["has_issues"], True) |
| 45 | + |
| 46 | + def test_adds_missing_key(self): |
| 47 | + out = deep_merge({"a": 1, "b": 2}, {"a": 9}) |
| 48 | + self.assertEqual(out, {"a": 9, "b": 2}) |
| 49 | + |
| 50 | + def test_nested_mapping_recurse(self): |
| 51 | + out = deep_merge( |
| 52 | + {"repository": {"has_issues": False, "default_branch": "main"}}, |
| 53 | + {"repository": {"has_issues": True}}, |
| 54 | + ) |
| 55 | + self.assertEqual(out["repository"], {"has_issues": True, "default_branch": "main"}) |
| 56 | + |
| 57 | + def test_rulesets_identity_merge_append_and_keep(self): |
| 58 | + source = { |
| 59 | + "rulesets": [ |
| 60 | + { |
| 61 | + "name": "protect", |
| 62 | + "bypass_actors": [ |
| 63 | + {"actor_id": 5, "actor_type": "RepositoryRole"}, |
| 64 | + {"actor_id": 99, "actor_type": "Integration"}, |
| 65 | + ], |
| 66 | + "rules": [{"type": "deletion"}, {"type": "non_fast_forward"}], |
| 67 | + }, |
| 68 | + {"name": "new-from-ui", "target": "branch"}, |
| 69 | + ] |
| 70 | + } |
| 71 | + target = { |
| 72 | + "rulesets": [ |
| 73 | + {"name": "pending-only", "target": "branch"}, # file-only, must stay |
| 74 | + { |
| 75 | + "name": "protect", |
| 76 | + "bypass_actors": [{"actor_id": 5, "actor_type": "RepositoryRole"}], |
| 77 | + "rules": [{"type": "deletion"}], |
| 78 | + }, |
| 79 | + ] |
| 80 | + } |
| 81 | + out = deep_merge(source, target) |
| 82 | + names = [r["name"] for r in out["rulesets"]] |
| 83 | + # pending-only kept, protect kept in place, new-from-ui appended at end. |
| 84 | + self.assertEqual(names, ["pending-only", "protect", "new-from-ui"]) |
| 85 | + protect = out["rulesets"][1] |
| 86 | + actor_ids = sorted(a["actor_id"] for a in protect["bypass_actors"]) |
| 87 | + self.assertEqual(actor_ids, [5, 99]) # missing actor added |
| 88 | + rule_types = sorted(r["type"] for r in protect["rules"]) |
| 89 | + self.assertEqual(rule_types, ["deletion", "non_fast_forward"]) # missing rule added |
| 90 | + |
| 91 | + def test_labels_collaborators_teams_identity(self): |
| 92 | + out = deep_merge( |
| 93 | + { |
| 94 | + "labels": [{"name": "bug", "color": "ff0000"}, {"name": "new", "color": "00ff00"}], |
| 95 | + "collaborators": [{"username": "alice", "permission": "admin"}], |
| 96 | + "teams": [{"name": "core", "permission": "push"}], |
| 97 | + }, |
| 98 | + { |
| 99 | + "labels": [{"name": "bug", "color": "d73a4a"}], # color kept (target wins) |
| 100 | + "collaborators": [{"username": "alice", "permission": "pull"}], |
| 101 | + "teams": [{"name": "core", "permission": "pull"}], |
| 102 | + }, |
| 103 | + ) |
| 104 | + labels = {label["name"]: label["color"] for label in out["labels"]} |
| 105 | + self.assertEqual(labels["bug"], "d73a4a") # target wins |
| 106 | + self.assertEqual(labels["new"], "00ff00") # appended |
| 107 | + self.assertEqual(out["collaborators"][0]["permission"], "pull") # target wins |
| 108 | + self.assertEqual(out["teams"][0]["permission"], "pull") # target wins |
| 109 | + |
| 110 | + def test_target_only_never_removed(self): |
| 111 | + out = deep_merge({"labels": []}, {"labels": [{"name": "keep", "color": "x"}]}) |
| 112 | + self.assertEqual([label["name"] for label in out["labels"]], ["keep"]) |
| 113 | + |
| 114 | + |
| 115 | +class CliBehavior(unittest.TestCase): |
| 116 | + def test_changed_true_when_adding(self): |
| 117 | + changed, text = merge_file( |
| 118 | + "repository:\n has_issues: true\n", |
| 119 | + {"repository": {"has_issues": False, "default_branch": "main"}}, |
| 120 | + ) |
| 121 | + self.assertTrue(changed) |
| 122 | + self.assertIn("default_branch: main", text) |
| 123 | + self.assertIn("has_issues: true", text) # not overwritten to false |
| 124 | + |
| 125 | + def test_changed_false_and_idempotent(self): |
| 126 | + yaml_text = ( |
| 127 | + "repository:\n has_issues: true\n" |
| 128 | + "labels:\n - name: bug\n color: \"d73a4a\"\n" |
| 129 | + ) |
| 130 | + changed, text = merge_file( |
| 131 | + yaml_text, |
| 132 | + {"repository": {"has_issues": True}, "labels": [{"name": "bug", "color": "d73a4a"}]}, |
| 133 | + ) |
| 134 | + self.assertFalse(changed) |
| 135 | + self.assertEqual(text, yaml_text) # byte-for-byte untouched when nothing to add |
| 136 | + |
| 137 | + def test_dry_run_does_not_write(self): |
| 138 | + yaml_text = "repository:\n has_issues: true\n" |
| 139 | + changed, text = merge_file( |
| 140 | + yaml_text, {"repository": {"default_branch": "main"}}, write=False |
| 141 | + ) |
| 142 | + self.assertTrue(changed) # would change |
| 143 | + self.assertEqual(text, yaml_text) # but file untouched (no --write) |
| 144 | + |
| 145 | + def test_comments_preserved(self): |
| 146 | + yaml_text = ( |
| 147 | + "# top-of-file comment\n" |
| 148 | + "rulesets:\n" |
| 149 | + " # keep this ruleset comment\n" |
| 150 | + " - name: protect # inline\n" |
| 151 | + " enforcement: active\n" |
| 152 | + ) |
| 153 | + changed, text = merge_file( |
| 154 | + yaml_text, |
| 155 | + {"rulesets": [{"name": "protect", "enforcement": "active", "target": "branch"}]}, |
| 156 | + ) |
| 157 | + self.assertTrue(changed) # added target: branch |
| 158 | + self.assertIn("# top-of-file comment", text) |
| 159 | + self.assertIn("# keep this ruleset comment", text) |
| 160 | + self.assertIn("# inline", text) |
| 161 | + self.assertIn("target: branch", text) |
| 162 | + |
| 163 | + |
| 164 | +if __name__ == "__main__": |
| 165 | + unittest.main() |
0 commit comments