From 9d4122076c3495607988d6b8abcc708c1856b9c9 Mon Sep 17 00:00:00 2001 From: Dan Wolfson Date: Fri, 16 Oct 2026 16:09:57 -0500 Subject: [PATCH 1/2] feat: GovernanceEngine client for moved engine-action operations (ISSUE-153) Egeria #9371 moved engine-action claim/status/completion/active-claimed to the governance-engine view service. Add the GovernanceEngine client, make the AutomatedCuration versions deprecated wrappers with a one-time HTTP-404 fallback to the old path, and add cancel_governance_action_process. Signed-off-by: Dan Wolfson --- pyegeria/__init__.py | 2 + pyegeria/egeria_client.py | 3 + pyegeria/egeria_tech_client.py | 3 + pyegeria/omvs/__init__.py | 2 + pyegeria/omvs/automated_curation.py | 308 +++++------------- pyegeria/omvs/governance_engine.py | 256 +++++++++++++++ .../test_issue_153_governance_engine.py | 181 ++++++++++ 7 files changed, 530 insertions(+), 225 deletions(-) create mode 100644 pyegeria/omvs/governance_engine.py create mode 100644 tests/micro-tests/test_issue_153_governance_engine.py diff --git a/pyegeria/__init__.py b/pyegeria/__init__.py index f0a72a6e..cf61f9f9 100644 --- a/pyegeria/__init__.py +++ b/pyegeria/__init__.py @@ -74,6 +74,7 @@ from pyegeria.omvs.glossary_manager import GlossaryManager from pyegeria.omvs.governance_officer import GovernanceOfficer from pyegeria.omvs.lineage_linker import LineageLinker +from pyegeria.omvs.governance_engine import GovernanceEngine from pyegeria.omvs.location_arena import LocationArena from pyegeria.omvs.metadata_expert import MetadataExpert from pyegeria.omvs.my_profile import MyProfile @@ -168,6 +169,7 @@ "AssetCatalog", "AssetMaker", "AutomatedCuration", + "GovernanceEngine", "ClassificationExplorer", "CollectionManager", "CommunityMatters", diff --git a/pyegeria/egeria_client.py b/pyegeria/egeria_client.py index 95de7952..526dee34 100644 --- a/pyegeria/egeria_client.py +++ b/pyegeria/egeria_client.py @@ -31,6 +31,7 @@ from pyegeria.omvs.glossary_manager import GlossaryManager from pyegeria.omvs.governance_officer import GovernanceOfficer from pyegeria.omvs.lineage_linker import LineageLinker +from pyegeria.omvs.governance_engine import GovernanceEngine from pyegeria.omvs.location_arena import LocationArena from pyegeria.omvs.metadata_expert import MetadataExpert from pyegeria.omvs.my_profile import MyProfile @@ -82,6 +83,8 @@ def __init__( "asset_maker": AssetMaker, "auto_curate": AutomatedCuration, "automated_curation": AutomatedCuration, + "gov_engine": GovernanceEngine, + "governance_engine": GovernanceEngine, "class_mgr": ClassificationExplorer, "classification_manager": ClassificationExplorer, "collections": CollectionManager, diff --git a/pyegeria/egeria_tech_client.py b/pyegeria/egeria_tech_client.py index a13cdc56..9cf39114 100644 --- a/pyegeria/egeria_tech_client.py +++ b/pyegeria/egeria_tech_client.py @@ -23,6 +23,7 @@ from pyegeria.omvs.glossary_manager import GlossaryManager from pyegeria.omvs.governance_officer import GovernanceOfficer from pyegeria.omvs.lineage_linker import LineageLinker +from pyegeria.omvs.governance_engine import GovernanceEngine from pyegeria.omvs.location_arena import LocationArena from pyegeria.omvs.metadata_expert import MetadataExpert from pyegeria.omvs.my_profile import MyProfile @@ -88,6 +89,8 @@ def __init__( self._subclient_map = { "auto_curate": AutomatedCuration, "automated_curation": AutomatedCuration, + "gov_engine": GovernanceEngine, + "governance_engine": GovernanceEngine, "class_mgr": ClassificationExplorer, "classification_manager": ClassificationExplorer, "reg_info": RegisteredInfo, diff --git a/pyegeria/omvs/__init__.py b/pyegeria/omvs/__init__.py index 4643dab9..7f110307 100644 --- a/pyegeria/omvs/__init__.py +++ b/pyegeria/omvs/__init__.py @@ -18,6 +18,7 @@ from pyegeria.omvs.glossary_manager import GlossaryManager from pyegeria.omvs.governance_officer import GovernanceOfficer from pyegeria.omvs.lineage_linker import LineageLinker +from pyegeria.omvs.governance_engine import GovernanceEngine from pyegeria.omvs.location_arena import LocationArena from pyegeria.omvs.metadata_expert import MetadataExpert from pyegeria.omvs.my_profile import MyProfile @@ -45,6 +46,7 @@ "AssetCatalog", "AssetMaker", "AutomatedCuration", + "GovernanceEngine", "ClassificationExplorer", "CollectionManager", "CommunityMatters", diff --git a/pyegeria/omvs/automated_curation.py b/pyegeria/omvs/automated_curation.py index 2ecea735..f1a7eacc 100644 --- a/pyegeria/omvs/automated_curation.py +++ b/pyegeria/omvs/automated_curation.py @@ -2300,6 +2300,9 @@ def delete_client_side_secret( async def _async_cancel_engine_action(self, engine_action_guid: str) -> None: """Request that an engine action request is cancelled and any running governance service is stopped. Async Ver. + + On Egeria 6.3 (odpi/egeria#9371) an engine action that has already finished (COMPLETED, INVALID, + IGNORED, FAILED, CANCELLED or ABANDONED) is left unchanged and the call still succeeds. Parameters ---------- engine_action_guid : str @@ -2351,279 +2354,134 @@ def cancel_engine_action(self, engine_action_guid: str) -> None: loop.run_until_complete(self._async_cancel_engine_action(engine_action_guid)) return - async def _async_claim_engine_action(self, engine_action_guid: str) -> None: - """Request that execution of an engine action is allocated to the caller. Async version. + # The operations below moved to the Governance Engine view service in Egeria odpi/egeria#9371. These + # wrappers are deprecated for one release: they delegate to + # pyegeria.omvs.governance_engine.GovernanceEngine on the new URLs. If (and only if) the new URL + # answers HTTP 404 - an Egeria built before #9371 - the call is retried once on the old + # automated-curation URL, and the URL family that worked is remembered for this client. No other + # error triggers a fallback. + + def _governance_engine_client(self): + from pyegeria.omvs.governance_engine import GovernanceEngine + client = getattr(self, "_gov_engine_client", None) + if client is None: + client = GovernanceEngine(self.view_server, self.platform_url, self.user_id, self.user_pwd, + token=self.token, timeout=getattr(self, "timeout", None), + legacy_fallback=True) + self._gov_engine_client = client + return client + + @staticmethod + def _warn_moved(name: str) -> None: + import warnings + warnings.warn( + f"AutomatedCuration.{name} is deprecated; the operation moved to the Governance Engine view " + f"service - use GovernanceEngine.{name} (egeria_client.gov_engine).", + DeprecationWarning, stacklevel=3) - Parameters - ---------- - engine_action_guid : str - The GUID of the engine action to claim. - - Returns - ------- - None - - Raises - ------ - PyegeriaException - - Notes - ----- - For more information see: https://egeria-project.org/concepts/engine-action - """ - validate_guid(engine_action_guid) - url = f"{self.ref_curation_command_base}/engine-actions/{engine_action_guid}/claim" - await self._async_make_request("POST", url) + async def _async_claim_engine_action(self, engine_action_guid: str) -> None: + """Deprecated - use GovernanceEngine. Claim an engine action for the caller. Async version.""" + self._warn_moved("claim_engine_action") + await self._governance_engine_client()._async_claim_engine_action(engine_action_guid) def claim_engine_action(self, engine_action_guid: str) -> None: - """Request that execution of an engine action is allocated to the caller.""" + """Deprecated - use GovernanceEngine. Claim an engine action for the caller.""" loop = asyncio.get_event_loop() loop.run_until_complete(self._async_claim_engine_action(engine_action_guid)) - return async def _async_update_engine_action_status(self, engine_action_guid: str, status: str, - body: Optional[dict] = None) -> None: - """Update the status of the engine action - providing the caller is permitted. Async version. - - Parameters - ---------- - engine_action_guid : str - The GUID of the engine action to update. - status : str - One of "Requested", "Approved", "Waiting", "Activating", "In Progress", "Paused", - "For Information", "Completed", "Invalid", "Ignored", "Failed", "Cancelled", "Abandoned" or "Other". - body : dict, optional - Request body to pass directly to the API - overrides `status` if supplied. - - Returns - ------- - None - - Raises - ------ - PyegeriaException - - Notes - ----- - Sample JSON body: - ```json - { - "class" : "EngineActionStatusRequestBody", - "status" : "IN_PROGRESS" - } - ``` - For more information see: https://egeria-project.org/concepts/engine-action - """ - validate_guid(engine_action_guid) - url = f"{self.ref_curation_command_base}/engine-actions/{engine_action_guid}/status/update" - payload = body if body is not None else {"class": "EngineActionStatusRequestBody", "status": status} - await self._async_make_request("POST", url, payload) + body: Optional[dict] = None) -> None: + """Deprecated - use GovernanceEngine. Update the status of an engine action. Async version.""" + self._warn_moved("update_engine_action_status") + await self._governance_engine_client()._async_update_engine_action_status( + engine_action_guid, status, body) def update_engine_action_status(self, engine_action_guid: str, status: str, body: Optional[dict] = None) -> None: - """Update the status of the engine action - providing the caller is permitted.""" + """Deprecated - use GovernanceEngine. Update the status of an engine action.""" loop = asyncio.get_event_loop() loop.run_until_complete(self._async_update_engine_action_status(engine_action_guid, status, body)) - return async def _async_update_action_target_status( self, action_target_guid: str, status: str, start_date: Optional[str] = None, completion_date: Optional[str] = None, completion_message: Optional[str] = None, body: Optional[dict] = None) -> None: - """Update the status of a specific action target. Async version. - - By default, these values are derived from the values for the governance action service. However, if - the governance action service has to process many target elements, then setting the status on each - individual target will show the progress of the governance action service. - - Parameters - ---------- - action_target_guid : str - The GUID of the action target to update. - status : str - The new status for the action target. - start_date : str, optional - The date processing of this target started. - completion_date : str, optional - The date processing of this target completed. - completion_message : str, optional - A message describing the outcome of processing this target. - body : dict, optional - Request body to pass directly to the API - overrides the other parameters if supplied. - - Returns - ------- - None - - Raises - ------ - PyegeriaException - - Notes - ----- - Sample JSON body: - ```json - { - "class" : "ActionTargetStatusRequestBody", - "actionTargetGUID" : "add guid here", - "status" : "COMPLETED", - "startDate" : "{{$isoTimestamp}}", - "completionDate" : "{{$isoTimestamp}}", - "completionMessage" : "add completion message here" - } - ``` - For more information see: https://egeria-project.org/concepts/engine-action - """ - validate_guid(action_target_guid) - url = f"{self.ref_curation_command_base}/engine-actions/action-targets/update" - payload = body if body is not None else { - "class": "ActionTargetStatusRequestBody", - "actionTargetGUID": action_target_guid, - "status": status, - "startDate": start_date, - "completionDate": completion_date, - "completionMessage": completion_message, - } - await self._async_make_request("POST", url, body_slimmer(payload)) + """Deprecated - use GovernanceEngine. Update the status of an action target. Async version.""" + self._warn_moved("update_action_target_status") + await self._governance_engine_client()._async_update_action_target_status( + action_target_guid, status, start_date, completion_date, completion_message, body) def update_action_target_status( self, action_target_guid: str, status: str, start_date: Optional[str] = None, completion_date: Optional[str] = None, completion_message: Optional[str] = None, body: Optional[dict] = None) -> None: - """Update the status of a specific action target.""" + """Deprecated - use GovernanceEngine. Update the status of an action target.""" loop = asyncio.get_event_loop() - loop.run_until_complete( - self._async_update_action_target_status( - action_target_guid, status, start_date, completion_date, completion_message, body - ) - ) - return + loop.run_until_complete(self._async_update_action_target_status( + action_target_guid, status, start_date, completion_date, completion_message, body)) async def _async_record_completion_status( self, engine_action_guid: str, status: str, request_parameters: Optional[dict] = None, output_guards: Optional[list[str]] = None, new_action_targets: Optional[list[dict]] = None, completion_message: Optional[str] = None, body: Optional[dict] = None) -> None: - """Declare that all the processing for the governance action service is finished and the status of the - work. Async version. - - Parameters - ---------- - engine_action_guid : str - The GUID of the engine action to record completion for. - status : str - One of "Actioned", "Invalid", "Failed" or "Other". - request_parameters : dict, optional - Additional request parameters. - output_guards : list[str], optional - Guards output by this governance action service, driving the next steps in a process. - new_action_targets : list[dict], optional - Additional action targets to add, each a dict with "actionTargetName"/"actionTargetGUID". - completion_message : str, optional - A message describing the completion outcome. - body : dict, optional - Request body to pass directly to the API - overrides the other parameters if supplied. - - Returns - ------- - None - - Raises - ------ - PyegeriaException - - Notes - ----- - Sample JSON body: - ```json - { - "class" : "CompletionStatusRequestBody", - "status" : "ACTIONED", - "requestParameters" : { - "propertyName1" : "propertyValue1", - "propertyName2" : "propertyValue2" - }, - "outputGuards" : [ "add guard here" ], - "newActionTargets" : [ - { - "actionTargetName" : "add action target name here", - "actionTargetGUID" : "add guid here" - } - ], - "completionMessage" : "add completion message here" - } - ``` - For more information see: https://egeria-project.org/concepts/engine-action - """ - validate_guid(engine_action_guid) - url = f"{self.ref_curation_command_base}/engine-actions/{engine_action_guid}/completion-status" - payload = body if body is not None else { - "class": "CompletionStatusRequestBody", - "status": status, - "requestParameters": request_parameters, - "outputGuards": output_guards, - "newActionTargets": new_action_targets, - "completionMessage": completion_message, - } - await self._async_make_request("POST", url, body_slimmer(payload)) + """Deprecated - use GovernanceEngine. Record the completion status of an engine action. Async version.""" + self._warn_moved("record_completion_status") + await self._governance_engine_client()._async_record_completion_status( + engine_action_guid, status, request_parameters, output_guards, new_action_targets, + completion_message, body) def record_completion_status( self, engine_action_guid: str, status: str, request_parameters: Optional[dict] = None, output_guards: Optional[list[str]] = None, new_action_targets: Optional[list[dict]] = None, completion_message: Optional[str] = None, body: Optional[dict] = None) -> None: - """Declare that all the processing for the governance action service is finished and the status of - the work.""" + """Deprecated - use GovernanceEngine. Record the completion status of an engine action.""" loop = asyncio.get_event_loop() - loop.run_until_complete( - self._async_record_completion_status( - engine_action_guid, status, request_parameters, output_guards, new_action_targets, - completion_message, body - ) - ) - return + loop.run_until_complete(self._async_record_completion_status( + engine_action_guid, status, request_parameters, output_guards, new_action_targets, + completion_message, body)) async def _async_get_active_claimed_engine_actions( self, governance_engine_guid: str, start_from: int = 0, page_size: int = 0) -> list | str: - """Retrieve the engine actions that are still in process and that have been claimed by this caller's - userId. This call is used when the caller restarts. Async version. + """Deprecated - use GovernanceEngine. Engine actions in process claimed by this caller. Async version.""" + self._warn_moved("get_active_claimed_engine_actions") + return await self._governance_engine_client()._async_get_active_claimed_engine_actions( + governance_engine_guid, start_from, page_size) - Parameters - ---------- - governance_engine_guid : str - The GUID of the governance engine to retrieve claimed engine actions for. - start_from : int, default 0 - The index from which to start fetching. - page_size : int, default 0 - The maximum number to fetch in a single request. + def get_active_claimed_engine_actions( + self, governance_engine_guid: str, start_from: int = 0, page_size: int = 0) -> list | str: + """Deprecated - use GovernanceEngine. Engine actions in process claimed by this caller.""" + loop = asyncio.get_event_loop() + return loop.run_until_complete(self._async_get_active_claimed_engine_actions( + governance_engine_guid, start_from, page_size)) - Returns - ------- - list | str - A list of dictionaries representing the retrieved engine actions, or a message if none found. + @dynamic_catch + async def _async_cancel_governance_action_process(self, process_instance_guid: str) -> None: + """Request that a running governance action process instance is cancelled. Async version. - Raises - ------ - PyegeriaException + Each of its engine actions that has not yet finished is cancelled and any governance service running + one is stopped; engine actions that have already finished are left unchanged. + + Parameters + ---------- + process_instance_guid : str + The GUID of the governance action process instance. Notes ----- - For more information see: https://egeria-project.org/concepts/engine-action + POST {automated-curation}/governance-action-process-instances/{processInstanceGUID}/cancel, no body. + Needs an Egeria that includes odpi/egeria#9371. + For more information see: https://egeria-project.org/concepts/governance-action-process """ - validate_guid(governance_engine_guid) - url = f"{self.ref_curation_command_base}/governance-engines/{governance_engine_guid}/engine-actions/active-claimed" - response = await self._async_make_request("GET", url, params={"startFrom": start_from, "pageSize": page_size}) - elements = response.json().get("elements", NO_ELEMENTS_FOUND) - return elements + validate_guid(process_instance_guid) + url = f"{self.ref_curation_command_base}/governance-action-process-instances/{process_instance_guid}/cancel" + await self._async_make_request("POST", url) - def get_active_claimed_engine_actions( - self, governance_engine_guid: str, start_from: int = 0, page_size: int = 0) -> list | str: - """Retrieve the engine actions that are still in process and that have been claimed by this caller's - userId. This call is used when the caller restarts.""" + def cancel_governance_action_process(self, process_instance_guid: str) -> None: + """Request that a running governance action process instance is cancelled.""" loop = asyncio.get_event_loop() - return loop.run_until_complete( - self._async_get_active_claimed_engine_actions(governance_engine_guid, start_from, page_size) - ) + loop.run_until_complete(self._async_cancel_governance_action_process(process_instance_guid)) async def _async_get_active_engine_actions( self, diff --git a/pyegeria/omvs/governance_engine.py b/pyegeria/omvs/governance_engine.py new file mode 100644 index 00000000..a6100006 --- /dev/null +++ b/pyegeria/omvs/governance_engine.py @@ -0,0 +1,256 @@ +""" +SPDX-License-Identifier: Apache-2.0 +Copyright Contributors to the ODPi Egeria project. + + Client for the Governance Engine view service. + + A governance engine that runs engine actions outside of Egeria claims the engine actions that are approved to + run on it, reports their progress and records their completion. These operations lived on the Automated + Curation view service until Egeria odpi/egeria#9371 moved them to + ``/servers/{serverName}/api/open-metadata/governance-engine``. +""" + +import asyncio +from typing import Optional + +from pyegeria.core._exceptions import PyegeriaClientException +from pyegeria.core._globals import NO_ELEMENTS_FOUND +from pyegeria.core._server_client import ServerClient +from pyegeria.core._validators import validate_guid +from pyegeria.core.utils import body_slimmer, dynamic_catch + + +class GovernanceEngine(ServerClient): + """Claim, progress and complete engine actions for a governance engine. + + Parameters + ---------- + view_server : str + The name of the View Server to connect to. + platform_url : str + URL of the server platform to connect to. + user_id : str + Default user identity for calls. + user_pwd : str, optional + Password for the user_id. If a token is supplied, this may be None. + legacy_fallback : bool, default False + If True, an operation whose governance-engine URL answers HTTP 404 (an Egeria built before #9371, which + served these operations from the automated-curation view service) is retried once on the old + automated-curation URL. Only an HTTP 404 on the URL triggers this; an Egeria-reported error (such as an + unknown engine action GUID) or any other failure is raised unchanged. Whichever URL family answered is + remembered for the life of this client (once the new URL has answered, a later 404 is raised, not retried). + """ + + def __init__( + self, + view_server: str, + platform_url: str, + user_id: str, + user_pwd: Optional[str] = None, + token: Optional[str] = None, + timeout: int = None, + legacy_fallback: bool = False, + ): + ServerClient.__init__(self, view_server, platform_url, user_id, user_pwd, token=token, timeout=timeout) + self.view_server = view_server + self.platform_url = platform_url + self.governance_engine_command_base = ( + f"{self.platform_url}/servers/{self.view_server}/api/open-metadata/governance-engine" + ) + self.command_root = self.governance_engine_command_base + self._legacy_base = ( + f"{self.platform_url}/servers/{self.view_server}/api/open-metadata/automated-curation" + ) + self.legacy_fallback = legacy_fallback + # None = not yet known, "new" or "legacy" once an attempt has succeeded / 404ed. + self._url_family: Optional[str] = None + + async def _request(self, method: str, suffix: str, body: Optional[dict] = None, params: Optional[dict] = None): + """Issue the request on the governance-engine URL, with the optional one-time 404 fallback.""" + + async def call(base: str): + url = f"{base}{suffix}" + kwargs = {"params": params} if params is not None else {} + if body is not None: + return await self._async_make_request(method, url, body, **kwargs) + return await self._async_make_request(method, url, **kwargs) + + if self._url_family == "legacy" and self.legacy_fallback: + return await call(self._legacy_base) + try: + response = await call(self.governance_engine_command_base) + except PyegeriaClientException as e: + if not (self.legacy_fallback and self._url_family is None + and getattr(e, "response_code", None) == 404): + raise + response = await call(self._legacy_base) + self._url_family = "legacy" + return response + self._url_family = "new" + return response + + @dynamic_catch + async def _async_get_approved_engine_actions( + self, governance_engine_guid: str, start_from: int = 0, page_size: int = 0) -> list | str: + """Retrieve the engine actions that are approved to run on a governance engine and that no engine has + claimed yet. Async version. + + Parameters + ---------- + governance_engine_guid : str + The GUID of the governance engine. + start_from : int, default 0 + The index from which to start fetching. + page_size : int, default 0 + The maximum number to fetch in a single request (0 = server default). + + Returns + ------- + list | str + The engine actions, or NO_ELEMENTS_FOUND if there are none. + + Notes + ----- + GET {base}/governance-engines/{governanceEngineGUID}/engine-actions/approved. There is no + automated-curation equivalent, so this never falls back. + For more information see: https://egeria-project.org/concepts/engine-action + """ + validate_guid(governance_engine_guid) + url = f"{self.governance_engine_command_base}/governance-engines/{governance_engine_guid}/engine-actions/approved" + response = await self._async_make_request("GET", url, params={"startFrom": start_from, "pageSize": page_size}) + return response.json().get("elements", NO_ELEMENTS_FOUND) + + def get_approved_engine_actions( + self, governance_engine_guid: str, start_from: int = 0, page_size: int = 0) -> list | str: + """Retrieve the engine actions approved to run on a governance engine that no engine has claimed yet.""" + loop = asyncio.get_event_loop() + return loop.run_until_complete( + self._async_get_approved_engine_actions(governance_engine_guid, start_from, page_size)) + + @dynamic_catch + async def _async_claim_engine_action(self, engine_action_guid: str) -> None: + """Request that execution of an engine action is allocated to the caller. Async version. + + POST {base}/engine-actions/{engineActionGUID}/claim (no body). + """ + validate_guid(engine_action_guid) + await self._request("POST", f"/engine-actions/{engine_action_guid}/claim") + + def claim_engine_action(self, engine_action_guid: str) -> None: + """Request that execution of an engine action is allocated to the caller.""" + loop = asyncio.get_event_loop() + loop.run_until_complete(self._async_claim_engine_action(engine_action_guid)) + + @dynamic_catch + async def _async_get_active_claimed_engine_actions( + self, governance_engine_guid: str, start_from: int = 0, page_size: int = 0) -> list | str: + """Retrieve the engine actions still in process that this caller's userId has claimed. Used when the + caller restarts. Async version. + + GET {base}/governance-engines/{governanceEngineGUID}/engine-actions/active-claimed + with startFrom / pageSize query parameters. + """ + validate_guid(governance_engine_guid) + response = await self._request( + "GET", f"/governance-engines/{governance_engine_guid}/engine-actions/active-claimed", + params={"startFrom": start_from, "pageSize": page_size}) + return response.json().get("elements", NO_ELEMENTS_FOUND) + + def get_active_claimed_engine_actions( + self, governance_engine_guid: str, start_from: int = 0, page_size: int = 0) -> list | str: + """Retrieve the engine actions still in process that this caller's userId has claimed.""" + loop = asyncio.get_event_loop() + return loop.run_until_complete( + self._async_get_active_claimed_engine_actions(governance_engine_guid, start_from, page_size)) + + @dynamic_catch + async def _async_update_engine_action_status(self, engine_action_guid: str, status: str, + body: Optional[dict] = None) -> None: + """Update the status of the engine action - providing the caller is permitted. Async version. + + POST {base}/engine-actions/{engineActionGUID}/status/update with an EngineActionStatusRequestBody. + + Parameters + ---------- + status : str + An ActivityStatus value such as "REQUESTED", "APPROVED", "IN_PROGRESS", "COMPLETED", "FAILED". + body : dict, optional + Request body passed as-is; overrides `status`. + """ + validate_guid(engine_action_guid) + payload = body if body is not None else {"class": "EngineActionStatusRequestBody", "status": status} + await self._request("POST", f"/engine-actions/{engine_action_guid}/status/update", payload) + + def update_engine_action_status(self, engine_action_guid: str, status: str, + body: Optional[dict] = None) -> None: + """Update the status of the engine action - providing the caller is permitted.""" + loop = asyncio.get_event_loop() + loop.run_until_complete(self._async_update_engine_action_status(engine_action_guid, status, body)) + + @dynamic_catch + async def _async_update_action_target_status( + self, action_target_guid: str, status: str, + start_date: Optional[str] = None, completion_date: Optional[str] = None, + completion_message: Optional[str] = None, body: Optional[dict] = None) -> None: + """Update the status of a specific action target. Async version. + + POST {base}/engine-actions/action-targets/update with an ActionTargetStatusRequestBody + (actionTargetGUID, status, startDate, completionDate, completionMessage). + `body`, if given, is passed as-is and overrides the other parameters. + """ + validate_guid(action_target_guid) + payload = body if body is not None else { + "class": "ActionTargetStatusRequestBody", + "actionTargetGUID": action_target_guid, + "status": status, + "startDate": start_date, + "completionDate": completion_date, + "completionMessage": completion_message, + } + await self._request("POST", "/engine-actions/action-targets/update", body_slimmer(payload)) + + def update_action_target_status( + self, action_target_guid: str, status: str, + start_date: Optional[str] = None, completion_date: Optional[str] = None, + completion_message: Optional[str] = None, body: Optional[dict] = None) -> None: + """Update the status of a specific action target.""" + loop = asyncio.get_event_loop() + loop.run_until_complete(self._async_update_action_target_status( + action_target_guid, status, start_date, completion_date, completion_message, body)) + + @dynamic_catch + async def _async_record_completion_status( + self, engine_action_guid: str, status: str, + request_parameters: Optional[dict] = None, output_guards: Optional[list[str]] = None, + new_action_targets: Optional[list[dict]] = None, completion_message: Optional[str] = None, + body: Optional[dict] = None) -> None: + """Declare that all processing for the engine action is finished, and the status of the work. Async + version. + + POST {base}/engine-actions/{engineActionGUID}/completion-status with a CompletionStatusRequestBody. + If the engine action is part of a governance action process, the output guards decide which steps run + next. `status` is one of "ACTIONED", "INVALID", "FAILED" or "OTHER". + `body`, if given, is passed as-is and overrides the other parameters. + """ + validate_guid(engine_action_guid) + payload = body if body is not None else { + "class": "CompletionStatusRequestBody", + "status": status, + "requestParameters": request_parameters, + "outputGuards": output_guards, + "newActionTargets": new_action_targets, + "completionMessage": completion_message, + } + await self._request("POST", f"/engine-actions/{engine_action_guid}/completion-status", + body_slimmer(payload)) + + def record_completion_status( + self, engine_action_guid: str, status: str, + request_parameters: Optional[dict] = None, output_guards: Optional[list[str]] = None, + new_action_targets: Optional[list[dict]] = None, completion_message: Optional[str] = None, + body: Optional[dict] = None) -> None: + """Declare that all processing for the engine action is finished, and the status of the work.""" + loop = asyncio.get_event_loop() + loop.run_until_complete(self._async_record_completion_status( + engine_action_guid, status, request_parameters, output_guards, new_action_targets, + completion_message, body)) diff --git a/tests/micro-tests/test_issue_153_governance_engine.py b/tests/micro-tests/test_issue_153_governance_engine.py new file mode 100644 index 00000000..fea6b4ee --- /dev/null +++ b/tests/micro-tests/test_issue_153_governance_engine.py @@ -0,0 +1,181 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright Contributors to the ODPi Egeria project. +""" +ISSUE-153: Egeria #9371 moved engine-action claim/status/completion operations from the automated-curation +view service to a new governance-engine one. No live server: `_async_make_request` is replaced with a recorder. +""" +import warnings +from unittest.mock import MagicMock, patch + +import pytest + +from pyegeria.core._exceptions import PyegeriaClientException, PyegeriaNotFoundException +from pyegeria.omvs.automated_curation import AutomatedCuration +from pyegeria.omvs.governance_engine import GovernanceEngine + +ROOT = "https://localhost:9443/servers/vs/api/open-metadata" +NEW = f"{ROOT}/governance-engine" +OLD = f"{ROOT}/automated-curation" +G = "11111111-1111-1111-1111-111111111111" +G2 = "22222222-2222-2222-2222-222222222222" + + +def _client(cls, **kw): + with patch("pyegeria.core._base_server_client.BaseServerClient.check_connection", return_value=""): + return cls(view_server="vs", platform_url="https://localhost:9443", user_id="u", user_pwd="p", **kw) + + +def _err(code): + resp = MagicMock() + resp.status_code = code + resp.text = "" + resp.json = MagicMock(return_value={}) + return PyegeriaClientException(resp, {}, {"userid": "u"}, Exception("boom")) + + +def _install(client, outcomes=None, elements=None): + """Record calls; `outcomes` is a list of exceptions/None consumed per call.""" + calls = [] + outcomes = list(outcomes or []) + + async def fake(method, url, body=None, **kwargs): + calls.append({"method": method, "url": url, "body": body, "params": kwargs.get("params")}) + out = outcomes.pop(0) if outcomes else None + if isinstance(out, Exception): + raise out + resp = MagicMock() + resp.json = MagicMock(return_value={"elements": elements} if elements is not None else {}) + return resp + + client._async_make_request = fake + return calls + + +def _all_ge_calls(ge): + return [ + (lambda: ge.claim_engine_action(G), "POST", f"/engine-actions/{G}/claim", None, None), + (lambda: ge.update_engine_action_status(G, "IN_PROGRESS"), "POST", f"/engine-actions/{G}/status/update", + {"class": "EngineActionStatusRequestBody", "status": "IN_PROGRESS"}, None), + (lambda: ge.update_action_target_status(G2, "COMPLETED", completion_message="ok"), "POST", + "/engine-actions/action-targets/update", + {"class": "ActionTargetStatusRequestBody", "actionTargetGUID": G2, "status": "COMPLETED", + "completionMessage": "ok"}, None), + (lambda: ge.record_completion_status(G, "ACTIONED", output_guards=["g1"]), "POST", + f"/engine-actions/{G}/completion-status", + {"class": "CompletionStatusRequestBody", "status": "ACTIONED", "outputGuards": ["g1"]}, None), + (lambda: ge.get_active_claimed_engine_actions(G, 2, 5), "GET", + f"/governance-engines/{G}/engine-actions/active-claimed", None, {"startFrom": 2, "pageSize": 5}), + ] + + +@pytest.mark.parametrize("i", range(5)) +def test_governance_engine_urls(i): + ge = _client(GovernanceEngine) + calls = _install(ge, elements=[]) + fn, method, suffix, body, params = _all_ge_calls(ge)[i] + fn() + assert len(calls) == 1 + c = calls[0] + assert (c["method"], c["url"], c["params"]) == (method, NEW + suffix, params) + if body is not None: + assert {k: v for k, v in c["body"].items() if v is not None} == body + + +def test_get_approved(): + ge = _client(GovernanceEngine) + calls = _install(ge, elements=[{"x": 1}]) + assert ge.get_approved_engine_actions(G) == [{"x": 1}] + assert calls[0]["method"] == "GET" + assert calls[0]["url"] == f"{NEW}/governance-engines/{G}/engine-actions/approved" + assert calls[0]["params"] == {"startFrom": 0, "pageSize": 0} + calls2 = _install(ge) + assert ge.get_approved_engine_actions(G, 1, 3) == "No elements found" + assert calls2[0]["params"] == {"startFrom": 1, "pageSize": 3} + + +def test_cancel_governance_action_process(): + ac = _client(AutomatedCuration) + calls = _install(ac) + ac.cancel_governance_action_process(G) + assert calls == [{"method": "POST", "url": f"{OLD}/governance-action-process-instances/{G}/cancel", + "body": None, "params": None}] + + +@pytest.mark.parametrize("i", range(5)) +def test_automated_curation_wrappers_use_new_url_and_warn(i): + ac = _client(AutomatedCuration) + ge = ac._governance_engine_client() + calls = _install(ge, elements=[]) + fn, method, suffix, body, params = _all_ge_calls(ac)[i] + with pytest.warns(DeprecationWarning): + fn() + assert calls[0]["url"] == NEW + suffix and calls[0]["method"] == method + + +def test_404_falls_back_once_and_caches(): + ac = _client(AutomatedCuration) + ge = ac._governance_engine_client() + calls = _install(ge, outcomes=[_err(404)]) + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + ac.claim_engine_action(G) + assert [c["url"] for c in calls] == [f"{NEW}/engine-actions/{G}/claim", f"{OLD}/engine-actions/{G}/claim"] + ac.update_engine_action_status(G, "COMPLETED") + # cached: the next call goes straight to the old path, no new-path probe + assert calls[2]["url"] == f"{OLD}/engine-actions/{G}/status/update" + assert len(calls) == 3 + assert calls[2]["body"]["status"] == "COMPLETED" + + +def test_new_path_success_is_cached_and_never_probes_old(): + ac = _client(AutomatedCuration) + ge = ac._governance_engine_client() + calls = _install(ge, outcomes=[None, _err(404)]) + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + ac.claim_engine_action(G) + with pytest.raises(PyegeriaClientException): + ac.claim_engine_action(G) # once the new URL has worked a 404 is raised, not retried + assert len(calls) == 2 and all(c["url"].startswith(NEW) for c in calls) + + +@pytest.mark.parametrize("exc", [lambda: _err(500), lambda: _err(403), lambda: _err(401)]) +def test_non_404_error_does_not_fall_back(exc): + ac = _client(AutomatedCuration) + ge = ac._governance_engine_client() + calls = _install(ge, outcomes=[exc()]) + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + with pytest.raises(PyegeriaClientException): + ac.claim_engine_action(G) + assert len(calls) == 1 and calls[0]["url"].startswith(NEW) + + +def test_egeria_reported_not_found_does_not_fall_back(): + ac = _client(AutomatedCuration) + ge = ac._governance_engine_client() + resp = MagicMock() + resp.status_code = 200 + resp.json = MagicMock(return_value={"relatedHTTPCode": 404}) + nf = PyegeriaNotFoundException(resp, {}, additional_info={"relatedHTTPCode": 404}) + calls = _install(ge, outcomes=[nf]) + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + with pytest.raises(PyegeriaNotFoundException): + ac.claim_engine_action(G) + assert len(calls) == 1 + + +def test_direct_governance_engine_never_falls_back(): + ge = _client(GovernanceEngine) + calls = _install(ge, outcomes=[_err(404)]) + with pytest.raises(PyegeriaClientException): + ge.claim_engine_action(G) + assert len(calls) == 1 + + +def test_egeria_client_proxy_exposes_gov_engine(): + from pyegeria.egeria_tech_client import EgeriaTech + assert EgeriaTech.__init__ # import works + import pyegeria + assert pyegeria.GovernanceEngine is GovernanceEngine From bfbcf3bbabda58cc9351ef71814c10b6192f6195 Mon Sep 17 00:00:00 2001 From: Dan Wolfson Date: Fri, 16 Oct 2026 16:27:32 -0500 Subject: [PATCH 2/2] test: ISSUE-153 tests skip the live connection check for every client they build The GovernanceEngine an AutomatedCuration wrapper builds lazily ran the real BaseServerClient.check_connection, so the tests passed only where an Egeria answered on localhost:9443 and failed in CI with httpx.ConnectError. An autouse fixture now patches check_connection for the whole file. Verified in a clean python:3.13 container: 20 passed (failed before), micro-tests -m unit 1192 passed. Signed-off-by: Dan Wolfson --- tests/micro-tests/test_issue_153_governance_engine.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/tests/micro-tests/test_issue_153_governance_engine.py b/tests/micro-tests/test_issue_153_governance_engine.py index fea6b4ee..d5fd9b1b 100644 --- a/tests/micro-tests/test_issue_153_governance_engine.py +++ b/tests/micro-tests/test_issue_153_governance_engine.py @@ -20,6 +20,14 @@ G2 = "22222222-2222-2222-2222-222222222222" +@pytest.fixture(autouse=True) +def _no_live_connection_check(): + """Every client this file builds, including the GovernanceEngine an AutomatedCuration wrapper creates lazily, + skips the live connection check: CI has no Egeria on localhost:9443.""" + with patch("pyegeria.core._base_server_client.BaseServerClient.check_connection", return_value=""): + yield + + def _client(cls, **kw): with patch("pyegeria.core._base_server_client.BaseServerClient.check_connection", return_value=""): return cls(view_server="vs", platform_url="https://localhost:9443", user_id="u", user_pwd="p", **kw)