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

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 

28class ResultsMigrator: 

29 """ 

30 Handles versioned migrations for PullRequestResults data. 

31 """ 

32 

33 # Ordered list of migration functions. 

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

35 migrations = [ 

36 migrate_resview_results_scopes, 

37 ] 

38 

39 @classmethod 

40 def current_version(cls) -> int: 

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

42 return len(cls.migrations) + 1 

43 

44 @classmethod 

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

46 """ 

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

48 

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

50 """ 

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

52 

53 # Apply migrations from current version to latest 

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

55 data = migration(data) 

56 

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

58 return data