From 539848d3a656d8fca3a97e90637a5db9cf22b2f2 Mon Sep 17 00:00:00 2001 From: Chris Busillo Date: Mon, 25 May 2026 21:34:34 -0400 Subject: [PATCH] Move Odoo generic post-deploy adapter out of service --- control_plane/service.py | 37 +----- .../workflows/odoo_generic_web_post_deploy.py | 52 ++++++++ tests/test_odoo_generic_web_post_deploy.py | 124 ++++++++++++++++++ tests/test_service.py | 88 +------------ 4 files changed, 185 insertions(+), 116 deletions(-) create mode 100644 control_plane/workflows/odoo_generic_web_post_deploy.py create mode 100644 tests/test_odoo_generic_web_post_deploy.py diff --git a/control_plane/service.py b/control_plane/service.py index 74538c6..b2f689b 100644 --- a/control_plane/service.py +++ b/control_plane/service.py @@ -283,8 +283,6 @@ ) from control_plane.workflows.generic_web_deploy import ( GenericWebDeployRequest, - GenericWebDeployStore, - GenericWebPostDeployContext, GenericWebPostDeployExecutor, execute_generic_web_deploy, ) @@ -350,9 +348,11 @@ ) from control_plane.workflows.odoo_post_deploy import ( OdooPostDeployRequest, - OdooPostDeployResult, execute_odoo_post_deploy, ) +from control_plane.workflows.odoo_generic_web_post_deploy import ( + generic_web_post_deploy_executor_for_driver_id, +) from control_plane.contracts.odoo_stable_bootstrap import ( OdooStableBootstrapRequest, OdooStableBootstrapResult, @@ -732,7 +732,6 @@ class _ResolvedProductDriverContext: lane: ProductLaneProfile | None = None -_ODOO_GENERIC_WEB_POST_DEPLOY_DRIVERS = frozenset({"odoo"}) _LAUNCHPLANE_IMAGE_REFERENCE_ENV_KEY = "DOCKER_IMAGE_REFERENCE" _LOGGER = logging.getLogger(__name__) _LAUNCHPLANE_SELF_DEPLOY_OAUTH_ENV_KEYS = frozenset( @@ -6426,38 +6425,10 @@ def _product_driver_route_compatible( ) -def _post_deploy_evidence_from_odoo_result( - result: OdooPostDeployResult, -) -> PostDeployUpdateEvidence: - return PostDeployUpdateEvidence( - attempted=True, - status=result.post_deploy_status, - detail=( - result.error_message - or "Odoo post-deploy completed through the generic-web extension hook." - ), - ) - - -def _execute_odoo_generic_web_post_deploy( - control_plane_root: Path, - record_store: GenericWebDeployStore, - context: GenericWebPostDeployContext, -) -> PostDeployUpdateEvidence: - result = execute_odoo_post_deploy( - control_plane_root=control_plane_root, - record_store=record_store, - request=OdooPostDeployRequest(context=context.context, instance=context.instance), - ) - return _post_deploy_evidence_from_odoo_result(result) - - def _generic_web_post_deploy_executor_for_profile( profile: LaunchplaneProductProfileRecord, ) -> GenericWebPostDeployExecutor | None: - if profile.driver_id.strip() in _ODOO_GENERIC_WEB_POST_DEPLOY_DRIVERS: - return _execute_odoo_generic_web_post_deploy - return None + return generic_web_post_deploy_executor_for_driver_id(profile.driver_id) def _find_product_profile_lane( diff --git a/control_plane/workflows/odoo_generic_web_post_deploy.py b/control_plane/workflows/odoo_generic_web_post_deploy.py new file mode 100644 index 0000000..33ebc5f --- /dev/null +++ b/control_plane/workflows/odoo_generic_web_post_deploy.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +from pathlib import Path + +from control_plane.contracts.promotion_record import PostDeployUpdateEvidence +from control_plane.workflows.generic_web_deploy import ( + GenericWebDeployStore, + GenericWebPostDeployContext, + GenericWebPostDeployExecutor, +) +from control_plane.workflows.odoo_post_deploy import ( + OdooPostDeployRequest, + OdooPostDeployResult, + execute_odoo_post_deploy, +) + + +_ODOO_GENERIC_WEB_POST_DEPLOY_DRIVERS = frozenset({"odoo"}) + + +def generic_web_post_deploy_executor_for_driver_id( + driver_id: str, +) -> GenericWebPostDeployExecutor | None: + if driver_id.strip() in _ODOO_GENERIC_WEB_POST_DEPLOY_DRIVERS: + return execute_odoo_generic_web_post_deploy + return None + + +def post_deploy_evidence_from_odoo_result( + result: OdooPostDeployResult, +) -> PostDeployUpdateEvidence: + return PostDeployUpdateEvidence( + attempted=True, + status=result.post_deploy_status, + detail=( + result.error_message + or "Odoo post-deploy completed through the generic-web extension hook." + ), + ) + + +def execute_odoo_generic_web_post_deploy( + control_plane_root: Path, + record_store: GenericWebDeployStore, + context: GenericWebPostDeployContext, +) -> PostDeployUpdateEvidence: + result = execute_odoo_post_deploy( + control_plane_root=control_plane_root, + record_store=record_store, + request=OdooPostDeployRequest(context=context.context, instance=context.instance), + ) + return post_deploy_evidence_from_odoo_result(result) diff --git a/tests/test_odoo_generic_web_post_deploy.py b/tests/test_odoo_generic_web_post_deploy.py new file mode 100644 index 0000000..14909b7 --- /dev/null +++ b/tests/test_odoo_generic_web_post_deploy.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +from pathlib import Path +from tempfile import TemporaryDirectory +from typing import cast +import unittest +from unittest.mock import patch + +from control_plane.contracts.product_profile_record import ( + LaunchplaneProductProfileRecord, + ProductImageProfile, + ProductLaneProfile, +) +from control_plane.storage.filesystem import FilesystemRecordStore +from control_plane.workflows.generic_web_deploy import ( + GenericWebDeployStore, + GenericWebPostDeployContext, +) +from control_plane.workflows.odoo_generic_web_post_deploy import ( + execute_odoo_generic_web_post_deploy, + generic_web_post_deploy_executor_for_driver_id, +) +from control_plane.workflows.odoo_post_deploy import OdooPostDeployResult + + +class OdooGenericWebPostDeployTests(unittest.TestCase): + def test_selects_odoo_driver_executor(self) -> None: + self.assertIs( + generic_web_post_deploy_executor_for_driver_id(" odoo "), + execute_odoo_generic_web_post_deploy, + ) + self.assertIsNone(generic_web_post_deploy_executor_for_driver_id("generic-web")) + + def test_returns_terminal_evidence(self) -> None: + context = _context() + + with TemporaryDirectory() as temporary_directory_name: + store = _store(Path(temporary_directory_name)) + with patch( + "control_plane.workflows.odoo_generic_web_post_deploy.execute_odoo_post_deploy", + return_value=OdooPostDeployResult( + context="cm", + instance="prod", + phase="deploy", + post_deploy_status="pass", + override_status="pass", + ), + ) as post_deploy: + evidence = execute_odoo_generic_web_post_deploy( + Path("."), + cast(GenericWebDeployStore, store), + context, + ) + + self.assertTrue(evidence.attempted) + self.assertEqual(evidence.status, "pass") + self.assertIn("generic-web extension hook", evidence.detail) + post_deploy.assert_called_once() + request = post_deploy.call_args.kwargs["request"] + self.assertEqual(request.context, "cm") + self.assertEqual(request.instance, "prod") + + def test_preserves_failure_detail(self) -> None: + context = _context() + + with TemporaryDirectory() as temporary_directory_name: + store = _store(Path(temporary_directory_name)) + with patch( + "control_plane.workflows.odoo_generic_web_post_deploy.execute_odoo_post_deploy", + return_value=OdooPostDeployResult( + context="cm", + instance="prod", + phase="deploy", + post_deploy_status="fail", + override_status="fail", + error_message="override failed", + ), + ): + evidence = execute_odoo_generic_web_post_deploy( + Path("."), + cast(GenericWebDeployStore, store), + context, + ) + + self.assertTrue(evidence.attempted) + self.assertEqual(evidence.status, "fail") + self.assertEqual(evidence.detail, "override failed") + + +def _context() -> GenericWebPostDeployContext: + return GenericWebPostDeployContext( + product="odoo-tenant-cm", + context="cm", + instance="prod", + deployment_record_id="deployment-cm-prod", + target_name="cm-prod", + target_type="compose", + target_id="compose-cm-prod", + artifact_id="ghcr.io/cbusillo/odoo-tenant-cm@sha256:abc123", + source_git_ref="abc123", + ) + + +def _store(root: Path) -> FilesystemRecordStore: + store = FilesystemRecordStore(state_dir=root / "state") + store.write_product_profile_record( + LaunchplaneProductProfileRecord( + product="odoo-tenant-cm", + display_name="Odoo Tenant CM", + repository="cbusillo/odoo-tenant-cm", + driver_id="odoo", + image=ProductImageProfile(repository="ghcr.io/cbusillo/odoo-tenant-cm"), + runtime_port=8069, + health_path="/web/health", + lanes=(ProductLaneProfile(instance="prod", context="cm"),), + updated_at="2026-05-26T00:00:00Z", + source="test", + ) + ) + return store + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_service.py b/tests/test_service.py index 1e06073..fc97611 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -143,6 +143,9 @@ from control_plane.workflows.odoo_artifact_publish import OdooArtifactPublishResult from control_plane.contracts.odoo_stable_bootstrap import OdooStableBootstrapResult from control_plane.workflows.odoo_post_deploy import OdooPostDeployResult +from control_plane.workflows.odoo_generic_web_post_deploy import ( + execute_odoo_generic_web_post_deploy, +) from control_plane.workflows.odoo_prod_backup_gate import OdooProdBackupGateResult from control_plane.workflows.odoo_prod_promotion import OdooProdPromotionResult from control_plane.workflows.odoo_prod_rollback import OdooProdRollbackResult @@ -165,10 +168,6 @@ GenericWebPreviewSmokeResult, ) from control_plane.workflows.odoo_preview_runtime import OdooPreviewDokployApplyResult -from control_plane.workflows.generic_web_deploy import ( - GenericWebDeployStore, - GenericWebPostDeployContext, -) StartResponse = Callable[[str, list[tuple[str, str]]], None] WsgiApp = Callable[[dict[str, object], StartResponse], Iterable[bytes]] @@ -13649,7 +13648,7 @@ def test_generic_web_deploy_route_accepts_base_driver_product(self) -> None: self.assertEqual(kwargs["lane"].context, "sellyouroutboard-testing") self.assertIs( kwargs["post_deploy_executor"], - control_plane_service._execute_odoo_generic_web_post_deploy, + execute_odoo_generic_web_post_deploy, ) def test_generic_web_deploy_route_keeps_literal_generic_products_without_post_deploy_adapter( @@ -14131,83 +14130,6 @@ def test_odoo_rollback_plan_alias_rejects_unauthorized_context(self) -> None: self.assertEqual(status_code, 403) self.assertEqual(payload["error"]["code"], "authorization_denied") - def test_odoo_generic_web_post_deploy_adapter_returns_terminal_evidence(self) -> None: - with TemporaryDirectory() as temporary_directory_name: - root = Path(temporary_directory_name) - store = FilesystemRecordStore(state_dir=root / "state") - context = GenericWebPostDeployContext( - product="odoo-tenant-cm", - context="cm", - instance="prod", - deployment_record_id="deployment-cm-prod", - target_name="cm-prod", - target_type="compose", - target_id="compose-cm-prod", - artifact_id="ghcr.io/cbusillo/odoo-tenant-cm@sha256:abc123", - source_git_ref="abc123", - ) - - with patch( - "control_plane.service.execute_odoo_post_deploy", - return_value=OdooPostDeployResult( - context="cm", - instance="prod", - phase="deploy", - post_deploy_status="pass", - override_status="pass", - ), - ) as post_deploy: - evidence = control_plane_service._execute_odoo_generic_web_post_deploy( - root, - cast(GenericWebDeployStore, store), - context, - ) - - self.assertTrue(evidence.attempted) - self.assertEqual(evidence.status, "pass") - self.assertIn("generic-web extension hook", evidence.detail) - post_deploy.assert_called_once() - request = post_deploy.call_args.kwargs["request"] - self.assertEqual(request.context, "cm") - self.assertEqual(request.instance, "prod") - - def test_odoo_generic_web_post_deploy_adapter_preserves_failure_detail(self) -> None: - with TemporaryDirectory() as temporary_directory_name: - root = Path(temporary_directory_name) - store = FilesystemRecordStore(state_dir=root / "state") - context = GenericWebPostDeployContext( - product="odoo-tenant-cm", - context="cm", - instance="prod", - deployment_record_id="deployment-cm-prod", - target_name="cm-prod", - target_type="compose", - target_id="compose-cm-prod", - artifact_id="ghcr.io/cbusillo/odoo-tenant-cm@sha256:abc123", - source_git_ref="abc123", - ) - - with patch( - "control_plane.service.execute_odoo_post_deploy", - return_value=OdooPostDeployResult( - context="cm", - instance="prod", - phase="deploy", - post_deploy_status="fail", - override_status="fail", - error_message="override failed", - ), - ): - evidence = control_plane_service._execute_odoo_generic_web_post_deploy( - root, - cast(GenericWebDeployStore, store), - context, - ) - - self.assertTrue(evidence.attempted) - self.assertEqual(evidence.status, "fail") - self.assertEqual(evidence.detail, "override failed") - def test_generic_web_rollback_route_applies_ready_plan(self) -> None: with TemporaryDirectory() as temporary_directory_name: root = Path(temporary_directory_name) @@ -14363,7 +14285,7 @@ def test_generic_web_rollback_route_passes_odoo_post_deploy_adapter_for_odoo_pro rollback.assert_called_once() self.assertIs( rollback.call_args.kwargs["post_deploy_executor"], - control_plane_service._execute_odoo_generic_web_post_deploy, + execute_odoo_generic_web_post_deploy, ) def test_generic_web_rollback_route_replays_idempotent_response_shape(self) -> None: