Coverage for frappe_manager / site_manager / modules / bench_info.py: 23%
147 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-07-02 18:13 +0530
1"""
2BenchInfo Module
4Handles information retrieval and display for the bench including:
5- Displaying comprehensive bench information
6- Reading config files (common_site_config.json, site_config.json)
7- Getting installed apps list
8- Getting log file paths
9"""
11import json
12from pathlib import Path
13from typing import TYPE_CHECKING, Any
15from rich.console import Console
16from rich.table import Table
18from frappe_manager.docker import DockerException
19from frappe_manager.output_manager import OutputHandler, temporary_stop
20from frappe_manager.output_manager.rich_output import RichOutputHandler
21from frappe_manager.site_manager.exceptions import BenchException
22from frappe_manager.ssl_manager import SUPPORTED_SSL_TYPES
23from frappe_manager.ssl_manager.letsencrypt_certificate import LetsencryptSSLCertificate
24from frappe_manager.utils.helpers import format_ssl_certificate_time_remaining
25from frappe_manager.utils.site import generate_services_table
27if TYPE_CHECKING:
28 from frappe_manager.services_manager.services import ServicesManager
29 from frappe_manager.site_manager.bench_config import BenchConfig
30 from frappe_manager.site_manager.modules.bench_admin_tools import BenchAdminTools
31 from frappe_manager.site_manager.modules.bench_workers import BenchWorkers
32 from frappe_manager.ssl_manager.ssl_certificate_manager import SSLCertificateManager
35class BenchInfo:
36 """
37 Manages information retrieval and display for a bench.
39 Responsibilities:
40 - Display comprehensive bench information
41 - Read configuration files
42 - Get installed apps list
43 - Get log file paths
44 """
46 def __init__(
47 self,
48 bench_name: str,
49 bench_path: Path,
50 bench_config: "BenchConfig",
51 services: "ServicesManager",
52 workers: "BenchWorkers",
53 admin_tools: "BenchAdminTools",
54 certificate_manager: "SSLCertificateManager",
55 get_db_connection_info_fn,
56 has_certificate_fn,
57 is_running_fn,
58 get_services_running_status_fn,
59 output_handler: OutputHandler | None = None,
60 ):
61 """
62 Initialize BenchInfo module.
64 Args:
65 bench_name: Name of the bench
66 bench_path: Path to bench directory
67 bench_config: Bench configuration object
68 services: Services manager instance
69 workers: Workers manager instance
70 admin_tools: Admin tools instance
71 certificate_manager: SSL certificate manager
72 get_db_connection_info_fn: Callable to get DB connection info
73 has_certificate_fn: Callable to check if certificate exists
74 is_running_fn: Callable to check if bench is running
75 get_services_running_status_fn: Callable to get services status
76 output_handler: Optional output handler for displaying information
77 """
78 self.bench_name = bench_name
79 self.bench_path = bench_path
80 self.bench_config = bench_config
81 self.services = services
82 self.workers = workers
83 self.admin_tools = admin_tools
84 self.certificate_manager = certificate_manager
85 self.get_db_connection_info = get_db_connection_info_fn
86 self.has_certificate = has_certificate_fn
87 self.is_running = is_running_fn
88 self.get_services_running_status = get_services_running_status_fn
89 self.output = output_handler or RichOutputHandler()
91 def get_common_config(self) -> dict:
92 """
93 Get common site configuration from common_site_config.json.
95 Returns:
96 dict: Common site configuration
98 Raises:
99 BenchException: If common_site_config.json not found
100 """
101 common_bench_config_path = self.bench_path / "workspace/frappe-bench/sites/common_site_config.json"
102 if not common_bench_config_path.exists():
103 raise BenchException(self.bench_name, message="common_site_config.json not found.")
104 return json.loads(common_bench_config_path.read_text())
106 def get_site_config(self) -> dict:
107 """
108 Get site-specific configuration from site_config.json.
110 Returns:
111 dict: Site configuration
113 Raises:
114 BenchException: If site_config.json not found
115 """
116 site_config_path = self.bench_path / "workspace/frappe-bench/sites" / self.bench_name / "site_config.json"
117 if not site_config_path.exists():
118 raise BenchException(self.bench_name, message="site_config.json not found.")
119 return json.loads(site_config_path.read_text())
121 def get_installed_apps_list(self) -> dict[str, Any]:
122 """
123 Get list of installed apps from apps.json.
125 Returns:
126 dict: Installed apps data with versions, or empty dict if not found
127 """
128 apps_json_file = self.bench_path / "workspace/frappe-bench/sites/apps.json"
129 if not apps_json_file.exists():
130 return {}
131 with open(apps_json_file) as f:
132 apps_data = json.load(f)
133 return apps_data
135 def get_python_version(self) -> str:
136 """
137 Read the active Python version from the uv python-default symlink.
139 Returns:
140 Version string (e.g. "3.12.9") or "N/A" if not resolvable.
141 """
142 symlink = self.bench_path / "workspace/frappe-bench/.uv/python-default"
143 try:
144 target = Path(symlink.readlink()).name # e.g. cpython-3.12.9-linux-x86_64-gnu
145 parts = target.split("-")
146 return parts[1] if len(parts) > 1 else "N/A"
147 except (OSError, IndexError):
148 return "N/A"
150 def get_node_version(self) -> str:
151 """
152 Read the active Node version from the fnm default alias symlink.
154 Returns:
155 Version string (e.g. "v22.11.0") or "N/A" if not resolvable.
156 """
157 symlink = self.bench_path / "workspace/frappe-bench/.fnm/aliases/default"
158 try:
159 target = symlink.readlink() # e.g. ../node-versions/v22.11.0/installation
160 version = next((p for p in target.parts if p.startswith("v")), None)
161 return version or "N/A"
162 except OSError:
163 return "N/A"
165 def get_log_file_paths(self) -> list[Path]:
166 """
167 Get log file paths based on environment type.
169 Returns:
170 list: List of log file paths
171 """
172 base_log_dir = self.bench_path / "workspace/frappe-bench/logs"
173 if self.bench_config.environment_type.value == "dev":
174 bench_dev_server_log_path = base_log_dir / "web.dev.log"
175 return [bench_dev_server_log_path]
176 bench_prod_server_log_path_stdout = base_log_dir / "web.log"
177 bench_prod_server_log_path_stderr = base_log_dir / "web.error.log"
178 return [bench_prod_server_log_path_stderr, bench_prod_server_log_path_stdout]
180 def display_info(self) -> None:
181 """
182 Retrieve and display comprehensive information about the bench.
184 This includes: URL, root path, status, credentials, database info,
185 SSL status, admin tools, installed apps, and running services.
186 """
187 self.output.change_head("Getting bench info")
188 bench_db_info = self.get_db_connection_info()
190 db_user = bench_db_info.get("name", "N/A")
191 db_pass = bench_db_info.get("password", "N/A")
193 services_db_info = self.services.database_manager.database_server_info
194 bench_info_table = Table(show_lines=True, show_header=False, highlight=True)
196 protocol = "https" if self.has_certificate() else "http"
198 # Get admin password from site_config.json if available
199 admin_pass = self.bench_config.admin_pass + " (default)"
200 site_config = self.get_site_config()
201 if "admin_password" in site_config:
202 admin_pass = site_config["admin_password"]
204 ssl_cert = self.bench_config.get_primary_certificate()
205 ssl_service_type = f"{ssl_cert.ssl_type.value}"
206 if ssl_cert.ssl_type == SUPPORTED_SSL_TYPES.le:
207 if isinstance(ssl_cert, LetsencryptSSLCertificate):
208 ssl_service_type = f"[{ssl_cert.challenge_type.value}] {ssl_cert.ssl_type.value}"
209 else:
210 ssl_service_type = f"{ssl_cert.ssl_type.value}"
212 is_running = self.is_running()
213 status = "Active" if is_running else "Inactive"
214 status_color = "green" if is_running else "red"
215 status_display = f"[{status_color}]{status}[/{status_color}]"
217 data = {
218 "Bench Url": f"{protocol}://{self.bench_name}",
219 "Bench Root": f"[link=file://{self.bench_path.absolute()}]{self.bench_path.absolute()}[/link]",
220 "Status": status_display,
221 "Python Version": self.get_python_version(),
222 "Node Version": self.get_node_version(),
223 "Frappe Username": "administrator",
224 "Frappe Password": admin_pass,
225 "Root DB User": services_db_info.user,
226 "Root DB Password": services_db_info.password,
227 "Root DB Host": services_db_info.host,
228 "DB Name": db_user,
229 "DB User": db_user,
230 "DB Password": db_pass,
231 "Environment": self.bench_config.environment_type.value,
232 "HTTPS": (
233 f"{ssl_service_type.upper()} ({format_ssl_certificate_time_remaining(self.certificate_manager.get_certificate_expiry())})"
234 if self.has_certificate()
235 else "Not Enabled"
236 ),
237 }
239 # Add alias domains if present (independent of SSL status)
240 if self.bench_config.alias_domains:
241 alias_list = "\n".join(sorted(self.bench_config.alias_domains))
242 data["Alias Domains"] = alias_list
244 if not self.bench_config.admin_tools:
245 data["Admin Tools"] = "Not Enabled"
246 else:
247 # Create main admin tools table
248 admin_tools_Table = Table(show_lines=False, show_edge=False, pad_edge=False, expand=True)
249 admin_tools_Table.add_column("Service", style="cyan")
250 admin_tools_Table.add_column("URL", style="blue")
252 # Get auth credentials
253 username = self.bench_config.admin_tools_username or "admin"
254 password = self.bench_config.admin_tools_password or "protected"
256 # Create auth info section
257 auth_info = f"\nAuthentication Required:\n Username: [cyan]{username}[/cyan]\n Password: [green]{password}[/green]"
259 admin_tools_Table.add_row("Mailpit", f"{protocol}://{self.bench_name}/mailpit")
260 admin_tools_Table.add_row("Adminer", f"{protocol}://{self.bench_name}/adminer")
262 # Combine table and auth info
263 from rich.console import Group
265 data["Admin Tools"] = Group(admin_tools_Table, auth_info)
267 bench_info_table.add_column(no_wrap=True)
268 bench_info_table.add_column(no_wrap=True)
270 for key in data:
271 bench_info_table.add_row(key, data[key])
273 # Get bench apps data
274 apps_json = self.get_installed_apps_list()
275 if apps_json:
276 bench_apps_list_table = Table(show_lines=True, show_edge=False, pad_edge=False, expand=True)
277 bench_apps_list_table.add_column("App")
278 bench_apps_list_table.add_column("Version")
279 for app in apps_json.keys():
280 bench_apps_list_table.add_row(app, apps_json[app]["version"])
281 bench_info_table.add_row("Bench Apps", bench_apps_list_table)
283 running_bench_services = self.get_services_running_status()
285 # Get workers status using docker_client
286 try:
287 services = self.workers.compose_file_manager.get_services_list()
288 containers = self.workers.compose_file_manager.get_container_names().values()
289 all_statuses = self.workers.docker_client.compose.get_all_services_status()
290 running_bench_workers = {
291 status["Service"]: status["State"] for status in all_statuses if status.get("Name") in containers
292 }
293 except DockerException:
294 running_bench_workers = {}
296 # Get admin tools services status directly from docker_client
297 running_bench_admin_tools = {}
298 if self.admin_tools.compose_file_manager.exists():
299 try:
300 services = self.admin_tools.compose_file_manager.get_services_list()
301 containers = self.admin_tools.compose_file_manager.get_container_names().values()
302 all_statuses = self.admin_tools.docker_client.compose.get_all_services_status()
303 running_bench_admin_tools = {
304 status["Service"]: status["State"] for status in all_statuses if status.get("Name") in containers
305 }
306 except Exception:
307 running_bench_admin_tools = {}
309 if running_bench_services:
310 bench_services_table = generate_services_table(running_bench_services)
311 bench_info_table.add_row("Bench Services", bench_services_table)
313 if running_bench_workers:
314 bench_workers_table = generate_services_table(running_bench_workers)
315 bench_info_table.add_row("Bench Workers", bench_workers_table)
317 if running_bench_admin_tools:
318 bench_admin_table = generate_services_table(running_bench_admin_tools)
319 bench_info_table.add_row("Bench Admin Tools", bench_admin_table)
321 # Stop spinner temporarily to print the table without corruption
322 with temporary_stop(self.output):
323 console = Console()
324 console.print(bench_info_table)