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"))