Coverage for src/pullapprove/results_migrations.py: 97%
99 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-07-28 16:50 -0500
1"""
2Versioned migrations for PullRequestResults data.
4When the schema changes, add a new migration function and append it to ResultsMigrator.migrations.
5Old stored data will be migrated on-the-fly when loaded via from_dict().
6"""
8from __future__ import annotations
10from typing import Any
13def migrate_resview_results_scopes(data: dict[str, Any]) -> dict[str, Any]:
14 """
15 Migrate v1 -> v2:
16 - Rename ReviewResult.scopes -> matched_scopes
17 """
18 if "review_results" in data:
19 for review_result in data["review_results"].values():
20 # Rename scopes -> matched_scopes
21 if "scopes" in review_result:
22 review_result["matched_scopes"] = review_result.pop("scopes")
24 return data
27def migrate_drop_config_branches(data: dict[str, Any]) -> dict[str, Any]:
28 """
29 Migrate v2 -> v3:
30 - Config-level `branches` was removed. Drop it from stored configs so old
31 results re-parse cleanly. Scope-level `branches` is unaffected.
32 """
33 for config_result in data.get("config_results", {}).values():
34 config = config_result.get("config")
35 if isinstance(config, dict):
36 config.pop("branches", None)
38 return data
41def migrate_drop_branch_fields(data: dict[str, Any]) -> dict[str, Any]:
42 """
43 Migrate v3 -> v4:
44 - Scope-level `branches` was removed.
45 - The PullRequest `base_branch`/`head_branch` fields were removed along with
46 the `Branch` domain object.
48 Both models forbid extra keys, so any leftover value in an old stored result
49 would break re-parsing. `base_branch`/`head_branch` were required (no
50 default), so every pre-v4 result carries them. Drop them from the pullrequest
51 and `branches` from every stored scope (configs and scope results).
52 """
53 pullrequest = data.get("pullrequest")
54 if isinstance(pullrequest, dict):
55 pullrequest.pop("base_branch", None)
56 pullrequest.pop("head_branch", None)
58 scopes: list[Any] = []
59 for config_result in data.get("config_results", {}).values():
60 if isinstance(config_result, dict) and isinstance(
61 config_result.get("config"), dict
62 ):
63 scopes.extend(config_result["config"].get("scopes") or [])
64 for scope_result in data.get("scope_results", {}).values():
65 if isinstance(scope_result, dict):
66 scopes.append(scope_result.get("scope"))
68 for scope in scopes:
69 if isinstance(scope, dict):
70 scope.pop("branches", None)
72 return data
75def migrate_drop_reviewed_for(data: dict[str, Any]) -> dict[str, Any]:
76 """
77 Migrate v4 -> v5:
78 - The scope-level `reviewed_for` setting was removed. ScopeModel forbids
79 extra keys, so any stored scope that set `reviewed_for` to a non-default
80 value (`required`/`ignored`) would break re-parsing. Drop it from every
81 stored scope (configs and scope results).
82 """
83 scopes: list[Any] = []
84 for config_result in data.get("config_results", {}).values():
85 if isinstance(config_result, dict) and isinstance(
86 config_result.get("config"), dict
87 ):
88 scopes.extend(config_result["config"].get("scopes") or [])
89 for scope_result in data.get("scope_results", {}).values():
90 if isinstance(scope_result, dict):
91 scopes.append(scope_result.get("scope"))
93 for scope in scopes:
94 if isinstance(scope, dict):
95 scope.pop("reviewed_for", None)
97 return data
100def migrate_unwrap_review_results(data: dict[str, Any]) -> dict[str, Any]:
101 """
102 Migrate v5 -> v6:
103 - The ReviewResult wrapper was removed; review_results now maps host_id ->
104 Review directly. Unwrap each stored {"review": {...}, ...} into the bare
105 review dict (which also drops the removed matched_scopes field).
106 """
107 review_results = data.get("review_results")
108 if isinstance(review_results, dict):
109 for host_id, review_result in review_results.items():
110 if isinstance(review_result, dict) and isinstance(
111 review_result.get("review"), dict
112 ):
113 review_results[host_id] = review_result["review"]
115 return data
118def migrate_drop_path_code_reviews(data: dict[str, Any]) -> dict[str, Any]:
119 """
120 Migrate v6 -> v7:
121 - PathResult.reviews / CodeResult.reviews were removed (write-only dead
122 state). Both models forbid extra keys, so strip `reviews` from every
123 stored path result and code result.
124 """
125 for key in ("path_results", "code_results"):
126 for result in data.get(key, {}).values():
127 if isinstance(result, dict):
128 result.pop("reviews", None)
130 return data
133def migrate_requested_reviews_to_flag(data: dict[str, Any]) -> dict[str, Any]:
134 """
135 Migrate v7 -> v8:
136 - Review requests used to be represented as synthetic PENDING reviews with
137 host_id "requested:<user id>". They are now a `requested` flag on the
138 Reviewer instead. Strip the synthetic reviews everywhere they were
139 stored (reviewer reviews, review_results, scope/LSC review id lists)
140 and set `requested` on the reviewers that carried one.
141 """
143 def is_placeholder(host_id: Any) -> bool:
144 return isinstance(host_id, str) and host_id.startswith("requested:")
146 pullrequest = data.get("pullrequest")
147 if isinstance(pullrequest, dict):
148 for reviewer in pullrequest.get("reviewers") or []:
149 if not isinstance(reviewer, dict):
150 continue
151 reviews = reviewer.get("reviews")
152 if not isinstance(reviews, list):
153 continue
154 kept = [
155 r
156 for r in reviews
157 if not (isinstance(r, dict) and is_placeholder(r.get("host_id")))
158 ]
159 if len(kept) != len(reviews):
160 reviewer["reviews"] = kept
161 reviewer["requested"] = True
163 review_results = data.get("review_results")
164 if isinstance(review_results, dict):
165 for host_id in [k for k in review_results if is_placeholder(k)]:
166 del review_results[host_id]
168 for scope_result in data.get("scope_results", {}).values():
169 if isinstance(scope_result, dict) and isinstance(
170 scope_result.get("reviews"), list
171 ):
172 scope_result["reviews"] = [
173 r for r in scope_result["reviews"] if not is_placeholder(r)
174 ]
176 lsc = data.get("large_scale_change_results")
177 if isinstance(lsc, dict) and isinstance(lsc.get("reviews"), list):
178 lsc["reviews"] = [r for r in lsc["reviews"] if not is_placeholder(r)]
180 return data
183def migrate_drop_agents(data: dict[str, Any]) -> dict[str, Any]:
184 """
185 Migrate v8 -> v9:
186 - `[[agents]]` was removed (replaced by scope-level `unless`). Results
187 stored while a config declared agents carry an `agents` list at the top
188 level and an `agents` key inside stored config dumps; the results model
189 forbids extra keys and ConfigModel now rejects any `agents` key
190 outright, so both must be stripped for old results to re-parse. Open
191 pull requests would self-heal on their next processing run, but
192 merged/closed ones never reprocess -- without this their stored history
193 is unrenderable forever.
194 """
195 data.pop("agents", None)
197 for config_result in data.get("config_results", {}).values():
198 if isinstance(config_result, dict) and isinstance(
199 config_result.get("config"), dict
200 ):
201 config_result["config"].pop("agents", None)
203 return data
206class ResultsMigrator:
207 """
208 Handles versioned migrations for PullRequestResults data.
209 """
211 # Ordered list of migration functions.
212 # Index 0 = v1->v2, index 1 = v2->v3, etc.
213 migrations = [
214 migrate_resview_results_scopes,
215 migrate_drop_config_branches,
216 migrate_drop_branch_fields,
217 migrate_drop_reviewed_for,
218 migrate_unwrap_review_results,
219 migrate_drop_path_code_reviews,
220 migrate_requested_reviews_to_flag,
221 migrate_drop_agents,
222 ]
224 @classmethod
225 def current_version(cls) -> int:
226 """Current version is always 1 more than the number of migrations."""
227 return len(cls.migrations) + 1
229 @classmethod
230 def migrate(cls, data: dict[str, Any]) -> dict[str, Any]:
231 """
232 Apply all necessary migrations to bring data to current version.
234 Data without a version field is assumed to be v1.
235 """
236 version = data.get("version", 1)
238 # Apply migrations from current version to latest
239 for migration in cls.migrations[version - 1 :]:
240 data = migration(data)
242 data["version"] = cls.current_version()
243 return data