Move Odoo generic post-deploy adapter out of service

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(

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)

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

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: