Coverage for frappe_manager / site_manager / modules / bench_admin_tools.py: 24%

1""" 

2Bench Admin Tools Module 

3 

4Handles admin tools (Mailpit, Adminer) management including: 

5- Docker compose generation and lifecycle 

6- Nginx location configuration 

7- HTTP authentication setup 

8- Mailpit integration with Frappe 

9""" 

10 

11import json 

12from pathlib import Path 

13from typing import TYPE_CHECKING, Any 

14 

15from frappe_manager import CLI_DEFAULT_DELIMETER 

16from frappe_manager.docker import ComposeFile, DockerClient, DockerException 

17from frappe_manager.output_manager import OutputHandler 

18from frappe_manager.output_manager.rich_output import RichOutputHandler 

19from frappe_manager.site_manager.exceptions import AdminToolsFailedToStart, BenchException 

20from frappe_manager.utils.helpers import get_container_name_prefix, get_current_fm_version, get_template_path 

21 

22if TYPE_CHECKING: 

23 from frappe_manager.site_manager.site import Bench 

24 

25 

26class BenchAdminTools: 

27 def __init__( 

28 self, 

29 bench: "Bench", 

30 nginx_proxy: Any, 

31 verbose: bool = True, 

32 output_handler: OutputHandler | None = None, 

33 ): 

34 """ 

35 Initialize BenchAdminTools. 

36 

37 Args: 

38 bench: The Bench instance 

39 nginx_proxy: Nginx proxy manager 

40 verbose: Whether to show verbose output 

41 output_handler: Optional output handler for displaying information 

42 """ 

43 self.bench = bench 

44 self.compose_path = bench.path / "docker-compose.admin-tools.yml" 

45 self.bench_name = bench.name 

46 self.output = output_handler or RichOutputHandler() 

47 

48 self.compose_file_manager = ComposeFile(self.compose_path, template_name="docker-compose.admin-tools.tmpl") 

49 

50 self.docker_client = DockerClient(compose_file_path=self.compose_path, output=self.output) 

51 

52 self.nginx_proxy = nginx_proxy 

53 self.nginx_config_location_path: Path = self.nginx_proxy.dirs.conf.host / "custom" / "admin-tools.conf" 

54 self.http_auth_path: Path = self.nginx_proxy.dirs.conf.host / "http_auth" 

55 

56 def generate_compose(self, db_host: str): 

57 self.compose_file_manager.yml = self.compose_file_manager.load_template() 

58 

59 self.compose_file_manager.configure_bench( 

60 prefix=get_container_name_prefix(self.bench_name), 

61 version=get_current_fm_version(), 

62 envs={"adminer": {"ADMINER_DEFAULT_SERVER": db_host}}, 

63 network_name="site-network", 

64 auto_save=False, 

65 ) 

66 

67 self.compose_file_manager.set_all_services_restart(self.bench.bench_config.restart_policy.value) 

68 self.compose_file_manager.write_to_file() 

69 

70 def create(self, db_host: str): 

71 self.output.change_head("Generating admin tools configuration") 

72 self.generate_compose(db_host) 

73 self.output.print("Generating admin tools configuration: Done") 

74 

75 def _generate_credentials(self) -> tuple[str, str]: 

76 """Generate or retrieve admin credentials""" 

77 import secrets 

78 

79 # Use existing credentials from bench config or generate new ones 

80 username = self.bench.bench_config.admin_tools_username or "admin" 

81 password = self.bench.bench_config.admin_tools_password 

82 

83 if not password: 

84 password = secrets.token_urlsafe(16) 

85 # Store new credentials in bench config 

86 self.bench.bench_config.admin_tools_username = username 

87 self.bench.bench_config.admin_tools_password = password 

88 self.bench.save_bench_config(print_message=False) 

89 

90 return username, password 

91 

92 def save_nginx_location_config(self): 

93 # Ensure http auth directory exists 

94 self.http_auth_path.mkdir(exist_ok=True) 

95 

96 # Generate and save htpasswd file 

97 from passlib.apache import HtpasswdFile 

98 

99 auth_file = self.http_auth_path / f"{self.bench_name}-admin-tools.htpasswd" 

100 

101 if not self.http_auth_path.exists(): 

102 self.http_auth_path.mkdir(exist_ok=True) 

103 

104 username, password = self._generate_credentials() 

105 ht = HtpasswdFile(str(auth_file), new=True) 

106 ht.set_password(username, password) 

107 ht.save() 

108 

109 data = { 

110 "mailpit_host": f"{get_container_name_prefix(self.bench_name)}{CLI_DEFAULT_DELIMETER}mailpit", 

111 "adminer_host": f"{get_container_name_prefix(self.bench_name)}{CLI_DEFAULT_DELIMETER}adminer", 

112 "auth_file": f"/etc/nginx/http_auth/{auth_file.name}", 

113 } 

114 

115 from jinja2 import Template 

116 

117 template_path: Path = get_template_path("admin-tools-location.tmpl") 

118 

119 template = Template(template_path.read_text()) 

120 output = template.render(data) 

121 

122 if not self.nginx_config_location_path.parent.exists(): 

123 self.nginx_config_location_path.mkdir(exist_ok=True) 

124 

125 self.nginx_config_location_path.write_text(output) 

126 

127 def remove_nginx_location_config(self): 

