diff --git a/scripts/langflow_deployments_api/adapters/deployment_watsonx_adapter_e2e.py b/scripts/e2e_deployment_tests/watsonx_orchestrate/adapter.py similarity index 82% rename from scripts/langflow_deployments_api/adapters/deployment_watsonx_adapter_e2e.py rename to scripts/e2e_deployment_tests/watsonx_orchestrate/adapter.py index 1b12c7e0f7..c5f2d88ad9 100644 --- a/scripts/langflow_deployments_api/adapters/deployment_watsonx_adapter_e2e.py +++ b/scripts/e2e_deployment_tests/watsonx_orchestrate/adapter.py @@ -41,14 +41,19 @@ Live lifecycle scenarios: - `live_status_after_delete_not_found_state`: confirms status on deleted deployment returns not found (expects DeploymentNotFoundError). -Live list-snapshots-by-ids scenarios: +Live snapshot/config listing scenarios: - `live_list_snapshots_by_ids_returns_known`: fetches known snapshot ids via snapshot_ids mode and confirms all are returned (expects Success). - `live_list_snapshots_by_ids_filters_unknown`: mixes a known id with a bogus id; confirms the provider returns only existing snapshots (expects Success). -- `live_list_snapshots_by_ids_empty_input`: passes an empty list; the adapter treats - this as "no snapshot filter" and falls through to the deployment-scoped branch, - which rejects because no deployment_id is provided (expects DeploymentError). +- `live_list_snapshots_by_ids_empty_input`: passes an empty list; this normalizes to + tenant-scoped snapshot listing and should still succeed (expects Success). +- `live_list_snapshots_tenant_scope`: lists tenant-scoped snapshots (expects Success). +- `live_list_configs_tenant_scope`: lists tenant-scoped configs (expects Success). +- `live_list_snapshots_by_names_returns_known`: queries snapshot_names mode and + confirms known names resolve to known snapshot ids (expects Success). +- `live_list_snapshots_by_names_ignored_with_deployment_scope`: passes both deployment_ids + and snapshot_names, and confirms deployment scope takes precedence (expects Success). Live negative scenarios: - `live_negative_create_seed`: creates a second seed deployment for negative-path checks (expects Success). @@ -58,9 +63,23 @@ Live negative scenarios: - `live_delete_missing_not_found`: delete on unknown deployment id returns not found (expects DeploymentNotFoundError). - `live_negative_delete_seed`: cleans up negative-path seed deployment (expects Success). +Live service-surface scenarios: +- `live_list_types_supports_agent`: lists supported deployment types and validates AGENT is present (expects Success). +- `live_list_llms_returns_models`: lists provider models and validates the normalized payload + is non-empty (expects Success). +- `live_verify_credentials_success`: verifies configured credentials against the provider instance (expects Success). +- `live_update_snapshot_success`: updates an existing snapshot artifact by id (expects Success). +- `live_rollback_create_result_cleans_up_created`: runs create rollback cleanup using a real create result + and verifies the created deployment is removed (expects Success). +- `live_redeploy_not_supported`: ensures redeploy returns operation-not-supported semantics + (expects InvalidDeploymentOperationError). +- `live_duplicate_not_supported`: ensures duplicate returns operation-not-supported semantics + (expects InvalidDeploymentOperationError). +- `live_teardown_noop`: calls adapter teardown and expects a successful no-op (expects Success). + Live update-matrix scenarios: -- Contract note: in provider_data operations, `app_ids` are unprefixed operation ids. - `resource_name_prefix` is applied only when raw resources are created in the provider. +- Contract note: in provider_data operations, `app_ids` are operation ids. + Raw connection `app_id` values are preserved exactly as declared. - `upd_spec_only_name_desc`: updates deployment metadata only (expects Success). - `upd_snapshot_remove_only_no_config`: removes an attached snapshot via provider_data operation (expects Success). @@ -146,6 +165,7 @@ from lfx.services.adapters.deployment.exceptions import ( InvalidContentError, InvalidDeploymentOperationError, InvalidDeploymentTypeError, + OperationNotSupportedError, ResourceConflictError, ) from lfx.services.adapters.deployment.schema import ( @@ -160,6 +180,7 @@ from lfx.services.adapters.deployment.schema import ( DeploymentUpdate, ExecutionCreate, SnapshotListParams, + VerifyCredentials, ) OUTCOME_SUCCESS = "Success" @@ -175,6 +196,7 @@ DEFAULT_CONCURRENCY_ITERATIONS = 1 EXECUTION_POLL_INTERVAL_SECS = 2 EXECUTION_POLL_MAX_ATTEMPTS = 10 EXECUTION_TERMINAL_STATUSES = {"completed", "failed", "cancelled", "async_completed", "expired", "requires_input"} +DEFAULT_WXO_LLM = "groq/openai/gpt-oss-120b" _INVALID_WXO_NAME_CHARS = re.compile(r"[^A-Za-z0-9_]") @@ -211,12 +233,14 @@ class WatsonxAdapterDirectE2E: project_id: str, mode: str, keep_resources: bool, + llm: str, ) -> None: self.provider_backend_url = provider_backend_url self.provider_api_key = provider_api_key self.project_id = project_id self.mode = mode self.keep_resources = keep_resources + self.llm = llm self.run_suffix = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S") + "-" + uuid4().hex[:8] self.user_id = str(uuid4()) @@ -249,7 +273,7 @@ class WatsonxAdapterDirectE2E: async def run(self) -> int: print("Starting watsonx direct adapter runner...") - print(f"mode={self.mode} project_id={self.project_id} keep_resources={self.keep_resources}") + print(f"mode={self.mode} project_id={self.project_id} keep_resources={self.keep_resources} llm={self.llm}") try: results: list[ScenarioResult] = [] if self.mode in {"live", "both"}: @@ -301,6 +325,8 @@ class WatsonxAdapterDirectE2E: results = await self._run_scenarios(scenarios) results.extend(await self._run_live_lifecycle_scenarios()) results.extend(await self._run_live_list_snapshots_by_ids_scenarios()) + results.extend(await self._run_live_listing_mode_scenarios()) + results.extend(await self._run_live_service_surface_scenarios()) results.extend(await self._run_live_update_matrix_scenarios()) results.extend(await self._run_live_concurrency_scenarios()) results.extend(await self._run_live_negative_scenarios()) @@ -529,11 +555,16 @@ class WatsonxAdapterDirectE2E: for target, attr_name, original in originals: setattr(target, attr_name, original) - async def _run_list_snapshots(self, deployment_id: str) -> tuple[str, str, Any | None]: + async def _run_list_snapshots_with_params( + self, + *, + params: SnapshotListParams | None, + detail_label: str = "snapshots_listed", + ) -> tuple[str, str, Any | None]: try: result = await self.service.list_snapshots( user_id=self.user_id, - params=SnapshotListParams(deployment_ids=[deployment_id]), + params=params, db=self.db, ) except DeploymentNotFoundError as exc: @@ -551,37 +582,36 @@ class WatsonxAdapterDirectE2E: except Exception as exc: # noqa: BLE001 return OUTCOME_FAILURE, str(exc), None else: - return OUTCOME_SUCCESS, "snapshots_listed", result + return OUTCOME_SUCCESS, detail_label, result + + async def _run_list_snapshots(self, deployment_id: str) -> tuple[str, str, Any | None]: + return await self._run_list_snapshots_with_params( + params=SnapshotListParams(deployment_ids=[deployment_id]), + detail_label="snapshots_listed", + ) async def _run_list_snapshots_by_ids(self, snapshot_ids: list[str]) -> tuple[str, str, Any | None]: - try: - result = await self.service.list_snapshots( - user_id=self.user_id, - params=SnapshotListParams(snapshot_ids=snapshot_ids), - db=self.db, - ) - except DeploymentNotFoundError as exc: - return OUTCOME_NOT_FOUND, str(exc), None - except ResourceConflictError as exc: - return OUTCOME_CONFLICT, exc.message, None - except InvalidContentError as exc: - return OUTCOME_INVALID_CONTENT, exc.message, None - except (InvalidDeploymentOperationError, InvalidDeploymentTypeError) as exc: - return OUTCOME_INVALID_OPERATION, exc.message, None - except DeploymentError as exc: - return OUTCOME_FAILURE, exc.message, None - except HTTPException as exc: - return self._outcome_from_http_exception(exc), str(exc.detail), None - except Exception as exc: # noqa: BLE001 - return OUTCOME_FAILURE, str(exc), None - else: - return OUTCOME_SUCCESS, "snapshots_by_ids_listed", result + return await self._run_list_snapshots_with_params( + params=SnapshotListParams(snapshot_ids=snapshot_ids), + detail_label="snapshots_by_ids_listed", + ) - async def _run_list_configs(self, deployment_id: str) -> tuple[str, str, Any | None]: + async def _run_list_snapshots_by_names(self, snapshot_names: list[str]) -> tuple[str, str, Any | None]: + return await self._run_list_snapshots_with_params( + params=SnapshotListParams(snapshot_names=snapshot_names), + detail_label="snapshots_by_names_listed", + ) + + async def _run_list_configs_with_params( + self, + *, + params: ConfigListParams | None, + detail_label: str = "configs_listed", + ) -> tuple[str, str, Any | None]: try: result = await self.service.list_configs( user_id=self.user_id, - params=ConfigListParams(deployment_ids=[deployment_id]), + params=params, db=self.db, ) except DeploymentNotFoundError as exc: @@ -599,7 +629,204 @@ class WatsonxAdapterDirectE2E: except Exception as exc: # noqa: BLE001 return OUTCOME_FAILURE, str(exc), None else: - return OUTCOME_SUCCESS, "configs_listed", result + return OUTCOME_SUCCESS, detail_label, result + + async def _run_list_configs(self, deployment_id: str) -> tuple[str, str, Any | None]: + return await self._run_list_configs_with_params( + params=ConfigListParams(deployment_ids=[deployment_id]), + detail_label="configs_listed", + ) + + async def _run_list_types(self) -> tuple[str, str, Any | None]: + try: + result = await self.service.list_types(user_id=self.user_id, db=self.db) + except DeploymentNotFoundError as exc: + return OUTCOME_NOT_FOUND, str(exc), None + except ResourceConflictError as exc: + return OUTCOME_CONFLICT, exc.message, None + except InvalidContentError as exc: + return OUTCOME_INVALID_CONTENT, exc.message, None + except (InvalidDeploymentOperationError, InvalidDeploymentTypeError, OperationNotSupportedError) as exc: + return OUTCOME_INVALID_OPERATION, exc.message, None + except DeploymentError as exc: + return OUTCOME_FAILURE, exc.message, None + except HTTPException as exc: + return self._outcome_from_http_exception(exc), str(exc.detail), None + except Exception as exc: # noqa: BLE001 + return OUTCOME_FAILURE, str(exc), None + else: + return OUTCOME_SUCCESS, "types_listed", result + + async def _run_list_llms(self) -> tuple[str, str, Any | None]: + try: + result = await self.service.list_llms(user_id=self.user_id, db=self.db) + except DeploymentNotFoundError as exc: + return OUTCOME_NOT_FOUND, str(exc), None + except ResourceConflictError as exc: + return OUTCOME_CONFLICT, exc.message, None + except InvalidContentError as exc: + return OUTCOME_INVALID_CONTENT, exc.message, None + except (InvalidDeploymentOperationError, InvalidDeploymentTypeError, OperationNotSupportedError) as exc: + return OUTCOME_INVALID_OPERATION, exc.message, None + except DeploymentError as exc: + return OUTCOME_FAILURE, exc.message, None + except HTTPException as exc: + return self._outcome_from_http_exception(exc), str(exc.detail), None + except Exception as exc: # noqa: BLE001 + return OUTCOME_FAILURE, str(exc), None + else: + return OUTCOME_SUCCESS, "llms_listed", result + + async def _run_verify_credentials(self, *, base_url: str, api_key: str) -> tuple[str, str, Any | None]: + try: + result = await self.service.verify_credentials( + user_id=self.user_id, + payload=VerifyCredentials(base_url=base_url, provider_data={"api_key": api_key}), + ) + except DeploymentNotFoundError as exc: + return OUTCOME_NOT_FOUND, str(exc), None + except ResourceConflictError as exc: + return OUTCOME_CONFLICT, exc.message, None + except InvalidContentError as exc: + return OUTCOME_INVALID_CONTENT, exc.message, None + except (InvalidDeploymentOperationError, InvalidDeploymentTypeError, OperationNotSupportedError) as exc: + return OUTCOME_INVALID_OPERATION, exc.message, None + except DeploymentError as exc: + return OUTCOME_FAILURE, exc.message, None + except HTTPException as exc: + return self._outcome_from_http_exception(exc), str(exc.detail), None + except Exception as exc: # noqa: BLE001 + return OUTCOME_FAILURE, str(exc), None + else: + return OUTCOME_SUCCESS, "credentials_verified", result + + async def _run_update_snapshot( + self, + *, + snapshot_id: str, + flow_artifact: BaseFlowArtifact[WatsonxFlowArtifactProviderData], + ) -> tuple[str, str, Any | None]: + try: + result = await self.service.update_snapshot( + user_id=self.user_id, + db=self.db, + snapshot_id=snapshot_id, + flow_artifact=flow_artifact, + ) + except DeploymentNotFoundError as exc: + return OUTCOME_NOT_FOUND, str(exc), None + except ResourceConflictError as exc: + return OUTCOME_CONFLICT, exc.message, None + except InvalidContentError as exc: + return OUTCOME_INVALID_CONTENT, exc.message, None + except (InvalidDeploymentOperationError, InvalidDeploymentTypeError, OperationNotSupportedError) as exc: + return OUTCOME_INVALID_OPERATION, exc.message, None + except DeploymentError as exc: + return OUTCOME_FAILURE, exc.message, None + except HTTPException as exc: + return self._outcome_from_http_exception(exc), str(exc.detail), None + except Exception as exc: # noqa: BLE001 + return OUTCOME_FAILURE, str(exc), None + else: + return OUTCOME_SUCCESS, "snapshot_updated", result + + async def _run_rollback_create_result( + self, + *, + deployment_id: str, + provider_result: object, + ) -> tuple[str, str, Any | None]: + try: + result = await self.service.rollback_create_result( + user_id=self.user_id, + deployment_id=deployment_id, + provider_result=provider_result, + db=self.db, + ) + except DeploymentNotFoundError as exc: + return OUTCOME_NOT_FOUND, str(exc), None + except ResourceConflictError as exc: + return OUTCOME_CONFLICT, exc.message, None + except InvalidContentError as exc: + return OUTCOME_INVALID_CONTENT, exc.message, None + except (InvalidDeploymentOperationError, InvalidDeploymentTypeError, OperationNotSupportedError) as exc: + return OUTCOME_INVALID_OPERATION, exc.message, None + except DeploymentError as exc: + return OUTCOME_FAILURE, exc.message, None + except HTTPException as exc: + return self._outcome_from_http_exception(exc), str(exc.detail), None + except Exception as exc: # noqa: BLE001 + return OUTCOME_FAILURE, str(exc), None + else: + return OUTCOME_SUCCESS, "rollback_create_result_done", result + + async def _run_redeploy(self, deployment_id: str) -> tuple[str, str, Any | None]: + try: + result = await self.service.redeploy( + user_id=self.user_id, + deployment_id=deployment_id, + db=self.db, + ) + except DeploymentNotFoundError as exc: + return OUTCOME_NOT_FOUND, str(exc), None + except ResourceConflictError as exc: + return OUTCOME_CONFLICT, exc.message, None + except InvalidContentError as exc: + return OUTCOME_INVALID_CONTENT, exc.message, None + except (InvalidDeploymentOperationError, InvalidDeploymentTypeError, OperationNotSupportedError) as exc: + return OUTCOME_INVALID_OPERATION, exc.message, None + except DeploymentError as exc: + return OUTCOME_FAILURE, exc.message, None + except HTTPException as exc: + return self._outcome_from_http_exception(exc), str(exc.detail), None + except Exception as exc: # noqa: BLE001 + return OUTCOME_FAILURE, str(exc), None + else: + return OUTCOME_SUCCESS, "redeployed", result + + async def _run_duplicate(self, deployment_id: str) -> tuple[str, str, Any | None]: + try: + result = await self.service.duplicate( + user_id=self.user_id, + deployment_id=deployment_id, + db=self.db, + ) + except DeploymentNotFoundError as exc: + return OUTCOME_NOT_FOUND, str(exc), None + except ResourceConflictError as exc: + return OUTCOME_CONFLICT, exc.message, None + except InvalidContentError as exc: + return OUTCOME_INVALID_CONTENT, exc.message, None + except (InvalidDeploymentOperationError, InvalidDeploymentTypeError, OperationNotSupportedError) as exc: + return OUTCOME_INVALID_OPERATION, exc.message, None + except DeploymentError as exc: + return OUTCOME_FAILURE, exc.message, None + except HTTPException as exc: + return self._outcome_from_http_exception(exc), str(exc.detail), None + except Exception as exc: # noqa: BLE001 + return OUTCOME_FAILURE, str(exc), None + else: + return OUTCOME_SUCCESS, "duplicated", result + + async def _run_teardown(self) -> tuple[str, str, Any | None]: + try: + result = await self.service.teardown() + except DeploymentNotFoundError as exc: + return OUTCOME_NOT_FOUND, str(exc), None + except ResourceConflictError as exc: + return OUTCOME_CONFLICT, exc.message, None + except InvalidContentError as exc: + return OUTCOME_INVALID_CONTENT, exc.message, None + except (InvalidDeploymentOperationError, InvalidDeploymentTypeError, OperationNotSupportedError) as exc: + return OUTCOME_INVALID_OPERATION, exc.message, None + except DeploymentError as exc: + return OUTCOME_FAILURE, exc.message, None + except HTTPException as exc: + return self._outcome_from_http_exception(exc), str(exc.detail), None + except Exception as exc: # noqa: BLE001 + return OUTCOME_FAILURE, str(exc), None + else: + return OUTCOME_SUCCESS, "teardown_done", result async def _run_status(self, deployment_id: str) -> tuple[str, str, Any | None]: try: @@ -1044,20 +1271,317 @@ class WatsonxAdapterDirectE2E: ) ) - # Empty snapshot_ids → SnapshotListParams.has_snapshot_ids is False → - # the service falls through to the deployment-scoped path which requires - # deployment_id. Since we don't provide one here, a DeploymentError is - # raised, making OUTCOME_FAILURE the correct expectation. + # Empty snapshot_ids currently normalizes to tenant-scoped listing. + # We still expect a successful response and ensure known seed IDs + # are visible in the returned set. print("[snap-ids/4] live_list_snapshots_by_ids_empty_input") status_code, detail, snap_result = await self._run_list_snapshots_by_ids([]) returned_ids = self._extract_snapshot_ids(snap_result) + has_known_subset = set(known_ids).issubset(returned_ids) results.append( self._build_result( name="live_list_snapshots_by_ids_empty_input", - expected={OUTCOME_FAILURE}, + expected={OUTCOME_SUCCESS}, actual_outcome=status_code, - detail=f"{detail} | returned_count={len(returned_ids)}", - ok=status_code == OUTCOME_FAILURE, + detail=(f"{detail} | returned_count={len(returned_ids)} has_known_subset={has_known_subset}"), + ok=status_code == OUTCOME_SUCCESS and has_known_subset, + ) + ) + + return results + + async def _run_live_listing_mode_scenarios(self) -> list[ScenarioResult]: + results: list[ScenarioResult] = [] + print("\n[list/1] creating seed for list mode checks") + deployment_id, config_id, seed_snapshot_ids, _ = await self._create_update_seed( + label="list_modes_seed", + snapshot_count=1, + ) + seed_snapshot_id = next(iter(seed_snapshot_ids), "") + if not seed_snapshot_id: + results.append( + self._build_result( + name="live_list_modes_seed_missing_snapshot", + expected={OUTCOME_SUCCESS}, + actual_outcome=OUTCOME_FAILURE, + detail="seed snapshot id missing", + ok=False, + ) + ) + return results + + print("[list/2] live_list_snapshots_tenant_scope") + status_code, detail, tenant_snapshots = await self._run_list_snapshots_with_params( + params=None, + detail_label="snapshots_tenant_listed", + ) + tenant_snapshot_ids = self._extract_snapshot_ids(tenant_snapshots) + tenant_has_seed = seed_snapshot_id in tenant_snapshot_ids + results.append( + self._build_result( + name="live_list_snapshots_tenant_scope", + expected={OUTCOME_SUCCESS}, + actual_outcome=status_code, + detail=( + f"{detail} | seed_snapshot_id={seed_snapshot_id} " + f"tenant_has_seed={tenant_has_seed} total={len(tenant_snapshot_ids)}" + ), + ok=status_code == OUTCOME_SUCCESS and tenant_has_seed, + ) + ) + + print("[list/3] live_list_configs_tenant_scope") + status_code, detail, tenant_configs = await self._run_list_configs_with_params( + params=None, + detail_label="configs_tenant_listed", + ) + tenant_config_ids = self._extract_config_ids(tenant_configs) + tenant_has_config = bool(config_id and config_id in tenant_config_ids) + results.append( + self._build_result( + name="live_list_configs_tenant_scope", + expected={OUTCOME_SUCCESS}, + actual_outcome=status_code, + detail=( + f"{detail} | seed_config_id={config_id} " + f"tenant_has_config={tenant_has_config} total={len(tenant_config_ids)}" + ), + ok=status_code == OUTCOME_SUCCESS and bool(config_id) and tenant_has_config, + ) + ) + + deployment_list_status, deployment_list_detail, deployment_snapshot_list = await self._run_list_snapshots( + deployment_id + ) + deployment_snapshots = getattr(deployment_snapshot_list, "snapshots", []) if deployment_snapshot_list else [] + seed_snapshot_name = "" + for snapshot in deployment_snapshots: + snapshot_id = str(getattr(snapshot, "id", "")).strip() + snapshot_name = str(getattr(snapshot, "name", "")).strip() + if snapshot_id == seed_snapshot_id and snapshot_name: + seed_snapshot_name = snapshot_name + break + if not seed_snapshot_name: + results.append( + self._build_result( + name="live_list_snapshots_by_names_seed_missing_name", + expected={OUTCOME_SUCCESS}, + actual_outcome=deployment_list_status, + detail=( + f"{deployment_list_detail} | seed_snapshot_id={seed_snapshot_id} " + "is missing from deployment-scoped snapshot names" + ), + ok=False, + ) + ) + return results + + print("[list/4] live_list_snapshots_by_names_returns_known") + status_code, detail, by_name_result = await self._run_list_snapshots_by_names([seed_snapshot_name]) + by_name_ids = self._extract_snapshot_ids(by_name_result) + by_name_has_seed = seed_snapshot_id in by_name_ids + results.append( + self._build_result( + name="live_list_snapshots_by_names_returns_known", + expected={OUTCOME_SUCCESS}, + actual_outcome=status_code, + detail=( + f"{detail} | seed_snapshot_name={seed_snapshot_name} " + f"seed_snapshot_id={seed_snapshot_id} by_name_has_seed={by_name_has_seed}" + ), + ok=status_code == OUTCOME_SUCCESS and by_name_has_seed, + ) + ) + + print("[list/5] live_list_snapshots_by_names_ignored_with_deployment_scope") + status_code, detail, mixed_filter_result = await self._run_list_snapshots_with_params( + params=SnapshotListParams( + deployment_ids=[deployment_id], + snapshot_names=[self._mk_name("snap_name_ignored")], + ), + detail_label="snapshots_mixed_filter_listed", + ) + mixed_filter_ids = self._extract_snapshot_ids(mixed_filter_result) + mixed_kept_seed = seed_snapshot_id in mixed_filter_ids + results.append( + self._build_result( + name="live_list_snapshots_by_names_ignored_with_deployment_scope", + expected={OUTCOME_SUCCESS}, + actual_outcome=status_code, + detail=( + f"{detail} | seed_snapshot_id={seed_snapshot_id} mixed_kept_seed={mixed_kept_seed} " + f"returned={sorted(mixed_filter_ids)}" + ), + ok=status_code == OUTCOME_SUCCESS and mixed_kept_seed, + ) + ) + + return results + + async def _run_live_service_surface_scenarios(self) -> list[ScenarioResult]: + results: list[ScenarioResult] = [] + + print("\n[surface/1] live_list_types_supports_agent") + status_code, detail, types_result = await self._run_list_types() + deployment_types = { + (dtype.value if hasattr(dtype, "value") else str(dtype)).strip() + for dtype in getattr(types_result, "deployment_types", []) + } + has_agent_type = DeploymentType.AGENT.value in deployment_types + results.append( + self._build_result( + name="live_list_types_supports_agent", + expected={OUTCOME_SUCCESS}, + actual_outcome=status_code, + detail=f"{detail} | deployment_types={sorted(deployment_types)}", + ok=status_code == OUTCOME_SUCCESS and has_agent_type, + ) + ) + + print("[surface/2] live_list_llms_returns_models") + status_code, detail, llms_result = await self._run_list_llms() + llm_provider_result = getattr(llms_result, "provider_result", {}) if llms_result else {} + models = llm_provider_result.get("models", []) if isinstance(llm_provider_result, dict) else [] + results.append( + self._build_result( + name="live_list_llms_returns_models", + expected={OUTCOME_SUCCESS}, + actual_outcome=status_code, + detail=f"{detail} | model_count={len(models)}", + ok=status_code == OUTCOME_SUCCESS and len(models) > 0, + ) + ) + + print("[surface/3] live_verify_credentials_success") + status_code, detail, _ = await self._run_verify_credentials( + base_url=self.provider_backend_url, + api_key=self.provider_api_key, + ) + results.append( + self._build_result( + name="live_verify_credentials_success", + expected={OUTCOME_SUCCESS}, + actual_outcome=status_code, + detail=detail, + ok=status_code == OUTCOME_SUCCESS, + ) + ) + + print("[surface/4] creating seed for update_snapshot + unsupported operations") + deployment_id, _config_id, surface_snapshot_ids, _ = await self._create_update_seed( + label="surface_seed", + snapshot_count=1, + ) + surface_snapshot_id = next(iter(surface_snapshot_ids), "") + if not surface_snapshot_id: + results.append( + self._build_result( + name="live_update_snapshot_seed_missing_snapshot", + expected={OUTCOME_SUCCESS}, + actual_outcome=OUTCOME_FAILURE, + detail="surface seed has no snapshot ids", + ok=False, + ) + ) + return results + + print("[surface/5] live_update_snapshot_success") + status_code, detail, update_snapshot_result = await self._run_update_snapshot( + snapshot_id=surface_snapshot_id, + flow_artifact=self._build_flow_payload(label="surface_update_snapshot_flow"), + ) + updated_snapshot_id = ( + str(getattr(update_snapshot_result, "snapshot_id", "")).strip() if update_snapshot_result else "" + ) + results.append( + self._build_result( + name="live_update_snapshot_success", + expected={OUTCOME_SUCCESS}, + actual_outcome=status_code, + detail=f"{detail} | expected_snapshot_id={surface_snapshot_id} got={updated_snapshot_id}", + ok=status_code == OUTCOME_SUCCESS and updated_snapshot_id == surface_snapshot_id, + ) + ) + + print("[surface/6] live_rollback_create_result_cleans_up_created") + rollback_create_status, rollback_create_detail, rollback_created = await self._run_create( + self._build_create_payload( + tool_payloads=[self._build_flow_payload(label="surface_rb_seed_snap")], + raw_connection=self._build_config_payload(label="surface_rb_seed_cfg"), + ) + ) + if rollback_create_status != OUTCOME_SUCCESS or rollback_created is None: + results.append( + self._build_result( + name="live_rollback_create_result_cleans_up_created", + expected={OUTCOME_SUCCESS}, + actual_outcome=rollback_create_status, + detail=f"seed_create={rollback_create_status}:{rollback_create_detail}", + ok=False, + ) + ) + else: + rollback_deployment_id = rollback_created.deployment_id + rollback_snapshot_ids = self._extract_create_snapshot_ids(rollback_created.provider_result) + rollback_app_ids = self._extract_create_app_ids(rollback_created.provider_result) + self.created_deployment_ids.add(rollback_deployment_id) + self.created_snapshot_ids.update(rollback_snapshot_ids) + self.created_config_ids.update(rollback_app_ids) + rollback_status, rollback_detail, _ = await self._run_rollback_create_result( + deployment_id=rollback_deployment_id, + provider_result=rollback_created.provider_result, + ) + post_status, post_detail, _ = await self._run_get(rollback_deployment_id) + cleaned_up = rollback_status == OUTCOME_SUCCESS and post_status == OUTCOME_NOT_FOUND + if cleaned_up: + self.created_deployment_ids.discard(rollback_deployment_id) + self.created_snapshot_ids.difference_update(rollback_snapshot_ids) + self.created_config_ids.difference_update(rollback_app_ids) + results.append( + self._build_result( + name="live_rollback_create_result_cleans_up_created", + expected={OUTCOME_SUCCESS}, + actual_outcome=rollback_status, + detail=(f"rollback={rollback_status}:{rollback_detail} post_get={post_status}:{post_detail}"), + ok=cleaned_up, + ) + ) + + print("[surface/7] live_redeploy_not_supported") + status_code, detail, _ = await self._run_redeploy(deployment_id) + redeploy_not_supported = "not supported" in detail.lower() + results.append( + self._build_result( + name="live_redeploy_not_supported", + expected={OUTCOME_INVALID_OPERATION}, + actual_outcome=status_code, + detail=detail, + ok=status_code == OUTCOME_INVALID_OPERATION and redeploy_not_supported, + ) + ) + + print("[surface/8] live_duplicate_not_supported") + status_code, detail, _ = await self._run_duplicate(deployment_id) + duplicate_not_supported = "not supported" in detail.lower() + results.append( + self._build_result( + name="live_duplicate_not_supported", + expected={OUTCOME_INVALID_OPERATION}, + actual_outcome=status_code, + detail=detail, + ok=status_code == OUTCOME_INVALID_OPERATION and duplicate_not_supported, + ) + ) + + print("[surface/9] live_teardown_noop") + status_code, detail, _ = await self._run_teardown() + results.append( + self._build_result( + name="live_teardown_noop", + expected={OUTCOME_SUCCESS}, + actual_outcome=status_code, + detail=detail, + ok=status_code == OUTCOME_SUCCESS, ) ) @@ -1214,6 +1738,16 @@ class WatsonxAdapterDirectE2E: snapshot_ids = getattr(provider_result, "created_snapshot_ids", []) if provider_result else [] return {str(snapshot_id) for snapshot_id in snapshot_ids if str(snapshot_id).strip()} + def _extract_update_added_snapshot_ids(self, update_result: Any) -> set[str]: + if update_result is None: + return set() + provider_result = getattr(update_result, "provider_result", None) + if isinstance(provider_result, dict): + snapshot_ids = provider_result.get("added_snapshot_ids", []) + else: + snapshot_ids = getattr(provider_result, "added_snapshot_ids", []) if provider_result else [] + return {str(snapshot_id) for snapshot_id in snapshot_ids if str(snapshot_id).strip()} + def _extract_update_created_app_ids(self, update_result: Any) -> set[str]: if update_result is None: return set() @@ -1350,7 +1884,7 @@ class WatsonxAdapterDirectE2E: DeploymentUpdate( provider_data={ "tools": {}, - "connections": {"existing_app_ids": [str(donor_config_id)]}, + "connections": {}, "operations": [ { "op": "bind", @@ -1364,8 +1898,8 @@ class WatsonxAdapterDirectE2E: ) list_status, _list_detail, list_after_config_only = await self._run_list_snapshots(primary_deployment_id) attached_after_config_only = self._extract_snapshot_ids(list_after_config_only) - config_only_snapshot_ids = self._extract_update_snapshot_ids(config_only_result) - config_only_snapshot_ids_ok = retained_snapshot_ids.issubset(config_only_snapshot_ids) + config_only_created_snapshot_ids = self._extract_update_snapshot_ids(config_only_result) + config_only_created_ok = len(config_only_created_snapshot_ids) == 0 results.append( self._build_result( name="upd_config_only_existing_tools_with_config_id", @@ -1376,7 +1910,7 @@ class WatsonxAdapterDirectE2E: status_code == OUTCOME_SUCCESS and list_status == OUTCOME_SUCCESS and retained_snapshot_ids.issubset(attached_after_config_only) - and config_only_snapshot_ids_ok + and config_only_created_ok ), ) ) @@ -1387,7 +1921,7 @@ class WatsonxAdapterDirectE2E: DeploymentUpdate( provider_data={ "tools": {}, - "connections": {"existing_app_ids": [str(donor_config_id)]}, + "connections": {}, "operations": [ { "op": "bind", @@ -1400,7 +1934,7 @@ class WatsonxAdapterDirectE2E: ) list_status, _list_detail, list_after_add_id = await self._run_list_snapshots(primary_deployment_id) attached_after_add_id = self._extract_snapshot_ids(list_after_add_id) - add_id_snapshot_ids = self._extract_update_snapshot_ids(add_id_result) + add_id_snapshot_ids = self._extract_update_added_snapshot_ids(add_id_result) results.append( self._build_result( name="upd_snapshot_add_ids_with_config_id", @@ -1422,9 +1956,8 @@ class WatsonxAdapterDirectE2E: primary_deployment_id, DeploymentUpdate( provider_data={ - "resource_name_prefix": f"e2e_upd_{uuid4().hex[:6]}_", "tools": {"raw_payloads": [raw_payload.model_dump(mode="json")]}, - "connections": {"existing_app_ids": [str(donor_config_id)]}, + "connections": {}, "operations": [ { "op": "bind", @@ -1462,11 +1995,10 @@ class WatsonxAdapterDirectE2E: primary_deployment_id, DeploymentUpdate( provider_data={ - "resource_name_prefix": f"e2e_upd_mix_{uuid4().hex[:6]}_", "tools": { "raw_payloads": [mixed_raw_payload.model_dump(mode="json")], }, - "connections": {"existing_app_ids": [str(donor_config_id)]}, + "connections": {}, "operations": [ { "op": "bind", @@ -1483,7 +2015,7 @@ class WatsonxAdapterDirectE2E: } ), ) - mixed_snapshot_ids = self._extract_update_snapshot_ids(mixed_result) + mixed_snapshot_ids = self._extract_update_added_snapshot_ids(mixed_result) self.created_snapshot_ids.update(mixed_snapshot_ids) list_status, _list_detail, list_after_mixed = await self._run_list_snapshots(primary_deployment_id) attached_after_mixed = self._extract_snapshot_ids(list_after_mixed) @@ -1588,7 +2120,7 @@ class WatsonxAdapterDirectE2E: primary_deployment_id, DeploymentUpdate( provider_data={ - "connections": {"existing_app_ids": [str(donor_config_id)]}, + "connections": {}, "operations": [ { "op": "unbind", @@ -1615,7 +2147,7 @@ class WatsonxAdapterDirectE2E: DeploymentUpdate( provider_data={ "tools": {}, - "connections": {"existing_app_ids": [str(donor_config_id)]}, + "connections": {}, "operations": [ { "op": "bind", @@ -1642,7 +2174,6 @@ class WatsonxAdapterDirectE2E: snapshot_count=1, ) conflict_suffix = uuid4().hex[:8] - conflict_prefix = f"e2e_upd_conflict_{conflict_suffix}_" conflict_name = f"dup_cfg_{conflict_suffix}" conflict_tool_id = next(iter(_conflict_snapshot_ids), "") if not conflict_tool_id: @@ -1658,7 +2189,6 @@ class WatsonxAdapterDirectE2E: return results conflict_payload = DeploymentUpdate( provider_data={ - "resource_name_prefix": conflict_prefix, "tools": {}, "connections": {"raw_payloads": [{"app_id": conflict_name, "environment_variables": {}}]}, "operations": [ @@ -1797,7 +2327,6 @@ class WatsonxAdapterDirectE2E: results: list[ScenarioResult] = [] print(f"[cc/{iteration}.1] cc_create_same_prefix_race") - shared_prefix = f"e2e_cc_shared_{uuid4().hex[:6]}_" shared_dep_name = self._mk_name("dep_cc_shared") shared_cfg_name = self._mk_name("cfg_cc_shared") shared_snap_name = self._mk_name("snap_cc_shared") @@ -1808,7 +2337,6 @@ class WatsonxAdapterDirectE2E: description="concurrency create collision", environment_variables={}, ), - resource_name_prefix=shared_prefix, ) shared_payload.spec = shared_payload.spec.model_copy(update={"name": shared_dep_name}, deep=True) create_race = await self._run_parallel_calls( @@ -1859,7 +2387,7 @@ class WatsonxAdapterDirectE2E: DeploymentUpdate( provider_data={ "tools": {}, - "connections": {"existing_app_ids": [str(donor_cfg_id)]}, + "connections": {}, "operations": [ { "op": "bind", @@ -1987,7 +2515,7 @@ class WatsonxAdapterDirectE2E: DeploymentUpdate( provider_data={ "tools": {}, - "connections": {"existing_app_ids": [str(bind_cfg_id)]}, + "connections": {}, "operations": [ { "op": "unbind", @@ -2028,7 +2556,6 @@ class WatsonxAdapterDirectE2E: ) ) return results - cfg_prefix = f"e2e_cc_del_cfg_{uuid4().hex[:6]}_" raw_cfg_name = self._mk_name("cc_raw_cfg") async def _delete_created_app_after_config_create(created_app_id: Any, **kwargs: Any) -> None: @@ -2045,7 +2572,6 @@ class WatsonxAdapterDirectE2E: delete_cfg_id, DeploymentUpdate( provider_data={ - "resource_name_prefix": cfg_prefix, "tools": {}, "connections": {"raw_payloads": [{"app_id": raw_cfg_name, "environment_variables": {}}]}, "operations": [ @@ -2077,7 +2603,6 @@ class WatsonxAdapterDirectE2E: ) print(f"[cc/{iteration}.7] cc_create_during_create_snapshots_stage") - create_race_prefix = f"e2e_cc_create_stage_{uuid4().hex[:6]}_" create_race_dep = self._mk_name("cc_stage_dep") create_race_cfg = self._mk_name("cc_stage_cfg") create_race_snap = self._mk_name("cc_stage_snap") @@ -2088,7 +2613,6 @@ class WatsonxAdapterDirectE2E: description="cc competing create", environment_variables={}, ), - resource_name_prefix=create_race_prefix, ) race_payload.spec = race_payload.spec.model_copy(update={"name": create_race_dep}, deep=True) competing_create_task: asyncio.Task[tuple[str, str, WxoCreatedDeploymentResult | None]] | None = None @@ -2146,13 +2670,11 @@ class WatsonxAdapterDirectE2E: ) ) return results - update_cfg_prefix = f"e2e_cc_upd_cfg_create_{uuid4().hex[:6]}_" update_cfg_name = self._mk_name("cc_upd_cfg_create") competing_update_task: asyncio.Task[tuple[str, str, Any | None]] | None = None competing_update_payload = DeploymentUpdate( provider_data={ - "resource_name_prefix": update_cfg_prefix, "tools": {}, "connections": {"raw_payloads": [{"app_id": update_cfg_name, "environment_variables": {}}]}, "operations": [ @@ -2223,7 +2745,6 @@ class WatsonxAdapterDirectE2E: ) ) return results - rollback_prefix = f"e2e_cc_rollback_{uuid4().hex[:6]}_" rollback_raw_flow = self._build_flow_payload(label=f"cc_rb_raw_{iteration}") rollback_raw_cfg_name = self._mk_name("cc_rb_cfg") rollback_status, rollback_detail, _ = await self._run_with_stage_hook( @@ -2233,12 +2754,10 @@ class WatsonxAdapterDirectE2E: DeploymentUpdate( spec=BaseDeploymentDataUpdate(description="cc rollback delete race"), provider_data={ - "resource_name_prefix": rollback_prefix, "tools": { "raw_payloads": [rollback_raw_flow.model_dump(mode="json")], }, "connections": { - "existing_app_ids": [str(rollback_cfg_id)], "raw_payloads": [{"app_id": rollback_raw_cfg_name, "environment_variables": {}}], }, "operations": [ @@ -2303,7 +2822,6 @@ class WatsonxAdapterDirectE2E: ) ) return results - rollback_create_prefix = f"e2e_cc_rb_create_{uuid4().hex[:6]}_" rollback_create_cfg_name = self._mk_name("cc_rb_create_cfg") competing_rollback_create_task: asyncio.Task[tuple[str, str, Any | None]] | None = None @@ -2316,7 +2834,6 @@ class WatsonxAdapterDirectE2E: rollback_create_id, DeploymentUpdate( provider_data={ - "resource_name_prefix": rollback_create_prefix, "tools": {}, "connections": { "raw_payloads": [{"app_id": rollback_create_cfg_name, "environment_variables": {}}] @@ -2342,12 +2859,10 @@ class WatsonxAdapterDirectE2E: DeploymentUpdate( spec=BaseDeploymentDataUpdate(description="cc rollback create race"), provider_data={ - "resource_name_prefix": rollback_create_prefix, "tools": { "raw_payloads": [rollback_create_raw_flow.model_dump(mode="json")], }, "connections": { - "existing_app_ids": [str(rollback_create_cfg_id)], "raw_payloads": [{"app_id": rollback_create_cfg_name, "environment_variables": {}}], }, "operations": [ @@ -2465,7 +2980,17 @@ class WatsonxAdapterDirectE2E: def _extract_config_ids(self, config_result: Any) -> set[str]: configs = getattr(config_result, "configs", []) if config_result else [] - return {str(config.id) for config in configs if config and getattr(config, "id", None)} + ids_or_names: set[str] = set() + for config in configs: + if not config: + continue + config_id = str(getattr(config, "id", "")).strip() + config_name = str(getattr(config, "name", "")).strip() + if config_id: + ids_or_names.add(config_id) + if config_name: + ids_or_names.add(config_name) + return ids_or_names def _stage_hook_mapping(self) -> dict[str, tuple[Any, str]]: return { @@ -2611,13 +3136,11 @@ class WatsonxAdapterDirectE2E: ) ) return results - failpoint_prefix = f"e2e_fp_upd_{uuid4().hex[:6]}_" failpoint_raw_app_id = self._mk_name("fp_upd_cfg") update_payload = DeploymentUpdate( spec=BaseDeploymentDataUpdate(description="trigger update failpoint"), provider_data={ - "resource_name_prefix": failpoint_prefix, "tools": {}, "connections": {"raw_payloads": [{"app_id": failpoint_raw_app_id, "environment_variables": {}}]}, "operations": [ @@ -2697,14 +3220,12 @@ class WatsonxAdapterDirectE2E: ) return results - restore_prefix = f"e2e_fp_restore_{uuid4().hex[:6]}_" restore_raw_cfg = self._mk_name("fp_restore_cfg") print("[fp-upd/3a] injecting update failure to corrupt tool list") inject_status, inject_detail, _ = await self._run_update( restore_id, DeploymentUpdate( provider_data={ - "resource_name_prefix": restore_prefix, "tools": {}, "connections": {"raw_payloads": [{"app_id": restore_raw_cfg, "environment_variables": {}}]}, "operations": [ @@ -2841,9 +3362,7 @@ class WatsonxAdapterDirectE2E: tool_payloads: list[BaseFlowArtifact[WatsonxFlowArtifactProviderData]], raw_connection: DeploymentConfig | None = None, existing_connection_app_id: str | None = None, - resource_name_prefix: str | None = None, ) -> DeploymentCreate: - prefix = resource_name_prefix or f"e2e_{uuid4().hex[:8]}_" spec = BaseDeploymentData( name=self._mk_name("dep_agent"), description="direct adapter scenario", @@ -2863,7 +3382,6 @@ class WatsonxAdapterDirectE2E: ] elif existing_connection_app_id: operation_app_id = str(existing_connection_app_id).strip() - connections["existing_app_ids"] = [operation_app_id] else: operation_app_id = self._mk_name("cfg_default_app") connections["raw_payloads"] = [{"app_id": operation_app_id, "environment_variables": {}}] @@ -2878,10 +3396,10 @@ class WatsonxAdapterDirectE2E: ] provider_data = { - "resource_name_prefix": prefix, "tools": {"raw_payloads": raw_tool_payloads}, "connections": connections, "operations": operations, + "llm": self.llm, } return DeploymentCreate(spec=spec, provider_data=provider_data) @@ -3062,6 +3580,7 @@ def _parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Run direct Watsonx adapter matrix (live + failpoints).") parser.add_argument("--project-id", default=os.getenv("WXO_PROJECT_ID", "e2e-project")) parser.add_argument("--mode", choices=["live", "failpoint", "both"], default=os.getenv("WXO_E2E_MODE", "both")) + parser.add_argument("--llm", default=os.getenv("WXO_DEFAULT_LLM", DEFAULT_WXO_LLM)) parser.add_argument("--keep-resources", action="store_true") return parser.parse_args() @@ -3075,6 +3594,7 @@ async def _main() -> int: project_id=args.project_id, mode=args.mode, keep_resources=args.keep_resources, + llm=args.llm, ) return await runner.run() diff --git a/scripts/e2e_deployment_tests/watsonx_orchestrate/api.py b/scripts/e2e_deployment_tests/watsonx_orchestrate/api.py new file mode 100644 index 0000000000..41533523b1 --- /dev/null +++ b/scripts/e2e_deployment_tests/watsonx_orchestrate/api.py @@ -0,0 +1,2149 @@ +"""Deployments API create/update matrix runner. + +This script exercises `/api/v1/deployments` over HTTP and focuses on +create/update payload-heavy scenarios for the Watsonx Orchestrate provider. + +Warning: +-------- +This script performs live integration calls and creates real resources in langflow +and Watsonx Orchestrate (agents, snapshots/tools, and configs/connections). +By default, cleanup runs at the end of execution, but cleanup is best-effort: +if the process is interrupted or provider deletes fail, resources may remain. +Use `--keep-resources` only when you intentionally want to inspect leftovers. + +Safety model: +- Destructive operations are only executed for resources created by this run. +- "Onboard existing agent" scenarios reuse an agent created by this run: + create deployment -> delete with `include_provider=false` -> re-onboard. +- End-of-run cleanup deletes all runner-owned deployment rows and provider + resources unless `--keep-resources` is set. +- Provider account resolution is deterministic: list deployment providers and + reuse the one whose URL matches `WXO_INSTANCE_URL`; create a provider account + when no matching record exists for the current user. + +Scenario catalog +---------------- +Live create/update happy-path scenarios: +- `create_new_agent_success`: creates a deployment using create-time `add_flows` + payload (expects Success). +- `create_onboard_existing_agent_without_mutation`: onboards an existing agent + created by this run (seeded and DB-deleted with `include_provider=false`) + without mutating provider state (expects Success). +- `create_onboard_existing_agent_with_mutation`: onboards a second owned + existing agent while applying create payload operations (expects Success). +- `update_metadata_only_success`: updates name/description only (expects Success). +- `update_provider_data_llm_only_success`: updates provider_data model only + (expects Success). + +Live payload-validation scenarios: +- `create_reject_missing_add_flows_and_upsert_tools`: rejects create + provider_data without operations for new agent creation (expects HTTP422). +- `create_reject_duplicate_connection_app_ids`: rejects create payload with + duplicate `connections[].app_id` values (expects HTTP422). +- `create_reject_unused_connection_app_ids`: rejects create payload where + declared connection app ids are not referenced by operations (expects HTTP422). +- `update_reject_empty_body`: rejects update with no changed fields + (expects HTTP422). +- `update_reject_add_remove_overlap`: rejects update when a flow item includes + overlapping `add_app_ids` and `remove_app_ids` (expects HTTP422). + +Live attachment/flow patching scenarios: +- `update_patch_upsert_flows_add_binding`: exercises update `upsert_flows` + attachment patch path (expects Success). +- `update_patch_add_second_flow_then_remove`: when a second flow-version id is + configured, chains add then remove operations for that flow + (expects Success). + +Live rollback/error-path scenarios: +- `create_duplicate_name_conflict`: validates duplicate name protection for the + same provider account (expects HTTP409). +- `update_unknown_deployment_not_found`: validates unknown deployment handling + for update calls (expects HTTP404). +- `update_remove_unknown_tool_id_noop_success`: validates remove-by-tool-id is + accepted as a no-op when the tool id is unknown (expects Success). + +Live concurrency/race scenarios: +- `cc_parallel_duplicate_create_`: runs two create calls in parallel with + the same payload; accepts deterministic outcomes of one success + one conflict + (or dual success when provider-side timing allows), and tracks owned results. +- `cc_parallel_update_`: runs two updates in parallel against the same + owned deployment and validates acceptable race outcomes. + +Live large/complex payload scenarios (executed at fixed tiers S/M/L): +- `create_large_payload_success_fanout_tier_`: validates large create + payload fanout success. +- `create_large_payload_reject_unused_connections_tier_`: validates + rejection of unreferenced connection app ids. +- `create_large_payload_reject_duplicate_connection_app_ids_tier_`: + validates duplicate connection app id rejection. +- `update_large_payload_success_mixed_ops_tier_`: validates mixed update + operations at large scale. +- `update_large_payload_success_tool_id_fanout_tier_`: validates large + tool-id upsert/remove operation fanout. +- `update_large_payload_reject_add_remove_overlap_tier_`: validates + overlap rejection within one upsert item. +- `update_large_payload_reject_remove_conflict_tier_`: validates + remove-vs-upsert conflict rejection. +- `update_large_payload_reject_unbind_raw_app_ids_tier_`: validates + rejection of remove_app_ids that target raw connection app ids. +- `update_llm_only_fast_path_control_tier_`: validates LLM-only fast + path remains successful. + +Failpoint-mode scenarios (API-deterministic failure points): +- `fp_create_missing_flow_version_in_project`: create payload references a + non-existent flow-version id (expects HTTP404). +- `fp_update_missing_deployment`: update call targets a non-existent + deployment id (expects HTTP404). +- `fp_create_invalid_conflicting_update_ops_shape`: create payload intentionally + uses conflicting update-shape operation fields and is rejected by schema + validation (expects HTTP422). +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import os +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import TYPE_CHECKING, Any +from uuid import UUID, uuid4 + +import httpx +from dotenv import load_dotenv +from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException +from langflow.services.adapters.deployment.context import DeploymentAdapterContext, DeploymentProviderIDContext +from langflow.services.adapters.deployment.watsonx_orchestrate import WxOCredentials + +if TYPE_CHECKING: + from collections.abc import Awaitable, Callable + +OUTCOME_SUCCESS = "Success" +OUTCOME_HTTP_404 = "HTTP404" +OUTCOME_HTTP_409 = "HTTP409" +OUTCOME_HTTP_422 = "HTTP422" +OUTCOME_HTTP_500 = "HTTP500" +OUTCOME_FAILURE = "Failure" + +DEFAULT_TIMEOUT_SECS = 90 +DEFAULT_CONCURRENCY_REPEAT = 2 +DEFAULT_WXO_LLM = "groq/openai/gpt-oss-120b" +LARGE_PAYLOAD_TIER_ORDER = ("S", "M", "L") +# Large success scenarios create connections three times per tier: +# - create_large_payload_success_fanout +# - update_large_payload_success_mixed_ops +# - update_large_payload_success_tool_id_fanout +LARGE_SUCCESS_CONNECTION_MULTIPLIER = 3 +MAX_LARGE_SUCCESS_CONNECTION_CREATES = 300 +LARGE_PAYLOAD_TIER_CONFIGS: dict[str, dict[str, int]] = { + "S": { + "connections": 16, + "credentials_per_connection": 2, + "create_flow_items": 1, + "update_flow_items": 1, + "update_tool_items": 4, + "remove_tool_items": 2, + "remove_app_ids_per_flow": 4, + }, + "M": { + "connections": 24, + "credentials_per_connection": 3, + "create_flow_items": 2, + "update_flow_items": 3, + "update_tool_items": 8, + "remove_tool_items": 4, + "remove_app_ids_per_flow": 8, + }, + "L": { + "connections": 48, + "credentials_per_connection": 4, + "create_flow_items": 6, + "update_flow_items": 6, + "update_tool_items": 24, + "remove_tool_items": 8, + "remove_app_ids_per_flow": 24, + }, +} +HTTP_STATUS_OK = 200 +HTTP_STATUS_CREATED = 201 +HTTP_STATUS_NO_CONTENT = 204 +HTTP_STATUS_BAD_REQUEST = 400 +HTTP_STATUS_NOT_FOUND = 404 +HTTP_STATUS_CONFLICT = 409 +HTTP_STATUS_UNPROCESSABLE = 422 +HTTP_STATUS_SERVER_ERROR = 500 +HTTP_STATUS_MULTIPLE_CHOICES = 300 + + +@dataclass(slots=True) +class ScenarioResult: + name: str + expected_outcomes: set[str] + actual_outcome: str + ok: bool + detail: str + + +@dataclass(slots=True) +class HttpResponseEnvelope: + status_code: int + payload: dict[str, Any] | list[Any] | None + detail: str + + +@dataclass(slots=True) +class OwnedDeployment: + deployment_id: str + resource_key: str + name: str + + +class DeploymentsApiParallelE2E: + def __init__( + self, + *, + base_url: str, + api_key: str, + instance_url: str, + provider_api_key: str, + provider_tenant_id: str | None, + provider_key: str, + mode: str, + test_subset: str, + keep_resources: bool, + llm: str, + flow_version_ids: list[str], + starter_project_files: list[str] | None, + starter_project_count: int, + project_id: str | None, + timeout_secs: int, + concurrency_repeat: int, + verify_tls: bool, + ) -> None: + self.base_url = base_url.rstrip("/") + self.api_key = api_key + self.instance_url = instance_url + self.provider_api_key = provider_api_key + self.provider_tenant_id = provider_tenant_id + self.provider_key = provider_key + self.provider_id: str | None = None + self.mode = mode + self.test_subset = test_subset + self.keep_resources = keep_resources + self.llm = llm + self.flow_version_ids = flow_version_ids + self.starter_project_files = starter_project_files or [] + self.starter_project_count = max(1, starter_project_count) + self.project_id = project_id + self.timeout_secs = timeout_secs + self.concurrency_repeat = concurrency_repeat + self.verify_tls = verify_tls + + self.run_suffix = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S") + "-" + uuid4().hex[:8] + self._name_counter = 0 + self.owned_deployments: dict[str, OwnedDeployment] = {} + self.orphaned_provider_resource_keys: set[str] = set() + self.created_snapshot_ids: set[str] = set() + self.created_config_ids: set[str] = set() + self.requested_raw_connection_app_ids: set[str] = set() + self.cleanup_issues: list[str] = [] + self.created_provider_account_id: str | None = None + self.created_flow_ids: set[str] = set() + self.user_id = str(uuid4()) + self.db = object() + self._client_mod: Any = None + self._deployment_context_token: Any = None + self._original_resolve_wxo_client_credentials: Any = None + + self._client = httpx.AsyncClient( + base_url=self.base_url, + timeout=httpx.Timeout(self.timeout_secs), + verify=self.verify_tls, + headers={ + "x-api-key": self.api_key, + "Content-Type": "application/json", + }, + ) + self._validate_large_payload_connection_budget() + + def _validate_large_payload_connection_budget(self) -> None: + total_tier_connections = sum(int(config["connections"]) for config in LARGE_PAYLOAD_TIER_CONFIGS.values()) + expected_large_success_creates = total_tier_connections * LARGE_SUCCESS_CONNECTION_MULTIPLIER + if expected_large_success_creates > MAX_LARGE_SUCCESS_CONNECTION_CREATES: + msg = ( + "large payload configuration exceeds connection-create budget: " + f"expected={expected_large_success_creates} max={MAX_LARGE_SUCCESS_CONNECTION_CREATES}" + ) + raise RuntimeError(msg) + + async def run(self) -> int: + print("Starting deployments API parallel E2E runner...") + results: list[ScenarioResult] = [] + try: + resolved_provider_id = await self._resolve_or_create_provider_account() + await self._setup_provider_clients_context() + await self._ensure_flow_versions() + print( + f"mode={self.mode} subset={self.test_subset} provider_id={resolved_provider_id} " + f"keep_resources={self.keep_resources} flow_versions={','.join(self.flow_version_ids)} " + f"project_id={self.project_id or ''}" + ) + total_tier_connections = sum(int(config["connections"]) for config in LARGE_PAYLOAD_TIER_CONFIGS.values()) + expected_large_success_creates = total_tier_connections * LARGE_SUCCESS_CONNECTION_MULTIPLIER + print( + "large payload connection budget: " + f"tiers_total={total_tier_connections} " + f"expected_success_creates={expected_large_success_creates} " + f"max={MAX_LARGE_SUCCESS_CONNECTION_CREATES}" + ) + if self.mode in {"live", "both"}: + results.extend(await self._run_live_scenarios()) + if self.mode in {"failpoint", "both"}: + results.extend(await self._run_failpoint_scenarios()) + finally: + if not self.keep_resources: + try: + await self._cleanup_resources() + except Exception as exc: # noqa: BLE001 + message = f"cleanup routine failed: {exc}" + print(f"cleanup warning: {message}") + self.cleanup_issues.append(message) + try: + await self._client.aclose() + except Exception as exc: # noqa: BLE001 + message = f"http client close failed: {exc}" + print(f"cleanup warning: {message}") + self.cleanup_issues.append(message) + try: + await self._teardown_provider_clients_context() + except Exception as exc: # noqa: BLE001 + message = f"provider context teardown failed: {exc}" + print(f"cleanup warning: {message}") + self.cleanup_issues.append(message) + + self._print_summary(results) + if self.cleanup_issues: + print("Cleanup verification failed:") + for issue in self.cleanup_issues: + print(f"- {issue}") + has_scenario_failures = any(not item.ok for item in results) + return 1 if has_scenario_failures or bool(self.cleanup_issues) else 0 + + async def _run_live_scenarios(self) -> list[ScenarioResult]: + if self.test_subset == "smoke-connections": + return await self._run_smoke_connection_scenarios() + if self.test_subset == "large-tier-s": + return await self._run_large_complex_payload_scenarios(tiers=["S"]) + results: list[ScenarioResult] = [] + results.extend(await self._run_create_update_happy_paths()) + results.extend(await self._run_payload_validation_scenarios()) + results.extend(await self._run_large_complex_payload_scenarios()) + results.extend(await self._run_attachment_patch_scenarios()) + results.extend(await self._run_rollback_and_error_scenarios()) + results.extend(await self._run_parallel_race_scenarios()) + return results + + async def _run_smoke_connection_scenarios(self) -> list[ScenarioResult]: + print("Running smoke-connections live subset ...") + + create_app_id = self._normalize_wxo_connection_app_id(f"smoke-create-{self._app_id_namespace()}-000") + update_app_id = self._normalize_wxo_connection_app_id(f"smoke-update-{self._app_id_namespace()}-001") + + create_payload = self._create_request_payload( + name=self._mk_name("smoke_create_conn"), + provider_data=self._provider_data_create( + add_flows=[{"flow_version_id": self.flow_version_ids[0], "app_ids": [create_app_id]}], + connections=[ + { + "app_id": create_app_id, + "credentials": [{"key": "SMOKE_KEY", "value": "smoke-value", "source": "raw"}], + } + ], + ), + ) + results = await self._run_http_scenarios( + [ + { + "name": "smoke_create_with_connection_success", + "expected": {OUTCOME_SUCCESS}, + "call": self._call_create, + "payload": create_payload, + "track_owned": True, + } + ] + ) + + update_seed = await self._create_owned_deployment( + name=self._mk_name("smoke_update_seed"), + provider_data=self._provider_data_create( + add_flows=[{"flow_version_id": self.flow_version_ids[0], "app_ids": []}], + ), + ) + update_payload = { + "deployment_id": update_seed.deployment_id, + "body": { + "provider_data": self._provider_data_update( + connections=[ + { + "app_id": update_app_id, + "credentials": [{"key": "SMOKE_UPD_KEY", "value": "smoke-update", "source": "raw"}], + } + ], + upsert_flows=[ + { + "flow_version_id": self.flow_version_ids[0], + "add_app_ids": [update_app_id], + "remove_app_ids": [], + } + ], + ) + }, + } + results.extend( + await self._run_http_scenarios( + [ + { + "name": "smoke_update_with_connection_success", + "expected": {OUTCOME_SUCCESS}, + "call": self._call_update, + "payload": update_payload, + "track_owned": False, + } + ] + ) + ) + return results + + async def _run_failpoint_scenarios(self) -> list[ScenarioResult]: + # API-level deterministic failure points (schema+validation+not-found). + scenarios = [ + { + "name": "fp_create_missing_flow_version_in_project", + "expected": {OUTCOME_HTTP_404}, + "call": self._call_create, + "payload": self._create_request_payload( + name=self._mk_name("fp_missing_flow"), + provider_data=self._provider_data_create( + add_flows=[{"flow_version_id": str(uuid4()), "app_ids": []}], + ), + ), + "track_owned": False, + "detail_contains": "not checkpoints of flows in the selected project", + }, + { + "name": "fp_update_missing_deployment", + "expected": {OUTCOME_HTTP_404}, + "call": self._call_update, + "payload": { + "deployment_id": str(uuid4()), + "body": {"provider_data": self._provider_data_update(llm=self.llm)}, + }, + "track_owned": False, + "detail_contains": "not found", + }, + { + "name": "fp_create_invalid_conflicting_update_ops_shape", + "expected": {OUTCOME_HTTP_422}, + "call": self._call_create, + "payload": self._create_request_payload( + name=self._mk_name("fp_invalid_shape"), + provider_data={ + "llm": self.llm, + "upsert_flows": [ + { + "flow_version_id": self.flow_version_ids[0], + "add_app_ids": ["cfg-a"], + "remove_app_ids": ["cfg-a"], + } + ], + }, + ), + "track_owned": False, + }, + ] + return await self._run_http_scenarios(scenarios) + + async def _run_create_update_happy_paths(self) -> list[ScenarioResult]: + results: list[ScenarioResult] = [] + onboard_seed_without_mutation = await self._create_owned_deployment( + name=self._mk_name("seed_existing_agent"), + provider_data=self._provider_data_create( + add_flows=[{"flow_version_id": self.flow_version_ids[0], "app_ids": []}], + ), + ) + await self._delete_owned_deployment(onboard_seed_without_mutation.deployment_id, include_provider=False) + self.orphaned_provider_resource_keys.add(onboard_seed_without_mutation.resource_key) + onboard_seed_with_mutation = await self._create_owned_deployment( + name=self._mk_name("seed_existing_agent_mutate"), + provider_data=self._provider_data_create( + add_flows=[{"flow_version_id": self.flow_version_ids[0], "app_ids": []}], + ), + ) + await self._delete_owned_deployment(onboard_seed_with_mutation.deployment_id, include_provider=False) + self.orphaned_provider_resource_keys.add(onboard_seed_with_mutation.resource_key) + + scenarios = [ + { + "name": "create_new_agent_success", + "expected": {OUTCOME_SUCCESS}, + "call": self._call_create, + "payload": self._create_request_payload( + name=self._mk_name("create_new"), + provider_data=self._provider_data_create( + add_flows=[{"flow_version_id": self.flow_version_ids[0], "app_ids": []}], + ), + ), + "track_owned": True, + }, + { + "name": "create_onboard_existing_agent_without_mutation", + "expected": {OUTCOME_SUCCESS}, + "call": self._call_create, + "payload": self._create_request_payload( + name=self._mk_name("onboard_existing_nomutate"), + provider_data=self._provider_data_create( + existing_agent_id=onboard_seed_without_mutation.resource_key + ), + ), + "track_owned": True, + }, + { + "name": "create_onboard_existing_agent_with_mutation", + "expected": {OUTCOME_SUCCESS}, + "call": self._call_create, + "payload": self._create_request_payload( + name=self._mk_name("onboard_existing_mutate"), + provider_data=self._provider_data_create( + existing_agent_id=onboard_seed_with_mutation.resource_key, + add_flows=[{"flow_version_id": self.flow_version_ids[0], "app_ids": []}], + ), + ), + "track_owned": True, + }, + { + "name": "update_metadata_only_success", + "expected": {OUTCOME_SUCCESS}, + "call": self._call_update, + "payload": { + "deployment_id": await self._ensure_seed_for_update("meta_only"), + "body": {"name": self._mk_name("updated_name"), "description": "metadata-only update"}, + }, + "track_owned": False, + }, + { + "name": "update_provider_data_llm_only_success", + "expected": {OUTCOME_SUCCESS}, + "call": self._call_update, + "payload": { + "deployment_id": await self._ensure_seed_for_update("provider_only"), + "body": {"provider_data": self._provider_data_update(llm=self.llm)}, + }, + "track_owned": False, + }, + ] + results.extend(await self._run_http_scenarios(scenarios)) + return results + + async def _run_payload_validation_scenarios(self) -> list[ScenarioResult]: + scenarios = [ + { + "name": "create_reject_missing_add_flows_and_upsert_tools", + "expected": {OUTCOME_HTTP_422}, + "call": self._call_create, + "payload": self._create_request_payload( + name=self._mk_name("invalid_create_no_ops"), + provider_data={"llm": self.llm, "connections": []}, + ), + "track_owned": False, + }, + { + "name": "create_reject_duplicate_connection_app_ids", + "expected": {OUTCOME_HTTP_422}, + "call": self._call_create, + "payload": self._create_request_payload( + name=self._mk_name("invalid_create_dup_conn"), + provider_data=self._provider_data_create( + add_flows=[{"flow_version_id": self.flow_version_ids[0], "app_ids": ["cfg-shared"]}], + connections=[ + {"app_id": "cfg-shared", "credentials": [{"key": "k1", "value": "v1", "source": "raw"}]}, + {"app_id": "cfg-shared", "credentials": [{"key": "k2", "value": "v2", "source": "raw"}]}, + ], + ), + ), + "track_owned": False, + }, + { + "name": "create_reject_unused_connection_app_ids", + "expected": {OUTCOME_HTTP_422}, + "call": self._call_create, + "payload": self._create_request_payload( + name=self._mk_name("invalid_create_unused_conn"), + provider_data=self._provider_data_create( + add_flows=[{"flow_version_id": self.flow_version_ids[0], "app_ids": []}], + connections=[ + { + "app_id": "cfg-unused", + "credentials": [{"key": "k", "value": "v", "source": "raw"}], + } + ], + ), + ), + "track_owned": False, + }, + { + "name": "update_reject_empty_body", + "expected": {OUTCOME_HTTP_422}, + "call": self._call_update, + "payload": {"deployment_id": str(uuid4()), "body": {}}, + "track_owned": False, + }, + { + "name": "update_reject_add_remove_overlap", + "expected": {OUTCOME_HTTP_422}, + "call": self._call_update, + "payload": { + "deployment_id": await self._ensure_seed_for_update("invalid_update_overlap"), + "body": { + "provider_data": self._provider_data_update( + upsert_flows=[ + { + "flow_version_id": self.flow_version_ids[0], + "add_app_ids": ["cfg-race"], + "remove_app_ids": ["cfg-race"], + } + ], + ) + }, + }, + "track_owned": False, + }, + ] + return await self._run_http_scenarios(scenarios) + + async def _run_large_complex_payload_scenarios(self, tiers: list[str] | None = None) -> list[ScenarioResult]: + results: list[ScenarioResult] = [] + tier_order = tiers or LARGE_PAYLOAD_TIER_ORDER + for tier in tier_order: + tier_config = LARGE_PAYLOAD_TIER_CONFIGS[tier] + tier_label = tier.lower() + print(f"Running large payload scenario tier={tier} ...") + + create_success_payload = self._create_request_payload( + name=self._mk_name(f"large_create_success_{tier_label}"), + provider_data=self._build_large_create_provider_data(tier=tier, tier_config=tier_config), + ) + create_reject_unused_payload = self._create_request_payload( + name=self._mk_name(f"large_create_unused_{tier_label}"), + provider_data=self._build_large_create_unused_connections_provider_data( + tier=tier, + tier_config=tier_config, + ), + ) + create_reject_duplicate_payload = self._create_request_payload( + name=self._mk_name(f"large_create_duplicate_{tier_label}"), + provider_data=self._build_large_create_duplicate_connections_provider_data( + tier=tier, + tier_config=tier_config, + ), + ) + create_scenarios = [ + { + "name": f"create_large_payload_success_fanout_tier_{tier_label}", + "expected": {OUTCOME_SUCCESS}, + "call": self._call_create, + "payload": create_success_payload, + "track_owned": True, + }, + { + "name": f"create_large_payload_reject_unused_connections_tier_{tier_label}", + "expected": {OUTCOME_HTTP_422}, + "call": self._call_create, + "payload": create_reject_unused_payload, + "track_owned": False, + "detail_contains": "not referenced by operations", + }, + { + "name": f"create_large_payload_reject_duplicate_connection_app_ids_tier_{tier_label}", + "expected": {OUTCOME_HTTP_422}, + "call": self._call_create, + "payload": create_reject_duplicate_payload, + "track_owned": False, + "detail_contains": "duplicate app_id", + }, + ] + results.extend(await self._run_http_scenarios(create_scenarios)) + + seed_flow_ids = self._large_seed_flow_ids() + primary_flow_id = seed_flow_ids[0] + remove_flow_id = seed_flow_ids[1] if len(seed_flow_ids) > 1 else None + + seeded_tool_ids = await self._ensure_large_tool_id_pool( + minimum_unique=max(2, int(tier_config["remove_tool_items"]) + 1), + tier=tier, + ) + if not seeded_tool_ids: + msg = f"Unable to seed provider tool ids for large payload scenarios (tier={tier})." + raise RuntimeError(msg) + mixed_seed = await self._create_large_update_seed_deployment(tier=tier, label="mixed") + fanout_seed = await self._create_large_update_seed_deployment(tier=tier, label="tool_fanout") + validation_seed = await self._create_large_update_seed_deployment(tier=tier, label="validation") + + mixed_provider_data = self._build_large_update_mixed_provider_data( + tier=tier, + tier_config=tier_config, + seeded_tool_ids=seeded_tool_ids, + primary_flow_id=primary_flow_id, + remove_flow_id=remove_flow_id, + ) + tool_fanout_provider_data = self._build_large_update_tool_fanout_provider_data( + tier=tier, + tier_config=tier_config, + seeded_tool_ids=seeded_tool_ids, + ) + overlap_provider_data = self._build_large_update_overlap_provider_data( + tier=tier, + tier_config=tier_config, + seeded_tool_ids=seeded_tool_ids, + primary_flow_id=primary_flow_id, + remove_flow_id=remove_flow_id, + ) + remove_conflict_provider_data = self._build_large_update_remove_conflict_provider_data( + tier=tier, + tier_config=tier_config, + seeded_tool_ids=seeded_tool_ids, + primary_flow_id=primary_flow_id, + remove_flow_id=remove_flow_id, + ) + unbind_raw_provider_data = self._build_large_update_unbind_raw_provider_data( + tier=tier, + tier_config=tier_config, + seeded_tool_ids=seeded_tool_ids, + primary_flow_id=primary_flow_id, + remove_flow_id=remove_flow_id, + ) + update_scenarios = [ + { + "name": f"update_large_payload_success_mixed_ops_tier_{tier_label}", + "expected": {OUTCOME_SUCCESS}, + "call": self._call_update, + "payload": { + "deployment_id": mixed_seed.deployment_id, + "body": {"provider_data": mixed_provider_data}, + }, + "track_owned": False, + }, + { + "name": f"update_large_payload_success_tool_id_fanout_tier_{tier_label}", + "expected": {OUTCOME_SUCCESS}, + "call": self._call_update, + "payload": { + "deployment_id": fanout_seed.deployment_id, + "body": {"provider_data": tool_fanout_provider_data}, + }, + "track_owned": False, + }, + { + "name": f"update_large_payload_reject_add_remove_overlap_tier_{tier_label}", + "expected": {OUTCOME_HTTP_422}, + "call": self._call_update, + "payload": { + "deployment_id": validation_seed.deployment_id, + "body": {"provider_data": overlap_provider_data}, + }, + "track_owned": False, + "detail_contains": "must not overlap", + }, + { + "name": f"update_large_payload_reject_remove_conflict_tier_{tier_label}", + "expected": {OUTCOME_HTTP_422}, + "call": self._call_update, + "payload": { + "deployment_id": validation_seed.deployment_id, + "body": {"provider_data": remove_conflict_provider_data}, + }, + "track_owned": False, + "detail_contains": "cannot be combined with upsert", + }, + { + "name": f"update_large_payload_reject_unbind_raw_app_ids_tier_{tier_label}", + "expected": {OUTCOME_HTTP_422}, + "call": self._call_update, + "payload": { + "deployment_id": validation_seed.deployment_id, + "body": {"provider_data": unbind_raw_provider_data}, + }, + "track_owned": False, + "detail_contains": "must not reference connections app_ids", + }, + { + "name": f"update_llm_only_fast_path_control_tier_{tier_label}", + "expected": {OUTCOME_SUCCESS}, + "call": self._call_update, + "payload": { + "deployment_id": validation_seed.deployment_id, + "body": {"provider_data": self._provider_data_update(llm=self.llm)}, + }, + "track_owned": False, + }, + ] + results.extend(await self._run_http_scenarios(update_scenarios)) + return results + + async def _run_attachment_patch_scenarios(self) -> list[ScenarioResult]: + update_seed = await self._create_owned_deployment( + name=self._mk_name("attachment_seed"), + provider_data=self._provider_data_create( + add_flows=[{"flow_version_id": self.flow_version_ids[0], "app_ids": []}], + ), + ) + scenarios = [ + { + "name": "update_patch_upsert_flows_add_binding", + "expected": {OUTCOME_SUCCESS}, + "call": self._call_update, + "payload": { + "deployment_id": update_seed.deployment_id, + "body": { + "provider_data": self._provider_data_update( + upsert_flows=[ + { + "flow_version_id": self.flow_version_ids[0], + "add_app_ids": [], + "remove_app_ids": [], + } + ], + ) + }, + }, + "track_owned": False, + }, + ] + if len(self.flow_version_ids) > 1: + scenarios.append( + { + "name": "update_patch_add_second_flow_then_remove", + "expected": {OUTCOME_SUCCESS}, + "call": self._call_update_chain_add_remove, + "payload": { + "deployment_id": update_seed.deployment_id, + "add_flow_version_id": self.flow_version_ids[1], + }, + "track_owned": False, + } + ) + return await self._run_http_scenarios(scenarios) + + async def _run_rollback_and_error_scenarios(self) -> list[ScenarioResult]: + duplicate_name = self._mk_name("dup_name") + created = await self._create_owned_deployment( + name=duplicate_name, + provider_data=self._provider_data_create( + add_flows=[{"flow_version_id": self.flow_version_ids[0], "app_ids": []}], + ), + ) + scenarios = [ + { + "name": "create_duplicate_name_conflict", + "expected": {OUTCOME_HTTP_409}, + "call": self._call_create, + "payload": self._create_request_payload( + name=duplicate_name, + provider_data=self._provider_data_create( + add_flows=[{"flow_version_id": self.flow_version_ids[0], "app_ids": []}], + ), + ), + "track_owned": False, + "detail_contains": "already exists", + }, + { + "name": "update_unknown_deployment_not_found", + "expected": {OUTCOME_HTTP_404}, + "call": self._call_update, + "payload": { + "deployment_id": str(uuid4()), + "body": {"provider_data": self._provider_data_update(llm=self.llm)}, + }, + "track_owned": False, + }, + { + "name": "update_remove_unknown_tool_id_noop_success", + "expected": {OUTCOME_SUCCESS}, + "call": self._call_update, + "payload": { + "deployment_id": created.deployment_id, + "body": { + "provider_data": self._provider_data_update( + remove_tools=["tool-not-owned-by-runner"], + ) + }, + }, + "track_owned": False, + }, + ] + return await self._run_http_scenarios(scenarios) + + async def _run_parallel_race_scenarios(self) -> list[ScenarioResult]: + results: list[ScenarioResult] = [] + for iteration in range(1, self.concurrency_repeat + 1): + race_name = self._mk_name(f"cc_dup_{iteration}") + create_body = self._create_request_payload( + name=race_name, + provider_data=self._provider_data_create( + add_flows=[{"flow_version_id": self.flow_version_ids[0], "app_ids": []}], + ), + ) + parallel_create_results = await self._run_parallel_calls( + { + "c1": (lambda body=create_body: self._call_create(body)), + "c2": (lambda body=create_body: self._call_create(body)), + } + ) + statuses = sorted(item.status_code for item in parallel_create_results.values()) + ok = statuses in ( + [HTTP_STATUS_CREATED, HTTP_STATUS_CONFLICT], + [HTTP_STATUS_CREATED, HTTP_STATUS_CREATED], + ) + detail = f"statuses={statuses}" + self._track_raw_connection_app_ids_from_request_payload(create_body) + for response in parallel_create_results.values(): + self._track_provider_artifacts_from_response(response) + if response.status_code == HTTP_STATUS_CREATED: + tracked = self._track_owned_from_create_response(response) + if tracked is None: + message = ( + f"parallel create iteration {iteration} returned HTTP 201 without " + "deployment ownership fields; cleanup may be incomplete" + ) + print(f"cleanup warning: {message}") + self.cleanup_issues.append(message) + results.append( + ScenarioResult( + name=f"cc_parallel_duplicate_create_{iteration}", + expected_outcomes={OUTCOME_SUCCESS}, + actual_outcome=OUTCOME_SUCCESS if ok else OUTCOME_FAILURE, + ok=ok, + detail=detail, + ) + ) + + update_seed = await self._create_owned_deployment( + name=self._mk_name(f"cc_update_seed_{iteration}"), + provider_data=self._provider_data_create( + add_flows=[{"flow_version_id": self.flow_version_ids[0], "app_ids": []}], + ), + ) + parallel_update_results = await self._run_parallel_calls( + { + "u1": ( + lambda deployment_id=update_seed.deployment_id, iteration_id=iteration: self._call_update( + { + "deployment_id": deployment_id, + "body": {"description": f"race-one-{iteration_id}"}, + } + ) + ), + "u2": ( + lambda deployment_id=update_seed.deployment_id, iteration_id=iteration: self._call_update( + { + "deployment_id": deployment_id, + "body": {"description": f"race-two-{iteration_id}"}, + } + ) + ), + } + ) + for response in parallel_update_results.values(): + self._track_provider_artifacts_from_response(response) + update_statuses = sorted(item.status_code for item in parallel_update_results.values()) + update_ok = update_statuses in ( + [HTTP_STATUS_OK, HTTP_STATUS_OK], + [HTTP_STATUS_OK, HTTP_STATUS_CONFLICT], + ) + results.append( + ScenarioResult( + name=f"cc_parallel_update_{iteration}", + expected_outcomes={OUTCOME_SUCCESS}, + actual_outcome=OUTCOME_SUCCESS if update_ok else OUTCOME_FAILURE, + ok=update_ok, + detail=f"statuses={update_statuses}", + ) + ) + return results + + async def _run_http_scenarios(self, scenarios: list[dict[str, Any]]) -> list[ScenarioResult]: + results: list[ScenarioResult] = [] + for index, scenario in enumerate(scenarios, start=1): + print(f"[{index}/{len(scenarios)}] {scenario['name']}") + try: + self._track_raw_connection_app_ids_from_request_payload(scenario["payload"]) + envelope = await scenario["call"](scenario["payload"]) + self._track_provider_artifacts_from_response(envelope) + if scenario.get("track_owned") or envelope.status_code == HTTP_STATUS_CREATED: + tracked = self._track_owned_from_create_response(envelope) + if envelope.status_code == HTTP_STATUS_CREATED and tracked is None: + message = ( + f"scenario {scenario['name']} returned HTTP 201 without deployment ownership fields; " + "cleanup may be incomplete" + ) + print(f"cleanup warning: {message}") + self.cleanup_issues.append(message) + outcome = self._to_outcome(envelope.status_code) + detail = envelope.detail + except Exception as exc: # noqa: BLE001 + outcome = OUTCOME_FAILURE + detail = str(exc) + detail_contains = str(scenario.get("detail_contains") or "").strip().lower() + detail_ok = not detail_contains or detail_contains in detail.lower() + ok = outcome in scenario["expected"] and detail_ok + results.append( + ScenarioResult( + name=scenario["name"], + expected_outcomes=scenario["expected"], + actual_outcome=outcome, + ok=ok, + detail=detail, + ) + ) + return results + + async def _call_create(self, payload: dict[str, Any]) -> HttpResponseEnvelope: + response = await self._client.post("/api/v1/deployments", json=payload) + return self._normalize_response(response) + + async def _call_update(self, payload: dict[str, Any]) -> HttpResponseEnvelope: + response = await self._client.patch(f"/api/v1/deployments/{payload['deployment_id']}", json=payload["body"]) + return self._normalize_response(response) + + async def _call_update_chain_add_remove(self, payload: dict[str, Any]) -> HttpResponseEnvelope: + add_response = await self._client.patch( + f"/api/v1/deployments/{payload['deployment_id']}", + json={ + "provider_data": self._provider_data_update( + upsert_flows=[ + { + "flow_version_id": payload["add_flow_version_id"], + "add_app_ids": [], + "remove_app_ids": [], + } + ] + ) + }, + ) + normalized_add = self._normalize_response(add_response) + self._track_provider_artifacts_from_response(normalized_add) + if normalized_add.status_code >= HTTP_STATUS_MULTIPLE_CHOICES: + return normalized_add + remove_response = await self._client.patch( + f"/api/v1/deployments/{payload['deployment_id']}", + json={ + "provider_data": self._provider_data_update( + remove_flows=[payload["add_flow_version_id"]], + ) + }, + ) + return self._normalize_response(remove_response) + + async def _create_owned_deployment(self, *, name: str, provider_data: dict[str, Any]) -> OwnedDeployment: + create_payload = self._create_request_payload(name=name, provider_data=provider_data) + self._track_raw_connection_app_ids_from_request_payload(create_payload) + result = await self._call_create(create_payload) + self._track_provider_artifacts_from_response(result) + if result.status_code != HTTP_STATUS_CREATED: + msg = f"create deployment failed: status={result.status_code} detail={result.detail}" + raise RuntimeError(msg) + tracked = self._track_owned_from_create_response(result) + if tracked is None: + msg = "create deployment succeeded but did not return deployment id/resource_key" + raise RuntimeError(msg) + return tracked + + async def _ensure_seed_for_update(self, suffix: str) -> str: + seed = await self._create_owned_deployment( + name=self._mk_name(f"upd_seed_{suffix}"), + provider_data=self._provider_data_create( + add_flows=[{"flow_version_id": self.flow_version_ids[0], "app_ids": []}], + ), + ) + return seed.deployment_id + + def _create_request_payload(self, *, name: str, provider_data: dict[str, Any]) -> dict[str, Any]: + if self.provider_id is None: + msg = "provider_id must be resolved before creating deployments" + raise RuntimeError(msg) + payload: dict[str, Any] = { + "provider_id": self.provider_id, + "name": name, + "description": "wxo deployments api e2e", + "type": "agent", + "provider_data": provider_data, + } + if self.project_id: + payload["project_id"] = self.project_id + return payload + + def _provider_data_create( + self, + *, + existing_agent_id: str | None = None, + add_flows: list[dict[str, Any]] | None = None, + upsert_tools: list[dict[str, Any]] | None = None, + connections: list[dict[str, Any]] | None = None, + ) -> dict[str, Any]: + normalized_add_flows: list[dict[str, Any]] = [] + for item in add_flows or []: + candidate = dict(item) + if not str(candidate.get("tool_name") or "").strip(): + flow_version_id = str(candidate.get("flow_version_id") or uuid4()).replace("-", "")[:12] + candidate["tool_name"] = self._mk_name(f"tool_{flow_version_id}") + normalized_add_flows.append(candidate) + payload: dict[str, Any] = { + "llm": self.llm, + "add_flows": normalized_add_flows, + "upsert_tools": upsert_tools or [], + "connections": connections or [], + } + if existing_agent_id: + payload["existing_agent_id"] = existing_agent_id + return payload + + def _provider_data_update( + self, + *, + llm: str | None = None, + connections: list[dict[str, Any]] | None = None, + upsert_flows: list[dict[str, Any]] | None = None, + upsert_tools: list[dict[str, Any]] | None = None, + remove_flows: list[str] | None = None, + remove_tools: list[str] | None = None, + ) -> dict[str, Any]: + return { + "llm": llm, + "connections": connections or [], + "upsert_flows": upsert_flows or [], + "upsert_tools": upsert_tools or [], + "remove_flows": remove_flows or [], + "remove_tools": remove_tools or [], + } + + def _large_seed_flow_ids(self) -> list[str]: + unique_flow_ids = list(dict.fromkeys(self.flow_version_ids)) + if not unique_flow_ids: + msg = "large payload scenarios require at least one flow_version_id." + raise RuntimeError(msg) + return unique_flow_ids[:2] if len(unique_flow_ids) > 1 else unique_flow_ids[:1] + + async def _create_large_update_seed_deployment(self, *, tier: str, label: str) -> OwnedDeployment: + add_flows = [ + { + "flow_version_id": flow_version_id, + "app_ids": [], + "tool_name": self._large_tool_name(tier=tier, scenario=f"seed_{label}", index=index), + } + for index, flow_version_id in enumerate(self._large_seed_flow_ids()) + ] + return await self._create_owned_deployment( + name=self._mk_name(f"large_update_seed_{tier.lower()}_{label}"), + provider_data=self._provider_data_create(add_flows=add_flows), + ) + + async def _ensure_large_tool_id_pool(self, *, minimum_unique: int, tier: str) -> list[str]: + target_unique = max(1, minimum_unique) + attempts = 0 + max_attempts = target_unique + 2 + while len(self.created_snapshot_ids) < target_unique and attempts < max_attempts: + attempts += 1 + await self._create_large_update_seed_deployment( + tier=tier, + label=f"tool_pool_{attempts:02d}", + ) + return sorted(self.created_snapshot_ids) + + def _app_id_namespace(self) -> str: + # Include seconds and random suffix to avoid cross-run app-id collisions. + return "".join(ch for ch in self.run_suffix.lower() if ch.isalnum())[:20] + + def _normalize_wxo_connection_app_id(self, app_id: str) -> str: + translated = str(app_id).strip().replace(" ", "_").replace("-", "_") + return "".join(ch for ch in translated if ch.isalnum() or ch == "_") + + def _build_large_app_ids(self, *, tier: str, prefix: str, count: int) -> list[str]: + namespace = self._app_id_namespace() + # Put the index before namespace so uniqueness survives provider-side truncation. + return [f"{prefix}-{tier.lower()}-{index:03d}-{namespace}" for index in range(max(1, count))] + + def _partition_evenly(self, values: list[str], partition_count: int) -> list[list[str]]: + normalized_count = max(1, partition_count) + buckets: list[list[str]] = [[] for _ in range(normalized_count)] + for index, value in enumerate(values): + buckets[index % normalized_count].append(value) + return buckets + + def _build_large_connections( + self, + *, + tier: str, + app_ids: list[str], + credentials_per_connection: int, + credential_prefix: str, + ) -> list[dict[str, Any]]: + connections: list[dict[str, Any]] = [] + normalized_cred_count = max(1, credentials_per_connection) + for connection_index, app_id in enumerate(app_ids): + credentials = [ + { + "key": f"{credential_prefix.upper()}_{credential_index:02d}", + "value": f"{tier.lower()}-{connection_index:03d}-{credential_index:02d}", + # Keep large-payload success scenarios self-contained and deterministic. + # Variable-sourced credentials require pre-existing variable records and + # can fail after connection creation, which then retries into 409 conflicts. + "source": "raw", + } + for credential_index in range(normalized_cred_count) + ] + connections.append({"app_id": app_id, "credentials": credentials}) + return connections + + def _large_tool_name(self, *, tier: str, scenario: str, index: int) -> str: + return f"wxo_{scenario}_{tier.lower()}_{index:03d}_tool" + + def _build_large_create_provider_data(self, *, tier: str, tier_config: dict[str, int]) -> dict[str, Any]: + app_ids = self._build_large_app_ids( + tier=tier, + prefix="cfg", + count=int(tier_config["connections"]), + ) + connections = self._build_large_connections( + tier=tier, + app_ids=app_ids, + credentials_per_connection=int(tier_config["credentials_per_connection"]), + credential_prefix="create", + ) + add_flow_count = max(1, int(tier_config["create_flow_items"])) + app_chunks = self._partition_evenly(app_ids, add_flow_count) + seed_flow_ids = self._large_seed_flow_ids() + add_flows = [] + for index, app_chunk in enumerate(app_chunks): + normalized_chunk = app_chunk or [app_ids[index % len(app_ids)]] + add_flows.append( + { + "flow_version_id": seed_flow_ids[index % len(seed_flow_ids)], + "app_ids": normalized_chunk, + "tool_name": self._large_tool_name(tier=tier, scenario="create_fanout", index=index), + } + ) + return self._provider_data_create( + add_flows=add_flows, + connections=connections, + ) + + def _build_large_create_unused_connections_provider_data( + self, + *, + tier: str, + tier_config: dict[str, int], + ) -> dict[str, Any]: + provider_data = self._build_large_create_provider_data(tier=tier, tier_config=tier_config) + extra_app_ids = self._build_large_app_ids( + tier=tier, + prefix="cfg-unused", + count=max(2, int(tier_config["credentials_per_connection"])), + ) + provider_data["connections"].extend( + self._build_large_connections( + tier=tier, + app_ids=extra_app_ids, + credentials_per_connection=int(tier_config["credentials_per_connection"]), + credential_prefix="unused", + ) + ) + return provider_data + + def _build_large_create_duplicate_connections_provider_data( + self, + *, + tier: str, + tier_config: dict[str, int], + ) -> dict[str, Any]: + provider_data = self._build_large_create_provider_data(tier=tier, tier_config=tier_config) + connections = provider_data.get("connections") or [] + if connections: + duplicate = dict(connections[0]) + duplicate_credentials = duplicate.get("credentials") or [] + duplicate["credentials"] = [dict(item) for item in duplicate_credentials] + connections.append(duplicate) + return provider_data + + def _split_tool_ids_for_upsert_remove( + self, + *, + tool_ids: list[str], + remove_target: int, + ) -> tuple[list[str], list[str]]: + unique_tool_ids = list(dict.fromkeys(item for item in tool_ids if str(item).strip())) + if not unique_tool_ids: + return [], [] + normalized_remove_target = max(0, remove_target) + remove_count = min(normalized_remove_target, max(0, len(unique_tool_ids) - 1)) + if remove_count == 0: + return unique_tool_ids, [] + remove_ids = unique_tool_ids[-remove_count:] + upsert_pool = unique_tool_ids[:-remove_count] or unique_tool_ids[:1] + return upsert_pool, remove_ids + + def _build_large_upsert_tools( + self, + *, + tier: str, + tool_ids: list[str], + app_ids: list[str], + item_count: int, + ) -> list[dict[str, Any]]: + if not tool_ids: + msg = f"No seeded tool ids available for tier={tier} upsert_tools payload." + raise RuntimeError(msg) + normalized_count = max(1, item_count) + add_app_chunks = self._partition_evenly(app_ids, normalized_count) + upsert_tools: list[dict[str, Any]] = [] + for index in range(normalized_count): + add_app_ids = add_app_chunks[index] if index < len(add_app_chunks) else [] + if not add_app_ids: + add_app_ids = [app_ids[index % len(app_ids)]] + upsert_tools.append( + { + "tool_id": tool_ids[index % len(tool_ids)], + "add_app_ids": add_app_ids, + "remove_app_ids": [], + } + ) + return upsert_tools + + def _build_legacy_remove_app_ids(self, *, tier: str, item_index: int, count: int) -> list[str]: + normalized_count = max(1, count) + return [f"legacy-{tier.lower()}-{item_index:03d}-{idx:03d}" for idx in range(normalized_count)] + + def _build_large_update_mixed_provider_data( + self, + *, + tier: str, + tier_config: dict[str, int], + seeded_tool_ids: list[str], + primary_flow_id: str, + remove_flow_id: str | None, + ) -> dict[str, Any]: + app_ids = self._build_large_app_ids( + tier=tier, + prefix="upd-mixed", + count=int(tier_config["connections"]), + ) + connections = self._build_large_connections( + tier=tier, + app_ids=app_ids, + credentials_per_connection=int(tier_config["credentials_per_connection"]), + credential_prefix="update", + ) + flow_item_count = max(1, int(tier_config["update_flow_items"])) + tool_item_count = max(1, int(tier_config["update_tool_items"]) // 2) + reference_chunks = self._partition_evenly(app_ids, flow_item_count + tool_item_count) + flow_add_chunks = reference_chunks[:flow_item_count] + tool_add_chunks = reference_chunks[flow_item_count:] + upsert_flows: list[dict[str, Any]] = [] + for index in range(flow_item_count): + add_app_ids = flow_add_chunks[index] if index < len(flow_add_chunks) else [] + if not add_app_ids: + add_app_ids = [app_ids[index % len(app_ids)]] + upsert_item: dict[str, Any] = { + "flow_version_id": primary_flow_id, + "add_app_ids": add_app_ids, + "remove_app_ids": [], + } + if index == 0: + upsert_item["tool_name"] = self._large_tool_name(tier=tier, scenario="update_mixed", index=index) + upsert_flows.append(upsert_item) + upsert_tool_pool, remove_tools = self._split_tool_ids_for_upsert_remove( + tool_ids=seeded_tool_ids, + remove_target=int(tier_config["remove_tool_items"]), + ) + tool_add_app_ids = [app_id for chunk in tool_add_chunks for app_id in chunk] or list(app_ids) + upsert_tools = self._build_large_upsert_tools( + tier=tier, + tool_ids=upsert_tool_pool or seeded_tool_ids, + app_ids=tool_add_app_ids, + item_count=tool_item_count, + ) + remove_flows = [remove_flow_id] if remove_flow_id else [] + return self._provider_data_update( + llm=self.llm, + connections=connections, + upsert_flows=upsert_flows, + upsert_tools=upsert_tools, + remove_flows=remove_flows, + remove_tools=remove_tools, + ) + + def _build_large_update_tool_fanout_provider_data( + self, + *, + tier: str, + tier_config: dict[str, int], + seeded_tool_ids: list[str], + ) -> dict[str, Any]: + app_ids = self._build_large_app_ids( + tier=tier, + prefix="upd-tools", + count=int(tier_config["connections"]), + ) + connections = self._build_large_connections( + tier=tier, + app_ids=app_ids, + credentials_per_connection=int(tier_config["credentials_per_connection"]), + credential_prefix="fanout", + ) + upsert_tool_pool, remove_tools = self._split_tool_ids_for_upsert_remove( + tool_ids=seeded_tool_ids, + remove_target=int(tier_config["remove_tool_items"]), + ) + upsert_tools = self._build_large_upsert_tools( + tier=tier, + tool_ids=upsert_tool_pool or seeded_tool_ids, + app_ids=app_ids, + item_count=int(tier_config["update_tool_items"]), + ) + return self._provider_data_update( + llm=self.llm, + connections=connections, + upsert_tools=upsert_tools, + remove_tools=remove_tools, + ) + + def _build_large_update_overlap_provider_data( + self, + *, + tier: str, + tier_config: dict[str, int], + seeded_tool_ids: list[str], + primary_flow_id: str, + remove_flow_id: str | None, + ) -> dict[str, Any]: + provider_data = self._build_large_update_mixed_provider_data( + tier=tier, + tier_config=tier_config, + seeded_tool_ids=seeded_tool_ids, + primary_flow_id=primary_flow_id, + remove_flow_id=remove_flow_id, + ) + connections = provider_data.get("connections") or [] + upsert_flows = provider_data.get("upsert_flows") or [] + if not connections or not upsert_flows: + return provider_data + overlap_app_id = f"overlap-{tier.lower()}-{self._app_id_namespace()}-000" + existing_add_app_ids = [ + str(item).strip() for item in upsert_flows[0].get("add_app_ids", []) if str(item).strip() + ] + if overlap_app_id not in existing_add_app_ids: + existing_add_app_ids.append(overlap_app_id) + upsert_flows[0]["add_app_ids"] = existing_add_app_ids + upsert_flows[0]["remove_app_ids"] = [overlap_app_id] + return provider_data + + def _build_large_update_remove_conflict_provider_data( + self, + *, + tier: str, + tier_config: dict[str, int], + seeded_tool_ids: list[str], + primary_flow_id: str, + remove_flow_id: str | None, + ) -> dict[str, Any]: + provider_data = self._build_large_update_mixed_provider_data( + tier=tier, + tier_config=tier_config, + seeded_tool_ids=seeded_tool_ids, + primary_flow_id=primary_flow_id, + remove_flow_id=remove_flow_id, + ) + upsert_flows = provider_data.get("upsert_flows") or [] + if upsert_flows: + provider_data["remove_flows"] = [upsert_flows[0]["flow_version_id"]] + return provider_data + + def _build_large_update_unbind_raw_provider_data( + self, + *, + tier: str, + tier_config: dict[str, int], + seeded_tool_ids: list[str], + primary_flow_id: str, + remove_flow_id: str | None, + ) -> dict[str, Any]: + provider_data = self._build_large_update_mixed_provider_data( + tier=tier, + tier_config=tier_config, + seeded_tool_ids=seeded_tool_ids, + primary_flow_id=primary_flow_id, + remove_flow_id=remove_flow_id, + ) + connections = provider_data.get("connections") or [] + upsert_flows = provider_data.get("upsert_flows") or [] + if not connections or not upsert_flows: + return provider_data + raw_app_id = str(connections[0].get("app_id", "")).strip() + if not raw_app_id: + return provider_data + first_upsert_flow = upsert_flows[0] + first_upsert_flow["remove_app_ids"] = [raw_app_id] + first_upsert_flow["add_app_ids"] = [ + item for item in first_upsert_flow.get("add_app_ids", []) if item != raw_app_id + ] + + is_referenced_on_add_side = any(raw_app_id in item.get("add_app_ids", []) for item in upsert_flows) or any( + raw_app_id in item.get("add_app_ids", []) for item in provider_data.get("upsert_tools", []) + ) + if not is_referenced_on_add_side: + upsert_tools = provider_data.get("upsert_tools") or [] + if upsert_tools: + first_tool_add_app_ids = list(upsert_tools[0].get("add_app_ids", [])) + if raw_app_id not in first_tool_add_app_ids: + first_tool_add_app_ids.append(raw_app_id) + upsert_tools[0]["add_app_ids"] = first_tool_add_app_ids + return provider_data + + def _normalize_response(self, response: httpx.Response) -> HttpResponseEnvelope: + payload: dict[str, Any] | list[Any] | None + try: + payload = response.json() + except ValueError: + payload = None + detail = self._detail_from_payload(payload) or (response.text[:500] if response.text else "") + return HttpResponseEnvelope(status_code=response.status_code, payload=payload, detail=detail) + + def _detail_from_payload(self, payload: dict[str, Any] | list[Any] | None) -> str: + if isinstance(payload, dict): + detail = payload.get("detail") + if isinstance(detail, str): + return detail + if isinstance(detail, list): + flattened = [str(item.get("msg", item)) if isinstance(item, dict) else str(item) for item in detail] + return "; ".join(flattened) + return str(payload)[:500] + if isinstance(payload, list): + return str(payload)[:500] + return "" + + def _track_owned_from_create_response(self, envelope: HttpResponseEnvelope) -> OwnedDeployment | None: + if not isinstance(envelope.payload, dict): + return None + deployment_id = envelope.payload.get("id") + resource_key = envelope.payload.get("resource_key") + name = envelope.payload.get("name") + if not deployment_id or not resource_key or not name: + return None + owned = OwnedDeployment(deployment_id=str(deployment_id), resource_key=str(resource_key), name=str(name)) + self.owned_deployments[owned.deployment_id] = owned + return owned + + def _track_provider_artifacts_from_response(self, envelope: HttpResponseEnvelope) -> None: + if envelope.status_code >= HTTP_STATUS_MULTIPLE_CHOICES: + return + if not isinstance(envelope.payload, dict): + return + provider_data = envelope.payload.get("provider_data") + if not isinstance(provider_data, dict): + return + created_app_ids = provider_data.get("created_app_ids") + if isinstance(created_app_ids, list): + for app_id in created_app_ids: + normalized = str(app_id).strip() + if normalized: + self.created_config_ids.add(normalized) + created_tools = provider_data.get("created_tools") + if isinstance(created_tools, list): + for item in created_tools: + if not isinstance(item, dict): + continue + tool_id = str(item.get("tool_id") or item.get("id") or "").strip() + if tool_id: + self.created_snapshot_ids.add(tool_id) + + def _track_raw_connection_app_ids_from_request_payload(self, payload: dict[str, Any]) -> None: + provider_data: Any = payload.get("provider_data") + if not isinstance(provider_data, dict): + body = payload.get("body") + if isinstance(body, dict): + provider_data = body.get("provider_data") + if not isinstance(provider_data, dict): + return + + connections = provider_data.get("connections") + if not isinstance(connections, list): + return + for connection in connections: + if not isinstance(connection, dict): + continue + app_id = str(connection.get("app_id") or "").strip() + if app_id: + self.requested_raw_connection_app_ids.add(app_id) + normalized = self._normalize_wxo_connection_app_id(app_id) + if normalized: + self.requested_raw_connection_app_ids.add(normalized) + + async def _delete_owned_deployment(self, deployment_id: str, *, include_provider: bool) -> int: + if deployment_id not in self.owned_deployments: + msg = f"refusing to delete unmanaged deployment id: {deployment_id}" + raise RuntimeError(msg) + response = await self._client.delete( + f"/api/v1/deployments/{deployment_id}", + params={"include_provider": str(include_provider).lower()}, + ) + if response.status_code not in {HTTP_STATUS_NO_CONTENT, HTTP_STATUS_NOT_FOUND}: + normalized = self._normalize_response(response) + msg = ( + f"delete deployment failed id={deployment_id} include_provider={include_provider} " + f"status={normalized.status_code} detail={normalized.detail}" + ) + raise RuntimeError(msg) + # Treat NOT_FOUND as already deleted and clear ownership tracking. + self.owned_deployments.pop(deployment_id, None) + return response.status_code + + async def _cleanup_resources(self) -> None: + print("Running cleanup...") + active_ids = list(self.owned_deployments.keys()) + orphaned_keys = list(self.orphaned_provider_resource_keys) + print( + "cleanup targets: " + f"deployments={len(active_ids)} " + f"orphaned_agents={len(orphaned_keys)} " + f"snapshots={len(self.created_snapshot_ids)} " + f"configs={len(self.created_config_ids | self.requested_raw_connection_app_ids)} " + f"flows={len(self.created_flow_ids)} " + f"provider_account={'1' if self.created_provider_account_id else '0'}" + ) + for index, deployment_id in enumerate(active_ids, start=1): + with_provider = True + try: + print(f"cleanup: deleting deployment {index}/{len(active_ids)} {deployment_id} ...") + status_code = await self._delete_owned_deployment(deployment_id, include_provider=with_provider) + if status_code == HTTP_STATUS_NO_CONTENT: + print(f"cleanup: deleted deployment {deployment_id}") + else: + print(f"cleanup: deployment already missing {deployment_id}") + except Exception as exc: # noqa: BLE001 + message = f"could not delete deployment {deployment_id}: {exc}" + print(f"cleanup warning: {message}") + self.cleanup_issues.append(message) + + for index, resource_key in enumerate(orphaned_keys, start=1): + try: + print(f"cleanup: deleting orphaned provider resource {index}/{len(orphaned_keys)} {resource_key} ...") + # Re-onboard owned orphan in DB, then delete with provider delete enabled. + tmp = await self._create_owned_deployment( + name=self._mk_name("cleanup_orphan"), + provider_data=self._provider_data_create(existing_agent_id=resource_key), + ) + await self._delete_owned_deployment(tmp.deployment_id, include_provider=True) + self.orphaned_provider_resource_keys.discard(resource_key) + print(f"cleanup: deleted orphaned provider resource {resource_key}") + except Exception as exc: # noqa: BLE001 + deleted_directly = await self._delete_provider_agent_direct(resource_key) + if deleted_directly: + self.orphaned_provider_resource_keys.discard(resource_key) + print(f"cleanup: deleted orphaned provider resource directly {resource_key}") + continue + message = f"could not cleanup orphaned provider resource {resource_key}: {exc}" + print(f"cleanup warning: {message}") + self.cleanup_issues.append(message) + + await self._cleanup_created_provider_artifacts() + + if self.owned_deployments: + message = f"owned deployment leftovers remain: {sorted(self.owned_deployments.keys())}" + print(f"cleanup warning: {message}") + self.cleanup_issues.append(message) + if self.orphaned_provider_resource_keys: + message = f"orphaned provider leftovers remain: {sorted(self.orphaned_provider_resource_keys)}" + print(f"cleanup warning: {message}") + self.cleanup_issues.append(message) + await self._cleanup_created_flows() + if self.created_provider_account_id: + await self._cleanup_created_provider_account() + + async def _cleanup_created_flows(self) -> None: + created_flow_ids = list(self.created_flow_ids) + for index, flow_id in enumerate(created_flow_ids, start=1): + print(f"cleanup: deleting flow {index}/{len(created_flow_ids)} {flow_id} ...") + response = await self._client.delete(f"/api/v1/flows/{flow_id}") + if response.status_code in {HTTP_STATUS_OK, HTTP_STATUS_NO_CONTENT, HTTP_STATUS_NOT_FOUND}: + self.created_flow_ids.discard(flow_id) + if response.status_code == HTTP_STATUS_NOT_FOUND: + print(f"cleanup: flow already missing {flow_id}") + else: + print(f"cleanup: deleted flow {flow_id}") + continue + normalized = self._normalize_response(response) + message = ( + f"could not delete runner-created flow {flow_id}: " + f"status={normalized.status_code} detail={normalized.detail}" + ) + print(f"cleanup warning: {message}") + self.cleanup_issues.append(message) + + async def _cleanup_created_provider_artifacts(self) -> None: + if self._client_mod is None or self.provider_id is None: + return + try: + clients = await self._client_mod.get_provider_clients(user_id=self.user_id, db=self.db) + except Exception as exc: # noqa: BLE001 + message = f"could not resolve provider clients for artifact cleanup: {exc}" + print(f"cleanup warning: {message}") + self.cleanup_issues.append(message) + return + + snapshot_ids = sorted(self.created_snapshot_ids) + for index, snapshot_id in enumerate(snapshot_ids, start=1): + try: + print(f"cleanup: deleting snapshot {index}/{len(snapshot_ids)} {snapshot_id} ...") + await asyncio.to_thread(clients.tool.delete, snapshot_id) + self.created_snapshot_ids.discard(snapshot_id) + print(f"cleanup: deleted snapshot {snapshot_id}") + except ClientAPIException as exc: + status_code = getattr(getattr(exc, "response", None), "status_code", None) + if status_code == HTTP_STATUS_NOT_FOUND: + self.created_snapshot_ids.discard(snapshot_id) + print(f"cleanup: snapshot already missing {snapshot_id}") + continue + message = f"could not delete snapshot {snapshot_id}: {exc}" + print(f"cleanup warning: {message}") + self.cleanup_issues.append(message) + except Exception as exc: # noqa: BLE001 + message = f"could not delete snapshot {snapshot_id}: {exc}" + print(f"cleanup warning: {message}") + self.cleanup_issues.append(message) + + config_cleanup_candidates = sorted(self.created_config_ids | self.requested_raw_connection_app_ids) + for index, config_id in enumerate(config_cleanup_candidates, start=1): + try: + print(f"cleanup: deleting config {index}/{len(config_cleanup_candidates)} {config_id} ...") + await asyncio.to_thread(clients.connections.delete, config_id) + self.created_config_ids.discard(config_id) + self.requested_raw_connection_app_ids.discard(config_id) + print(f"cleanup: deleted config {config_id}") + except ClientAPIException as exc: + status_code = getattr(getattr(exc, "response", None), "status_code", None) + if status_code == HTTP_STATUS_NOT_FOUND: + self.created_config_ids.discard(config_id) + self.requested_raw_connection_app_ids.discard(config_id) + print(f"cleanup: config already missing {config_id}") + continue + message = f"could not delete config {config_id}: {exc}" + print(f"cleanup warning: {message}") + self.cleanup_issues.append(message) + except Exception as exc: # noqa: BLE001 + message = f"could not delete config {config_id}: {exc}" + print(f"cleanup warning: {message}") + self.cleanup_issues.append(message) + + if self.created_snapshot_ids: + message = f"snapshot leftovers remain: {sorted(self.created_snapshot_ids)}" + print(f"cleanup warning: {message}") + self.cleanup_issues.append(message) + if self.created_config_ids: + message = f"config leftovers remain: {sorted(self.created_config_ids)}" + print(f"cleanup warning: {message}") + self.cleanup_issues.append(message) + if self.requested_raw_connection_app_ids: + message = f"raw connection app-id leftovers remain: {sorted(self.requested_raw_connection_app_ids)}" + print(f"cleanup warning: {message}") + self.cleanup_issues.append(message) + + async def _delete_provider_agent_direct(self, resource_key: str) -> bool: + if self._client_mod is None: + return False + try: + clients = await self._client_mod.get_provider_clients(user_id=self.user_id, db=self.db) + await asyncio.to_thread(clients.agent.delete, resource_key) + except ClientAPIException as exc: + status_code = getattr(getattr(exc, "response", None), "status_code", None) + return status_code == HTTP_STATUS_NOT_FOUND + except Exception: # noqa: BLE001 + return False + else: + return True + + async def _setup_provider_clients_context(self) -> None: + if self.provider_id is None: + msg = "provider_id must be resolved before setting adapter client context" + raise RuntimeError(msg) + if self._client_mod is not None: + return + + import langflow.services.adapters.deployment.watsonx_orchestrate.client as client_mod + + self._client_mod = client_mod + self._original_resolve_wxo_client_credentials = client_mod.resolve_wxo_client_credentials + deployment_context = DeploymentAdapterContext(provider_id=UUID(self.provider_id)) + self._deployment_context_token = DeploymentProviderIDContext.set_current(deployment_context) + + async def _resolve_credentials(*, user_id, db, provider_id): # noqa: ARG001 + authenticator = client_mod.get_authenticator( + instance_url=self.instance_url, + api_key=self.provider_api_key, + ) + return WxOCredentials(instance_url=self.instance_url, authenticator=authenticator) + + client_mod.resolve_wxo_client_credentials = _resolve_credentials # type: ignore[assignment] + + async def _teardown_provider_clients_context(self) -> None: + if self._client_mod is not None and self._original_resolve_wxo_client_credentials is not None: + self._client_mod.resolve_wxo_client_credentials = self._original_resolve_wxo_client_credentials + self._client_mod.clear_provider_clients_request_context() + if self._deployment_context_token is not None: + DeploymentProviderIDContext.reset_current(self._deployment_context_token) + self._client_mod = None + self._original_resolve_wxo_client_credentials = None + self._deployment_context_token = None + if self.created_flow_ids: + message = f"flow leftovers remain: {sorted(self.created_flow_ids)}" + print(f"cleanup warning: {message}") + self.cleanup_issues.append(message) + + async def _cleanup_created_provider_account(self) -> None: + provider_id = self.created_provider_account_id + if provider_id is None: + return + response = await self._client.delete(f"/api/v1/deployments/providers/{provider_id}") + if response.status_code in {HTTP_STATUS_NO_CONTENT, HTTP_STATUS_NOT_FOUND}: + self.created_provider_account_id = None + return + normalized = self._normalize_response(response) + message = ( + f"could not delete runner-created provider account {provider_id}: " + f"status={normalized.status_code} detail={normalized.detail}" + ) + print(f"cleanup warning: {message}") + self.cleanup_issues.append(message) + + async def _resolve_or_create_provider_account(self) -> str: + existing_id = await self._find_provider_account_id_by_url() + if existing_id: + self.provider_id = existing_id + return existing_id + created_id = await self._create_provider_account_for_instance_url() + self.provider_id = created_id + self.created_provider_account_id = created_id + return created_id + + async def _find_provider_account_id_by_url(self) -> str | None: + wanted_url = self._normalize_url(self.instance_url) + page = 1 + size = 50 + while True: + response = await self._client.get( + "/api/v1/deployments/providers", + params={"page": page, "size": size}, + ) + normalized = self._normalize_response(response) + if normalized.status_code != HTTP_STATUS_OK: + msg = f"listing provider accounts failed: status={normalized.status_code} detail={normalized.detail}" + raise RuntimeError(msg) + payload = normalized.payload if isinstance(normalized.payload, dict) else {} + provider_accounts = payload.get("provider_accounts") if isinstance(payload, dict) else [] + if not isinstance(provider_accounts, list): + provider_accounts = [] + for account in provider_accounts: + if not isinstance(account, dict): + continue + if str(account.get("provider_key", "")).strip() != self.provider_key: + continue + provider_data = account.get("provider_data") + if not isinstance(provider_data, dict): + continue + account_url = self._normalize_url(str(provider_data.get("url", ""))) + if account_url != wanted_url: + continue + provider_id = str(account.get("id", "")).strip() + if provider_id: + return provider_id + total = payload.get("total") if isinstance(payload, dict) else None + if not isinstance(total, int): + if len(provider_accounts) < size: + return None + elif page * size >= total: + return None + page += 1 + + async def _create_provider_account_for_instance_url(self) -> str: + create_payload: dict[str, Any] = { + "name": self._mk_name("provider-account"), + "provider_key": self.provider_key, + "provider_data": { + "url": self.instance_url, + "api_key": self.provider_api_key, + }, + } + if self.provider_tenant_id: + create_payload["provider_data"]["tenant_id"] = self.provider_tenant_id + response = await self._client.post("/api/v1/deployments/providers", json=create_payload) + normalized = self._normalize_response(response) + if normalized.status_code != HTTP_STATUS_CREATED: + msg = f"creating provider account failed: status={normalized.status_code} detail={normalized.detail}" + raise RuntimeError(msg) + payload = normalized.payload if isinstance(normalized.payload, dict) else {} + provider_id = str(payload.get("id", "")).strip() + if not provider_id: + msg = "provider account create succeeded but response did not include id" + raise RuntimeError(msg) + return provider_id + + def _normalize_url(self, value: str) -> str: + return value.strip().rstrip("/").lower() + + async def _ensure_flow_versions(self) -> None: + if self.flow_version_ids: + return + provisioned_ids = await self._provision_flow_versions_from_starter_projects(self.starter_project_count) + if not provisioned_ids: + msg = "no flow versions were provisioned from starter projects" + raise RuntimeError(msg) + self.flow_version_ids = provisioned_ids + + async def _provision_flow_versions_from_starter_projects(self, count: int) -> list[str]: + starter_paths = self._resolve_starter_project_paths(count=count) + version_ids: list[str] = [] + for starter_path in starter_paths: + starter_payload = self._load_starter_project_payload(starter_path) + flow_payload = self._build_flow_create_payload(starter_payload=starter_payload, starter_path=starter_path) + flow_response = await self._client.post("/api/v1/flows/", json=flow_payload) + flow_envelope = self._normalize_response(flow_response) + if flow_envelope.status_code != HTTP_STATUS_CREATED: + msg = ( + f"creating flow from starter project failed ({starter_path.name}): " + f"status={flow_envelope.status_code} detail={flow_envelope.detail}" + ) + raise RuntimeError(msg) + flow_payload_body = flow_envelope.payload if isinstance(flow_envelope.payload, dict) else {} + flow_id = str(flow_payload_body.get("id", "")).strip() + if not flow_id: + msg = f"flow create response missing id for starter project {starter_path.name}" + raise RuntimeError(msg) + self.created_flow_ids.add(flow_id) + + snapshot_response = await self._client.post( + f"/api/v1/flows/{flow_id}/versions/", + json={"description": f"e2e version from {starter_path.stem}"}, + ) + snapshot_envelope = self._normalize_response(snapshot_response) + if snapshot_envelope.status_code != HTTP_STATUS_CREATED: + msg = ( + f"creating flow version failed for flow {flow_id} ({starter_path.name}): " + f"status={snapshot_envelope.status_code} detail={snapshot_envelope.detail}" + ) + raise RuntimeError(msg) + snapshot_payload = snapshot_envelope.payload if isinstance(snapshot_envelope.payload, dict) else {} + version_id = str(snapshot_payload.get("id", "")).strip() + if not version_id: + msg = f"flow version create response missing id for flow {flow_id}" + raise RuntimeError(msg) + version_ids.append(version_id) + return version_ids + + def _resolve_starter_project_paths(self, *, count: int) -> list[Path]: + starter_root = Path(__file__).resolve().parents[3] / "src/backend/base/langflow/initial_setup/starter_projects" + if not starter_root.is_dir(): + msg = f"starter projects directory not found: {starter_root}" + raise RuntimeError(msg) + if self.starter_project_files: + paths = [starter_root / item for item in self.starter_project_files] + else: + paths = sorted(starter_root.glob("*.json"))[:count] + missing = [str(path) for path in paths if not path.is_file()] + if missing: + msg = f"starter project file(s) not found: {missing}" + raise RuntimeError(msg) + if len(paths) < count: + msg = f"not enough starter project files to provision {count} flow versions" + raise RuntimeError(msg) + return paths + + def _load_starter_project_payload(self, starter_path: Path) -> dict[str, Any]: + try: + raw = starter_path.read_text(encoding="utf-8") + payload = json.loads(raw) + except Exception as exc: + msg = f"failed to load starter project JSON at {starter_path}: {exc}" + raise RuntimeError(msg) from exc + if not isinstance(payload, dict): + msg = f"starter project payload must be a JSON object: {starter_path}" + raise TypeError(msg) + return payload + + def _build_flow_create_payload(self, *, starter_payload: dict[str, Any], starter_path: Path) -> dict[str, Any]: + data = starter_payload.get("data") + if not isinstance(data, dict): + msg = f"starter project is missing object `data`: {starter_path}" + raise TypeError(msg) + payload: dict[str, Any] = { + "name": self._mk_name(starter_path.stem.lower().replace(" ", "_")), + "description": str(starter_payload.get("description") or f"e2e flow from {starter_path.stem}"), + "data": data, + "is_component": bool(starter_payload.get("is_component", False)), + "endpoint_name": None, + "tags": starter_payload.get("tags") if isinstance(starter_payload.get("tags"), list) else [], + } + for optional_key in ("icon", "icon_bg_color", "gradient", "webhook"): + if optional_key in starter_payload: + payload[optional_key] = starter_payload[optional_key] + return payload + + async def _run_parallel_calls( + self, + calls: dict[str, Callable[[], Awaitable[HttpResponseEnvelope]]], + ) -> dict[str, HttpResponseEnvelope]: + tasks = {name: asyncio.create_task(call()) for name, call in calls.items()} + gathered = await asyncio.gather(*tasks.values(), return_exceptions=True) + results: dict[str, HttpResponseEnvelope] = {} + for name, outcome in zip(tasks, gathered, strict=False): + if isinstance(outcome, Exception): + results[name] = HttpResponseEnvelope( + status_code=HTTP_STATUS_SERVER_ERROR, + payload=None, + detail=str(outcome), + ) + else: + results[name] = outcome + return results + + def _to_outcome(self, status_code: int) -> str: + if HTTP_STATUS_OK <= status_code < HTTP_STATUS_MULTIPLE_CHOICES: + return OUTCOME_SUCCESS + if status_code == HTTP_STATUS_NOT_FOUND: + return OUTCOME_HTTP_404 + if status_code == HTTP_STATUS_CONFLICT: + return OUTCOME_HTTP_409 + if status_code == HTTP_STATUS_UNPROCESSABLE: + return OUTCOME_HTTP_422 + if status_code >= HTTP_STATUS_SERVER_ERROR: + return OUTCOME_HTTP_500 + return OUTCOME_FAILURE + + def _mk_name(self, label: str) -> str: + sanitized = "".join(ch if ch.isalnum() or ch in {"_", "-"} else "-" for ch in label).strip("-") + self._name_counter += 1 + return f"wxo-api-{sanitized}-{self.run_suffix}-{self._name_counter:04d}" + + def _print_summary(self, results: list[ScenarioResult]) -> None: + print("\n=== Deployments API E2E Summary ===") + for result in results: + status = "PASS" if result.ok else "FAIL" + expected = ",".join(sorted(result.expected_outcomes)) + print(f"[{status}] {result.name}: expected={expected} got={result.actual_outcome} detail={result.detail}") + passed = sum(1 for item in results if item.ok) + failed = len(results) - passed + print(f"Totals: passed={passed} failed={failed} total={len(results)}") + + +def _parse_uuid_list(raw: str) -> list[str]: + values = [item.strip() for item in raw.split(",") if item.strip()] + if not values: + return [] + return [str(UUID(value)) for value in values] + + +def _parse_csv(raw: str) -> list[str]: + return [item.strip() for item in raw.split(",") if item.strip()] + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Run deployments API create/update matrix against /api/v1/deployments." + ) + parser.add_argument("--base-url", default=os.getenv("LANGFLOW_BASE_URL", "http://localhost:7860")) + parser.add_argument("--api-key", default=os.getenv("LANGFLOW_API_KEY", "")) + parser.add_argument("--instance-url", default=os.getenv("WXO_INSTANCE_URL", "")) + parser.add_argument("--provider-api-key", default=os.getenv("WXO_API_KEY", "")) + parser.add_argument("--provider-tenant-id", default=os.getenv("WXO_TENANT_ID")) + parser.add_argument("--provider-key", default=os.getenv("WXO_PROVIDER_KEY", "watsonx-orchestrate")) + parser.add_argument( + "--flow-version-ids", + default=os.getenv("WXO_E2E_FLOW_VERSION_IDS", ""), + help="Comma-separated flow version UUIDs. First is required; second enables add/remove patch scenario.", + ) + parser.add_argument("--project-id", default=os.getenv("LANGFLOW_PROJECT_ID")) + parser.add_argument( + "--starter-project-files", + default=os.getenv("WXO_E2E_STARTER_PROJECT_FILES", ""), + help=( + "Optional comma-separated starter project filenames from " + "src/backend/base/langflow/initial_setup/starter_projects." + ), + ) + parser.add_argument( + "--starter-project-count", + type=int, + default=int(os.getenv("WXO_E2E_STARTER_PROJECT_COUNT", "2")), + help="Number of starter projects to provision when --flow-version-ids is not provided.", + ) + parser.add_argument("--mode", choices=["live", "failpoint", "both"], default=os.getenv("WXO_E2E_MODE", "both")) + parser.add_argument("--llm", default=os.getenv("WXO_DEFAULT_LLM", DEFAULT_WXO_LLM)) + parser.add_argument( + "--timeout-secs", + type=int, + default=int(os.getenv("WXO_E2E_TIMEOUT_SECS", str(DEFAULT_TIMEOUT_SECS))), + ) + parser.add_argument( + "--concurrency-repeat", + type=int, + default=int(os.getenv("WXO_CONCURRENCY_REPEAT", str(DEFAULT_CONCURRENCY_REPEAT))), + ) + parser.add_argument( + "--test-subset", + choices=["full", "smoke-connections", "large-tier-s"], + default=os.getenv("WXO_E2E_TEST_SUBSET", "full"), + help="Run full matrix, a connection smoke subset, or only large tier-S scenarios.", + ) + parser.add_argument("--keep-resources", action="store_true") + parser.add_argument("--insecure", action="store_true", help="Disable TLS verification.") + return parser.parse_args() + + +def _require(value: str, env_name: str) -> str: + normalized = value.strip() + if not normalized: + msg = f"Missing required value for {env_name}" + raise ValueError(msg) + return normalized + + +async def _main() -> int: + load_dotenv() + args = _parse_args() + api_key = _require(args.api_key, "LANGFLOW_API_KEY/--api-key") + instance_url = _require(args.instance_url, "WXO_INSTANCE_URL/--instance-url") + provider_api_key = _require(args.provider_api_key, "WXO_API_KEY/--provider-api-key") + flow_version_ids = _parse_uuid_list(args.flow_version_ids) + starter_project_files = _parse_csv(args.starter_project_files) + runner = DeploymentsApiParallelE2E( + base_url=args.base_url, + api_key=api_key, + instance_url=instance_url, + provider_api_key=provider_api_key, + provider_tenant_id=args.provider_tenant_id, + provider_key=args.provider_key, + mode=args.mode, + test_subset=args.test_subset, + keep_resources=args.keep_resources, + llm=args.llm, + flow_version_ids=flow_version_ids, + starter_project_files=starter_project_files, + starter_project_count=max(1, args.starter_project_count), + project_id=str(UUID(args.project_id)) if args.project_id else None, + timeout_secs=args.timeout_secs, + concurrency_repeat=max(1, args.concurrency_repeat), + verify_tls=not args.insecure, + ) + return await runner.run() + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(_main())) diff --git a/src/backend/base/langflow/api/v1/deployments.py b/src/backend/base/langflow/api/v1/deployments.py index cc53922635..3c01ed8216 100644 --- a/src/backend/base/langflow/api/v1/deployments.py +++ b/src/backend/base/langflow/api/v1/deployments.py @@ -485,10 +485,14 @@ async def create_deployment( db=session, ) else: + # Existing-resource create starts as DB-only onboarding: no provider + # mutation is performed and created_* response fields stay empty. provider_create_result = deployment_mapper.util_create_result_from_existing_resource( existing_resource_key=str(existing_resource_key), ) if should_mutate_existing_resource: + # When create payload includes add_flows/upsert_tools, run provider + # update and normalize the update result into create-style created_*. adapter_payload = await deployment_mapper.resolve_deployment_update_for_existing_create( user_id=current_user.id, project_id=project_id, diff --git a/src/backend/base/langflow/api/v1/mappers/deployments/watsonx_orchestrate/mapper.py b/src/backend/base/langflow/api/v1/mappers/deployments/watsonx_orchestrate/mapper.py index 1851289b4d..45b9c237ea 100644 --- a/src/backend/base/langflow/api/v1/mappers/deployments/watsonx_orchestrate/mapper.py +++ b/src/backend/base/langflow/api/v1/mappers/deployments/watsonx_orchestrate/mapper.py @@ -504,6 +504,28 @@ class WatsonxOrchestrateDeploymentMapper(BaseDeploymentMapper): provider_result=create_provider_result, ) + def util_create_result_from_existing_resource( + self, + *, + existing_resource_key: str, + ) -> DeploymentCreateResult: + """Build a create-style result payload for DB-only onboarding. + + This path is used when create request includes ``existing_agent_id`` + without create-time mutation operations. ``created_*`` fields represent + what this request created, so they are intentionally empty here. + """ + create_provider_result = self._parse_required_payload_slot( + slot=WXO_ADAPTER_PAYLOAD_SCHEMAS.deployment_create_result, + slot_name="deployment_create_result", + raw={"app_ids": [], "tools_with_refs": []}, + operation="building the create response for the existing resource", + ) + return DeploymentCreateResult( + id=existing_resource_key, + provider_result=create_provider_result.model_dump(mode="json"), + ) + async def _resolve_provider_payload_from_create_api( self, *, diff --git a/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/config.py b/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/config.py index ca4c7e7102..3f005ec5c8 100644 --- a/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/config.py +++ b/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/config.py @@ -61,13 +61,21 @@ async def create_config( config: DeploymentConfig, user_id: IdLike, db: AsyncSession, + created_app_ids_journal: list[str] | None = None, ) -> str: - """Create/update a wxO draft key-value connection config plus runtime credentials.""" + """Create/update a wxO draft key-value connection config plus runtime credentials. + + When ``created_app_ids_journal`` is provided, ``app_id`` is appended + immediately after provider connection creation succeeds so rollback can + clean up partially completed creates. + """ app_id = validate_wxo_name(config.name) env_var_keys = list((config.environment_variables or {}).keys()) logger.debug("create_config: app_id='%s', env_var_keys=%s", app_id, env_var_keys) await asyncio.to_thread(clients.connections.create, payload={"app_id": app_id}) + if created_app_ids_journal is not None: + created_app_ids_journal.append(app_id) wxo_config = ConnectionConfiguration( app_id=app_id, diff --git a/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/create.py b/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/create.py index caa3d55c9f..9e7c93572d 100644 --- a/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/create.py +++ b/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/create.py @@ -28,7 +28,6 @@ from langflow.services.adapters.deployment.watsonx_orchestrate.core.shared impor OrderedUniqueStrs, RawConnectionCreatePlan, RawToolCreatePlan, - create_connection_with_conflict_mapping, create_raw_tools_with_bindings, log_batch_errors, resolve_connections_for_operations, @@ -199,11 +198,15 @@ async def apply_provider_create_plan_with_rollback( # - operation_to_provider_app_id: operation app_id → provider app_id # (identity mapping for both existing and raw-created connections). # - resolved_connections: provider_app_id → connection_id map for bind calls. + # - created_app_ids_journal: app_ids recorded immediately after successful + # provider connection creation; used to ensure rollback sees partial + # successes even if create later fails before returning. created_snapshot_bindings: list[WatsonxToolRefBinding] = [] created_tool_app_bindings: list[WatsonxToolAppBinding] = [] agent_create_response = None operation_to_provider_app_id: dict[str, str] = {} resolved_connections: dict[str, str] = {} + created_app_ids_journal: list[str] = [] try: try: @@ -215,7 +218,7 @@ async def apply_provider_create_plan_with_rollback( raw_connections_to_create=plan.raw_connections_to_create, error_prefix=ErrorPrefix.CREATE.value, validate_connection_fn=validate_connection, - create_connection_fn=create_connection_with_conflict_mapping, + created_app_ids_journal=created_app_ids_journal, ) operation_to_provider_app_id = connection_result.operation_to_provider_app_id resolved_connections = connection_result.resolved_connections diff --git a/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/shared.py b/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/shared.py index eff05d5111..048470df26 100644 --- a/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/shared.py +++ b/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/shared.py @@ -22,7 +22,11 @@ from langflow.services.adapters.deployment.watsonx_orchestrate.core.tools import create_and_upload_wxo_flow_tools_with_bindings, ) from langflow.services.adapters.deployment.watsonx_orchestrate.payloads import WatsonxResultToolRefBinding -from langflow.services.adapters.deployment.watsonx_orchestrate.utils import extract_error_detail +from langflow.services.adapters.deployment.watsonx_orchestrate.utils import ( + dedupe_list, + extract_error_detail, + validate_wxo_name, +) if TYPE_CHECKING: from collections.abc import Awaitable, Callable, Iterator @@ -72,6 +76,12 @@ class RawConnectionCreatePlan: provider_app_id: str payload: WatsonxConnectionRawPayload + def __post_init__(self) -> None: + # operations[*].app_ids use operation_app_id as the caller-visible key. + # provider_app_id is used for provider calls and must follow wxO rules. + # Normalizing here keeps create/validate/rollback on one canonical id. + self.provider_app_id = validate_wxo_name(self.provider_app_id) + @dataclass(slots=True) class RawToolCreatePlan: @@ -116,6 +126,7 @@ async def create_connection_with_conflict_mapping( user_id: IdLike, db: AsyncSession, error_prefix: str, + created_app_ids_journal: list[str] | None = None, ) -> str: from lfx.services.adapters.deployment.schema import DeploymentConfig @@ -133,6 +144,7 @@ async def create_connection_with_conflict_mapping( config=config_payload, user_id=user_id, db=db, + created_app_ids_journal=created_app_ids_journal, ) except (ClientAPIException, HTTPException) as exc: if isinstance(exc, ClientAPIException): @@ -163,7 +175,7 @@ async def resolve_connections_for_operations( raw_connections_to_create: list[RawConnectionCreatePlan], error_prefix: str, validate_connection_fn: Callable[..., Awaitable[object]] = validate_connection, - create_connection_fn: Callable[..., Awaitable[str]] = create_connection_with_conflict_mapping, + created_app_ids_journal: list[str] | None = None, ) -> ConnectionResolutionResult: logger.debug( "resolve_connections_for_operations: existing_app_ids=%s, raw_to_create=%d", @@ -187,15 +199,17 @@ async def resolve_connections_for_operations( created_app_ids=[], ) + journal = created_app_ids_journal if created_app_ids_journal is not None else [] created_connections_results = await asyncio.gather( *( - create_connection_fn( + create_connection_with_conflict_mapping( clients=clients, app_id=create_plan.provider_app_id, payload=create_plan.payload, user_id=user_id, db=db, error_prefix=error_prefix, + created_app_ids_journal=journal, ) for create_plan in raw_connections_to_create ), @@ -203,7 +217,7 @@ async def resolve_connections_for_operations( ) create_connection_errors: list[Exception] = [] - created_app_ids_journal: list[str] = [] + created_app_ids: list[str] = [] for result in created_connections_results: if isinstance(result, BaseException): if isinstance(result, Exception): @@ -213,26 +227,34 @@ async def resolve_connections_for_operations( RuntimeError(f"Connection create failed with non-standard exception: {type(result).__name__}") ) continue - created_app_ids_journal.append(result) - created_app_ids = list(dict.fromkeys(created_app_ids_journal)) + created_app_ids.append(result) if create_connection_errors: + rollback_app_ids = dedupe_list([*journal, *created_app_ids]) logger.debug( "resolve_connections_for_operations: %d errors, created_app_ids=%s", len(create_connection_errors), - created_app_ids, + rollback_app_ids, ) - raise ConnectionCreateBatchError(created_app_ids=created_app_ids, errors=create_connection_errors) + raise ConnectionCreateBatchError(created_app_ids=rollback_app_ids, errors=create_connection_errors) - validated_created_connections: list[object] = await asyncio.gather( - *( - retry_create( - validate_connection_fn, - clients.connections, - app_id=create_plan.provider_app_id, + try: + validated_created_connections: list[object] = await asyncio.gather( + *( + retry_create( + validate_connection_fn, + clients.connections, + app_id=create_plan.provider_app_id, + ) + for create_plan in raw_connections_to_create ) - for create_plan in raw_connections_to_create ) - ) + except Exception as exc: + rollback_app_ids = dedupe_list([*journal, *created_app_ids]) + logger.debug( + "resolve_connections_for_operations: validation error, created_app_ids=%s", + rollback_app_ids, + ) + raise ConnectionCreateBatchError(created_app_ids=rollback_app_ids, errors=[exc]) from exc for create_plan, connection in zip(raw_connections_to_create, validated_created_connections, strict=True): operation_to_provider_app_id[create_plan.operation_app_id] = create_plan.provider_app_id resolved_connections[create_plan.provider_app_id] = connection.connection_id # type: ignore[attr-defined] diff --git a/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/update.py b/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/update.py index 004f955a33..a839eb1821 100644 --- a/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/update.py +++ b/src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/update.py @@ -25,7 +25,6 @@ from langflow.services.adapters.deployment.watsonx_orchestrate.core.shared impor OrderedUniqueStrs, RawConnectionCreatePlan, RawToolCreatePlan, - create_connection_with_conflict_mapping, create_raw_tools_with_bindings, log_batch_errors, resolve_connections_for_operations, @@ -449,6 +448,9 @@ async def apply_provider_update_plan_with_rollback( # - referenced_snapshot_bindings: full operation correlation set. # - final_update_payload: outbound agent patch payload (spec + tools). # - rollback_agent_payload: best-effort restore payload for agent rollback. + # - created_app_ids_journal: app_ids recorded immediately after successful + # provider connection creation; used to ensure rollback sees partial + # successes even if create later fails before returning. resolved_connections: dict[str, str] = {} operation_to_provider_app_id: dict[str, str] = {app_id: app_id for app_id in plan.existing_app_ids} created_snapshot_ids: list[str] = [] @@ -456,6 +458,7 @@ async def apply_provider_update_plan_with_rollback( created_snapshot_bindings: list[WatsonxResultToolRefBinding] = [] final_update_payload = dict(update_payload) rollback_agent_payload: dict[str, Any] = {} + created_app_ids_journal: list[str] = [] # Pre-seed resolved_connections with bindings already attached to the # agent's existing tools. This lets new tools reuse the same connections @@ -499,7 +502,7 @@ async def apply_provider_update_plan_with_rollback( raw_connections_to_create=plan.raw_connections_to_create, error_prefix=ErrorPrefix.UPDATE.value, validate_connection_fn=validate_connection, - create_connection_fn=create_connection_with_conflict_mapping, + created_app_ids_journal=created_app_ids_journal, ) operation_to_provider_app_id.update(connection_result.operation_to_provider_app_id) resolved_connections.update(connection_result.resolved_connections) diff --git a/src/backend/tests/unit/api/v1/test_deployment_mapper_watsonx.py b/src/backend/tests/unit/api/v1/test_deployment_mapper_watsonx.py index 965bfd2e14..55c1e445a4 100644 --- a/src/backend/tests/unit/api/v1/test_deployment_mapper_watsonx.py +++ b/src/backend/tests/unit/api/v1/test_deployment_mapper_watsonx.py @@ -896,6 +896,16 @@ def test_watsonx_mapper_create_result_from_existing_update_normalizes_slot_paylo } +def test_watsonx_mapper_create_result_from_existing_resource_includes_empty_payload() -> None: + mapper = WatsonxOrchestrateDeploymentMapper() + create_result = mapper.util_create_result_from_existing_resource(existing_resource_key="existing-agent-1") + + assert create_result.id == "existing-agent-1" + assert isinstance(create_result.provider_result, dict) + assert create_result.provider_result.get("app_ids") == [] + assert create_result.provider_result.get("tools_with_refs") == [] + + def test_watsonx_mapper_resolve_verify_credentials_for_update_returns_none_without_provider_data() -> None: from langflow.services.database.models.deployment_provider_account.model import DeploymentProviderAccount diff --git a/src/backend/tests/unit/services/deployment/test_watsonx_orchestrate.py b/src/backend/tests/unit/services/deployment/test_watsonx_orchestrate.py index a2a0782989..dfb1c0b344 100644 --- a/src/backend/tests/unit/services/deployment/test_watsonx_orchestrate.py +++ b/src/backend/tests/unit/services/deployment/test_watsonx_orchestrate.py @@ -1621,6 +1621,125 @@ async def test_apply_provider_create_plan_rolls_back_mutated_existing_tools_with assert rollback_payload["binding"]["langflow"]["connections"] == {"old": "conn-old"} +@pytest.mark.anyio +async def test_resolve_connections_for_operations_normalizes_provider_app_id_and_uses_created_id_for_validation( + monkeypatch, +): + raw_payload = payloads_module.WatsonxConnectionRawPayload.model_validate( + {"app_id": "cfg-s-003", "environment_variables": {"API_KEY": {"source": "raw", "value": "x"}}} + ) + raw_connections_to_create = [ + shared_core_module.RawConnectionCreatePlan( + operation_app_id="cfg-s-003", + provider_app_id="cfg-s-003", + payload=raw_payload, + ) + ] + fake_clients = SimpleNamespace(connections=FakeConnectionsClient()) + validated_app_ids: list[str] = [] + + async def mock_create_connection_with_conflict_mapping( + *, + clients, + app_id, + payload, + user_id, + db, + error_prefix, + created_app_ids_journal=None, + ): + _ = clients, app_id, payload, user_id, db, error_prefix + created_id = "cfg_s_003" + if created_app_ids_journal is not None: + created_app_ids_journal.append(created_id) + return created_id + + async def mock_validate_connection(connections_client, *, app_id): # noqa: ARG001 + validated_app_ids.append(app_id) + return SimpleNamespace(connection_id=f"conn-{app_id}") + + monkeypatch.setattr( + shared_core_module, + "create_connection_with_conflict_mapping", + mock_create_connection_with_conflict_mapping, + ) + + result = await shared_core_module.resolve_connections_for_operations( + clients=fake_clients, + user_id="user-1", + db=object(), + existing_app_ids=[], + raw_connections_to_create=raw_connections_to_create, + error_prefix="CREATE", + validate_connection_fn=mock_validate_connection, + ) + + assert raw_connections_to_create[0].provider_app_id == "cfg_s_003" + assert validated_app_ids == ["cfg_s_003"] + assert result.operation_to_provider_app_id == {"cfg-s-003": "cfg_s_003"} + assert result.resolved_connections == {"cfg_s_003": "conn-cfg_s_003"} + assert result.created_app_ids == ["cfg_s_003"] + + +@pytest.mark.anyio +async def test_resolve_connections_for_operations_wraps_validation_failure_with_rollback_metadata(monkeypatch): + raw_payload = payloads_module.WatsonxConnectionRawPayload.model_validate( + {"app_id": "cfg-s-003", "environment_variables": {"API_KEY": {"source": "raw", "value": "x"}}} + ) + raw_connections_to_create = [ + shared_core_module.RawConnectionCreatePlan( + operation_app_id="cfg-s-003", + provider_app_id="cfg-s-003", + payload=raw_payload, + ) + ] + fake_clients = SimpleNamespace(connections=FakeConnectionsClient()) + created_app_ids_journal: list[str] = [] + + async def mock_create_connection_with_conflict_mapping( + *, + clients, + app_id, + payload, + user_id, + db, + error_prefix, + created_app_ids_journal=None, + ): + _ = clients, app_id, payload, user_id, db, error_prefix + created_id = "cfg_s_003" + if created_app_ids_journal is not None: + created_app_ids_journal.append(created_id) + return created_id + + async def mock_validate_connection(connections_client, *, app_id): # noqa: ARG001 + msg = f"Connection '{app_id}' not found. Ensure the connection exists with a draft configuration." + raise InvalidContentError(message=msg) + + monkeypatch.setattr( + shared_core_module, + "create_connection_with_conflict_mapping", + mock_create_connection_with_conflict_mapping, + ) + + with pytest.raises(shared_core_module.ConnectionCreateBatchError) as exc_info: + await shared_core_module.resolve_connections_for_operations( + clients=fake_clients, + user_id="user-1", + db=object(), + existing_app_ids=[], + raw_connections_to_create=raw_connections_to_create, + error_prefix="CREATE", + validate_connection_fn=mock_validate_connection, + created_app_ids_journal=created_app_ids_journal, + ) + + assert created_app_ids_journal == ["cfg_s_003"] + assert exc_info.value.created_app_ids == ["cfg_s_003"] + assert len(exc_info.value.errors) == 1 + assert isinstance(exc_info.value.errors[0], InvalidContentError) + + @pytest.mark.anyio async def test_apply_provider_create_plan_rolls_back_successfully_created_raw_connections_on_partial_batch_failure( monkeypatch, @@ -1651,7 +1770,17 @@ async def test_apply_provider_create_plan_rolls_back_successfully_created_raw_co fake_clients = SimpleNamespace(connections=FakeConnectionsClient()) captured: dict[str, Any] = {} - async def mock_create_connection_with_conflict_mapping(*, clients, app_id, payload, user_id, db, error_prefix): # noqa: ARG001 + async def mock_create_connection_with_conflict_mapping( + *, + clients, + app_id, + payload, + user_id, + db, + error_prefix, + created_app_ids_journal=None, + ): + _ = clients, payload, user_id, db, error_prefix, created_app_ids_journal if app_id.endswith("cfg-a"): return app_id msg = "boom-create-connection" @@ -1661,7 +1790,7 @@ async def test_apply_provider_create_plan_rolls_back_successfully_created_raw_co captured["rollback_app_ids"] = list(app_ids or []) monkeypatch.setattr( - create_core_module, + shared_core_module, "create_connection_with_conflict_mapping", mock_create_connection_with_conflict_mapping, ) @@ -1714,7 +1843,17 @@ async def test_apply_provider_create_plan_rolls_back_all_journaled_raw_connectio fake_clients = SimpleNamespace(connections=FakeConnectionsClient()) captured: dict[str, Any] = {} - async def mock_create_connection_with_conflict_mapping(*, clients, app_id, payload, user_id, db, error_prefix): # noqa: ARG001 + async def mock_create_connection_with_conflict_mapping( + *, + clients, + app_id, + payload, + user_id, + db, + error_prefix, + created_app_ids_journal=None, + ): + _ = clients, payload, user_id, db, error_prefix, created_app_ids_journal if app_id.endswith("cfg-c"): msg = "boom-create-connection" raise RuntimeError(msg) @@ -1724,7 +1863,7 @@ async def test_apply_provider_create_plan_rolls_back_all_journaled_raw_connectio captured["rollback_app_ids"] = list(app_ids or []) monkeypatch.setattr( - create_core_module, + shared_core_module, "create_connection_with_conflict_mapping", mock_create_connection_with_conflict_mapping, ) @@ -1746,6 +1885,77 @@ async def test_apply_provider_create_plan_rolls_back_all_journaled_raw_connectio assert captured["rollback_app_ids"] == ["cfg-a", "cfg-b"] +@pytest.mark.anyio +async def test_apply_provider_create_plan_rolls_back_journaled_app_ids_when_create_fails_after_provider_create( + monkeypatch, +): + provider_create = payloads_module.WatsonxDeploymentCreatePayload.model_validate( + { + "tools": {}, + "connections": { + "raw_payloads": [ + {"app_id": "cfg-a", "environment_variables": {"API_KEY": {"source": "raw", "value": "x"}}}, + ] + }, + "llm": TEST_WXO_LLM, + "operations": [ + { + "op": "bind", + "tool": {"tool_id_with_ref": _tool_ref("tool-existing-1")}, + "app_ids": ["cfg-a"], + }, + ], + } + ) + plan = create_core_module.build_provider_create_plan( + deployment_name="my deployment", + provider_create=provider_create, + ) + fake_clients = SimpleNamespace(connections=FakeConnectionsClient()) + captured: dict[str, Any] = {} + + async def mock_create_connection_with_conflict_mapping( + *, + clients, + app_id, + payload, + user_id, + db, + error_prefix, + created_app_ids_journal=None, + ): + _ = clients, payload, user_id, db, error_prefix + if created_app_ids_journal is not None: + created_app_ids_journal.append(app_id) + msg = "boom-after-provider-create" + raise RuntimeError(msg) + + async def mock_rollback_created_resources(*, clients, agent_id, tool_ids, app_ids=None): # noqa: ARG001 + captured["rollback_app_ids"] = list(app_ids or []) + + monkeypatch.setattr( + shared_core_module, + "create_connection_with_conflict_mapping", + mock_create_connection_with_conflict_mapping, + ) + monkeypatch.setattr(create_core_module, "rollback_created_resources", mock_rollback_created_resources) + + with pytest.raises(RuntimeError, match="boom-after-provider-create"): + await create_core_module.apply_provider_create_plan_with_rollback( + clients=fake_clients, + user_id="user-1", + db=object(), + deployment_spec=BaseDeploymentData( + name="my deployment", + description="desc", + type=DeploymentType.AGENT, + ), + plan=plan, + ) + + assert captured["rollback_app_ids"] == ["cfg-a"] + + @pytest.mark.anyio async def test_apply_provider_update_plan_rolls_back_successfully_created_raw_connections_on_partial_batch_failure( monkeypatch, @@ -1780,7 +1990,17 @@ async def test_apply_provider_update_plan_rolls_back_successfully_created_raw_co ) captured: dict[str, Any] = {} - async def mock_create_connection_with_conflict_mapping(*, clients, app_id, payload, user_id, db, error_prefix): # noqa: ARG001 + async def mock_create_connection_with_conflict_mapping( + *, + clients, + app_id, + payload, + user_id, + db, + error_prefix, + created_app_ids_journal=None, + ): + _ = clients, payload, user_id, db, error_prefix, created_app_ids_journal if app_id.endswith("cfg-a"): return app_id msg = "boom-update-connection" @@ -1793,7 +2013,7 @@ async def test_apply_provider_update_plan_rolls_back_successfully_created_raw_co captured["rolled_back_app_ids"] = list(created_app_ids) monkeypatch.setattr( - update_core_module, + shared_core_module, "create_connection_with_conflict_mapping", mock_create_connection_with_conflict_mapping, ) @@ -1849,7 +2069,17 @@ async def test_apply_provider_update_plan_rolls_back_all_journaled_raw_connectio ) captured: dict[str, Any] = {} - async def mock_create_connection_with_conflict_mapping(*, clients, app_id, payload, user_id, db, error_prefix): # noqa: ARG001 + async def mock_create_connection_with_conflict_mapping( + *, + clients, + app_id, + payload, + user_id, + db, + error_prefix, + created_app_ids_journal=None, + ): + _ = clients, payload, user_id, db, error_prefix, created_app_ids_journal if app_id.endswith("cfg-c"): msg = "boom-update-connection" raise RuntimeError(msg) @@ -1862,7 +2092,7 @@ async def test_apply_provider_update_plan_rolls_back_all_journaled_raw_connectio captured["rolled_back_app_ids"] = list(created_app_ids) monkeypatch.setattr( - update_core_module, + shared_core_module, "create_connection_with_conflict_mapping", mock_create_connection_with_conflict_mapping, ) @@ -1883,6 +2113,83 @@ async def test_apply_provider_update_plan_rolls_back_all_journaled_raw_connectio assert captured["rolled_back_app_ids"] == ["cfg-a", "cfg-b"] +@pytest.mark.anyio +async def test_apply_provider_update_plan_rolls_back_journaled_app_ids_when_create_fails_after_provider_create( + monkeypatch, +): + provider_update = payloads_module.WatsonxDeploymentUpdatePayload.model_validate( + { + "tools": {}, + "connections": { + "raw_payloads": [ + {"app_id": "cfg-a", "environment_variables": {"API_KEY": {"source": "raw", "value": "x"}}}, + ] + }, + "llm": TEST_WXO_LLM, + "operations": [ + { + "op": "bind", + "tool": {"tool_id_with_ref": _tool_ref("tool-existing-1")}, + "app_ids": ["cfg-a"], + }, + ], + } + ) + plan = update_core_module.build_provider_update_plan( + agent={"id": "dep-1", "tools": ["tool-existing-1"]}, + provider_update=provider_update, + ) + fake_clients = SimpleNamespace( + agent=FakeAgentClient({"id": "dep-1", "tools": ["tool-existing-1"]}), + tool=FakeToolClient([]), + connections=FakeConnectionsClient(), + ) + captured: dict[str, Any] = {} + + async def mock_create_connection_with_conflict_mapping( + *, + clients, + app_id, + payload, + user_id, + db, + error_prefix, + created_app_ids_journal=None, + ): + _ = clients, payload, user_id, db, error_prefix + if created_app_ids_journal is not None: + created_app_ids_journal.append(app_id) + msg = "boom-after-provider-create" + raise RuntimeError(msg) + + async def mock_rollback_update_resources(*, clients, created_tool_ids, created_app_id, original_tools): # noqa: ARG001 + _ = (created_tool_ids, created_app_id, original_tools) + + async def mock_rollback_created_app_ids(*, clients, created_app_ids): # noqa: ARG001 + captured["rolled_back_app_ids"] = list(created_app_ids) + + monkeypatch.setattr( + shared_core_module, + "create_connection_with_conflict_mapping", + mock_create_connection_with_conflict_mapping, + ) + monkeypatch.setattr(update_core_module, "rollback_update_resources", mock_rollback_update_resources) + monkeypatch.setattr(update_core_module, "rollback_created_app_ids", mock_rollback_created_app_ids) + + with pytest.raises(RuntimeError, match="boom-after-provider-create"): + await update_core_module.apply_provider_update_plan_with_rollback( + clients=fake_clients, + user_id="user-1", + db=object(), + agent_id="dep-1", + agent={"id": "dep-1", "tools": ["tool-existing-1"]}, + update_payload={}, + plan=plan, + ) + + assert captured["rolled_back_app_ids"] == ["cfg-a"] + + @pytest.mark.anyio async def test_create_provider_data_prefixes_tool_and_deployment_names_but_not_connection_app_ids(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) @@ -2351,7 +2658,19 @@ async def test_create_provider_data_rolls_back_partially_created_raw_tools(monke async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients - async def mock_create_connection_with_conflict_mapping(*, clients, app_id, payload, user_id, db, error_prefix): # noqa: ARG001 + async def mock_create_connection_with_conflict_mapping( + *, + clients, + app_id, + payload, + user_id, + db, + error_prefix, # noqa: ARG001 + created_app_ids_journal=None, + ): + _ = clients, payload, user_id, db + if created_app_ids_journal is not None: + created_app_ids_journal.append(app_id) fake_connections._connections_by_app_id[app_id] = f"conn-{app_id}" return app_id @@ -2368,7 +2687,7 @@ async def test_create_provider_data_rolls_back_partially_created_raw_tools(monke monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) monkeypatch.setattr( - create_core_module, + shared_core_module, "create_connection_with_conflict_mapping", mock_create_connection_with_conflict_mapping, )