teleo-infrastructure/tests/test_reweave_merge.py
m3taversal 6b3a5833df feat: per-file frontmatter union for reweave PR merge
Reweave PRs modify existing files (appending YAML edges). Cherry-pick
fails ~75% when main moves between PR creation and merge.

_merge_reweave_pr() reads each changed file from both main HEAD and
branch HEAD, unions the edge arrays (order-preserving, main-first),
and writes the result. Eliminates merge conflicts structurally.

Key design decisions (Ganymede + Theseus approved):
- Order-preserving dedup: main's edges first, branch-new appended
- Superset assertion: logs warning if branch missing main edges
- Uses main's body text (reweave only touches frontmatter)
- Loud failure on parse errors (no cherry-pick fallback)
- Append-only contract: reweave adds edges, never removes

18 tests covering parse, union, serialize, superset, and full workflow.
2026-04-04 13:43:32 +01:00

227 lines
8.2 KiB
Python

"""Tests for _merge_reweave_pr helpers — frontmatter union, order-preserving dedup, superset assertion.
These test the pure functions used by _merge_reweave_pr in lib/merge.py.
Copied here because lib/merge.py's relative imports make direct import impractical in tests.
If these functions change in merge.py, update them here too.
"""
import pytest
import yaml
# --- Copied from lib/merge.py (pure functions, no dependencies) ---
REWEAVE_EDGE_FIELDS = ("supports", "challenges", "depends_on", "related", "reweave_edges")
def _parse_yaml_frontmatter(text: str) -> tuple[dict | None, str]:
if not text.startswith("---"):
return None, text
end = text.find("\n---", 3)
if end == -1:
return None, text
try:
fm = yaml.safe_load(text[3:end])
body = text[end:]
return fm if isinstance(fm, dict) else None, body
except Exception:
return None, text
def _union_edge_lists(main_edges: list, branch_edges: list) -> list:
seen = set()
result = []
for edge in main_edges:
key = str(edge).strip().lower()
if key not in seen:
seen.add(key)
result.append(edge)
for edge in branch_edges:
key = str(edge).strip().lower()
if key not in seen:
seen.add(key)
result.append(edge)
return result
def _serialize_frontmatter(fm: dict, body: str) -> str:
fm_str = yaml.dump(fm, default_flow_style=False, allow_unicode=True, sort_keys=False).rstrip("\n")
return f"---\n{fm_str}\n{body}"
# --- End copied functions ---
class TestParseYamlFrontmatter:
def test_basic(self):
text = "---\ntitle: Test Claim\nsupports:\n - claim-a\n---\nBody text here."
fm, body = _parse_yaml_frontmatter(text)
assert fm is not None
assert fm["title"] == "Test Claim"
assert fm["supports"] == ["claim-a"]
assert body.startswith("\n---")
def test_no_frontmatter(self):
text = "Just plain text"
fm, body = _parse_yaml_frontmatter(text)
assert fm is None
assert body == text
def test_malformed_yaml(self):
text = "---\n: invalid: yaml: {{{\n---\nBody"
fm, body = _parse_yaml_frontmatter(text)
assert fm is None
class TestUnionEdgeLists:
def test_no_overlap(self):
main = ["claim-a", "claim-b"]
branch = ["claim-c", "claim-d"]
result = _union_edge_lists(main, branch)
assert result == ["claim-a", "claim-b", "claim-c", "claim-d"]
def test_overlap_preserves_main_order(self):
main = ["claim-b", "claim-a"]
branch = ["claim-a", "claim-c"]
result = _union_edge_lists(main, branch)
assert result == ["claim-b", "claim-a", "claim-c"]
def test_case_insensitive_dedup(self):
main = ["Claim A"]
branch = ["claim a", "Claim B"]
result = _union_edge_lists(main, branch)
assert len(result) == 2
assert result[0] == "Claim A"
assert result[1] == "Claim B"
def test_empty_main(self):
result = _union_edge_lists([], ["claim-a", "claim-b"])
assert result == ["claim-a", "claim-b"]
def test_empty_branch(self):
result = _union_edge_lists(["claim-a"], [])
assert result == ["claim-a"]
def test_both_empty(self):
assert _union_edge_lists([], []) == []
def test_duplicates_within_branch(self):
main = ["claim-a"]
branch = ["claim-b", "claim-b"]
result = _union_edge_lists(main, branch)
assert result == ["claim-a", "claim-b"]
class TestSerializeFrontmatter:
def test_roundtrip(self):
fm = {"title": "Test", "supports": ["claim-a", "claim-b"]}
body = "\n---\nBody text here."
text = _serialize_frontmatter(fm, body)
assert text.startswith("---\n")
assert "title: Test" in text
assert "Body text here." in text
fm2, body2 = _parse_yaml_frontmatter(text)
assert fm2["title"] == "Test"
assert fm2["supports"] == ["claim-a", "claim-b"]
class TestSupersetDetection:
def test_branch_is_superset(self):
main_edges = {"claim-a", "claim-b"}
branch_edges = {"claim-a", "claim-b", "claim-c"}
assert len(main_edges - branch_edges) == 0
def test_branch_missing_edge(self):
main_edges = {"claim-a", "claim-b"}
branch_edges = {"claim-a", "claim-c"}
assert "claim-b" in (main_edges - branch_edges)
def test_equal_sets(self):
main_edges = {"claim-a", "claim-b"}
branch_edges = {"claim-a", "claim-b"}
assert len(main_edges - branch_edges) == 0
class TestEdgeFieldsCoverage:
def test_standard_fields_present(self):
assert "supports" in REWEAVE_EDGE_FIELDS
assert "challenges" in REWEAVE_EDGE_FIELDS
assert "related" in REWEAVE_EDGE_FIELDS
assert "reweave_edges" in REWEAVE_EDGE_FIELDS
assert "depends_on" in REWEAVE_EDGE_FIELDS
class TestFullUnionWorkflow:
def test_main_evolved_branch_stale(self):
"""Main got new edges after branch was created. Union includes both."""
main_text = (
"---\ntitle: Test Claim\nconfidence: 0.8\n"
"supports:\n - claim-a\n - claim-b\n"
"related:\n - claim-x\n"
"---\nBody text."
)
branch_text = (
"---\ntitle: Test Claim\nconfidence: 0.8\n"
"supports:\n - claim-a\n"
"related:\n - claim-x\n - claim-y\n"
"reweave_edges:\n - \"claim-y|related|2026-04-04\"\n"
"---\nBody text."
)
main_fm, main_body = _parse_yaml_frontmatter(main_text)
branch_fm, _ = _parse_yaml_frontmatter(branch_text)
merged_fm = dict(main_fm)
for field in REWEAVE_EDGE_FIELDS:
main_list = main_fm.get(field, [])
branch_list = branch_fm.get(field, [])
if not isinstance(main_list, list):
main_list = [main_list] if main_list else []
if not isinstance(branch_list, list):
branch_list = [branch_list] if branch_list else []
if main_list or branch_list:
merged_fm[field] = _union_edge_lists(main_list, branch_list)
assert merged_fm["supports"] == ["claim-a", "claim-b"]
assert "claim-x" in merged_fm["related"]
assert "claim-y" in merged_fm["related"]
assert len(merged_fm.get("reweave_edges", [])) == 1
assert merged_fm["confidence"] == 0.8
def test_no_edge_fields_untouched(self):
"""Non-edge fields (title, confidence, type) come from main unchanged."""
main_text = "---\ntitle: Original\nconfidence: 0.9\ntype: claim\n---\nBody."
branch_text = "---\ntitle: Original\nconfidence: 0.9\ntype: claim\nrelated:\n - new-claim\n---\nBody."
main_fm, main_body = _parse_yaml_frontmatter(main_text)
branch_fm, _ = _parse_yaml_frontmatter(branch_text)
merged_fm = dict(main_fm)
for field in REWEAVE_EDGE_FIELDS:
main_list = main_fm.get(field, [])
branch_list = branch_fm.get(field, [])
if not isinstance(main_list, list):
main_list = [main_list] if main_list else []
if not isinstance(branch_list, list):
branch_list = [branch_list] if branch_list else []
if main_list or branch_list:
merged_fm[field] = _union_edge_lists(main_list, branch_list)
assert merged_fm["title"] == "Original"
assert merged_fm["confidence"] == 0.9
assert merged_fm["type"] == "claim"
assert merged_fm["related"] == ["new-claim"]
def test_scalar_edge_field_converted_to_list(self):
"""Edge fields stored as scalars (not lists) are handled gracefully."""
main_fm = {"supports": "single-claim"}
branch_fm = {"supports": ["single-claim", "new-claim"]}
main_list = main_fm.get("supports", [])
branch_list = branch_fm.get("supports", [])
if not isinstance(main_list, list):
main_list = [main_list] if main_list else []
if not isinstance(branch_list, list):
branch_list = [branch_list] if branch_list else []
result = _union_edge_lists(main_list, branch_list)
assert result == ["single-claim", "new-claim"]