Coverage for src/pullapprove/migrations.py: 27%

1""" 

2Versioned migrations for PullRequestResults data. 

3 

4When the schema changes, add a new migration function and append it to 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 _state_from_review_state(review_state: str) -> str: 

14 """Map Review.state to ReviewResultState for migration.""" 

15 if review_state == "APPROVED": 

16 return "APPROVED" 

17 elif review_state == "CHANGES_REQUESTED": 

18 return "CHANGES_REQUESTED" 

19 elif review_state == "PENDING": 

20 return "PENDING" 

21 else: 

22 return "COMMENTED" 

23 

24 

25def _migrate_v1_to_v2(data: dict[str, Any]) -> dict[str, Any]: 

26 """ 

27 v1 -> v2: Add ReviewResult.state, rename scopes -> reviewed_for. 

28 

29 Old schema had ReviewResult with: 

30 - scopes: list[str] (explicit Reviewed-for declarations) 

31 - no state field 

32 

33 New schema has: 

34 - reviewed_for: list[str] (renamed from scopes) 

35 - state: ReviewResultState (computed from review.state) 

36 """ 

37 for review_result in data.get("review_results", {}).values(): 

38 # Rename scopes -> reviewed_for 

39 if "scopes" in review_result: 

40 review_result["reviewed_for"] = review_result.pop("scopes") 

41 

42 # Add state if missing (compute from review.state) 

43 if "state" not in review_result: 

44 review_state = review_result.get("review", {}).get("state", "") 

45 review_result["state"] = _state_from_review_state(review_state) 

46 

47 return data 

48 

49 

50# Ordered list of migrations. 

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

52MIGRATIONS = [ 

53 _migrate_v1_to_v2, 

54] 

55 

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

57CURRENT_VERSION = len(MIGRATIONS) + 1 

58 

59 

60def migrate_data(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 MIGRATIONS[version - 1 :]: 

70 data = migration(data) 

71 

72 data["_version"] = CURRENT_VERSION 

73 return data