diff --git a/pyegeria/view/analytic_registry.py b/pyegeria/view/analytic_registry.py index 8982552f..e2dfe80e 100644 --- a/pyegeria/view/analytic_registry.py +++ b/pyegeria/view/analytic_registry.py @@ -26,8 +26,8 @@ string a FormatSet's `extra_find` (compiled into `ActionParameter.analytic_function`) carries, and exactly what `format_set_executor._resolve_analytic_function` resolves at call time. `params` documents the function's real keyword -arguments (excluding its leading client argument(s) -- `format_set_executor` -binds those automatically, see `_bind_client_args`) for anything that wants +arguments (excluding its leading client argument(s) -- those are its declared +`inputs`, which the executor supplies by role, see `role_supply`) for anything that wants to render a param-entry form (Egeria Advisor) or a reference doc (Egeria Explorer), not for validation -- there is deliberately no fixed required/optional whitelist enforced at call time (see `ActionParameter`'s @@ -119,10 +119,10 @@ class AnalyticActionSpec(BaseModel): a relationship to enrich -- exactly what `ai_ready_assets` already does inline, just as an undeclared implementation detail rather than a named step). Both `fetch` and `analytic` are dotted import paths, resolved and - client-bound the same way a plain `AnalyticFunctionSpec.function` always - has been (see `format_set_executor._resolve_analytic_function`/ - `_bind_client_args`, and its `run_analytic_action` for the two-step - invocation itself). + client-bound the same way a plain `AnalyticFunctionSpec.function` is: + by the fetch routine's declared `inputs` (see + `format_set_executor.run_analytic_action` for the two-step invocation + itself, and `role_supply` for how inputs are supplied). Deliberately *not* concerned with which of the resulting attributes appear in a given output (TABLE/REPORT/KPI/...) -- that selection is @@ -450,6 +450,10 @@ def _measured_kinds_declare_measures(self) -> "AnalyticDeclaration": PARAM_CATEGORIES = ("content", "shape", "performance") #: The role words a routine's `inputs` use. `egeria_client` and `reader` take a qualifier. +#: The executor supplies each declared input from the environment's `role_supply.RoleSupply` and +#: binds it to the routine's leading positional parameters in declared order (RS-W2-3). An +#: `egeria_client:` names the class whose methods that argument calls; `EgeriaTech` only for +#: an argument that calls across several classes. Each input is declared once. INPUT_ROLES = ("egeria_client", "reader", "viewer", "subject") _INPUT_ROLE = re.compile(r"^(?:egeria_client:[A-Za-z_][A-Za-z0-9_]*|reader:[a-z_][a-z0-9_]*|viewer|subject)$") @@ -833,6 +837,7 @@ def _always_applies(what: str) -> str: "count_elements": AnalyticFunctionSpec( name="count_elements", function="pyegeria.view.overview_metrics.count_elements", + inputs=["egeria_client:MetadataExpert"], description="Count active elements of a given type, optionally filtered by classification. " "Cost: one native count, sub-second in the repository population (0.2-0.8 s " "measured); a viewer-population count reads every element and can take minutes " @@ -867,6 +872,7 @@ def _always_applies(what: str) -> str: "counts_by_type": AnalyticFunctionSpec( name="counts_by_type", function="pyegeria.view.overview_metrics.counts_by_type", + inputs=["egeria_client:MetadataExpert"], description="Per-type element counts for a set of (label, type_name) pairs -- a breakdown/" "composition metric, e.g. the assets-by-type chart. " "Cost: one native count per type, sub-second each in the repository population; a viewer-population count can take minutes for a large type.", @@ -897,6 +903,7 @@ def _always_applies(what: str) -> str: "governed_coverage": AnalyticFunctionSpec( name="governed_coverage", function="pyegeria.view.overview_metrics.governed_coverage", + inputs=["egeria_client:MetadataExpert"], description="Share of assets carrying at least one governance classification, plus a " "by-classification / top-zones breakdown. " "Cost: one capped find (one page of elements), seconds.", @@ -941,6 +948,7 @@ def _always_applies(what: str) -> str: "ownership_coverage": AnalyticFunctionSpec( name="ownership_coverage", function="pyegeria.view.overview_metrics.ownership_coverage", + inputs=["egeria_client:MetadataExpert"], description="Share of assets carrying an Ownership classification (a named owner " "responsible for management/governance decisions), plus a by-owner-type " "breakdown. Distinct from governed_coverage -- data mesh literature names " @@ -972,6 +980,7 @@ def _always_applies(what: str) -> str: "business_value_signals": AnalyticFunctionSpec( name="business_value_signals", function="pyegeria.view.overview_metrics.business_value_signals", + inputs=["egeria_client:MetadataExpert"], description="Real signals behind the four Overview dashboard Business Value tiles " "(Risk & Compliance, Productivity, Trust & Adoption, Cost Avoidance): " "Confidentiality-classified asset count, described-asset count, " @@ -1017,6 +1026,7 @@ def _always_applies(what: str) -> str: "certifications_summary": AnalyticFunctionSpec( name="certifications_summary", function="pyegeria.view.overview_metrics.certifications_summary", + inputs=["egeria_client:ClassificationExplorer"], description="Active/expiring Certification relationships, licenses, and open Exception counts. " "Cost: three capped relationship pages, seconds.", returns="dict (active, expiring90, soon, licenses, exceptions)", @@ -1051,6 +1061,7 @@ def _always_applies(what: str) -> str: "people_counts": AnalyticFunctionSpec( name="people_counts", function="pyegeria.view.overview_metrics.people_counts", + inputs=["egeria_client:MetadataExpert"], description="Counts of Person / Team / Organization / ITProfile / Community actor profiles. " "Cost: five native counts, sub-second each in the repository population.", returns="dict (persons, teams, organizations, itProfiles, communities)", @@ -1086,6 +1097,7 @@ def _always_applies(what: str) -> str: "feedback_summary": AnalyticFunctionSpec( name="feedback_summary", function="pyegeria.view.overview_metrics.feedback_summary", + inputs=["egeria_client:ClassificationExplorer"], description="Crowd-sourced feedback counts (ratings/comments/likes/tags/noteLogs) by type, " "via Collaboration OMAS relationships. " "Cost: five relationship pages (native counts when an expert client is passed), seconds.", @@ -1120,6 +1132,7 @@ def _always_applies(what: str) -> str: "usage_context_counts": AnalyticFunctionSpec( name="usage_context_counts", function="pyegeria.view.overview_metrics.usage_context_counts", + inputs=["egeria_client:MetadataExpert"], description="InformationSupplyChain and SolutionBlueprint counts -- the two structures that " "put assets in a business usage context (supply chain membership, blueprint " "realization). " @@ -1151,6 +1164,7 @@ def _always_applies(what: str) -> str: "semantic_grounding": AnalyticFunctionSpec( name="semantic_grounding", function="pyegeria.view.overview_metrics.semantic_grounding", + inputs=["egeria_client:MetadataExpert", "egeria_client:ClassificationExplorer"], description="SemanticAssignment relationship count and coverage percent -- the meaning layer " "that grounds AI. Takes two leading clients (mgr, ce) -- the executor supplies the " "same EgeriaTech instance for both. " @@ -1184,6 +1198,7 @@ def _always_applies(what: str) -> str: "context_readiness_funnel": AnalyticFunctionSpec( name="context_readiness_funnel", function="pyegeria.view.overview_metrics.context_readiness_funnel", + inputs=["egeria_client:MetadataExpert", "egeria_client:ClassificationExplorer"], description="Cataloged -> Documented -> Classified -> Lineage-traced -> AI-Ready funnel counts. " "Takes two leading clients (mgr, ce) -- the executor supplies the same EgeriaTech " "instance for both, matching semantic_grounding's convention. This function's own " @@ -1226,6 +1241,7 @@ def _always_applies(what: str) -> str: "ai_ready_assets": AnalyticFunctionSpec( name="ai_ready_assets", function="pyegeria.view.overview_metrics.ai_ready_assets", + inputs=["egeria_client:MetadataExpert", "egeria_client:ClassificationExplorer"], description="The true 'AI-Ready' composite: Asset elements that are governed AND documented " "AND lineage-traced simultaneously, not three independent counts intersected after " "the fact. First real implementation of the composite/derived analytic metric " @@ -1261,6 +1277,7 @@ def _always_applies(what: str) -> str: "drl_readiness_gates": AnalyticFunctionSpec( name="drl_readiness_gates", function="pyegeria.view.overview_metrics.drl_readiness_gates", + inputs=["egeria_client:MetadataExpert", "egeria_client:ClassificationExplorer"], description="A recency-narrowed view of ai_ready_assets plus a modality breakdown " "(tabular / text_media / other) of the Asset population -- the two Data " "Readiness Level pieces computable today. Not a DRL band distribution: " @@ -1312,6 +1329,7 @@ def _always_applies(what: str) -> str: "growth_series": AnalyticFunctionSpec( name="growth_series", function="pyegeria.view.overview_metrics.growth_series", + inputs=["egeria_client:MetadataExpert"], description="Snapshot-based growth series (assets/terms/governed/products by default) via " "asOfTime, one point per window step. The one shape SERIES output format " "(a Vega-Lite line chart) is proven against today. " @@ -1359,6 +1377,7 @@ def _always_applies(what: str) -> str: "term_definition_completeness": AnalyticFunctionSpec( name="term_definition_completeness", function="pyegeria.view.overview_metrics.term_definition_completeness", + inputs=["egeria_client:MetadataExpert"], description="Share of GlossaryTerms carrying a non-empty description -- a definitions-" "coverage metric, distinct from semantic_grounding (term<->asset linkage, not " "whether the term itself is actually defined). " @@ -1394,6 +1413,7 @@ def _always_applies(what: str) -> str: "active_contributors": AnalyticFunctionSpec( name="active_contributors", function="pyegeria.view.overview_metrics.active_contributors", + inputs=["egeria_client:ClassificationExplorer"], description="Distinct usernames behind at least one feedback relationship (ratings/comments/" "likes/tags/noteLogs) -- an engagement signal feedback_summary's raw relationship " "counts don't give you (one prolific commenter vs. many occasional ones). " @@ -1432,6 +1452,7 @@ def _always_applies(what: str) -> str: "sum_type_counts": AnalyticFunctionSpec( name="sum_type_counts", function="pyegeria.view.overview_metrics.sum_type_counts", + inputs=["egeria_client:MetadataExpert"], description="Sum of counts_by_type() across a caller-given (label, type_name) list -- " "e.g. the Overview dashboard's 'Cataloged Assets' total. First real registered " "user of the fetch+analytic action shape (see AnalyticActionSpec) -- fetch is " @@ -1474,6 +1495,7 @@ def _always_applies(what: str) -> str: "count_elements_by_property": AnalyticFunctionSpec( name="count_elements_by_property", function="pyegeria.view.overview_metrics.count_elements_by_property", + inputs=["egeria_client:MetadataExpert"], description="Count active elements of a given type whose named string property equals a " "given value -- e.g. DigitalProduct elements with deploymentStatus=ACTIVE. Same " "native-count-with-fallback seam as count_elements, one cheap native COUNT call " @@ -1509,6 +1531,7 @@ def _always_applies(what: str) -> str: "count_relationships": AnalyticFunctionSpec( name="count_relationships", function="pyegeria.view.overview_metrics.count_relationships", + inputs=["egeria_client:ClassificationExplorer"], description="Count active relationships of one type. By default one capped " "get_relationships page (the path the rest of the portal uses); a native count " "when a MetadataExpert is passed as `expert` (opt-in: the two paths can disagree " @@ -1539,6 +1562,7 @@ def _always_applies(what: str) -> str: "contextualised_coverage": AnalyticFunctionSpec( name="contextualised_coverage", function="pyegeria.view.overview_metrics.contextualised_coverage", + inputs=["egeria_client:MetadataExpert", "egeria_client:ClassificationExplorer"], description="Percent of Assets given business/solution-design context via an ImplementedBy " "relationship to a SolutionComponent. A proxy, not the literal 'participates in " "an ISC/blueprint' metric -- confirms some solution-design context exists, not " @@ -1575,6 +1599,7 @@ def _always_applies(what: str) -> str: "solution_component_realisation": AnalyticFunctionSpec( name="solution_component_realisation", function="pyegeria.view.overview_metrics.solution_component_realisation", + inputs=["egeria_client:MetadataExpert", "egeria_client:ClassificationExplorer"], description="SolutionComponent count and how many of them are realised (have at least one " "ImplementedBy implementation) -- the Overview's Solution Blueprints drill-down. A " "single-hop proxy: the implementation need not itself sit in the component's " @@ -1607,6 +1632,7 @@ def _always_applies(what: str) -> str: "karma_leaderboard": AnalyticFunctionSpec( name="karma_leaderboard", function="pyegeria.view.overview_metrics.karma_leaderboard", + inputs=["egeria_client:MetadataExpert"], description="Top-N people by karma -- one bounded ContributionRecord fetch, karmaPoints is a " "scalar property on the record itself (not derived from counting related things), " "filtered to the given anchor type(s) via each record's own Anchors classification. " @@ -1646,6 +1672,7 @@ def _always_applies(what: str) -> str: "engagement_series": AnalyticFunctionSpec( name="engagement_series", function="pyegeria.view.overview_metrics.engagement_series", + inputs=["egeria_client:ClassificationExplorer"], description="Weekly-bucketed feedback-event trend (comments/ratings/likes/tags/noteLogs), " "zero-filled across the trailing window -- reuses the same 5 relationship-type " "queries feedback_summary() already makes, keeping createTime instead of only " @@ -1695,6 +1722,7 @@ def _always_applies(what: str) -> str: "orphan_glossary_terms": AnalyticFunctionSpec( name="orphan_glossary_terms", function="pyegeria.view.overview_metrics.orphan_glossary_terms", + inputs=["egeria_client:MetadataExpert", "egeria_client:ClassificationExplorer"], description="Approved-but-unassigned glossary terms -- a term with no SemanticAssignment " "relationship to anything was authored but never put to use grounding the " "catalog. One bounded SemanticAssignment fetch, distinct GlossaryTerm-end GUID " @@ -1731,6 +1759,7 @@ def _always_applies(what: str) -> str: "stale_assets": AnalyticFunctionSpec( name="stale_assets", function="pyegeria.view.overview_metrics.stale_assets", + inputs=["egeria_client:MetadataExpert"], description="Assets with no update in the last N days (default 180) -- candidates for " "archival review. One bounded Asset element fetch, each element's own version " "metadata compared against the cutoff, no relationship traversal. " @@ -1765,6 +1794,7 @@ def _always_applies(what: str) -> str: "metric_trend": AnalyticFunctionSpec( name="metric_trend", function="pyegeria.view.overview_metrics.metric_trend", + inputs=["egeria_client:EgeriaTech", "egeria_client:ClassificationExplorer"], description="Turn any single-snapshot function in this module into a time series, by " "re-calling it once per asOfTime snapshot -- the same windowing growth_series " "uses, generalized to any function here instead of one hardcoded type_map. Takes " @@ -1806,6 +1836,7 @@ def _always_applies(what: str) -> str: "asset_composition": AnalyticFunctionSpec( name="asset_composition", function="pyegeria.view.overview_metrics.asset_composition", + inputs=["egeria_client:EgeriaTech"], description="The composition of a type (default Asset) by subtype: one row per type in its " "subtype tree to `depth`, each counted in viewer and/or repository scope, an " "`other` row per parent (elements typed as exactly the parent), and not_visible = " @@ -1870,6 +1901,7 @@ def _always_applies(what: str) -> str: "control_implementation_summary": AnalyticFunctionSpec( name="control_implementation_summary", function="pyegeria.view.overview_metrics.control_implementation_summary", + inputs=["egeria_client:SecurityOfficer"], description="Applicable controls in a statement of applicability, counted by implementation " "status (NOT_STARTED, PLANNED, PARTIALLY_IMPLEMENTED, IMPLEMENTED, OTHER). " "Cost: one read of one statement of applicability's control relationships, seconds.", diff --git a/pyegeria/view/format_set_executor.py b/pyegeria/view/format_set_executor.py index 4e60213d..ac5e9ea0 100644 --- a/pyegeria/view/format_set_executor.py +++ b/pyegeria/view/format_set_executor.py @@ -37,6 +37,7 @@ from pyegeria.view.analytic_registry import ABSENCE_STATES, AnalyticActionSpec, canonical_absence_state from pyegeria.view.analytic_registry import get_analytic_registry as get_analytic_registry_functions from pyegeria.egeria_tech_client import EgeriaTech +from pyegeria.view.role_supply import PlainSupply, RoleRefused, RoleSupply, supply_inputs _CLIENT_CLASS_MAP = { "CollectionManager": "pyegeria.omvs.collection_manager.CollectionManager", @@ -203,38 +204,24 @@ def resolve_report_spec_function(func_decl: str): f"from pyegeria 6.2.19).") -# Parameter names an analytic function uses for its leading client argument(s) -# -- pyegeria.view.overview_metrics's functions take one (`mgr` or `ce`) or, -# for semantic_grounding, two (`mgr`, `ce`) positional clients before their -# real parameters. EgeriaTech is a facade that proxies to every subclient via -# __getattr__ (see egeria_tech_client.py), so the same instance satisfies -# either role -- no need to construct a second client. -_ANALYTIC_CLIENT_PARAM_NAMES = {"mgr", "ce", "expert", "client"} - - -def _count_client_params(func) -> int: - """Number of leading parameters named like a client (mgr/ce/expert/client).""" - count = 0 - for name, param in inspect.signature(func).parameters.items(): - if name in _ANALYTIC_CLIENT_PARAM_NAMES and param.kind in ( - inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD, - ): - count += 1 - else: - break - return count - - -def _bind_client_args(func, client: Any) -> list: - """Return the positional args to pass before **kwargs: one copy of `client` - per leading parameter named like a client (mgr/ce/expert/client).""" - return [client] * _count_client_params(func) +def _plain_client(view_server: str, view_url: str, user: str, user_pass: str, + token: Optional[str] = None) -> Any: + """The caller's EgeriaTech, authenticated: with `token` when given (ISSUE-86), else a token + minted from `user`/`user_pass`. `PlainSupply` calls this at most once per run, and only when a + routine declares an egeria_client input (ISSUE-137: no client, no connection).""" + client = EgeriaTech(view_server, view_url, user_id=user, user_pwd=user_pass) + if token: + client.set_bearer_token(token) + else: + client.create_egeria_bearer_token() + return client def run_analytic_action( - action: AnalyticActionSpec, client: Any, *, + action: AnalyticActionSpec, client: Any = None, *, fetch_kwargs: Optional[Dict[str, Any]] = None, analytic_kwargs: Optional[Dict[str, Any]] = None, + supply: Optional[RoleSupply] = None, ) -> Any: """Run an `analytic_registry.AnalyticActionSpec`'s fetch step, then (if declared) its analytic step over the fetch's raw result -- the executor @@ -246,24 +233,25 @@ def run_analytic_action( below exactly as before -- this is a separate, additive entry point, not a replacement. - `client` is resolved once by the caller (same convention as - `_run_analytic_function`) and bound positionally to both steps' leading - client-shaped parameters -- `fetch` via the usual signature introspection - (`_bind_client_args`), `analytic` via the SAME resolved client args - (not re-introspected from `analytic`'s own signature, since its first - positional parameter is the fetch result, not a client) -- so `analytic` - can itself make a further client call (e.g. follow a relationship to - enrich) using the identical already-authenticated client `fetch` used. + The fetch step must be a registered routine (by name or by its python binding's path): its + declared `inputs` are supplied, in declared order, from `supply` -- or, when none is given, + from `PlainSupply(client)`, the caller's already-authenticated client (RS-W2-3: inputs by + role, never by parameter name). The analytic step gets the fetch's raw result followed by the + SAME supplied arguments, so it can itself make a further client call (e.g. follow a + relationship to enrich) with the identical client the fetch used. """ + from pyegeria.view.spec_resolver import declared_inputs # lazy: the resolver imports this module fetch_func = _resolve_analytic_function(action.fetch) - fetch_clients = _bind_client_args(fetch_func, client) - raw_result = fetch_func(*fetch_clients, **(fetch_kwargs or {})) + if supply is None: + supply = PlainSupply(client) + fetch_args = supply_inputs(action.fetch, declared_inputs(action.fetch, fetch_func), supply) + raw_result = fetch_func(*fetch_args, **(fetch_kwargs or {})) if not action.analytic: return raw_result analytic_func = _resolve_analytic_function(action.analytic) - return analytic_func(raw_result, *fetch_clients, **(analytic_kwargs or {})) + return analytic_func(raw_result, *fetch_args, **(analytic_kwargs or {})) def _apply_analytic_binders(func, binders: Dict[str, str], call_params: Dict[str, Any]) -> Dict[str, Any]: @@ -413,6 +401,7 @@ def _run_analytic_function( action: dict, *, params: Dict[str, Any], view_server: str, view_url: str, user: str, user_pass: str, token: Optional[str] = None, + supply: Optional[RoleSupply] = None, ) -> Any: """Resolve and call a report spec action's `analytic_function` (extra_find), returning the function's raw result -- no chart-wrapping, no output_format @@ -427,6 +416,12 @@ def _run_analytic_function( as that user instead of falling back to the `user`/`user_pass` service account. Backward compatible: `user`/`user_pass` keep working exactly as before when no token is given. + + RS-W2-3: the routine's leading arguments are its declared `inputs`, in declared order, each + asked of `supply` (default: `PlainSupply` over the caller's client built from the arguments + above). Every input is obtained before the routine runs, so a role the supply cannot provide + -- or a declaration that does not match its binding (check 5) -- raises `RoleRefused` with + its words before any stage. No parameter name is read to decide what to pass. """ func_decl = action.get("analytic_function") if not func_decl: @@ -446,17 +441,15 @@ def _run_analytic_function( call_params.update({k: v for k, v in params.items() if v not in (None, "")}) func = resolve_report_spec_function(func_decl) + from pyegeria.view.spec_resolver import declared_inputs # lazy: the resolver imports this module + inputs = declared_inputs(func_decl, func) # check 5: the same function the resolver runs at load call_params = _apply_analytic_binders(func, action.get("analytic_binders") or {}, call_params) - n_clients = _count_client_params(func) - if n_clients == 0: - # The routine takes no client, so don't connect to Egeria or mint a token (ISSUE-137). - return func(**call_params) - client = EgeriaTech(view_server, view_url, user_id=user, user_pwd=user_pass) - if token: - client.set_bearer_token(token) - else: - client.create_egeria_bearer_token() - return func(*([client] * n_clients), **call_params) + if supply is None: + supply = PlainSupply(client_factory=lambda: _plain_client(view_server, view_url, user, user_pass, token)) + # Every role is obtained here, before the routine runs; a routine that declares no + # egeria_client input never connects or mints a token (ISSUE-137). + args = supply_inputs(getattr(func_decl, "__name__", func_decl), inputs, supply) + return func(*args, **call_params) def _resolve_action_target_client(egeria_client: EgeriaTech, client_class: type) -> Any: @@ -870,6 +863,7 @@ def _exec_analytic_chart( user: str, user_pass: str, token: Optional[str] = None, + supply: Optional[RoleSupply] = None, ) -> Dict[str, Any]: """Run a report spec's `analytic_function` (extra_find) and wrap the result as a Vega-Lite chart -- `chart_kind` picks which: @@ -904,7 +898,7 @@ def _exec_analytic_chart( result = _run_analytic_function( action, params=params, view_server=view_server, view_url=view_url, - user=user, user_pass=user_pass, token=token, + user=user, user_pass=user_pass, token=token, supply=supply, ) envelope = _result_envelope(result) result = _jsonable(result) @@ -1003,6 +997,7 @@ def exec_report_spec( user: str = settings.User_Profile.user_name, user_pass: str = settings.User_Profile.user_pwd, token: Optional[str] = None, + supply: Optional[RoleSupply] = None, ) -> Dict[str, Any]: """ Execute the action for a given format set and return a normalized result structure. @@ -1026,6 +1021,13 @@ def exec_report_spec( report execution -- and its provenance -- reflects that user, not a fallback service account. `user`/`user_pass` remain fully backward compatible when `token` is not given. + + RS-W2-3: `supply` (a `role_supply.RoleSupply`) provides an analytic routine's declared inputs + -- its Egeria client(s), and any reader, viewer or subject it declares. Omitted, plain + pyegeria's `PlainSupply` is built from the arguments above and supplies the caller's + EgeriaTech only. A role the supply cannot provide raises `role_supply.RoleRefused`, with the + environment's words, before the routine runs. A find-method spec (no analytic routine) does + not read `supply`: it builds its client class from the arguments above, as before. """ output_format = (output_format or "DICT").upper() params = _normalize_report_params(dict(params or {}), action_mode="find") @@ -1041,7 +1043,7 @@ def exec_report_spec( return _exec_analytic_chart( format_set_name, chart_kind=_CHART_KINDS[output_format], params=params, view_server=view_server, view_url=view_url, user=user, user_pass=user_pass, - token=token, + token=token, supply=supply, ) # Resolve the format set and action @@ -1103,7 +1105,7 @@ def exec_report_spec( if action.get("analytic_function") and not func_decl: result = _run_analytic_function( action, params=params, view_server=view_server, view_url=view_url, - user=user, user_pass=user_pass, token=token, + user=user, user_pass=user_pass, token=token, supply=supply, ) envelope = _result_envelope(result) result = _jsonable(result) diff --git a/pyegeria/view/role_supply.py b/pyegeria/view/role_supply.py new file mode 100644 index 00000000..492c630c --- /dev/null +++ b/pyegeria/view/role_supply.py @@ -0,0 +1,219 @@ +""" +SPDX-License-Identifier: Apache-2.0 +Copyright Contributors to the ODPi Egeria project. + +Role supply (RS-W2-3; report-spec pipeline model §7): the executor builds a routine's call from +the inputs the routine DECLARES, asking a `RoleSupply` for each one. It never reads parameter +names to decide what to pass. + +A routine declares its inputs as role words (`analytic_registry.INPUT_ROLES`): + + - ``egeria_client:`` -> ``supply.client()``: a client whose methods the routine calls + - ``reader:`` -> ``supply.reader()``: a reader the environment owns + - ``viewer`` -> ``supply.viewer()``: who is looking (identity words) + - ``subject`` -> ``supply.subject()``: the resource in view + +The inputs are bound to the routine's leading positional parameters in declared order (each +input's position is its index in ``inputs``); declared parameters follow by name. + +Every input is obtained before the routine's first stage runs, so a role the environment cannot +provide refuses the run before anything is called, in the environment's own words +(`RoleSupply.refusal_words`). Each environment provides its own supply: `PlainSupply` (here) is +plain pyegeria's and supplies the caller's Egeria client only; the Portal and Resource Explorer +provide theirs. + +Wave 2 boundary: `PlainSupply.client(cls)` hands back the caller's `EgeriaTech` itself for +EgeriaTech and for every class it proxies to, never a separate subclient, so behaviour is +byte-identical with the name-introspecting executor it replaces. Handing out per-class clients, +and the call indirection around them, are Wave 3. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Callable, List, Optional, Protocol, Sequence, runtime_checkable + +__all__ = [ + "RoleRefused", + "NoSuchReader", + "CallerTokenExpired", + "Viewer", + "Subject", + "RoleSupply", + "PlainSupply", + "egeria_tech_classes", + "role_kind", + "supply_inputs", +] + + +class RoleRefused(ValueError): + """A declared input the environment cannot supply. Raised before any stage runs. `words` is + the whole sentence a person is shown; `role` the input that could not be supplied.""" + + def __init__(self, words: str, *, role: str = "", routine: str = "", reason: str = ""): + super().__init__(words) + self.words = words + self.role = role + self.routine = routine + self.reason = reason or words + + +class NoSuchReader(RoleRefused): + """The environment has no reader of the declared name.""" + + +class CallerTokenExpired(RoleRefused): + """The environment cannot mint or refresh a client for the caller (e.g. an expired token). + Raised by an environment's supply; plain pyegeria never raises it.""" + + +@dataclass(frozen=True) +class Viewer: + """Who is looking. A field the environment does not know is None, never a guessed value.""" + user_id: str + is_admin: Optional[bool] = None + display_name: Optional[str] = None + + +@dataclass(frozen=True) +class Subject: + """The resource in view.""" + guid: str + type_name: Optional[str] = None + display_name: Optional[str] = None + + +@runtime_checkable +class RoleSupply(Protocol): + """What an environment supplies a routine. Each method returns the role or raises + `RoleRefused` (or a subclass) whose words say why, in the environment's own terms.""" + + def client(self, cls: type) -> Any: ... + + def viewer(self) -> Viewer: ... + + def reader(self, source: str) -> Any: ... + + def subject(self) -> Optional[Subject]: ... + + def refusal_words(self, kind: str) -> str: ... + + +# ── EgeriaTech and the classes it proxies to ── + +_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). + The resolver and `PlainSupply` both read this list, so "a class EgeriaTech serves" means the + same thing at load and at run.""" + 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("role-supply", "https://localhost:9443", user_id="role-supply", user_pwd="") + _EGERIA_TECH_CLASSES = [EgeriaTech] + list(dict.fromkeys(probe._subclient_map.values())) + return _EGERIA_TECH_CLASSES + + +def _client_class(name: str) -> Optional[type]: + return next((c for c in egeria_tech_classes() if c.__name__ == name), None) + + +# ── Plain pyegeria's supply ── + +_PLAIN_WORDS = { + "egeria_client": "Plain pyegeria supplies one Egeria client, the caller's EgeriaTech, for EgeriaTech " + "and the clients it proxies to; it supplies no other client class.", + "reader": "Plain pyegeria supplies no readers; run this routine where its reader is supplied " + "(Resource Explorer or the Portal).", + "viewer": "Plain pyegeria does not know who is viewing; run this routine where the viewer is supplied " + "(Resource Explorer or the Portal).", + "subject": "Plain pyegeria has no resource in view; run this routine where a subject is supplied " + "(Resource Explorer or the Portal).", +} + + +class PlainSupply: + """Plain pyegeria's supply: the caller's EgeriaTech, nothing else. + + Give it the client (`egeria_client`) or a factory that builds an authenticated one + (`client_factory`, called at most once, on the first `client()` request, so a routine that + declares no client never connects or mints a token, ISSUE-137). + + `client(cls)` returns that one EgeriaTech when `cls` is EgeriaTech or a class EgeriaTech + proxies to (Wave 2: byte-identical, no subclients), or when the client already is a `cls`. + Any other class, and every reader, viewer and subject, is refused with `refusal_words`.""" + + def __init__(self, egeria_client: Any = None, *, client_factory: Optional[Callable[[], Any]] = None): + self._client = egeria_client + self._factory = client_factory + + def _the_client(self) -> Any: + if self._client is None: + if self._factory is None: + raise RoleRefused("no Egeria client was given to plain pyegeria's supply", role="egeria_client") + self._client = self._factory() + return self._client + + def client(self, cls: type) -> Any: + if cls in egeria_tech_classes(): + return self._the_client() + if self._client is not None and isinstance(self._client, cls): + return self._client + raise RoleRefused(f"no {getattr(cls, '__name__', cls)} client: " + self.refusal_words("egeria_client"), + role=f"egeria_client:{getattr(cls, '__name__', cls)}") + + def viewer(self) -> Viewer: + raise RoleRefused(self.refusal_words("viewer"), role="viewer") + + def reader(self, source: str) -> Any: + raise NoSuchReader(f"no reader '{source}': " + self.refusal_words("reader"), role=f"reader:{source}") + + def subject(self) -> Optional[Subject]: + return None + + def refusal_words(self, kind: str) -> str: + return _PLAIN_WORDS.get(kind, f"Plain pyegeria does not supply '{kind}'.") + + +# ── Supplying a routine's declared inputs ── + +def role_kind(role: str) -> str: + """The role word without its qualifier: egeria_client, reader, viewer or subject.""" + return role.split(":", 1)[0] + + +def supply_inputs(routine: str, inputs: Sequence[str], supply: RoleSupply) -> List[Any]: + """The positional arguments for a routine's declared `inputs`, in declared order, each asked of + `supply`. Every input is obtained before this returns, so a role the supply cannot provide + raises `RoleRefused` before the routine's first stage runs. The words name the routine, the + input and the environment's reason.""" + args: List[Any] = [] + for role in inputs: + kind, _, qualifier = role.partition(":") + try: + if kind == "egeria_client": + cls = _client_class(qualifier) + if cls is None: + raise RoleRefused(f"no client class named {qualifier} is known to pyegeria: " + + supply.refusal_words("egeria_client"), role=role) + value = supply.client(cls) + elif kind == "reader": + value = supply.reader(qualifier) + elif kind == "viewer": + value = supply.viewer() + elif kind == "subject": + value = supply.subject() + if value is None: + raise RoleRefused(supply.refusal_words("subject"), role=role) + else: + raise RoleRefused(f"'{role}' is not a role word", role=role) + except RoleRefused as exc: + reason = f"its declared input '{role}' is not supplied ({exc.words})" + raise type(exc)(f"Routine '{routine}' cannot run: {reason}", role=role, routine=routine, + reason=reason) from exc + args.append(value) + return args diff --git a/pyegeria/view/spec_resolver.py b/pyegeria/view/spec_resolver.py index c32a6f2e..dbbb7d50 100644 --- a/pyegeria/view/spec_resolver.py +++ b/pyegeria/view/spec_resolver.py @@ -24,8 +24,11 @@ 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). + 5. role inputs: the routine's declared inputs, read by `declared_inputs`, the function the + executor calls before it binds them (RS-W2-3). A declaration whose egeria_client inputs do + not match the leading client arguments its python binding takes is refused, at load and at + run, with the same words. Which environment supplies each role is the run's: a role the + supply cannot provide refuses the run before its first stage (`role_supply.supply_inputs`). 6. admission and tile_safe: recorded from the routine's declaration. 7. seeding: the find method a spec may be seeded by is recorded. @@ -53,6 +56,8 @@ "OUTPUT_FORMATS", "CHART_FORMATS", "FUNCTION_VALUED_PARAMS", + "declared_inputs", + "routine_entry", ] #: Output formats the executor renders from a Format row ("ALL" means every one of them). @@ -113,18 +118,83 @@ def unrunnable_reason(self) -> str: # ── 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: the one list `PlainSupply` serves.""" + from pyegeria.view.role_supply import egeria_tech_classes + return egeria_tech_classes() -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 +# ── Check 5: a routine's declared inputs, for the resolver and the executor alike ── + +#: The leading parameter names the pre-W2-3 executor bound a client to by name. Read here ONLY to +#: check a declaration against its python binding (check 5); nothing binds by these names. +_BINDING_CLIENT_PARAM_NAMES = ("mgr", "ce", "expert", "client") + + +def _binding_client_slots(func: Any) -> int: + """How many leading client arguments a python binding takes (check 5's consistency count).""" + try: + params = inspect.signature(func).parameters.values() + except (TypeError, ValueError): + return 0 + count = 0 + for param in params: + if param.name in _BINDING_CLIENT_PARAM_NAMES and param.kind in ( + inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD): + count += 1 + else: + break + return count + + +def routine_entry(func_decl: Any) -> Any: + """The registry entry a routine reference names: by registered name, else the entry whose + python binding is that dotted path (a fetch step names its function by path). None if none.""" + from pyegeria.view.analytic_registry import get_analytic_registry + registry = get_analytic_registry() + if isinstance(func_decl, str): + if func_decl in registry: + return registry[func_decl] + return next((s for s in registry.values() if s.python_target() == func_decl), None) + return None + + +def declared_inputs(func_decl: Any, func: Any = None) -> List[str]: + """The inputs a routine declares, in declared order: what the executor supplies and binds to + its leading positional parameters. The resolver's check 5 and the executor both call this, so + a routine refused at load is refused at run with the same words. + + Raises `role_supply.RoleRefused` when the routine is not registered (so it declares nothing), + when an input is not a role word or is repeated, or when its python binding (`func`, imported + from the entry when not given) takes a different number of leading client arguments than the + declaration lists egeria_client inputs.""" + from pyegeria.view.analytic_registry import _INPUT_ROLE + from pyegeria.view.role_supply import RoleRefused, role_kind + + name = getattr(func_decl, "__name__", func_decl) + def refuse(reason: str) -> None: + raise RoleRefused(f"Routine '{name}' cannot run: {reason}.", routine=str(name), reason=reason) + + entry = routine_entry(func_decl) + if entry is None: + refuse(f"'{name}' is not a registered routine, so it declares no inputs for the executor to supply") + name = entry.name + inputs = list(entry.inputs) + bad = [i for i in inputs if not _INPUT_ROLE.fullmatch(i or "")] + if bad: + refuse(f"routine {name}'s input {', '.join(repr(b) for b in bad)} is not a role word " + "(egeria_client:, reader:, viewer, subject)") + if len(set(inputs)) != len(inputs): + refuse(f"routine {name} declares an input more than once") + if func is None: + from pyegeria.view.format_set_executor import _import_callable + func = _import_callable(entry.python_target()) + slots = _binding_client_slots(func) + clients = sum(1 for i in inputs if role_kind(i) == "egeria_client") + if slots != clients: + refuse(f"routine {name}'s binding takes {slots} leading client argument(s) and its declaration " + f"lists {clients} egeria_client input(s)") + return inputs def _find_method(func_decl: str) -> Tuple[Optional[Any], str]: @@ -216,6 +286,7 @@ def _check_analytic(name: str, action: Dict[str, Any], fail, notes, warns, out: 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) + from pyegeria.view.role_supply import RoleRefused func_decl = action["analytic_function"] registry = get_analytic_registry() @@ -269,18 +340,23 @@ def _check_analytic(name: str, action: Dict[str, Any], fail, notes, warns, out: fail(1, f"its {param} '{value}' is not a registered routine name (a report spec names routines by " f"registered name only)") + try: + out["role_inputs"] = declared_inputs(func_decl) + except RoleRefused as exc: + fail(5, exc.reason) + else: + notes.append("check 5: " + (f"role inputs {', '.join(out['role_inputs'])}" if out["role_inputs"] else + f"routine '{func_decl}' declares no role inputs") + + "; the environment's supply provides each before the first stage runs") + 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, " + notes.append(f"check 6: routine '{func_decl}' is not a valid routine declaration, so its " 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 diff --git a/tests/micro-tests/test_exec_report_spec_token.py b/tests/micro-tests/test_exec_report_spec_token.py index 99df9e5a..f9193b65 100644 --- a/tests/micro-tests/test_exec_report_spec_token.py +++ b/tests/micro-tests/test_exec_report_spec_token.py @@ -45,21 +45,31 @@ def factory(*args, **kwargs): return instances +def _dummy(client): + return {"count": 1} + + +@pytest.fixture +def dummy_routine(monkeypatch): + """A registered routine that declares one Egeria client input (RS-W2-3: the executor supplies + declared inputs; it no longer reads parameter names).""" + from pyegeria.view.analytic_registry import (AnalyticFunctionSpec, clear_runtime_analytic_functions, + register_analytic_functions) + register_analytic_functions({"token_probe": AnalyticFunctionSpec( + name="token_probe", function="pyegeria.view.overview_metrics.count_elements", + inputs=["egeria_client:EgeriaTech"])}) + monkeypatch.setattr(fse, "resolve_report_spec_function", lambda decl: _dummy) + yield + clear_runtime_analytic_functions() + + def _dummy_analytic_action(): - def _dummy(client): - return {"count": 1} - return {"analytic_function": _dummy, "analytic_spec_params": {}} + return {"analytic_function": "token_probe", "analytic_spec_params": {}} -def test_run_analytic_function_uses_token_when_given(fake_egeria_tech, monkeypatch): +def test_run_analytic_function_uses_token_when_given(fake_egeria_tech, dummy_routine): """When a token is supplied, set_bearer_token(token) is used -- not create_egeria_bearer_token().""" - monkeypatch.setattr( - fse, "resolve_report_spec_function", lambda decl: decl - ) - monkeypatch.setattr( - fse, "_bind_client_args", lambda func, client: (client,) - ) action = _dummy_analytic_action() result = fse._run_analytic_function( @@ -74,15 +84,9 @@ def test_run_analytic_function_uses_token_when_given(fake_egeria_tech, monkeypat assert client.create_egeria_bearer_token_calls == 0 -def test_run_analytic_function_mints_token_when_none_given(fake_egeria_tech, monkeypatch): +def test_run_analytic_function_mints_token_when_none_given(fake_egeria_tech, dummy_routine): """Backward compatible: with no token, the old create_egeria_bearer_token() path from user/user_pass is used unchanged.""" - monkeypatch.setattr( - fse, "resolve_report_spec_function", lambda decl: decl - ) - monkeypatch.setattr( - fse, "_bind_client_args", lambda func, client: (client,) - ) action = _dummy_analytic_action() result = fse._run_analytic_function( diff --git a/tests/micro-tests/test_format_set_executor_resolution.py b/tests/micro-tests/test_format_set_executor_resolution.py index 5c346b75..1ab68069 100644 --- a/tests/micro-tests/test_format_set_executor_resolution.py +++ b/tests/micro-tests/test_format_set_executor_resolution.py @@ -224,6 +224,21 @@ async def test_async_run_report_coerces_inherited_list_filters_from_strings(monk # dotted-path resolution) rather than a local test-module path, since # "micro-tests" isn't a valid dotted package segment. +def _register_fetch(path): + """A fetch step is a registered routine: its declared inputs are what run_analytic_action + supplies (RS-W2-3), here the caller's client from PlainSupply.""" + from pyegeria.view.analytic_registry import AnalyticFunctionSpec, register_analytic_functions + register_analytic_functions({"fetch_probe": AnalyticFunctionSpec( + name="fetch_probe", function=path, inputs=["egeria_client:MetadataExpert"])}) + + +@pytest.fixture(autouse=True) +def _clear_runtime_routines(): + yield + from pyegeria.view.analytic_registry import clear_runtime_analytic_functions + clear_runtime_analytic_functions() + + def test_run_analytic_action_runs_fetch_then_analytic_with_same_client(monkeypatch): import pyegeria.view.overview_metrics as om_module @@ -239,6 +254,7 @@ def _analytic(by_type, mgr): monkeypatch.setattr(om_module, "_fetch_for_test", _fetch, raising=False) monkeypatch.setattr(om_module, "_analytic_for_test", _analytic, raising=False) + _register_fetch("pyegeria.view.overview_metrics._fetch_for_test") spec = AnalyticActionSpec( fetch="pyegeria.view.overview_metrics._fetch_for_test", @@ -260,6 +276,7 @@ def _fetch_only(mgr, type_map): return [{"label": l, "type": t} for l, t in type_map] monkeypatch.setattr(om_module, "_fetch_only_for_test", _fetch_only, raising=False) + _register_fetch("pyegeria.view.overview_metrics._fetch_only_for_test") spec = AnalyticActionSpec(fetch="pyegeria.view.overview_metrics._fetch_only_for_test") result = fse.run_analytic_action(spec, "CLIENT-42", fetch_kwargs={"type_map": [("A", "Alpha")]}) diff --git a/tests/micro-tests/test_format_set_executor_token.py b/tests/micro-tests/test_format_set_executor_token.py index 1c45e132..78e6fd1f 100644 --- a/tests/micro-tests/test_format_set_executor_token.py +++ b/tests/micro-tests/test_format_set_executor_token.py @@ -53,7 +53,7 @@ def _reset_instances(): } _ANALYTIC_FMT = { - "action": {"analytic_function": "fake.analytic", "analytic_spec_params": {}}, + "action": {"analytic_function": "fake_analytic", "analytic_spec_params": {}}, "target_type": "Referenceable", } @@ -65,7 +65,23 @@ def _patch_find_path(monkeypatch): lambda decl: (_RecordingClient, "find_things")) -def _patch_analytic_path(monkeypatch): +def _register_fake_analytic(inputs): + """The routine is registered with its declared inputs: the executor supplies those (RS-W2-3).""" + from pyegeria.view.analytic_registry import AnalyticFunctionSpec, register_analytic_functions + register_analytic_functions({"fake_analytic": AnalyticFunctionSpec( + name="fake_analytic", function="pyegeria.view.overview_metrics.count_elements", inputs=list(inputs))}) + + +@pytest.fixture(autouse=True) +def _clear_runtime_routines(): + yield + from pyegeria.view.analytic_registry import clear_runtime_analytic_functions + clear_runtime_analytic_functions() + + +def _patch_analytic_path(monkeypatch, inputs=("egeria_client:EgeriaTech",)): + _register_fake_analytic(inputs) + def _analytic(client): return {"count": 3, "token_seen": client.token} monkeypatch.setattr(fse, "select_report_spec", lambda name, out: _ANALYTIC_FMT) @@ -139,10 +155,10 @@ def test_chart_path_threads_token(monkeypatch): def test_analytic_routine_without_client_param_builds_no_client(monkeypatch): - """ISSUE-137: a routine that takes no mgr/ce/expert/client parameter must not + """ISSUE-137: a routine that declares no egeria_client input must not cause an EgeriaTech to be built or a token minted -- an unreachable view_url must not stop it running.""" - _patch_analytic_path(monkeypatch) + _patch_analytic_path(monkeypatch, inputs=()) def _local_only(window=0): return {"count": 7, "window": window} diff --git a/tests/micro-tests/test_routine_declarations_match_library.py b/tests/micro-tests/test_routine_declarations_match_library.py index b1c9a958..5f115dbe 100644 --- a/tests/micro-tests/test_routine_declarations_match_library.py +++ b/tests/micro-tests/test_routine_declarations_match_library.py @@ -55,7 +55,11 @@ def test_builtin_routine_loads_in_the_library(name): @pytest.mark.parametrize("name", sorted(_BUILTINS)) def test_the_library_reads_a_pyegeria_entry_as_pyegeria_does(name): spec = _BUILTINS[name] - theirs = routine_lib.routine_declaration(spec).model_dump(by_alias=True, exclude={"declaration"}) + # The library's routine_declaration() takes what a manifest does not carry through `extra`, and + # 0.4 lists `inputs` there (RS-W2-3 declares them on every built-in), so they are handed over + # the documented way. A release that reads them from the manifest gets the same value. + theirs = routine_lib.routine_declaration(spec, inputs=list(spec.inputs)).model_dump( + by_alias=True, exclude={"declaration"}) ours = spec.routine().model_dump(by_alias=True, exclude={"declaration"}) assert theirs == ours diff --git a/tests/micro-tests/test_rs_w2_2_spec_resolver.py b/tests/micro-tests/test_rs_w2_2_spec_resolver.py index a261131d..21b2f431 100644 --- a/tests/micro-tests/test_rs_w2_2_spec_resolver.py +++ b/tests/micro-tests/test_rs_w2_2_spec_resolver.py @@ -138,7 +138,8 @@ def test_check_2_analytic_param_not_declared_is_refused_with_the_declared_list() 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")}) + name="silent", function="pyegeria.view.overview_metrics.count_elements", + inputs=["egeria_client:MetadataExpert"])}) 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) @@ -186,7 +187,9 @@ def test_an_analytic_spec_records_role_inputs_admission_and_tile_safe(): 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 result.role_inputs == ["egeria_client:MetadataExpert"] + assert "check 5: role inputs egeria_client:MetadataExpert; the environment's supply provides each before the " \ + "first stage runs" in result.notes assert any("declares no tile_safe" in n for n in result.notes) assert "type_name" in result.params @@ -228,7 +231,8 @@ def test_results_are_cached_and_a_registry_change_invalidates(): 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")}) + name="not_yet_registered", function="pyegeria.view.overview_metrics.count_elements", + inputs=["egeria_client:MetadataExpert"])}) assert sr.resolve_all()["Probe-Spec"].runnable brf.unregister_report_spec("Probe-Spec") assert "Probe-Spec" not in sr.resolve_all() diff --git a/tests/micro-tests/test_rs_w2_3_inputs_by_role.py b/tests/micro-tests/test_rs_w2_3_inputs_by_role.py new file mode 100644 index 00000000..84a5875d --- /dev/null +++ b/tests/micro-tests/test_rs_w2_3_inputs_by_role.py @@ -0,0 +1,305 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright Contributors to the ODPi Egeria project. +"""RS-W2-3 PR-A: the executor supplies a routine's DECLARED inputs from the environment's role +supply, in declared order; it never reads parameter names to decide what to pass. A role the +supply cannot provide, or a declaration that does not match its binding (check 5), refuses the +run before any stage, in words. Plain pyegeria's PlainSupply hands back the caller's EgeriaTech +for every class it proxies to, so Wave 2 is byte-identical.""" + +import pytest + +import pyegeria.view.base_report_formats as brf +import pyegeria.view.overview_metrics as om +from pyegeria.egeria_tech_client import EgeriaTech +from pyegeria.omvs.classification_explorer import ClassificationExplorer +from pyegeria.omvs.metadata_expert import MetadataExpert +from pyegeria.view import format_set_executor as fse +from pyegeria.view import spec_resolver as sr +from pyegeria.view._output_format_models import ActionParameter, Format, FormatSet +from pyegeria.view.analytic_demo_specs import get_analytic_demo_specs +from pyegeria.view.analytic_registry import (AnalyticFunctionSpec, _BUILTINS as BUILTIN_ANALYTIC_FUNCTIONS, + clear_runtime_analytic_functions, register_analytic_functions) +from pyegeria.view.role_supply import (CallerTokenExpired, NoSuchReader, PlainSupply, RoleRefused, Subject, + Viewer, egeria_tech_classes, supply_inputs) + +MOD = "pyegeria.view.overview_metrics" + + +# ── A fake Egeria client: the only thing faked in the end-to-end tests ── + +class _FakeTech: + """Stands in for EgeriaTech inside format_set_executor; records construction and auth.""" + made = [] + + def __init__(self, view_server=None, view_url=None, user_id=None, user_pwd=None, **kw): + self.view_server, self.platform_url, self.user_id = view_server, view_url, user_id + self.minted = 0 + self.tokens = [] + self.calls = [] + _FakeTech.made.append(self) + + def set_bearer_token(self, token): + self.tokens.append(token) + + def create_egeria_bearer_token(self): + self.minted += 1 + + def find_metadata_elements(self, body): + self.calls.append(("find_metadata_elements", body.get("metadataElementTypeName"))) + return [{"elementHeader": {"guid": "g1", "classifications": []}, "elementProperties": {}}] + + def count_metadata_elements(self, body, **kw): + self.calls.append(("count_metadata_elements", body.get("metadataElementTypeName"))) + return 7 + + def get_relationships(self, relationship_type=None, **kw): + self.calls.append(("get_relationships", relationship_type)) + return [{"relationshipHeader": {"guid": "r1"}, "end1": {"guid": "g1"}, "end2": {"guid": "g2"}}] + + +@pytest.fixture +def fake_tech(monkeypatch): + _FakeTech.made = [] + monkeypatch.setattr(fse, "EgeriaTech", _FakeTech) + om._count_caps.clear() + om._native_server_ok.clear() + yield _FakeTech.made + om._count_caps.clear() + om._native_server_ok.clear() + + +@pytest.fixture(autouse=True) +def _clean(): + yield + clear_runtime_analytic_functions() + brf.clear_runtime_report_specs() + + +def _routine(monkeypatch, name, func, inputs): + """Register `func` as routine `name` (bound under overview_metrics, an allowed module).""" + monkeypatch.setattr(om, f"_w23_{name}", func, raising=False) + register_analytic_functions({name: AnalyticFunctionSpec(name=name, function=f"{MOD}._w23_{name}", + inputs=list(inputs))}) + + +def _spec_for(routine, label="W23-Probe"): + brf.register_report_specs({label: FormatSet( + heading=label, description="probe", target_type="Referenceable", + action=ActionParameter(analytic_function=routine), formats=[Format(types=["DICT"], attributes=[])])}, + source="test") + return label + + +def _run(label, **kw): + return fse.exec_report_spec(label, output_format="DICT", view_server="vs", view_url="https://x", + user="u", user_pass="p", **kw) + + +class _Supply: + """An environment's supply with a distinct object per client class, a viewer and a subject.""" + + def __init__(self, fail_client=None): + self.asked = [] + self.fail_client = fail_client + + def client(self, cls): + self.asked.append(cls.__name__) + if self.fail_client: + raise self.fail_client + return f"client<{cls.__name__}>" + + def viewer(self): + return Viewer(user_id="peterprofile", is_admin=False, display_name="Peter") + + def reader(self, source): + return f"reader<{source}>" + + def subject(self): + return Subject(guid="g-subject", type_name="Asset") + + def refusal_words(self, kind): + return f"the test environment cannot supply {kind}" + + +# ── The 28 built-in declarations ── + +def test_every_builtin_declares_its_client_inputs_and_they_match_its_binding(): + assert len(BUILTIN_ANALYTIC_FUNCTIONS) == 28 + served = {c.__name__ for c in egeria_tech_classes()} + for name, entry in BUILTIN_ANALYTIC_FUNCTIONS.items(): + assert entry.inputs, f"{name} declares no inputs" + for role in entry.inputs: + kind, _, cls = role.partition(":") + assert kind == "egeria_client" and cls in served, f"{name}: {role}" + assert sr.declared_inputs(name) == entry.inputs # check 5 passes for every built-in + entry.routine() # a valid RoutineDeclaration (no repeated input) + + +def test_two_client_slots_are_declared_by_the_class_each_one_calls(): + assert BUILTIN_ANALYTIC_FUNCTIONS["semantic_grounding"].inputs == [ + "egeria_client:MetadataExpert", "egeria_client:ClassificationExplorer"] + assert BUILTIN_ANALYTIC_FUNCTIONS["certifications_summary"].inputs == ["egeria_client:ClassificationExplorer"] + assert BUILTIN_ANALYTIC_FUNCTIONS["control_implementation_summary"].inputs == ["egeria_client:SecurityOfficer"] + # Calls across several classes -> EgeriaTech (asset_composition also reads ValidMetadata; metric_trend + # forwards its first client to any target routine). + assert BUILTIN_ANALYTIC_FUNCTIONS["asset_composition"].inputs == ["egeria_client:EgeriaTech"] + assert BUILTIN_ANALYTIC_FUNCTIONS["metric_trend"].inputs == [ + "egeria_client:EgeriaTech", "egeria_client:ClassificationExplorer"] + + +# ── PlainSupply ── + +def test_plain_supply_hands_back_the_callers_egeria_tech_for_every_class_it_proxies(): + tech = EgeriaTech("vs", "https://localhost:9443", user_id="u", user_pwd="p") + supply = PlainSupply(tech) + for cls in egeria_tech_classes(): + assert supply.client(cls) is tech # never a subclient in Wave 2 + assert tech._instantiated_clients == {} + + +def test_plain_supply_builds_the_client_once_and_only_when_asked(): + built = [] + supply = PlainSupply(client_factory=lambda: built.append(1) or "THE-CLIENT") + assert built == [] + assert supply.client(MetadataExpert) == supply.client(ClassificationExplorer) == "THE-CLIENT" + assert built == [1] + + +def test_plain_supply_supplies_no_reader_viewer_or_subject(): + supply = PlainSupply("C") + with pytest.raises(NoSuchReader, match="no reader 'ods': Plain pyegeria supplies no readers"): + supply.reader("ods") + with pytest.raises(RoleRefused, match="Plain pyegeria does not know who is viewing"): + supply.viewer() + assert supply.subject() is None + with pytest.raises(RoleRefused, match="no Egeria client was given"): + PlainSupply().client(MetadataExpert) + + +def test_inputs_are_supplied_in_declared_order(): + supply = _Supply() + args = supply_inputs("r", ["egeria_client:ClassificationExplorer", "egeria_client:MetadataExpert", + "viewer", "subject", "reader:ods"], supply) + assert args[:2] == ["client", "client"] + assert args[2].user_id == "peterprofile" and args[3].guid == "g-subject" and args[4] == "reader" + + +# ── Refusals, each with its words, before any stage runs ── + +def test_refusal_a_missing_role(monkeypatch, fake_tech): + """The routine declares a role (viewer) plain pyegeria does not supply.""" + called = [] + _routine(monkeypatch, "who_is_looking", lambda mgr, viewer, **kw: called.append(1), + ["egeria_client:MetadataExpert", "viewer"]) + with pytest.raises(RoleRefused) as info: + _run(_spec_for("who_is_looking")) + assert str(info.value) == ( + "Routine 'who_is_looking' cannot run: its declared input 'viewer' is not supplied (Plain pyegeria " + "does not know who is viewing; run this routine where the viewer is supplied (Resource Explorer or " + "the Portal).)") + assert info.value.role == "viewer" + assert called == [] # no stage ran + + +def test_refusal_an_undeclared_client_slot_at_load_and_at_run_with_the_same_words(monkeypatch, fake_tech): + """The routine's binding takes a client argument its declaration does not list (check 5).""" + called = [] + _routine(monkeypatch, "takes_a_client", lambda mgr, as_of=None: called.append(1), []) + words = ("routine takes_a_client's binding takes 1 leading client argument(s) and its declaration lists " + "0 egeria_client input(s)") + label = _spec_for("takes_a_client") + resolved = sr.resolve_all(refresh=True)[label] + assert not resolved.runnable and resolved.checks == [5] + assert resolved.unrunnable_reason == f"Spec '{label}' cannot run (check 5): {words}." + with pytest.raises(RoleRefused) as info: + _run(label) + assert str(info.value) == f"Routine 'takes_a_client' cannot run: {words}." + assert called == [] and fake_tech == [] # refused before connecting, before any stage + + +def test_refusal_a_declared_input_the_supply_lacks(monkeypatch, fake_tech): + """The environment's supply cannot mint a client for the caller (e.g. an expired token).""" + called = [] + _routine(monkeypatch, "counts", lambda mgr, ce, **kw: called.append(1), + ["egeria_client:MetadataExpert", "egeria_client:ClassificationExplorer"]) + expired = CallerTokenExpired("Your Egeria session has expired; sign in again.") + with pytest.raises(CallerTokenExpired) as info: + _run(_spec_for("counts"), supply=_Supply(fail_client=expired)) + assert str(info.value) == ( + "Routine 'counts' cannot run: its declared input 'egeria_client:MetadataExpert' is not supplied " + "(Your Egeria session has expired; sign in again.)") + assert called == [] and fake_tech == [] + + +def test_refusal_a_client_class_plain_pyegeria_does_not_serve(monkeypatch, fake_tech): + _routine(monkeypatch, "odd_client", lambda client: 1, ["egeria_client:NotAPyegeriaClient"]) + with pytest.raises(RoleRefused, match="no client class named NotAPyegeriaClient is known to pyegeria"): + _run(_spec_for("odd_client")) + + +def test_a_fetch_step_must_be_a_registered_routine(monkeypatch): + from pyegeria.view.analytic_registry import AnalyticActionSpec + monkeypatch.setattr(om, "_w23_unregistered_fetch", lambda mgr: [], raising=False) + with pytest.raises(RoleRefused, match="is not a registered routine, so it declares no inputs"): + fse.run_analytic_action(AnalyticActionSpec(fetch=f"{MOD}._w23_unregistered_fetch"), "C") + + +# ── End to end through the real executor, only the Egeria client faked ── + +def test_end_to_end_a_two_client_routine_gets_the_callers_one_client_in_both_slots(fake_tech): + seen = {} + real = om.semantic_grounding + + def spy(mgr, ce, **kw): + seen["mgr"], seen["ce"] = mgr, ce + return real(mgr, ce, **kw) + + om_spy = pytest.MonkeyPatch() + om_spy.setattr(om, "semantic_grounding", spy) + try: + result = _run("Analytic Demo - Semantic Grounding", token="caller-token") + finally: + om_spy.undo() + assert result["kind"] == "json" and set(result["data"]) == {"groundingLinks", "groundingPct"} + assert len(fake_tech) == 1 and fake_tech[0].tokens == ["caller-token"] and fake_tech[0].minted == 0 + assert seen["mgr"] is seen["ce"] is fake_tech[0] + assert {c[0] for c in fake_tech[0].calls} == {"count_metadata_elements", "get_relationships"} + + +def test_end_to_end_an_environment_supply_is_asked_by_declared_class_in_order(fake_tech): + supply = _Supply() + seen = {} + + def spy(mgr, ce, **kw): + seen["args"] = (mgr, ce) + return {"groundingLinks": 1, "groundingPct": 1} + + mp = pytest.MonkeyPatch() + mp.setattr(om, "semantic_grounding", spy) + try: + result = _run("Analytic Demo - Semantic Grounding", supply=supply) + finally: + mp.undo() + assert supply.asked == ["MetadataExpert", "ClassificationExplorer"] + assert seen["args"] == ("client", "client") + assert fake_tech == [] # the environment's supply, not a client built from user/password + assert result == {"kind": "json", "data": {"groundingLinks": 1, "groundingPct": 1}} + + +def test_the_run_and_the_resolver_read_inputs_through_one_function(monkeypatch, fake_tech): + """Rule 1: what the resolver shows (check 5) and what the run binds come from declared_inputs.""" + calls = [] + real = sr.declared_inputs + monkeypatch.setattr(sr, "declared_inputs", lambda *a, **k: calls.append(a[0]) or real(*a, **k)) + _run("Analytic Demo - Element Count by Type") + sr.resolve(get_analytic_demo_specs()["Analytic Demo - Element Count by Type"]) + assert calls == ["count_elements", "count_elements"] + + +def test_every_demo_spec_lists_its_declared_inputs_at_load(): + results = sr.resolve_all(refresh=True) + for label, spec in get_analytic_demo_specs().items(): + result = results[label] + assert result.runnable, result.unrunnable_reason + entry = BUILTIN_ANALYTIC_FUNCTIONS[spec.action.analytic_function] + assert result.role_inputs == entry.inputs