Coverage for little_loops / cli / migrate_relationships.py: 85%

1"""ll-migrate-relationships: Rename parent_issue: -> parent: and related: -> relates_to: in issue files.""" 

2 

3from __future__ import annotations 

4 

5import argparse 

6import re 

7from pathlib import Path 

8 

9from little_loops.cli_args import add_config_arg, add_dry_run_arg 

10from little_loops.frontmatter import parse_frontmatter 

11 

12_FM_FIELD_RE = re.compile(r"^---\s*$", re.MULTILINE) 

13 

14_PARENT_ISSUE_RE = re.compile(r"^parent_issue:(.*)$", re.MULTILINE) 

15_RELATED_RE = re.compile(r"^related:(.*)$", re.MULTILINE) 

16 

17 

18def _set_fields(content: str, fields: dict[str, str]) -> str: 

19 """Set frontmatter fields via direct string manipulation (avoids yaml roundtrip). 

20 

21 Replaces existing fields in-place; inserts missing fields before the closing 

22 ``---`` marker. Prepends a new frontmatter block if none exists. 

23 """ 

24 if not content.startswith("---\n"): 

25 lines = "\n".join(f"{k}: {v}" for k, v in fields.items()) 

26 return f"---\n{lines}\n---\n{content}" 

27 

28 result = content 

29 for key, value in fields.items(): 

30 line = f"{key}: {value}" 

31 key_re = re.compile(rf"^{re.escape(key)}:.*$", re.MULTILINE) 

32 if key_re.search(result): 

33 result = key_re.sub(line, result) 

34 else: 

35 # Insert before the closing --- of the frontmatter 

36 markers = list(_FM_FIELD_RE.finditer(result)) 

37 if len(markers) >= 2: 

38 pos = markers[1].start() 

39 result = result[:pos] + f"{line}\n" + result[pos:] 

40 return result 

41 

42 

43def _migrate_content(content: str) -> tuple[str, list[str]]: 

44 """Rename relationship keys in frontmatter. Returns (updated_content, list_of_renames).""" 

45 fm = parse_frontmatter(content) 

46 if not fm: 

47 return content, [] 

48 

49 renames: list[str] = [] 

50 result = content 

51 

52 if "parent_issue" in fm: 

53 value = str(fm["parent_issue"]) 

54 result = _set_fields(result, {"parent": value}) 

55 result = _PARENT_ISSUE_RE.sub("", result) 

56 # Clean up blank line left by removal 

57 result = re.sub(r"\n{3,}", "\n\n", result) 

58 renames.append(f"parent_issue: {value!r} → parent: {value!r}") 

59 

60 if "related" in fm: 

61 value_raw = _RELATED_RE.search(content) 

62 raw_suffix = value_raw.group(1) if value_raw else "" 

63 result = _set_fields(result, {"relates_to": fm["related"]}) 

64 result = _RELATED_RE.sub("", result) 

65 result = re.sub(r"\n{3,}", "\n\n", result) 

66 renames.append(f"related:{raw_suffix} → relates_to:{raw_suffix}") 

67 

68 return result, renames 

69 

70 

71def main_migrate_relationships() -> int: 

72 """Entry point for ll-migrate-relationships command. 

73 

74 Renames parent_issue: -> parent: and related: -> relates_to: in all issue files. 

75 

76 Returns: 

77 Exit code (0 = success, 1 = error) 

78 """ 

79 parser = argparse.ArgumentParser( 

80 prog="ll-migrate-relationships", 

81 description=( 

82 "Rename parent_issue: → parent: and related: → relates_to: " 

83 "in all issue frontmatter files. One-time migration for ENH-1431." 

84 ), 

85 formatter_class=argparse.RawDescriptionHelpFormatter, 

86 epilog=""" 

87Examples: 

88 %(prog)s --dry-run # Preview all planned renames (strongly advised first) 

89 %(prog)s # Execute migration 

90""", 

91 ) 

92 add_dry_run_arg(parser) 

93 add_config_arg(parser) 

94 args = parser.parse_args() 

95 

96 dry_run: bool = args.dry_run 

97 repo_root: Path = args.config or Path.cwd() 

98 

99 issues_dir = repo_root / ".issues" 

100 if not issues_dir.exists(): 

101 print(f"No .issues/ directory found at {repo_root}") 

102 return 1 

103 

104 if dry_run: 

105 print("[DRY RUN] No files will be modified.") 

106 

107 renamed = 0 

108 errors: list[str] = [] 

109 

110 for file_path in sorted(issues_dir.rglob("*.md")): 

111 try: 

112 content = file_path.read_text(encoding="utf-8") 

113 except OSError as exc: 

114 errors.append(str(file_path)) 

115 print(f" [ERROR] {file_path}: {exc}") 

116 continue 

117 

118 updated, renames = _migrate_content(content) 

119 if not renames: 

120 continue 

121 

122 prefix = "[DRY RUN] " if dry_run else "" 

123 rel = file_path.relative_to(repo_root) 

124 for rename in renames: 

125 print(f" {prefix}RENAME {rel}: {rename}") 

126 

127 if not dry_run: 

128 try: 

129 file_path.write_text(updated, encoding="utf-8") 

130 renamed += 1 

131 except OSError as exc: 

132 errors.append(str(file_path)) 

133 print(f" [ERROR] {file_path}: {exc}") 

134 else: 

135 renamed += 1 

136 

137 print() 

138 print(f"Results: {renamed} files {'would be ' if dry_run else ''}updated.") 

139 if errors: 

140 print(f" Errors: {len(errors)}") 

141 return 1 

142 return 0