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

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 - ReviewResult.scopes -> reviewed_for 

17 - ReviewResult.state computed from review.state 

18 """ 

19 if "review_results" in data: 

20 for review_result in data["review_results"].values(): 

21 # Rename scopes -> reviewed_for 

22 if "scopes" in review_result: 

23 review_result["matched_scopes"] = review_result.pop("scopes") 

24 

25 return data 

26 

27 

28def migrate_drop_config_branches(data: dict[str, Any]) -> dict[str, Any]: 

29 """ 

30 Migrate v2 -> v3: 

31 - Config-level `branches` was removed. Drop it from stored configs so old 

32 results re-parse cleanly. Scope-level `branches` is unaffected. 

33 """ 

34 for config_result in data.get("config_results", {}).values(): 

35 config = config_result.get("config") 

36 if isinstance(config, dict): 

37 config.pop("branches", None) 

38 

39 return data 

40 

41 

42class ResultsMigrator: 

43 """ 

44 Handles versioned migrations for PullRequestResults data. 

45 """ 

46 

47 # Ordered list of migration functions. 

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

49 migrations = [ 

50 migrate_resview_results_scopes, 

51 migrate_drop_config_branches, 

52 ] 

53 

54 @classmethod 

55 def current_version(cls) -> int: 

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

57 return len(cls.migrations) + 1 

58 

59 @classmethod 

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

61 """ 

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

63 

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

65 """ 

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

67 

68 # Apply migrations from current version to latest 

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

70 data = migration(data) 

71 

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

73 return data