128 if self.nginx_config_location_path.exists(): 

129 self.nginx_config_location_path.unlink() 

130 

131 # Remove htpasswd file if exists 

132 auth_file = self.http_auth_path / f"{self.bench_name}-admin-tools.htpasswd" 

133 if auth_file.exists(): 

134 auth_file.unlink() 

135 

136 # Remove credentials from bench config 

137 self.bench.bench_config.admin_tools_username = None 

138 self.bench.bench_config.admin_tools_password = None 

139 self.bench.save_bench_config(print_message=False) 

140 

141 def _get_common_site_config_path(self) -> Path: 

142 return self.compose_path.parent / "workspace/frappe-bench/sites/common_site_config.json" 

143 

144 def _get_common_site_config(self) -> dict: 

145 config_path = self._get_common_site_config_path() 

146 if not config_path.exists(): 

147 raise BenchException(self.bench_name, message="common_site_config.json not found.") 

148 return json.loads(config_path.read_bytes()) 

149 

150 def _save_common_site_config(self, config: dict): 

151 self._get_common_site_config_path().write_text(json.dumps(config)) 

152 

153 def configure_mailpit_as_default_server(self): 

154 self.output.change_head("Configuring Mailpit as default mail server") 

155 current_common_site_config = self._get_common_site_config() 

156 

157 new_conf = { 

158 "mail_port": 1025, 

159 "mail_server": f"{get_container_name_prefix(self.bench_name)}{CLI_DEFAULT_DELIMETER}mailpit", 

160 "disable_mail_smtp_authentication": 1, 

161 } 

162 

163 for key, value in new_conf.items(): 

164 if key not in current_common_site_config or not current_common_site_config[key] == value: 

165 current_common_site_config[key] = value 

166 

167 self._save_common_site_config(current_common_site_config) 

168 self.output.print("Configured Mailpit as default mail server") 

169 

170 def remove_mailpit_as_default_server(self): 

171 self.output.change_head("Removing Mailpit as default mail server") 

172 current_common_site_config = self._get_common_site_config() 

173 

174 new_conf = { 

175 "mail_port": 1025, 

176 "mail_server": f"{get_container_name_prefix(self.bench_name)}{CLI_DEFAULT_DELIMETER}mailpit", 

177 "disable_mail_smtp_authentication": 1, 

178 } 

179 

180 for key, value in new_conf.items(): 

181 if key not in current_common_site_config: 

182 continue 

183 

184 if not current_common_site_config[key] == value: 

185 continue 

186 

187 del current_common_site_config[key] 

188 

189 self._save_common_site_config(current_common_site_config) 

190 self.output.print("Removed Mailpit as default mail server") 

191 

192 def wait_till_services_started(self, interval=2, timeout=30): 

193 admin_tools_services = ["mailpit:8025", "adminer:8080"] 

194 

195 for tool in admin_tools_services: 

196 running = False 

197 for i in range(timeout): 

198 try: 

199 check_command = f"wait-for-it -t {interval} {get_container_name_prefix(self.bench_name)}{CLI_DEFAULT_DELIMETER}{tool}" 

200 self.bench.docker_client.compose.exec(service="nginx", command=check_command, stream=False) 

201 

202 running = True 

203 break 

204 except DockerException as e: 

205 continue 

206 

207 if not running: 

208 raise AdminToolsFailedToStart(self.bench_name) 

209 

210 def enable(self, force_recreate_container: bool = False, force_configure: bool = False): 

211 """Enable admin tools by starting services.""" 

212 # Use docker_client directly instead of compose_project wrapper 

213 try: 

214 self.docker_client.compose.up( 

215 services=[], 

216 detach=True, 

217 pull="never", 

218 force_recreate=force_recreate_container, 

219 ) 

220 except DockerException as e: 

221 from frappe_manager.compose_project.exceptions import DockerComposeProjectFailedToStartError 

222 

223 raise DockerComposeProjectFailedToStartError(self.compose_path, []) 

224 

225 self.wait_till_services_started() 

226 self.save_nginx_location_config() 

227 self.nginx_proxy.reload() 

228 

229 if force_configure: 

230 self.configure_mailpit_as_default_server() 

231 

232 def stop(self): 

233 """Stop admin tools containers without removing configuration.""" 

234 try: 

235 self.docker_client.compose.stop(services=[], timeout=2) 

236 except DockerException as e: 

237 from frappe_manager.compose_project.exceptions import DockerComposeProjectFailedToStopError 

238 

239 raise DockerComposeProjectFailedToStopError( 

240 self.compose_path, 

241 self.compose_file_manager.get_services_list(), 

242 ) 

243 

244 def disable(self): 

245 """Disable admin tools by stopping services and removing all configuration.""" 

246 self.stop() 

247 

248 self.remove_nginx_location_config() 

249 self.nginx_proxy.reload() 

250 

251 self.remove_mailpit_as_default_server() 

252 

253 def is_running(self) -> bool: 

254 """Check if all admin tools services are running.""" 

255 try: 

256 services = self.compose_file_manager.get_services_list() 

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

258 all_statuses = self.docker_client.compose.get_all_services_status() 

259 

260 running_statuses = { 

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

262 } 

263 

264 if not services: 

265 return False 

266 

267 return all(running_statuses.get(service) == "running" for service in services) 

268 except Exception: 

269 return False