From f74be9bdba83cda94d6fa2f06d1db5d01c4d85ab Mon Sep 17 00:00:00 2001 From: Dan Wolfson Date: Sat, 7 Nov 2026 23:06:30 +0000 Subject: [PATCH] feat(report-specs): resolve every report spec at load; list unrunnable specs with words (RS-W2-2 PR-B) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pyegeria/view/spec_resolver.py runs the report-spec pipeline model's §2c checks over each spec: 1 names (find method resolves on the executor's client class; analytic routine registered, python binding imports; binders and metric_path), 2 parameters (what the executor always sends is accepted by the find method; analytic_spec_params are declared by the routine), 3 single stage recorded, 4 formatter vs out (known output formats, charts need a routine, analytic columns are declared out fields, detail_spec registered), 5 role inputs listed (supply is W2-3), 6 admission/tile_safe recorded, 7 seeding method recorded. A failing spec is an Unrunnable carrying each failing check's sentence; it is still listed and never deleted, and the executor is unchanged. Results are cached per spec and invalidated when the spec object, the set of spec names or the analytic registry changes. Hooked into startup (first get_report_registry), refresh_report_specs, load_egeria_report_specs and the refresh_specs CLI: each logs the count of unrunnable specs and their words. list_report_specs(with_status=True) gives {name, runnable, unrunnable_reason}; the default return is unchanged. FormatSet accepts target_type_name (Egeria's property) as target_type's canonical key. First resolution over all 437 specs: 2 unrunnable (Asset-Graph, Data-Classes: find methods that do not exist), 61 format-only (no action). Co-Authored-By: Claude Opus 5.5 Signed-off-by: Dan Wolfson --- commands/tech/refresh_specs.py | 7 + pyegeria/view/_output_format_models.py | 14 + pyegeria/view/base_report_formats.py | 29 +- pyegeria/view/spec_resolver.py | 439 ++++++++++++++++++ .../micro-tests/test_rs_w2_2_spec_resolver.py | 276 +++++++++++ 5 files changed, 762 insertions(+), 3 deletions(-) create mode 100644 pyegeria/view/spec_resolver.py create mode 100644 tests/micro-tests/test_rs_w2_2_spec_resolver.py diff --git a/commands/tech/refresh_specs.py b/commands/tech/refresh_specs.py index 26eebd59..ee958244 100644 --- a/commands/tech/refresh_specs.py +++ b/commands/tech/refresh_specs.py @@ -99,6 +99,13 @@ def main(): run_command(cmd_reports, "Report Specifications Generation") + # 4. Resolve every report spec (RS-W2-2 PR-B, note §2c) and say which cannot run, and why. + # Imported here, after the generation subprocess, so the freshly written specs are the ones read. + from pyegeria.view.spec_resolver import log_resolution_summary + unrunnable = log_resolution_summary(force=True, where="refresh_specs") + logger.info(f"Report spec resolution: {len(unrunnable)} spec(s) cannot run" + + ("; their reasons are logged above." if unrunnable else ".")) + logger.info("All regeneration tasks have completed successfully.") if __name__ == "__main__": diff --git a/pyegeria/view/_output_format_models.py b/pyegeria/view/_output_format_models.py index aea0c6c8..ab3deb51 100644 --- a/pyegeria/view/_output_format_models.py +++ b/pyegeria/view/_output_format_models.py @@ -333,8 +333,22 @@ def _migrate_legacy_fields(cls, values): if isinstance(values, dict): if 'entity_type' in values and 'target_type' not in values: values['target_type'] = values.pop('entity_type') + # Egeria's ReportSpec property, and the canonical JSON key (ws #794, RS-W2-2 PR-B). + # `target_type` stays this model's field name until the generator moves (§9.6). + if 'target_type_name' in values: + values = dict(values) + canonical = values.pop('target_type_name') + if values.get('target_type') not in (None, canonical): + raise ValueError(f"target_type_name '{canonical}' and target_type '{values['target_type']}' " + "disagree: they are one value (target_type is its old key)") + values['target_type'] = canonical return values + @property + def target_type_name(self) -> Optional[str]: + """Egeria's name for `target_type` (the ReportSpec property targetTypeName).""" + return self.target_type + @field_validator('formats', mode='before') @classmethod def validate_formats(cls, v): diff --git a/pyegeria/view/base_report_formats.py b/pyegeria/view/base_report_formats.py index b8cf41ca..2e5778f7 100644 --- a/pyegeria/view/base_report_formats.py +++ b/pyegeria/view/base_report_formats.py @@ -4045,10 +4045,23 @@ def refresh_report_specs() -> List[Tuple[str, str]]: except Exception as exc: # noqa: BLE001 -- one bad entry must not block the rest logger.warning(f"Skipping report spec source {source}: {exc}") skipped.append((source, str(exc))) + if _startup_resolution_done: # an explicit refresh after startup re-resolves and says so + _log_spec_resolution("refresh_report_specs", force=True) return skipped _config_report_specs_loaded = False +_startup_resolution_done = False + + +def _log_spec_resolution(where: str, *, force: bool = False) -> None: + """Resolve every spec at load (RS-W2-2 PR-B, note §2c) and log how many cannot run, with + each one's words. Never raises: resolution lists a spec as unrunnable, it never blocks loading.""" + try: + from pyegeria.view.spec_resolver import log_resolution_summary + log_resolution_summary(force=force, where=where) + except Exception as exc: # noqa: BLE001 -- e.g. a partial import during module initialisation + logger.debug(f"Report spec resolution at {where} skipped: {exc}") def get_report_registry() -> FormatSetDict: @@ -4057,13 +4070,15 @@ def get_report_registry() -> FormatSetDict: (`refresh_report_specs()`) is auto-loaded on first call -- no consumer needs to remember to call it explicitly. """ - global _config_report_specs_loaded + global _config_report_specs_loaded, _startup_resolution_done if not _config_report_specs_loaded: _config_report_specs_loaded = True try: refresh_report_specs() except Exception as exc: # noqa: BLE001 -- a misconfigured entry shouldn't break the registry logger.warning(f"refresh_report_specs() failed during auto-load: {exc}") + _log_spec_resolution("startup") + _startup_resolution_done = True combined = FormatSetDict() _add_with_collision_check(combined, base_report_specs, source="BUILTINS") @@ -4280,8 +4295,15 @@ def clear_runtime_report_specs() -> None: _RUNTIME_REPORT_SOURCES.clear() -def list_report_specs() -> list[str]: - return list(get_report_registry().keys()) +def list_report_specs(*, with_status: bool = False) -> list: + """The registered report spec labels. With `with_status`, one dict per spec instead: + `{"name", "runnable", "unrunnable_reason"}` from load-time resolution (RS-W2-2 PR-B). An + unrunnable spec is still listed, with the words saying why it cannot run.""" + if not with_status: + return list(get_report_registry().keys()) + from pyegeria.view.spec_resolver import resolve_all + return [{"name": label, "runnable": result.runnable, "unrunnable_reason": result.unrunnable_reason} + for label, result in resolve_all().items()] # ── Built-in report specs for viewing question spec entities ────────────────── @@ -4590,6 +4612,7 @@ def load_egeria_report_specs( _EGERIA_SPECS_LOADED_AT = datetime.now(timezone.utc) logger.debug(f"load_egeria_report_specs: registry updated, cache set at {_EGERIA_SPECS_LOADED_AT.isoformat()}") + _log_spec_resolution("Egeria archive load", force=True) return True diff --git a/pyegeria/view/spec_resolver.py b/pyegeria/view/spec_resolver.py new file mode 100644 index 00000000..c32a6f2e --- /dev/null +++ b/pyegeria/view/spec_resolver.py @@ -0,0 +1,439 @@ +""" +SPDX-License-Identifier: Apache-2.0 +Copyright Contributors to the ODPi Egeria project. + +Load-time resolution of report specs (RS-W2-2 PR-B; report-spec pipeline model §2c). + +A spec is resolved when it is loaded, not when it is run: `resolve(spec)` runs the §2c checks +and returns a `ResolvedSpec`, or an `Unrunnable` carrying the failing checks' words. A spec that +fails is still listed (`list_report_specs(with_status=True)`), so a person sees why it cannot +run instead of a blank result. No spec is deleted, and resolution never stops a run: the +executor is unchanged here. + +Today's specs are single-stage: one find method (`action.function`) or one analytic routine +(`action.analytic_function`). For them the §2c checks are: + + 1. names: the find method resolves on the client class the executor would use; the analytic + routine is a registered name with a python binding that imports; binder and function-valued + parameters name what they must. + 2. parameters: every parameter the executor always sends (required params, spec_params, + output_format, report_spec) is one the find method accepts; every analytic_spec_params key + is a parameter the routine declares. (No spec declares constants or `$param`s yet.) + 3. stage shapes: a single stage has no previous stage to check against (recorded, not run). + 4. formatter vs out: every output format is one the executor knows; a chart format needs an + analytic routine; every column an analytic spec lists is a field the routine declares in + `out`; every column's detail_spec names a registered spec. A find method declares no `out`, + so a find spec's columns are not checked (recorded). + 5. role inputs: listed from the routine's declaration; which environment supplies each is + W2-3's (declared, not checked here). + 6. admission and tile_safe: recorded from the routine's declaration. + 7. seeding: the find method a spec may be seeded by is recorded. + +Results are cached per spec and invalidated when the spec object or the analytic registry +changes (both are compared by identity, so re-registering either is a change). +""" + +from __future__ import annotations + +import inspect +import warnings +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Tuple, Union + +from loguru import logger + +__all__ = [ + "ResolvedSpec", + "Unrunnable", + "resolve", + "resolve_all", + "unrunnable_specs", + "spec_status", + "log_resolution_summary", + "OUTPUT_FORMATS", + "CHART_FORMATS", + "FUNCTION_VALUED_PARAMS", +] + +#: Output formats the executor renders from a Format row ("ALL" means every one of them). +OUTPUT_FORMATS = ("ALL", "DICT", "JSON", "TABLE", "LIST", "REPORT", "REPORT-GRAPH", "MD", "FORM", "HTML", + "MERMAID", "GRAPH") +#: Chart formats: dispatched before any Format row, and only for an analytic routine. +CHART_FORMATS = ("SERIES", "BAR", "PIE") +#: Routine parameters whose value names another routine (by registered name only, ISSUE-173). +FUNCTION_VALUED_PARAMS = ("metric_path",) + +#: What the executor always passes a find method, besides the spec's own parameters. +_ALWAYS_SENT = ("output_format", "report_spec") + + +@dataclass +class ResolvedSpec: + """A spec that passed every check that could run. `notes` say which checks did not run and why; + `warnings` are defects that bite only when a caller uses an optional parameter.""" + name: str + kind: str # "find" | "analytic" | "find+analytic" + target_type_name: Optional[str] = None + stages: List[Dict[str, str]] = field(default_factory=list) + params: List[str] = field(default_factory=list) + output_formats: List[str] = field(default_factory=list) + role_inputs: List[str] = field(default_factory=list) + tile_safe: Optional[Dict[str, Any]] = None + admission: Dict[str, Any] = field(default_factory=dict) + seeds: List[str] = field(default_factory=list) + notes: List[str] = field(default_factory=list) + warnings: List[str] = field(default_factory=list) + + runnable = True + unrunnable_reason = "" + + +@dataclass +class Unrunnable: + """A spec that failed a check. `words` holds one sentence per failure, each naming its check; + `checks` the failing check numbers. `kind` is "format_only" for a spec with no action (it + formats elements another call fetched).""" + name: str + kind: str + words: List[str] + checks: List[int] + target_type_name: Optional[str] = None + notes: List[str] = field(default_factory=list) + warnings: List[str] = field(default_factory=list) + + runnable = False + + @property + def unrunnable_reason(self) -> str: + return " ".join(self.words) + + +Resolution = Union[ResolvedSpec, Unrunnable] + + +# ── Client classes, as the executor resolves them ── + +_EGERIA_TECH_CLASSES: Optional[List[type]] = None + + +def _egeria_tech_classes() -> List[type]: + """EgeriaTech plus every subclient class it proxies to (its `__getattr__` looks there).""" + global _EGERIA_TECH_CLASSES + if _EGERIA_TECH_CLASSES is None: + from pyegeria.egeria_tech_client import EgeriaTech + # Construction sets attributes only; nothing connects. + probe = EgeriaTech("spec-resolver", "https://localhost:9443", user_id="spec-resolver", user_pwd="") + _EGERIA_TECH_CLASSES = [EgeriaTech] + list(dict.fromkeys(probe._subclient_map.values())) + return _EGERIA_TECH_CLASSES + + +def _find_method(func_decl: str) -> Tuple[Optional[Any], str]: + """The method the executor would call for `Class.method`, or (None, the words why not).""" + from pyegeria.view.format_set_executor import _resolve_client_and_method + + parts = func_decl.split(".") + if len(parts) != 2 or not all(parts): + return None, (f"its find method '{func_decl}' is not Class.method, which the executor needs " + f"(exactly one dot)") + class_name, method_name = parts + if not method_name.startswith("_async_"): + method_name = "_async_" + method_name + try: + client_class, method_name = _resolve_client_and_method(f"{class_name}.{method_name}") + except Exception as exc: # noqa: BLE001 -- an unimportable mapped class is the finding + return None, f"its find method's client class '{class_name}' cannot be imported: {exc}" + from pyegeria.egeria_tech_client import EgeriaTech + classes = _egeria_tech_classes() if client_class is EgeriaTech else [client_class] + for cls in classes: + method = getattr(cls, method_name, None) + if callable(method): + return method, "" + where = ("EgeriaTech or any client it proxies to" if client_class is EgeriaTech + else f"client class {client_class.__name__}") + return None, f"its find method '{func_decl}' does not exist: {where} has no '{method_name}'" + + +def _accepts(method: Any) -> Tuple[set, bool]: + try: + sig = inspect.signature(method) + except (TypeError, ValueError): + return set(), True # unknowable: treat as accepting anything + names = {n for n in sig.parameters if n != "self"} + var_kw = any(p.kind == inspect.Parameter.VAR_KEYWORD for p in sig.parameters.values()) + return names, var_kw + + +# ── The checks ── + +def _as_dict(spec: Any) -> Dict[str, Any]: + if hasattr(spec, "model_dump"): + return spec.model_dump() + return dict(spec or {}) + + +def _target_type_name(data: Dict[str, Any]) -> Optional[str]: + """Egeria's property name (`target_type_name`), with `target_type` read as its alias.""" + return data.get("target_type_name") or data.get("target_type") + + +def _check_find(name: str, action: Dict[str, Any], fail, notes, warns, out: Dict[str, Any]) -> None: + func_decl = action.get("function") + if not isinstance(func_decl, str) or "." not in func_decl: + fail(1, f"its find method {func_decl!r} is not Class.method") + return + method, why = _find_method(func_decl) + if method is None: + fail(1, why) + return + out["stages"].append({"kind": "producer", "name": func_decl, "binding": "python (client method)"}) + out["seeds"].append(func_decl) + accepted, var_kw = _accepts(method) + if var_kw: + notes.append(f"check 2: {func_decl} takes **kwargs, so its parameter names are not checked") + return + required = list(action.get("required_params") or []) + spec_params = list((action.get("spec_params") or {}).keys()) + always = [p for p in _ALWAYS_SENT if p not in accepted] + if always: + fail(2, f"the executor always passes {', '.join(always)} and {func_decl} does not accept " + f"{'it' if len(always) == 1 else 'them'}") + for label, names in (("required parameter", required), ("spec parameter", spec_params)): + bad = [p for p in names if p not in accepted] + if bad: + fail(2, f"its {label}{'s' if len(bad) > 1 else ''} {', '.join(bad)} " + f"{'is' if len(bad) == 1 else 'are'} sent on every run and {func_decl} does not accept " + f"{'it' if len(bad) == 1 else 'them'}") + optional = [p for p in (action.get("optional_params") or []) if p not in accepted] + if optional: + warns.append(f"check 2: optional parameter{'s' if len(optional) > 1 else ''} {', '.join(optional)} " + f"{'is' if len(optional) == 1 else 'are'} not accepted by {func_decl}: a caller who " + f"passes {'it' if len(optional) == 1 else 'one'} gets a TypeError") + out["params"].extend(p for p in required + list(action.get("optional_params") or []) if p not in out["params"]) + + +def _check_analytic(name: str, action: Dict[str, Any], fail, notes, warns, out: Dict[str, Any]) -> Any: + """Checks 1, 2, 5, 6 for the analytic stage. Returns the routine's declaration, or None.""" + from pyegeria.view.analytic_registry import get_analytic_registry + from pyegeria.view.format_set_executor import (AnalyticFunctionRefused, _import_callable, + _resolve_analytic_function, resolve_report_spec_function) + + func_decl = action["analytic_function"] + registry = get_analytic_registry() + entry = registry.get(func_decl) + if entry is None: + try: + resolve_report_spec_function(func_decl) + except AnalyticFunctionRefused as exc: + fail(1, str(exc)) + return None + target = entry.python_target() + if not target: + kinds = ", ".join(f"{b.kind} '{b.target}'" for b in entry.bindings) or "none" + fail(1, f"routine '{func_decl}' has no binding this executor can honour (its bindings: {kinds}; " + f"pyegeria runs python bindings only)") + return None + try: + _import_callable(target) + except Exception as exc: # noqa: BLE001 -- an unimportable binding is the finding + fail(1, f"routine '{func_decl}' binds python target '{target}', which cannot be imported: {exc}") + return None + out["stages"].append({"kind": entry.kind, "name": func_decl, "binding": f"python {target}"}) + if entry.action is not None: + notes.append(f"check 3: routine '{func_decl}' runs fetch {entry.action.fetch} then analytic " + f"{entry.action.analytic}; neither step declares its shapes, so the hand-off is not checked") + + for param, binder in (action.get("analytic_binders") or {}).items(): + try: + _resolve_analytic_function(binder) + except AnalyticFunctionRefused as exc: + fail(1, f"its binder for {param} is refused: {exc}") + except Exception as exc: # noqa: BLE001 + fail(1, f"its binder for {param} ('{binder}') cannot be imported: {exc}") + + declared = [p.name for p in entry.params] + out["params"].extend(p for p in declared if p not in out["params"]) + spec_params = action.get("analytic_spec_params") or {} + binders = set((action.get("analytic_binders") or {}).keys()) + unknown = [k for k in spec_params if k not in declared and k not in binders] + if unknown and not declared: + # A routine that declares no parameters at all is silent about them, not declaring none: + # the spec runs today, so this is the declaration's gap, said as a warning. + warns.append(f"check 2: routine '{func_decl}' declares no parameters, so analytic_spec_params " + f"{', '.join(unknown)} cannot be checked; the routine's declaration should list them") + elif unknown: + fail(2, f"its analytic_spec_params {', '.join(unknown)} {'is' if len(unknown) == 1 else 'are'} not " + f"declared by routine '{func_decl}' (declared: {', '.join(declared)})") + for param in FUNCTION_VALUED_PARAMS: + value = spec_params.get(param) + if param in declared and value is not None and value not in registry: + fail(1, f"its {param} '{value}' is not a registered routine name (a report spec names routines by " + f"registered name only)") + + try: + routine = entry.routine() + except ValueError as exc: + notes.append(f"checks 5-6: routine '{func_decl}' is not a valid routine declaration, so its inputs, " + f"admission and tile_safe are not recorded: {exc}") + return entry.declaration + out["role_inputs"] = list(routine.inputs) + out["tile_safe"] = routine.tile_safe.model_dump() if routine.tile_safe else None + out["admission"] = routine.admission.model_dump() + notes.append("check 5: " + (f"role inputs {', '.join(routine.inputs)}" if routine.inputs else + f"routine '{func_decl}' declares no role inputs") + + "; which environment supplies each is not checked here (W2-3)") + if routine.tile_safe is None: + notes.append(f"check 6: routine '{func_decl}' declares no tile_safe") + return routine.declaration + + +def _check_formats(data: Dict[str, Any], has_analytic: bool, decl: Any, fail, notes, out: Dict[str, Any], + registry_names: set) -> None: + seen: List[str] = [] + out_names = None + if decl is not None: + fields = decl.out_fields() + out_names = {f.name for f in fields} if fields else None + analytic_name = (data.get("action") or {}).get("analytic_function") + for fmt in data.get("formats") or []: + types = [str(t).upper() for t in (fmt.get("types") or [])] + for t in types: + if t not in seen: + seen.append(t) + unknown = [t for t in types if t not in OUTPUT_FORMATS + CHART_FORMATS] + if unknown: + fail(4, f"its output format{'s' if len(unknown) > 1 else ''} {', '.join(unknown)} " + f"{'is' if len(unknown) == 1 else 'are'} not one the executor renders") + charts = [t for t in types if t in CHART_FORMATS] + if charts and not has_analytic: + fail(4, f"it offers {', '.join(charts)}, a chart, and has no analytic routine to chart") + for col in fmt.get("attributes") or []: + key = str(col.get("key") or "") + detail = col.get("detail_spec") + if detail and detail not in registry_names: + fail(4, f"its column '{col.get('name') or key}' drills into detail_spec '{detail}', which is not " + f"a registered report spec") + if has_analytic and out_names is not None and key and key.split(".", 1)[0] not in out_names: + fail(4, f"its column '{col.get('name') or key}' reads field '{key}', which routine " + f"'{analytic_name}' does not declare in out (declared: {', '.join(sorted(out_names))})") + if has_analytic and decl is not None and out_names is None: + notes.append(f"check 4: routine '{analytic_name}' declares no out, so its columns are not checked") + if (data.get("action") or {}).get("function"): + notes.append("check 4: a find method declares no out, so the columns (element properties) are not checked") + out["output_formats"] = seen + + +def resolve(spec: Any, name: Optional[str] = None, *, registry_names: Optional[set] = None) -> Resolution: + """Resolve one report spec (a FormatSet or its dict) through the §2c checks.""" + data = _as_dict(spec) + name = name or data.get("heading") or "?" + if registry_names is None: + from pyegeria.view.base_report_formats import get_report_registry + registry_names = set(get_report_registry().keys()) + target = _target_type_name(data) + words: List[str] = [] + checks: List[int] = [] + notes: List[str] = [] + warns: List[str] = [] + + def fail(check: int, sentence: str) -> None: + words.append(f"Spec '{name}' cannot run (check {check}): {sentence}.") + if check not in checks: + checks.append(check) + + action = data.get("action") + if not action: + return Unrunnable(name=name, kind="format_only", checks=[1], target_type_name=target, words=[ + f"Spec '{name}' cannot run on its own: it has no action, so it formats elements another call " + f"fetched (it is a detail or a command's output format)."]) + + out: Dict[str, Any] = {"stages": [], "params": [], "seeds": [], "role_inputs": [], "tile_safe": None, + "admission": {}} + has_find = bool(action.get("function")) + has_analytic = bool(action.get("analytic_function")) + kind = "+".join(k for k, on in (("find", has_find), ("analytic", has_analytic)) if on) or "none" + if has_find: + _check_find(name, action, fail, notes, warns, out) + decl = _check_analytic(name, action, fail, notes, warns, out) if has_analytic else None + if not has_find and not has_analytic: + fail(1, "its action names neither a find method nor an analytic routine") + notes.append("check 3: a single stage, so there is no previous stage's shape to check against") + _check_formats(data, has_analytic, decl, fail, notes, out, registry_names) + + if words: + return Unrunnable(name=name, kind=kind, words=words, checks=sorted(checks), target_type_name=target, + notes=notes, warnings=warns) + return ResolvedSpec(name=name, kind=kind, target_type_name=target, notes=notes, warnings=warns, **out) + + +# ── All specs, cached ── + +_CACHE: Dict[str, Tuple[int, Tuple, Resolution]] = {} +_SUMMARY_LOGGED = False + + +def _analytic_fingerprint() -> Tuple: + from pyegeria.view.analytic_registry import get_analytic_registry + return tuple(sorted((n, id(s)) for n, s in get_analytic_registry().items())) + + +def resolve_all(*, refresh: bool = False) -> Dict[str, Resolution]: + """Every registered report spec, resolved. Cached per spec: a spec is re-resolved when its + FormatSet object, the set of registered spec names, or the analytic registry changes.""" + from pyegeria.view.base_report_formats import get_report_registry + registry = get_report_registry() + names = set(registry.keys()) + fingerprint = (_analytic_fingerprint(), tuple(sorted(names))) + results: Dict[str, Resolution] = {} + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + for label, spec in registry.items(): + cached = _CACHE.get(label) + if not refresh and cached and cached[0] == id(spec) and cached[1] == fingerprint: + results[label] = cached[2] + continue + try: + result = resolve(spec, label, registry_names=names) + except Exception as exc: # noqa: BLE001 -- the resolver must not take the listing down + result = Unrunnable(name=label, kind="unknown", checks=[], words=[ + f"Spec '{label}' could not be resolved: {type(exc).__name__}: {exc}."]) + _CACHE[label] = (id(spec), fingerprint, result) + results[label] = result + for gone in set(_CACHE) - names: + _CACHE.pop(gone, None) + return results + + +def unrunnable_specs(*, include_format_only: bool = False) -> Dict[str, Unrunnable]: + """The specs that cannot run, with their words. Specs with no action (format-only) are left + out unless asked for: they are not meant to run on their own.""" + return {n: r for n, r in resolve_all().items() + if isinstance(r, Unrunnable) and (include_format_only or r.kind != "format_only")} + + +def spec_status(label: str) -> Dict[str, Any]: + """{name, runnable, unrunnable_reason} for one registered spec.""" + result = resolve_all().get(label) + if result is None: + return {"name": label, "runnable": False, "unrunnable_reason": f"Spec '{label}' is not registered."} + return {"name": label, "runnable": result.runnable, "unrunnable_reason": result.unrunnable_reason} + + +def log_resolution_summary(*, force: bool = False, where: str = "load") -> Dict[str, Unrunnable]: + """Resolve every spec and log the count of unrunnable specs and each one's words. Logged once + per process unless `force` (refresh_specs and archive loads force it).""" + global _SUMMARY_LOGGED + bad = unrunnable_specs() + if _SUMMARY_LOGGED and not force: + return bad + _SUMMARY_LOGGED = True + total = len(_CACHE) + format_only = sum(1 for r in resolve_all().values() if isinstance(r, Unrunnable) and r.kind == "format_only") + if bad: + logger.warning(f"Report specs resolved at {where}: {len(bad)} of {total} cannot run " + f"({format_only} more are format-only and not run on their own).") + for result in bad.values(): + logger.warning(result.unrunnable_reason) + else: + logger.info(f"Report specs resolved at {where}: all {total - format_only} runnable specs resolve " + f"({format_only} format-only).") + return bad diff --git a/tests/micro-tests/test_rs_w2_2_spec_resolver.py b/tests/micro-tests/test_rs_w2_2_spec_resolver.py new file mode 100644 index 00000000..a261131d --- /dev/null +++ b/tests/micro-tests/test_rs_w2_2_spec_resolver.py @@ -0,0 +1,276 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright Contributors to the ODPi Egeria project. +"""RS-W2-2 PR-B: every report spec is resolved at load through the §2c checks; a spec that fails +is listed as unrunnable with the failing check's words, never deleted and never run blank.""" + +import pytest +from loguru import logger + +import pyegeria.view.base_report_formats as brf +from pyegeria.view import spec_resolver as sr +from pyegeria.view._output_format_models import ActionParameter, Column, Format, FormatSet +from pyegeria.view.analytic_registry import (AnalyticFunctionSpec, Binding, + clear_runtime_analytic_functions, register_analytic_functions) + +FIND = "ValidMetadataManager.get_valid_metadata_values" # takes no **kwargs, so its names are checked + + +def _spec(action=None, types=("DICT",), columns=(), **kw) -> FormatSet: + return FormatSet(heading="Probe", description="probe", target_type="Referenceable", action=action, + formats=[Format(types=list(types), attributes=list(columns))], **kw) + + +def _find(**kw) -> ActionParameter: + return ActionParameter(function=kw.pop("function", FIND), **kw) + + +def _analytic(name="count_elements", **kw) -> ActionParameter: + return ActionParameter(analytic_function=name, **kw) + + +def _words(result) -> str: + assert isinstance(result, sr.Unrunnable), f"expected unrunnable, got {result}" + return result.unrunnable_reason + + +@pytest.fixture(autouse=True) +def _clean(): + yield + clear_runtime_analytic_functions() + brf.clear_runtime_report_specs() + + +# ── The census over every registered spec (base_report_formats + analytic demo specs) ── + +#: The unrunnable specs the first resolution found, each a real defect (see the PR body). +KNOWN_UNRUNNABLE = { + "Asset-Graph": "its find method 'ServerClient.get_asset_graph' does not exist", + "Data-Classes": "its find method 'DataDesigner.find_data_classes' does not exist", +} + + +def test_every_registered_spec_resolves_and_the_unrunnable_ones_are_the_known_defects(): + results = sr.resolve_all(refresh=True) + assert set(results) == set(brf.get_report_registry()) + assert not [n for n, r in results.items() if r.kind == "unknown"], "the resolver itself failed on a spec" + bad = sr.unrunnable_specs() + assert set(bad) == set(KNOWN_UNRUNNABLE) + for name, words in KNOWN_UNRUNNABLE.items(): + assert words in bad[name].unrunnable_reason and bad[name].checks == [1] + + +def test_format_only_specs_are_the_specs_with_no_action(): + registry = brf.get_report_registry() + format_only = {n for n, r in sr.resolve_all().items() if isinstance(r, sr.Unrunnable) and r.kind == "format_only"} + assert format_only == {n for n, s in registry.items() if not s.action} + assert "Referenceable" in format_only + assert "it has no action" in sr.spec_status("Referenceable")["unrunnable_reason"] + + +def test_every_analytic_demo_spec_is_runnable(): + from pyegeria.view.analytic_demo_specs import get_analytic_demo_specs + results = sr.resolve_all() + for label in get_analytic_demo_specs(): + assert results[label].runnable, results[label].unrunnable_reason + + +# ── Check 1: names ── + +@pytest.mark.parametrize("action, words", [ + (_find(function="DataDesigner.find_nothing_at_all"), "client class DataDesigner has no '_async_find_nothing_at_all'"), + (_find(function="NoSuchClass.find_nothing_at_all"), "EgeriaTech or any client it proxies to has no"), + (_find(function="a.b.c"), "is not Class.method"), + (_analytic("not_registered"), "'not_registered', which is not a registered name"), + (_analytic("pyegeria.view.overview_metrics.count_elements"), "use its registered name 'count_elements'"), + (_analytic("metric_trend", analytic_spec_params={"metric_path": "nope"}), + "its metric_path 'nope' is not a registered routine name"), + (_analytic("count_elements", analytic_binders={"type_name": "os.getcwd"}), "its binder for type_name is refused"), +]) +def test_check_1_names(action, words): + result = sr.resolve(_spec(action), "Probe") + assert words in _words(result) and 1 in result.checks + + +def test_check_1_a_routine_with_no_python_binding(): + register_analytic_functions({"saved_census": AnalyticFunctionSpec( + name="saved_census", bindings=[Binding(kind="egeria_query", target="Census query")])}) + assert ("routine 'saved_census' has no binding this executor can honour (its bindings: egeria_query " + "'Census query'; pyegeria runs python bindings only)") in _words(sr.resolve(_spec(_analytic("saved_census")))) + + +def test_check_1_an_unimportable_python_binding(): + register_analytic_functions({"gone": AnalyticFunctionSpec(name="gone", function="pyegeria.no_such_module.f")}) + assert "binds python target 'pyegeria.no_such_module.f', which cannot be imported" in _words( + sr.resolve(_spec(_analytic("gone")))) + + +# ── Check 2: parameters ── + +def test_check_2_a_spec_param_the_method_does_not_accept(): + words = _words(sr.resolve(_spec(_find(required_params=["property_name"], spec_params={"colour": "red"})))) + assert f"its spec parameter colour is sent on every run and {FIND} does not accept it" in words + + +def test_check_2_a_required_param_the_method_does_not_accept(): + words = _words(sr.resolve(_spec(_find(required_params=["property_name", "search_string"])))) + assert "its required parameter search_string is sent on every run" in words + + +def test_check_2_output_format_and_report_spec_are_always_sent(monkeypatch): + async def _async_find_things(self, search_string: str): + return [] + monkeypatch.setattr(sr, "_find_method", lambda decl: (_async_find_things, "")) + words = _words(sr.resolve(_spec(_find(function="Things.find_things", required_params=["search_string"])))) + assert "the executor always passes output_format, report_spec and Things.find_things does not accept them" in words + + +def test_check_2_an_optional_param_the_method_does_not_accept_is_a_warning(): + result = sr.resolve(_spec(_find(required_params=["property_name"], optional_params=["as_of_time"]))) + assert result.runnable + assert any("optional parameter as_of_time is not accepted" in w for w in result.warnings) + + +def test_check_2_analytic_param_not_declared_is_refused_with_the_declared_list(): + words = _words(sr.resolve(_spec(_analytic(analytic_spec_params={"typename": "Asset"})))) + assert ("its analytic_spec_params typename is not declared by routine 'count_elements' " + "(declared: type_name, as_of, strict, classifications, population)") in words + + +def test_check_2_a_routine_that_declares_no_params_warns_instead(): + register_analytic_functions({"silent": AnalyticFunctionSpec( + name="silent", function="pyegeria.view.overview_metrics.count_elements")}) + result = sr.resolve(_spec(_analytic("silent", analytic_spec_params={"type_name": "Asset"}))) + assert result.runnable + assert any("routine 'silent' declares no parameters" in w for w in result.warnings) + + +# ── Check 4: formatter vs out ── + +@pytest.mark.parametrize("spec, words", [ + (_spec(_find(required_params=["property_name"]), types=("DICT", "SPREADSHEET")), + "its output format SPREADSHEET is not one the executor renders"), + (_spec(_find(required_params=["property_name"]), types=("BAR",)), + "it offers BAR, a chart, and has no analytic routine to chart"), + (_spec(_find(required_params=["property_name"]), + columns=[Column(name="Terms", key="terms", detail_spec="No-Such-Spec")]), + "drills into detail_spec 'No-Such-Spec', which is not a registered report spec"), + (_spec(_analytic(), columns=[Column(name="Count", key="count")]), + "its column 'Count' reads field 'count', which routine 'count_elements' does not declare in out " + "(declared: value)"), +]) +def test_check_4_formatter_against_out(spec, words): + result = sr.resolve(spec, "Probe") + assert words in _words(result) and 4 in result.checks + + +def test_check_4_a_declared_column_and_a_chart_pass(): + result = sr.resolve(_spec(_analytic("counts_by_type"), types=("DICT", "BAR"), + columns=[Column(name="Count", key="count"), Column(name="Type", key="type")])) + assert result.runnable and result.output_formats == ["DICT", "BAR"] + + +# ── What a resolved spec records (checks 3, 5, 6, 7) ── + +def test_a_find_spec_records_its_stage_and_seed(): + result = sr.resolve(_spec(_find(required_params=["property_name"]))) + assert result.runnable and result.kind == "find" + assert result.seeds == [FIND] + assert result.stages == [{"kind": "producer", "name": FIND, "binding": "python (client method)"}] + assert any(n.startswith("check 3: a single stage") for n in result.notes) + assert any("a find method declares no out" in n for n in result.notes) + + +def test_an_analytic_spec_records_role_inputs_admission_and_tile_safe(): + result = sr.resolve(_spec(_analytic())) + assert result.runnable and result.kind == "analytic" and result.seeds == [] + assert result.stages == [{"kind": "routine", "name": "count_elements", + "binding": "python pyegeria.view.overview_metrics.count_elements"}] + assert result.admission == {"resource_types": [], "test_only": False} and result.tile_safe is None + assert any(n.startswith("check 5:") and "(W2-3)" in n for n in result.notes) + assert any("declares no tile_safe" in n for n in result.notes) + assert "type_name" in result.params + + +def test_a_composite_routine_says_its_hand_off_is_not_checked(): + result = sr.resolve(_spec(_analytic("sum_type_counts"))) + assert any("runs fetch pyegeria.view.overview_metrics.counts_by_type then analytic" in n for n in result.notes) + + +def test_target_type_name_is_read_with_target_type_as_its_alias(): + assert sr.resolve(_spec(_analytic())).target_type_name == "Referenceable" + assert sr.resolve({"heading": "h", "description": "d", "formats": [], "target_type_name": "Asset", + "action": {"analytic_function": "count_elements"}}).target_type_name == "Asset" + + +def test_formatset_accepts_target_type_name(): + fs = FormatSet(heading="h", description="d", formats=[], target_type_name="Asset") + assert fs.target_type == fs.target_type_name == "Asset" + with pytest.raises(ValueError, match="disagree"): + FormatSet(heading="h", description="d", formats=[], target_type_name="Asset", target_type="Glossary") + + +# ── Listing, caching, load hooks ── + +def test_list_report_specs_is_unchanged_by_default_and_gains_status_on_request(): + names = brf.list_report_specs() + assert names and all(isinstance(n, str) for n in names) + status = {d["name"]: d for d in brf.list_report_specs(with_status=True)} + assert set(status) == set(names) + assert status["Data-Classes"]["runnable"] is False + assert "check 1" in status["Data-Classes"]["unrunnable_reason"] + assert status["Analytic Demo - Element Count by Type"] == { + "name": "Analytic Demo - Element Count by Type", "runnable": True, "unrunnable_reason": ""} + + +def test_results_are_cached_and_a_registry_change_invalidates(): + first = sr.resolve_all() + assert sr.resolve_all()["Data-Classes"] is first["Data-Classes"] + brf.register_report_specs({"Probe-Spec": _spec(_analytic("not_yet_registered"))}, source="test") + assert not sr.resolve_all()["Probe-Spec"].runnable + register_analytic_functions({"not_yet_registered": AnalyticFunctionSpec( + name="not_yet_registered", function="pyegeria.view.overview_metrics.count_elements")}) + assert sr.resolve_all()["Probe-Spec"].runnable + brf.unregister_report_spec("Probe-Spec") + assert "Probe-Spec" not in sr.resolve_all() + + +def test_the_resolver_never_raises_on_a_spec(): + brf.register_report_specs({"Probe-Broken": _spec(_analytic())}, source="test") + real = sr.resolve + + def boom(spec, name=None, **kw): + if name == "Probe-Broken": + raise RuntimeError("bad spec") + return real(spec, name, **kw) + + sr.resolve = boom + try: + result = sr.resolve_all(refresh=True)["Probe-Broken"] + finally: + sr.resolve = real + assert result.kind == "unknown" and "could not be resolved: RuntimeError: bad spec" in result.unrunnable_reason + + +def _captured(fn) -> str: + lines = [] + logger.enable("pyegeria") + sink = logger.add(lambda m: lines.append(str(m)), level="INFO", format="{message}") + try: + fn() + finally: + logger.remove(sink) + logger.disable("pyegeria") + return "".join(lines) + + +def test_refresh_report_specs_logs_the_unrunnable_count_and_words(): + brf.get_report_registry() # startup has happened + text = _captured(brf.refresh_report_specs) + assert "Report specs resolved at refresh_report_specs: 2 of" in text + assert "Spec 'Data-Classes' cannot run (check 1)" in text + + +def test_the_summary_is_logged_once_unless_forced(): + sr.log_resolution_summary() + assert _captured(sr.log_resolution_summary) == "" + assert "cannot run" in _captured(lambda: sr.log_resolution_summary(force=True, where="test"))