Coverage for frappe_manager / migration_manager / bench_migration_state.py: 22%

36 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-07-02 18:13 +0530

1""" 

2Bench migration state management. 

3 

4Tracks migration version for individual benches. 

5""" 

6 

7from datetime import datetime 

8from pathlib import Path 

9 

10from frappe_manager.migration_manager.version import Version 

11from frappe_manager.site_manager.bench_config import BenchConfig, MigrationState 

12 

13 

14def get_bench_migration_version(bench_path: Path) -> Version: 

15 """ 

16 Get the version bench is migrated to. 

17 

18 Args: 

19 bench_path: Path to bench directory 

20 

21 Returns: 

22 Version object representing bench migration state 

23 """ 

24 bench_config_path = bench_path / "bench_config.toml" 

25 

26 if not bench_config_path.exists(): 

27 return Version("0.0.0") 

28 

29 try: 

30 config = BenchConfig.import_from_toml(bench_config_path) 

31 if config.migration_state and config.migration_state.migrated_to: 

32 return Version(config.migration_state.migrated_to) 

33 except Exception: 

34 pass 

35 

36 return Version("0.0.0") 

37 

38 

39def set_bench_migration_version(bench_path: Path, version: Version) -> None: 

40 """ 

41 Update bench migration version. 

42 

43 Args: 

44 bench_path: Path to bench directory 

45 version: Version to set 

46 """ 

47 bench_config_path = bench_path / "bench_config.toml" 

48 

49 if not bench_config_path.exists(): 

50 raise FileNotFoundError(f"Bench config not found: {bench_config_path}") 

51 

52 config = BenchConfig.import_from_toml(bench_config_path) 

53 config.migration_state = MigrationState( 

54 migrated_to=str(version.version), 

55 last_migration_date=datetime.now().isoformat(), 

56 ) 

57 config.export_to_toml(bench_config_path) 

58 

59 

60def bench_needs_migration(bench_path: Path, target_version: Version) -> bool: 

61 """ 

62 Check if bench needs migration to target version. 

63 

64 Args: 

65 bench_path: Path to bench directory 

66 target_version: Target migration version 

67 

68 Returns: 

69 True if bench needs migration, False otherwise 

70 """ 

71 current = get_bench_migration_version(bench_path) 

72 return current < target_version 

73 

74 

75def get_bench_migration_date(bench_path: Path) -> str | None: 

76 """ 

77 Get last migration date for bench. 

78 

79 Args: 

80 bench_path: Path to bench directory 

81 

82 Returns: 

83 ISO format date string or None 

84 """ 

85 bench_config_path = bench_path / "bench_config.toml" 

86 

87 if not bench_config_path.exists(): 

88 return None 

89 

90 try: 

91 config = BenchConfig.import_from_toml(bench_config_path) 

92 if config.migration_state: 

93 return config.migration_state.last_migration_date 

94 except Exception: 

95 return None 

96 

97 return None