Coverage for frappe_manager / migration_manager / migration_helpers.py: 64%

1import json 

2import platform 

3from pathlib import Path 

4 

5from frappe_manager import CLI_SERVICES_DIRECTORY 

6from frappe_manager.docker import ComposeFile, DockerClient 

7from frappe_manager.migration_manager.migration_constants import MIGRATION_BENCH_STOP_TIMEOUT_SECONDS 

8from frappe_manager.output_manager.base import OutputHandler 

9 

10 

11class MigrationBench: 

12 def __init__(self, name: str, path: Path, output: OutputHandler | None = None) -> None: 

13 self.name = name 

14 self.path = path 

15 self.output = output 

16 self.compose_file_manager = ComposeFile( 

17 self.path / "docker-compose.yml", 

18 "docker-compose.tmpl", 

19 ) 

20 self.docker = DockerClient(compose_file_path=self.compose_file_manager.compose_path, output=output) 

21 

22 self.workers_compose_file_manager = ComposeFile( 

23 self.path / "docker-compose.workers.yml", 

24 "docker-compose.workers.tmpl", 

25 ) 

26 self.workers_docker = DockerClient( 

27 compose_file_path=self.workers_compose_file_manager.compose_path, 

28 output=output, 

29 ) 

30 

31 @property 

32 def compose(self): 

33 assert self.docker.compose is not None 

34 return self.docker.compose 

35 

36 @property 

37 def running(self) -> bool: 

38 services = self.compose_file_manager.get_services_list() 

39 running_status = self.get_services_running_status() 

40 

41 if not running_status: 

42 return False 

43 

44 for service in services: 

45 try: 

46 if not running_status[service] == "running": 

47 return False 

48 except KeyError: 

49 return False 

50 return True 

51 

52 @property 

53 def workers_running(self) -> bool: 

54 from frappe_manager.docker import DockerException 

55 

56 services = self.workers_compose_file_manager.get_services_list() 

57 

58 try: 

59 all_statuses = self.workers_docker.compose.get_all_services_status() 

60 containers = self.workers_compose_file_manager.get_container_names().values() 

61 running_status = { 

62 status["Service"]: status["State"] for status in all_statuses if status.get("Name") in containers 

63 } 

64 

65 for service in services: 

66 if running_status.get(service) != "running": 

67 return False 

68 return True 

69 except (DockerException, KeyError): 

70 return False 

71 

72 def get_services_running_status(self) -> dict: 

73 from frappe_manager.docker import DockerException 

74 

75 services = self.compose_file_manager.get_services_list() 

76 containers = self.compose_file_manager.get_container_names().values() 

77 

78 try: 

79 all_statuses = self.compose.get_all_services_status() 

80 return {status["Service"]: status["State"] for status in all_statuses if status.get("Name") in containers} 

81 except DockerException: 

82 return {} 

83 

84 def get_db_connection_info(self): 

85 from frappe_manager.utils.site import get_bench_db_connection_info 

86 

87 return get_bench_db_connection_info(self.name, self.path) 

88 

89 def common_bench_config_set(self, config: dict): 

90 """ 

91 Sets the values in the common_site_config.json file. 

92 Args: 

93 config (dict): A dictionary containing the key-value pairs to be set in the common_site_config.json file. 

94 Returns: 

95 bool: True if the values are successfully set, False otherwise. 

96 """ 

97 common_bench_config_path = self.path / "workspace/frappe-bench/sites/common_site_config.json" 

98 

99 if not common_bench_config_path.exists(): 

100 return False 

101 

102 common_site_config = {} 

103 

104 with open(common_bench_config_path) as f: 

105 common_site_config = json.load(f) 

106 

107 try: 

108 for key, value in config.items(): 

109 common_site_config[key] = value 

110 with open(common_bench_config_path, "w") as f: 

111 json.dump(common_site_config, f) 

112 return True 

113 except KeyError as e: 

114 return False 

115 

116 

117class MigrationBenches: 

118 def __init__(self, benches_path: Path) -> None: 

119 self.benches_path = benches_path 

120 

121 def get_all_benches(self, exclude: list[str] | None = None): 

122 exclude = exclude or [] 

123 benches: dict[str, Path] = {} 

124 for dir in self.benches_path.iterdir(): 

125 if dir.is_dir() and dir.parts[-1] not in exclude: 

126 name = dir.parts[-1] 

127 dir = dir / "docker-compose.yml" 

128 if dir.exists(): 

129 benches[name] = dir 

130 return benches 

131 

132 def stop_benches(self, timeout: int = MIGRATION_BENCH_STOP_TIMEOUT_SECONDS): 

133 compose_list = self.get_all_benches() 

134 for name, compose_path in compose_list.items(): 

135 bench = MigrationBench(name, compose_path.parent) 

136 bench.docker.compose.stop(timeout=timeout, stream=False) 

137 

138 

139class MigrationServicesManager: 

140 def __init__(self, services_path: Path = CLI_SERVICES_DIRECTORY): 

141 self.services_path = services_path 

142 

143 template_name = "docker-compose.services.tmpl" 

144 

145 if platform.system() == "Darwin": 

146 template_name = "docker-compose.services.osx.tmpl" 

147 

148 self.compose_file_manager = ComposeFile( 

149 self.services_path / "docker-compose.yml", 

150 template_name, 

151 ) 

152 self.docker = DockerClient(compose_file_path=self.compose_file_manager.compose_path) 

153 

154 @property 

155 def compose(self): 

156 assert self.docker.compose is not None 

157 return self.docker.compose