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
« prev ^ index » next coverage.py v7.13.5, created at 2026-07-02 18:13 +0530
1"""
2Bench migration state management.
4Tracks migration version for individual benches.
5"""
7from datetime import datetime
8from pathlib import Path
10from frappe_manager.migration_manager.version import Version
11from frappe_manager.site_manager.bench_config import BenchConfig, MigrationState
14def get_bench_migration_version(bench_path: Path) -> Version:
15 """
16 Get the version bench is migrated to.
18 Args:
19 bench_path: Path to bench directory
21 Returns:
22 Version object representing bench migration state
23 """
24 bench_config_path = bench_path / "bench_config.toml"
26 if not bench_config_path.exists():
27 return Version("0.0.0")
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
36 return Version("0.0.0")
39def set_bench_migration_version(bench_path: Path, version: Version) -> None:
40 """
41 Update bench migration version.
43 Args:
44 bench_path: Path to bench directory
45 version: Version to set
46 """
47 bench_config_path = bench_path / "bench_config.toml"
49 if not bench_config_path.exists():
50 raise FileNotFoundError(f"Bench config not found: {bench_config_path}")
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)
60def bench_needs_migration(bench_path: Path, target_version: Version) -> bool:
61 """
62 Check if bench needs migration to target version.
64 Args:
65 bench_path: Path to bench directory
66 target_version: Target migration version
68 Returns:
69 True if bench needs migration, False otherwise
70 """
71 current = get_bench_migration_version(bench_path)
72 return current < target_version
75def get_bench_migration_date(bench_path: Path) -> str | None:
76 """
77 Get last migration date for bench.
79 Args:
80 bench_path: Path to bench directory
82 Returns:
83 ISO format date string or None
84 """
85 bench_config_path = bench_path / "bench_config.toml"
87 if not bench_config_path.exists():
88 return None
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
97 return None