Coverage for src / ocarina / opinionated / plugins / reports / docx_tests_proofs.py: 91.12%
172 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-04 16:22 +0200
1"""Generate test proof documents in DOCX format.
3Transforms text log files from automated tests into formatted Word documents,
4including screenshots.
6Architecture:
7 - Recursive reading of log tree (campaigns > suites > cases)
8 - UTC metadata parsing and local time conversion
9 - Automatic screenshot detection and insertion
10 - Robust error handling with logging
12Output format:
13 - Heading 1: Test campaign
14 - Heading 2: Test suite
15 - Heading 3: Test case
16 - Body: Logs with inserted screenshots
18Example:
19 >>> generate_docx_proof(
20 ... logs_root=Path("logs/e2e"),
21 ... docx_root=Path(".docx_test_proofs_xxx"),
22 ... logger=logger,
23 ... )
25"""
27import os
28import re
29import uuid
30from concurrent.futures import ThreadPoolExecutor
31from contextlib import suppress
32from datetime import datetime
33from functools import partial
34from pathlib import Path
35from typing import TYPE_CHECKING, NamedTuple, TypedDict, final
37from docx import Document
38from docx.enum.text import WD_PARAGRAPH_ALIGNMENT
39from docx.shared import Inches
41if TYPE_CHECKING:
42 from collections.abc import Callable, Iterator
44 from docx.document import Document as DocxDocument
46 from ocarina.ports.ilogger import ILogger
48_DEFAULT_SCREENSHOT_NEEDLE = "Screenshot: "
49_DEFAULT_UTC_DATE_REGEX = re.compile(r"\[UTC_DATE::([^]]+)]")
52def _default_format_date(local_dt: datetime) -> str:
53 """Render a parsed UTC marker as local time: ``[MM/DD/YYYY | HHhMM:SS.ffffff]``.
55 Receives the marker's datetime already converted to the local timezone, and
56 returns the full text that replaces the ``[UTC_DATE::...]`` marker (brackets
57 included). Override it via ``generate_docx_proof(format_date=...)`` to render
58 another layout, e.g. a French ``[%d/%m/%Y à %Hh%M:%S]``.
59 """
60 return local_dt.strftime("[%m/%d/%Y | %Hh%M:%S.%f]")
63def _replace_utc_date(
64 line: str,
65 *,
66 utc_date_regex: re.Pattern[str],
67 format_date: Callable[[datetime], str],
68) -> str:
69 def _repl(m: re.Match[str]) -> str:
70 with suppress(Exception):
71 local_dt = datetime.fromisoformat(
72 m.group(1).replace("Z", "+00:00") # noqa: FURB162
73 ).astimezone()
74 return format_date(local_dt)
75 return m.group(0)
77 return utc_date_regex.sub(_repl, line)
80def _shorten_docx_path(docx_path: Path) -> Path:
81 """Atomically reserve a short unique path if the stem exceeds 8 chars.
83 When shortening is required, the candidate name is created with
84 ``O_CREAT | O_EXCL``, so the filesystem itself guarantees uniqueness in a
85 single reserve-or-fail syscall (exactly like ``mkdir(exist_ok=False)``) —
86 no check-then-act race between two concurrent workers. The reserved file is
87 an empty placeholder the caller is expected to overwrite.
89 When the stem already fits, the path is returned unchanged and nothing is
90 reserved.
92 Raises:
93 RuntimeError: If a unique filename cannot be reserved after 500 attempts.
95 """
96 max_stem_length = 8
97 retries = 500
99 if len(docx_path.stem) <= max_stem_length:
100 return docx_path
102 parent = docx_path.parent
103 for _ in range(retries):
104 short_name = uuid.uuid4().hex[:max_stem_length]
105 new_path = parent / f"{short_name}.docx"
106 try:
107 fd = os.open(new_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
108 except FileExistsError:
109 continue
110 os.close(fd)
111 return new_path
113 msg = ( # pragma: no cover
114 f"Cannot generate a unique short filename for:"
115 " "
116 f"{docx_path} after {retries} attempts."
117 )
118 raise RuntimeError(msg) # pragma: no cover
121def _create_unique_output_dir(output_root: Path) -> Path:
122 """Create a uniquely named subdirectory (4-char hex) inside output_root.
124 Raises:
125 RuntimeError: If a unique directory cannot be created after 500 attempts.
127 """
128 retries = 500
130 for _ in range(retries): 130 ↛ 139line 130 didn't jump to line 139 because the loop on line 130 didn't complete
131 candidate = output_root / uuid.uuid4().hex[:4]
132 try:
133 candidate.mkdir(parents=True, exist_ok=False)
134 except FileExistsError:
135 continue
136 else:
137 return candidate
139 msg = (
140 "Cannot generate a unique output subdirectory under:"
141 " "
142 f"{output_root} after {retries} attempts."
143 )
144 raise RuntimeError(msg)
147class _TestCaseEntry(TypedDict):
148 test_campaign_name: str
149 test_suite_name: str
150 test_case_name: str
151 file_path: Path
154class _GenerationStats(NamedTuple):
155 """Outcome of a generation run.
157 ``attempted`` counts the test cases discovered in the log tree;
158 ``succeeded`` counts those whose DOCX was actually written to disk.
159 """
161 attempted: int
162 succeeded: int
165def _safe_iter_lines(file_path: Path, logger: ILogger) -> Iterator[str]:
166 try:
167 with file_path.open("r", encoding="utf-8", errors="replace") as f:
168 yield from f
169 except Exception as exc:
170 msg = f"Cannot read: {file_path}"
171 logger.exception(msg, exc=exc)
174def _save_docx(*, doc: DocxDocument, docx_path: Path, logger: ILogger) -> bool:
175 try:
176 doc.save(str(docx_path))
177 except Exception as exc:
178 msg = f"Cannot save file: {docx_path}"
179 logger.exception(msg, exc=exc)
180 return False
181 else:
182 logger.success(f"Successfully written: {docx_path}")
183 return True
186@final
187class _DocxProofGenerator:
188 """Generate DOCX test proofs from log files.
190 Expected log tree:
191 logs_root/
192 ├── campaign_1/
193 │ ├── suite_A/
194 │ │ ├── test_case_1.log
195 │ │ └── test_case_2.log
196 │ └── suite_B/
197 │ └── test_case_3.log
198 └── campaign_2/
199 └── ...
200 """
202 def __init__(
203 self,
204 *,
205 logs_root: Path,
206 docx_root: Path,
207 screenshot_needle: str = _DEFAULT_SCREENSHOT_NEEDLE,
208 utc_date_regex: re.Pattern[str] = _DEFAULT_UTC_DATE_REGEX,
209 format_date: Callable[[datetime], str] = _default_format_date,
210 ) -> None:
211 self.logs_root = logs_root
212 self.docx_root = docx_root
213 self._screenshot_needle = screenshot_needle
214 self._utc_date_regex = utc_date_regex
215 self._format_date = format_date
217 def _iter_test_cases(self, logger: ILogger) -> Iterator[_TestCaseEntry]:
218 try:
219 for campaign_dir in self.logs_root.iterdir():
220 if not campaign_dir.is_dir():
221 continue
223 for suite_dir in campaign_dir.iterdir():
224 if not suite_dir.is_dir():
225 continue
227 for file in suite_dir.iterdir():
228 if file.is_file(): 228 ↛ 227line 228 didn't jump to line 227 because the condition on line 228 was always true
229 yield _TestCaseEntry(
230 test_campaign_name=campaign_dir.name,
231 test_suite_name=suite_dir.name,
232 test_case_name=file.stem,
233 file_path=file,
234 )
235 except OSError as exc:
236 msg = f"Failed to iterate over: {self.logs_root}"
237 logger.exception(msg, exc=exc)
239 def _create_docx_from_case(
240 self, test_case: _TestCaseEntry, logger: ILogger
241 ) -> bool:
242 campaign_name = test_case["test_campaign_name"]
243 suite_name = test_case["test_suite_name"]
244 case_name = test_case["test_case_name"]
245 file_path = test_case["file_path"]
247 doc = Document()
248 doc.add_heading(
249 campaign_name, level=1
250 ).alignment = WD_PARAGRAPH_ALIGNMENT.CENTER
251 doc.add_heading(suite_name, level=2).alignment = WD_PARAGRAPH_ALIGNMENT.CENTER
252 doc.add_paragraph()
253 doc.add_heading(case_name, level=3)
255 for line in _safe_iter_lines(file_path, logger):
256 normalized_line = _replace_utc_date(
257 line.rstrip("\n"),
258 utc_date_regex=self._utc_date_regex,
259 format_date=self._format_date,
260 )
262 if self._screenshot_needle in normalized_line:
263 parts = normalized_line.split(self._screenshot_needle, 1)
264 if len(parts) > 1 and parts[1].strip(): 264 ↛ 255line 264 didn't jump to line 255 because the condition on line 264 was always true
265 image_path = Path(parts[1].strip())
266 if image_path.exists() and image_path.is_file():
267 doc.add_paragraph()
268 try:
269 doc.add_picture(str(image_path), width=Inches(6))
270 except Exception as exc:
271 msg = f"Cannot add image {image_path}"
272 logger.exception(msg, exc=exc)
273 else:
274 p = doc.add_paragraph()
275 p.paragraph_format.space_before = 0
276 p.paragraph_format.space_after = 0
277 p.add_run(normalized_line)
279 relative_path = file_path.relative_to(self.logs_root)
280 docx_path = (self.docx_root / relative_path).with_suffix(".docx").resolve()
282 try:
283 docx_path.parent.mkdir(parents=True, exist_ok=True)
284 doc.save(str(docx_path))
285 except OSError:
286 original_docx_path = docx_path
287 docx_path = _shorten_docx_path(docx_path)
288 msg = (
289 f"Filename too long or access error for: {original_docx_path}."
290 f" Retrying with: {docx_path}"
291 )
292 logger.warning(msg)
293 saved = _save_docx(docx_path=docx_path, doc=doc, logger=logger)
294 if not saved:
295 # Drop the empty placeholder reserved above (or any partial
296 # file) so a failed retry never leaves a 0-byte .docx behind.
297 with suppress(OSError):
298 docx_path.unlink()
299 return saved
300 else:
301 return True
303 def generate_docx_proofs(
304 self, logger: ILogger, *, max_workers: int = 1
305 ) -> _GenerationStats:
306 """Generate one DOCX per test case, optionally in parallel.
308 Cases are independent units of disk I/O (each reads its own log, builds
309 its own Document, and writes a unique output path), so they parallelise
310 cleanly. The only shared object is ``logger``; its writes may interleave
311 across threads, which is harmless for this best-effort reporter.
313 Args:
314 logger: Logger for progress and error reporting.
315 max_workers: Upper bound on worker threads. ``<= 1`` runs the original
316 sequential path verbatim — no list materialisation, no pool, no
317 thread. Above 1, it is clamped to at most the number of cases.
319 Returns:
320 A ``_GenerationStats`` pair: the number of cases discovered
321 (``attempted``) and the number whose DOCX was actually written
322 to disk (``succeeded``).
324 """
325 if max_workers <= 1:
326 attempted = 0
327 succeeded = 0
328 for case in self._iter_test_cases(logger):
329 attempted += 1
330 succeeded += self._create_docx_from_case(case, logger)
331 return _GenerationStats(attempted=attempted, succeeded=succeeded)
333 cases = list(self._iter_test_cases(logger))
334 if not cases: 334 ↛ 335line 334 didn't jump to line 335 because the condition on line 334 was never true
335 return _GenerationStats(attempted=0, succeeded=0)
337 workers = min(max_workers, len(cases))
338 create = partial(self._create_docx_from_case, logger=logger)
339 with ThreadPoolExecutor(max_workers=workers) as executor:
340 succeeded = sum(executor.map(create, cases))
341 return _GenerationStats(attempted=len(cases), succeeded=succeeded)
344def generate_docx_proof( # noqa: PLR0913
345 *,
346 logs_root: Path,
347 output_root: Path,
348 logger: ILogger,
349 screenshot_needle: str = _DEFAULT_SCREENSHOT_NEEDLE,
350 utc_date_regex: re.Pattern[str] = _DEFAULT_UTC_DATE_REGEX,
351 format_date: Callable[[datetime], str] = _default_format_date,
352 auto_create_unique_directory: bool = True,
353 max_workers: int = 1,
354) -> None:
355 """Generate DOCX test proofs from log files.
357 Args:
358 logs_root: Root directory of the log tree to process.
359 output_root: Root directory where DOCX files will be written.
360 logger: Logger for progress and error reporting.
361 screenshot_needle: used to detect screenshot lines. Default: "Screenshot: ".
362 utc_date_regex: used to detect and replace UTC dates. Default: [UTC_DATE::...].
363 format_date: Renders each matched date marker. Receives the marker's
364 datetime already converted to local time and returns the full
365 replacement text (brackets included). Default: a US-style
366 ``[MM/DD/YYYY | HHhMM:SS.ffffff]``. Pass your own, e.g.
367 ``lambda dt: dt.strftime("[%d/%m/%Y à %Hh%M:%S]")`` for a French layout.
368 auto_create_unique_directory: creates automatically a random-named unique dir.
369 max_workers: Worker threads used to generate documents in parallel.
370 Default: 1 (sequential). Clamped to at least 1 and at most the total
371 number of documents to generate; raise it to parallelise.
373 """
374 if not logs_root.is_dir():
375 msg = f"Directory not accessible: {logs_root}. Generation skipped."
376 logger.warning(msg)
377 return
379 docx_root = (
380 _create_unique_output_dir(output_root)
381 if auto_create_unique_directory
382 else output_root
383 )
385 stats = _DocxProofGenerator(
386 logs_root=logs_root,
387 docx_root=docx_root,
388 screenshot_needle=screenshot_needle,
389 utc_date_regex=utc_date_regex,
390 format_date=format_date,
391 ).generate_docx_proofs(logger, max_workers=max_workers)
393 if stats.attempted == 0:
394 msg = f"No test case found under: {logs_root}. Nothing was generated."
395 logger.warning(msg)
396 if auto_create_unique_directory: 396 ↛ 399line 396 didn't jump to line 399 because the condition on line 396 was always true
397 with suppress(Exception):
398 docx_root.rmdir()
399 return
401 if stats.succeeded == 0:
402 msg = (
403 f"Found {stats.attempted} test case(s) under: {logs_root}, but every"
404 f" DOCX generation failed. Output: {docx_root}"
405 )
406 logger.warning(msg)
407 return
409 if stats.succeeded < stats.attempted:
410 msg = (
411 f"Plugin execution done with errors. Generated"
412 f" {stats.succeeded}/{stats.attempted} DOCX. Output: {docx_root}"
413 )
414 logger.warning(msg)
415 return
417 msg = (
418 f"Plugin execution done. Generated {stats.succeeded} DOCX. Output: {docx_root}"
419 )
420 logger.info(msg)