Coverage for frappe_manager / migration_manager / migration_validator.py: 53%

1""" 

2Migration validation and filtering logic. 

3""" 

4 

5from frappe_manager import CLI_BENCHES_DIRECTORY 

6from frappe_manager.migration_manager.bench_migration_state import get_bench_migration_version 

7from frappe_manager.migration_manager.migration_constants import MINIMUM_SUPPORTED_VERSION 

8from frappe_manager.migration_manager.migration_helpers import MigrationBenches 

9from frappe_manager.migration_manager.version import Version 

10from frappe_manager.output_manager import OutputHandler 

11 

12 

13class BenchFilter: 

14 """ 

15 Handles bench filtering logic based on target_benches and exclude_benches. 

16 

17 Single source of truth for "should this bench be processed?" logic. 

18 """ 

19 

20 def __init__( 

21 self, 

22 target_benches: list[str] | None, 

23 exclude_benches: list[str], 

24 ): 

25 self.target_benches = target_benches 

26 self.exclude_benches = exclude_benches 

27 

28 def should_process_bench(self, bench_name: str) -> bool: 

29 """ 

30 Determine if a bench should be processed during migration. 

31 

32 Returns False if: 

33 - target_benches is None (infrastructure-only migration) 

34 - bench_name not in target_benches (when targeting specific benches) 

35 - bench_name in exclude_benches (explicitly excluded) 

36 """ 

37 if self.target_benches is None: 

38 return False 

39 

40 if bench_name not in self.target_benches: 

41 return False 

42 

43 if bench_name in self.exclude_benches: 

44 return False 

45 

46 return True 

47 

48 

49class MigrationValidator: 

50 """ 

51 Validates migration preconditions and checks version compatibility. 

52 """ 

53 

54 def __init__( 

55 self, 

56 prev_version: Version, 

57 current_version: Version, 

58 bench_filter: BenchFilter, 

59 output_handler: OutputHandler, 

60 ): 

61 self.prev_version = prev_version 

62 self.current_version = current_version 

63 self.bench_filter = bench_filter 

64 self.output = output_handler 

65 

66 def get_minimum_bench_version(self) -> Version: 

67 """ 

68 Get the minimum migration version across all target benches. 

69 

70 Returns the lowest version that needs migration. This determines 

71 which migration classes need to be loaded. 

72 """ 

73 if self.bench_filter.target_benches is None: 

74 return self.current_version 

75 

76 benches_manager = MigrationBenches(CLI_BENCHES_DIRECTORY) 

77 all_benches = benches_manager.get_all_benches() 

78 min_version = self.current_version 

79 

80 for bench_name, bench_path in all_benches.items(): 

81 if not self.bench_filter.should_process_bench(bench_name): 

82 continue 

83 

84 bench_version = get_bench_migration_version(bench_path.parent) 

85 

86 min_version = min(min_version, bench_version) 

87 

88 return min_version 

89 

90 def check_benches_need_migration(self) -> bool: 

91 """Check if any target benches need migration to current version.""" 

92 if self.bench_filter.target_benches is None: 

93 return False 

94 

95 benches_manager = MigrationBenches(CLI_BENCHES_DIRECTORY) 

96 all_benches = benches_manager.get_all_benches() 

97 

98 for bench_name, bench_path in all_benches.items(): 

99 if not self.bench_filter.should_process_bench(bench_name): 

100 continue 

101 

102 bench_version = get_bench_migration_version(bench_path.parent) 

103 if bench_version < self.current_version: 

104 return True 

105 

106 return False 

107 

108 def validate_version_support(self, effective_prev_version: Version) -> bool: 

109 """ 

110 Check if migration from effective_prev_version is supported. 

111 

112 Returns False and displays error if version is too old. 

113 """ 

114 if effective_prev_version == Version("0.0.0"): 

115 return True 

116 

117 if effective_prev_version < MINIMUM_SUPPORTED_VERSION: 

118 self.output.display_error( 

119 f"Cannot migrate from v{effective_prev_version.version}. " 

120 f"Minimum supported version is v{MINIMUM_SUPPORTED_VERSION.version}.", 

121 ) 

122 self.output.display_error( 

123 f"\nPlease upgrade to v{MINIMUM_SUPPORTED_VERSION.version} first, " 

124 f"then upgrade to v{self.current_version.version}.", 

125 ) 

126 self.output.display_error( 

127 f"\nMigration path: v{effective_prev_version.version} → " 

128 f"v{MINIMUM_SUPPORTED_VERSION.version} → v{self.current_version.version}", 

129 ) 

130 return False 

131 

132 return True