"""A broad handler must not overwrite a precise diagnosis with a generic one. ``except Exception: raise AppError(...)`` also catches ``AppError``, so a service that raised INVALID_CRS with a 409 comes out as whatever generic code the outer handler chose. For a product whose whole claim is that a result can be traced back to its cause, that is the wrong direction in which to lose information. Wrapping a narrow call whose failure the outer message describes better is legitimate — coercing a CRS string, parsing one geometry, calling one's own private helper. Relabelling what *another* component reported is not. """ from __future__ import annotations import re from pathlib import Path APP = Path(__file__).resolve().parents[1] / "app" GENERIC_HANDLER = re.compile(r"except Exception as \w+:\s*\n\s*raise AppError\(", re.M) OTHER_SERVICE = re.compile(r"\b([A-Z]\w*Service)\.") # Cross-component entry points that carry their own considered diagnosis. GOVERNED_CALLS = re.compile(r"assert_eligible|assert_within_storage_root|canonicalize_|_load_dataset_payload") def _enclosing_service(text: str, handler_line: int) -> str | None: """The class the handler sits in, not merely the first one in the file.""" enclosing = None for match in re.finditer(r"^class (\w+)[:(]", text, re.M): if text[: match.start()].count(chr(10)) > handler_line: break enclosing = match.group(1) return enclosing def _try_body(lines: list[str], handler_line: int) -> str: index = handler_line while index > 0 and lines[index].strip() != "try:": index -= 1 return "\n".join(lines[index:handler_line]) def calls_another_component(body: str, own_service: str | None) -> bool: if GOVERNED_CALLS.search(body): return True return any(name != own_service for name in OTHER_SERVICE.findall(body)) def _offenders() -> list[str]: found: list[str] = [] for path in sorted(APP.rglob("*.py")): text = path.read_text(encoding="utf-8") lines = text.splitlines() for match in GENERIC_HANDLER.finditer(text): preceding = text[max(0, match.start() - 240):match.start() + 40] if "except AppError:" in preceding: continue handler_line = text[:match.start()].count("\n") own = _enclosing_service(text, handler_line) if calls_another_component(_try_body(lines, handler_line), own): found.append(f"{path.relative_to(APP)}:{handler_line + 1}") return found def test_a_call_into_another_component_keeps_the_error_it_raised() -> None: offenders = _offenders() assert not offenders, ( "These catch Exception around a call into another component and relabel " "whatever it raised, including its AppError. Add `except AppError: " f"raise` before the generic handler: {offenders}" ) def test_the_check_recognises_the_shape_it_guards_against() -> None: """Without this the guard could pass because its pattern never matches.""" other = "try:\n DatasetService.get_dataset(db, dataset_id)" own = "try:\n GrbAcquisitionService._extract_dimension(geometry)" governed = "try:\n DatasetConsumptionGate.assert_eligible(dataset)" assert calls_another_component(other, "GrbAcquisitionService") assert calls_another_component(governed, "GrbAcquisitionService") # Wrapping one's own private helper is the legitimate case. assert not calls_another_component(own, "GrbAcquisitionService") assert GENERIC_HANDLER.search("except Exception as exc:\n raise AppError(")