Coverage for src/pullapprove/results_migrations.py: 99%

1""" 

2Versioned migrations for PullRequestResults data. 

3 

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""" 

7 

8from __future__ import annotations 

9 

10from typing import Any 

11 

12 

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") 

23 

24 return data 

25 

26 

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) 

37 

38 return data 

39 

40 

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. 

47 

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) 

57 

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")) 

67 

68 for scope in scopes: 

69 if isinstance(scope, dict): 

70 scope.pop("branches", None) 

71 

72 return data 

73 

74 

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")) 

92 

93 for scope in scopes: 

94 if isinstance(scope, dict): 

95 scope.pop("reviewed_for", None) 

96 

97 return data 

98 

99 

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"] 

114 

115 return data 

116 

117 

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) 

129 

130 return data 

131 

132 

133class ResultsMigrator: 

134 """ 

135 Handles versioned migrations for PullRequestResults data. 

136 """ 

137 

138 # Ordered list of migration functions. 

139 # Index 0 = v1->v2, index 1 = v2->v3, etc. 

140 migrations = [ 

141 migrate_resview_results_scopes, 

142 migrate_drop_config_branches, 

143 migrate_drop_branch_fields, 

144 migrate_drop_reviewed_for, 

145 migrate_unwrap_review_results, 

146 migrate_drop_path_code_reviews, 

147 ] 

148 

149 @classmethod 

150 def current_version(cls) -> int: 

151 """Current version is always 1 more than the number of migrations.""" 

152 return len(cls.migrations) + 1 

153 

154 @classmethod 

155 def migrate(cls, data: dict[str, Any]) -> dict[str, Any]: 

156 """ 

157 Apply all necessary migrations to bring data to current version. 

158 

159 Data without a version field is assumed to be v1. 

160 """ 

161 version = data.get("version", 1) 

162 

163 # Apply migrations from current version to latest 

164 for migration in cls.migrations[version - 1 :]: 

165 data = migration(data) 

166 

167 data["version"] = cls.current_version() 

168 return data