Coverage for src / mafw / devtools / documentation / server.py: 86%
159 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-06-28 13:34 +0000
1# Copyright 2025–2026 European Union
2# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
3# SPDX-License-Identifier: EUPL-1.2
4"""
5Local documentation server management for MAFw.
7This module provides the :class:`~.ServerStatusEnum`, helper functions, and
8orchestration logic for managing a ``python -m http.server`` subprocess used
9to serve locally built versioned documentation.
10"""
12from __future__ import annotations
14import json
15import subprocess
16import sys
17from datetime import datetime, timezone
18from enum import StrEnum
19from pathlib import Path
20from typing import Any, List, Tuple, cast
21from urllib.error import HTTPError, URLError
22from urllib.request import urlopen
24import psutil
26from mafw.devtools import DevtoolsError
28SERVER_DEFAULT_DIRECTORY = Path('docs') / 'build'
29"""Default directory served by the local documentation server."""
31SERVER_METADATA_FILENAME = '.multiversion-doc-server.json'
32"""Filename for server metadata stored inside the served directory."""
35class ServerStatusEnum(StrEnum):
36 """Possible status values for the multiversion-doc local documentation server.
38 :cvar running: The server process exists, matches the expected signature and responds over HTTP.
39 :cvar stale: The server process exists and matches the expected signature, but does not respond over HTTP.
40 :cvar stopped: The recorded PID does not exist or does not match the expected signature.
41 :cvar unknown: The metadata file is missing, so no status can be determined.
42 """
44 running = 'running'
45 stale = 'stale'
46 stopped = 'stopped'
47 unknown = 'unknown'
50def server_default_directory() -> Path:
51 """Return the default directory served by the local documentation server.
53 :return: Default documentation build directory
54 :rtype: Path
55 """
56 return SERVER_DEFAULT_DIRECTORY.resolve()
59def server_metadata_path(directory: Path) -> Path:
60 """Return the metadata path associated to a served directory.
62 :param directory: Served directory
63 :type directory: Path
64 :return: Metadata JSON file path
65 :rtype: Path
66 """
67 return Path(directory) / SERVER_METADATA_FILENAME
70def read_server_metadata(directory: Path) -> dict[str, Any] | None:
71 """Read local documentation server metadata if available.
73 :param directory: Served directory to locate metadata under
74 :type directory: Path
75 :return: Parsed metadata, or None when the metadata file is missing
76 :rtype: dict[str, Any] | None
77 :raises DevtoolsError: If the metadata file exists but cannot be parsed
78 """
79 meta_path = server_metadata_path(directory)
80 if not meta_path.exists():
81 return None
82 try:
83 loaded = json.loads(meta_path.read_text(encoding='utf-8'))
84 return cast(dict[str, Any], loaded)
85 except (OSError, json.JSONDecodeError) as e:
86 raise DevtoolsError(f'Invalid server metadata file: {meta_path}') from e
89def write_server_metadata(directory: Path, metadata: dict[str, Any]) -> None:
90 """Write local documentation server metadata.
92 :param directory: Served directory where metadata is stored
93 :type directory: Path
94 :param metadata: Metadata dictionary to serialize as JSON
95 :type metadata: dict[str, Any]
96 :raises DevtoolsError: If the metadata file cannot be written
97 """
98 meta_path = server_metadata_path(directory)
99 try:
100 meta_path.write_text(json.dumps(metadata, indent=2, sort_keys=True) + '\n', encoding='utf-8')
101 except OSError as e:
102 raise DevtoolsError(f'Cannot write server metadata: {meta_path}') from e
105def process_matches_http_server(proc: Any, address: str, port: int, directory: Path) -> bool:
106 """Check whether a process resembles the expected http.server invocation.
108 :param proc: psutil Process instance
109 :type proc: Any
110 :param address: Expected bind address
111 :type address: str
112 :param port: Expected port
113 :type port: int
114 :param directory: Expected served directory
115 :type directory: Path
116 :return: True when the process signature matches
117 :rtype: bool
118 """
119 try:
120 cmdline: List[str] = list(proc.cmdline())
121 except Exception:
122 return False
124 if '-m' not in cmdline:
125 return False
126 m_index = cmdline.index('-m')
127 if m_index + 1 >= len(cmdline) or cmdline[m_index + 1] != 'http.server': 127 ↛ 128line 127 didn't jump to line 128 because the condition on line 127 was never true
128 return False
130 directory_str = str(Path(directory).resolve())
131 checks = [
132 ('-b', address),
133 ('-d', directory_str),
134 ]
135 for flag, expected in checks:
136 if flag not in cmdline: 136 ↛ 137line 136 didn't jump to line 137 because the condition on line 136 was never true
137 return False
138 i = cmdline.index(flag)
139 if i + 1 >= len(cmdline) or cmdline[i + 1] != expected: 139 ↛ 140line 139 didn't jump to line 140 because the condition on line 139 was never true
140 return False
142 if str(port) not in cmdline: 142 ↛ 143line 142 didn't jump to line 143 because the condition on line 142 was never true
143 return False
145 return True
148def http_probe(address: str, port: int, timeout_s: float = 1.0) -> bool:
149 """Probe whether the server responds at the given address:port.
151 :param address: Server address
152 :type address: str
153 :param port: Server port
154 :type port: int
155 :param timeout_s: Timeout in seconds
156 :type timeout_s: float
157 :return: True if a GET to the server root returns HTTP 200
158 :rtype: bool
159 """
160 url = f'http://{address}:{port}/'
161 try:
162 with urlopen(url, timeout=timeout_s) as resp:
163 return getattr(resp, 'status', None) == 200
164 except (HTTPError, URLError, OSError):
165 return False
168def format_started_at_utc(now: datetime) -> str:
169 """Format a timestamp for server metadata.
171 :param now: A datetime instance (timezone-aware is preferred)
172 :type now: datetime
173 :return: UTC ISO-8601 timestamp with ``Z`` suffix
174 :rtype: str
175 """
176 utc = now.astimezone(timezone.utc).replace(microsecond=0)
177 return utc.isoformat().replace('+00:00', 'Z')
180def get_server_status(directory: Path) -> Tuple[ServerStatusEnum, dict[str, Any] | None]:
181 """Determine the current server status from metadata and process/HTTP checks."""
182 metadata = read_server_metadata(directory)
183 if metadata is None:
184 return ServerStatusEnum.unknown, None
186 pid = metadata.get('pid')
187 address = metadata.get('address')
188 port = metadata.get('port')
189 dir_value = metadata.get('directory')
191 if (
192 not isinstance(pid, int)
193 or not isinstance(address, str)
194 or not isinstance(port, int)
195 or not isinstance(dir_value, str)
196 ):
197 raise DevtoolsError(f'Invalid server metadata contents: {server_metadata_path(directory)}')
199 if not psutil.pid_exists(pid):
200 return ServerStatusEnum.stopped, metadata
202 try:
203 proc = psutil.Process(pid)
204 except Exception:
205 return ServerStatusEnum.stopped, metadata
207 if not process_matches_http_server(proc, address=address, port=port, directory=Path(dir_value)):
208 return ServerStatusEnum.stopped, metadata
210 if http_probe(address=address, port=port):
211 return ServerStatusEnum.running, metadata
212 return ServerStatusEnum.stale, metadata
215def server_stop_by_metadata(directory: Path, require_metadata: bool) -> None:
216 """Stop the server based on metadata found under a served directory."""
217 metadata = read_server_metadata(directory)
218 meta_path = server_metadata_path(directory)
219 if metadata is None:
220 if require_metadata:
221 raise DevtoolsError(f'Server metadata not found: {meta_path}')
222 return
224 pid = metadata.get('pid')
225 address = metadata.get('address')
226 port = metadata.get('port')
227 dir_value = metadata.get('directory')
229 if (
230 not isinstance(pid, int)
231 or not isinstance(address, str)
232 or not isinstance(port, int)
233 or not isinstance(dir_value, str)
234 ):
235 raise DevtoolsError(f'Invalid server metadata contents: {meta_path}')
237 if not psutil.pid_exists(pid):
238 try:
239 meta_path.unlink(missing_ok=True)
240 except OSError:
241 pass
242 print('ℹ️ Server already stopped; removed stale metadata.')
243 return
245 try:
246 proc = psutil.Process(pid)
247 except Exception:
248 try:
249 meta_path.unlink(missing_ok=True)
250 except OSError:
251 pass
252 print('ℹ️ Server already stopped; removed stale metadata.')
253 return
255 if not process_matches_http_server(proc, address=address, port=port, directory=Path(dir_value)):
256 try:
257 meta_path.unlink(missing_ok=True)
258 except OSError:
259 pass
260 print('ℹ️ Recorded PID does not match the documentation server; removed metadata.')
261 return
263 try:
264 proc.terminate()
265 try:
266 proc.wait(timeout=3.0)
267 except Exception:
268 proc.kill()
269 proc.wait(timeout=3.0)
270 finally:
271 try:
272 meta_path.unlink(missing_ok=True)
273 except OSError:
274 pass
276 print('✅ Documentation server stopped.')
279def server_start_impl(address: str, port: int, directory: Path, log_file: Path, force_restart: bool) -> None:
280 """Implementation for starting a local documentation server."""
281 directory = Path(directory).resolve()
282 log_file = Path(log_file).resolve()
284 if not directory.exists():
285 raise DevtoolsError(f'Served directory does not exist: {directory}')
287 status, metadata = get_server_status(directory)
288 if status in (ServerStatusEnum.running, ServerStatusEnum.stale):
289 assert metadata is not None
290 url = f'http://{metadata["address"]}:{metadata["port"]}'
291 print(f'ℹ️ Documentation server already present ({status}) at {url}')
292 if not force_restart:
293 return
294 server_stop_by_metadata(directory, require_metadata=False)
296 log_file.parent.mkdir(parents=True, exist_ok=True)
297 with open(log_file, 'a', encoding='utf-8') as log_handle:
298 process = subprocess.Popen(
299 [
300 sys.executable,
301 '-m',
302 'http.server',
303 '-b',
304 address,
305 '-d',
306 str(directory),
307 str(port),
308 ],
309 stdout=log_handle,
310 stderr=subprocess.STDOUT,
311 start_new_session=True,
312 )
314 metadata_to_write = {
315 'pid': int(process.pid),
316 'address': address,
317 'port': int(port),
318 'directory': str(directory),
319 'log_file': str(log_file),
320 'started_at': format_started_at_utc(datetime.now(timezone.utc)),
321 }
322 write_server_metadata(directory, metadata_to_write)
324 print(f'✅ Documentation server launched and responding at http://{address}:{port}')