Coverage for src/retemplar/core.py: 100%
67 statements
« prev ^ index » next coverage.py v7.10.5, created at 2025-08-30 17:51 -0500
1# src/retemplar/core.py
2"""Core retemplar operations (MVP)."""
4from pathlib import Path
6from retemplar.lockfile import LockfileManager, LockfileNotFoundError
7from retemplar.logging import RetemplarError, get_logger
8from retemplar.schema import RetemplarLock, parse_template_ref
9from retemplar.utils import fs_utils, merge_utils
10from retemplar.utils.apply_utils import apply_file_changes
11from retemplar.utils.plan_utils import plan_file_changes
13logger = get_logger(__name__)
16class RetemplarCore:
17 """Core orchestrator for retemplar operations (refactored MVP)."""
19 repo_path: Path
21 def __init__(self, repo_path: Path):
22 self.repo_path = repo_path
24 def adopt_template(
25 self,
26 lock: RetemplarLock,
27 dry_run: bool = False,
28 ) -> dict[str, str | list | bool]:
29 """Create initial .retemplar.lock (no baseline yet)."""
30 logger.debug(
31 'adopt_template_started',
32 template_ref=lock.template_ref,
33 dry_run=dry_run,
34 managed_paths_count=len(lock.managed_paths),
35 )
37 with LockfileManager(self.repo_path) as lockfile_manager:
38 if lockfile_manager.exists():
39 raise RetemplarError(
40 'Repository already has .retemplar.lock',
41 repo_path=str(self.repo_path),
42 )
44 if not dry_run:
45 try:
46 lockfile_manager.write(lock)
47 logger.debug('lockfile_created')
48 except Exception as e:
49 raise RetemplarError(
50 'Failed to create lockfile',
51 repo_path=str(self.repo_path),
52 error=str(e),
53 ) from e
55 return {
56 'template': lock.template_ref,
57 'managed_paths': [mp.path for mp in lock.managed_paths],
58 'ignore_paths': lock.ignore_paths,
59 'render_rules': lock.render_rules,
60 'lockfile_created': not dry_run,
61 }
63 def plan_upgrade(
64 self,
65 target_ref: str,
66 ) -> dict[str, str | int | list[dict]]:
67 """Compute a human-readable plan. 2-way semantics for now."""
68 logger.debug('plan_upgrade_started', target_ref=target_ref)
70 with LockfileManager(self.repo_path) as lockfile_manager:
71 if not lockfile_manager.exists():
72 raise LockfileNotFoundError(
73 "No .retemplar.lock found. Run 'retemplar adopt' first.",
74 )
76 lock = lockfile_manager.read()
77 logger.debug(
78 'lockfile_loaded',
79 current_ref=lock.template_ref,
80 managed_paths_count=len(lock.managed_paths),
81 )
83 target_src = parse_template_ref(target_ref)
84 tpl_root = fs_utils.resolve_template_path(target_src.repo)
85 managed_files = fs_utils.get_managed_files(lock, tpl_root)
87 logger.debug(
88 'template_analysis',
89 template_root=str(tpl_root),
90 managed_files_count=len(managed_files),
91 )
93 items, conflicts = plan_file_changes(managed_files, tpl_root, lock)
94 block_events = merge_utils.scan_block_protection(
95 lock.managed_paths or [],
96 )
98 return {
99 'current_version': lock.template_ref,
100 'target_version': target_ref,
101 'changes': [
102 {
103 'file': item.path,
104 'strategy': item.strategy,
105 'type': item.kind,
106 'note': item.note,
107 'had_conflict': item.had_conflict,
108 'matched_rule': getattr(
109 managed_files.get(item.path),
110 'path',
111 item.path,
112 ),
113 }
114 for item in items
115 ],
116 'conflicts': conflicts,
117 'block_protection': block_events,
118 }
120 def apply_changes(
121 self,
122 target_ref: str,
123 dry_run: bool = False,
124 ) -> dict[str, str | int]:
125 """Apply the plan (non-interactive). 2-way for now; conflict markers on merge."""
126 logger.debug(
127 'apply_changes_started',
128 target_ref=target_ref,
129 dry_run=dry_run,
130 )
132 with LockfileManager(self.repo_path) as lockfile_manager:
133 if not lockfile_manager.exists():
134 raise LockfileNotFoundError(
135 "No .retemplar.lock found. Run 'retemplar adopt' first.",
136 )
137 if not target_ref:
138 raise ValueError('target_ref is required for apply')
140 plan = self.plan_upgrade(target_ref)
141 lock = lockfile_manager.read()
142 target_src = parse_template_ref(target_ref)
143 tpl_root = fs_utils.resolve_template_path(target_src.repo)
145 logger.debug(
146 'applying_file_changes',
147 changes_count=len(plan['changes']),
148 dry_run=dry_run,
149 )
151 files_changed, conflicts = apply_file_changes(
152 plan['changes'],
153 tpl_root,
154 lock,
155 dry_run,
156 )
158 if not dry_run:
159 try:
160 updated = lock.model_copy(
161 update={
162 'template_ref': target_ref,
163 'applied_ref': target_src.ref,
164 'applied_commit': target_src.commit,
165 },
166 )
167 lockfile_manager.write(updated)
168 logger.debug(
169 'lockfile_updated_after_apply',
170 target_ref=target_ref,
171 )
172 except Exception as e:
173 raise RetemplarError(
174 'Failed to update lockfile after applying changes - manual intervention required',
175 target_ref=target_ref,
176 files_changed=files_changed,
177 error=str(e),
178 repo_path=str(self.repo_path),
179 ) from e
181 return {
182 'applied_version': target_ref,
183 'files_changed': files_changed,
184 'conflicts_resolved': conflicts,
185 }
187 def detect_drift(self) -> dict[str, str | list]:
188 """Detect drift between repo and baseline (placeholder for 3-way)."""
189 logger.debug('detect_drift_started')
191 with LockfileManager(self.repo_path) as lockfile_manager:
192 if not lockfile_manager.exists():
193 raise LockfileNotFoundError(
194 "No .retemplar.lock found. Run 'retemplar adopt' first.",
195 )
197 current_lock = lockfile_manager.read()
198 logger.debug(
199 'lockfile_loaded_for_drift',
200 current_ref=current_lock.template_ref,
201 applied_ref=current_lock.applied_ref,
202 )
204 # TODO: Real implementation needs:
205 # - Baseline resolution from applied_ref
206 # - 3-way comparison: Base vs Ours vs Theirs
207 # - Categorization of changes (template-only, local-only, conflicts)
209 return {
210 'baseline_version': current_lock.applied_ref
211 or getattr(current_lock.template, 'ref', 'unknown'),
212 'template_only_changes': [], # TODO: implement with baseline
213 'local_only_changes': [], # TODO: implement with baseline
214 'conflicts': [], # TODO: implement with baseline
215 'unmanaged_files': [], # TODO: implement
216 }