"""Direct Watsonx adapter scenario runner. Runs scenario matrices against `WatsonxOrchestrateDeploymentService` directly (no `/api/v1/deployments` calls). Warning: -------- This script performs live integration calls and creates real resources in 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. Scenario catalog ---------------- Live create scenarios: - `live_create_success`: creates config + snapshot + agent successfully (expects Success). - `live_invalid_config_reference`: rejects create payload when provider_data references a non-existent existing connection id (expects InvalidContentError). - `live_duplicate_snapshot_names_conflict`: duplicate raw snapshot names are deduped by schema normalization (expects Success). Live lifecycle scenarios: - `live_lifecycle_create_seed`: creates a seed deployment for lifecycle checks (expects Success). - `live_list_contains_seed`: verifies list-by-id includes the seed deployment (expects Success). - `live_get_seed`: fetches deployment details by id (expects Success). - `live_update_seed_name_description`: updates deployment name/description (expects Success). - `live_get_after_update_reflects_name`: confirms updated name is persisted (expects Success). - `live_get_status_connected`: confirms status endpoint reports connected deployment (expects Success). - `live_create_execution_success`: starts an execution run with valid message payload and validates provider_result contains execution_id (not run_id) and status (expects Success). - `live_create_execution_input_string`: creates execution using plain string input (expects Success). - `live_create_execution_input_dict_content`: creates execution using dict input with content key (expects Success). - `live_get_execution_poll_terminal`: polls execution until terminal status and validates provider_result fields including execution_id, status, and absence of run_id (expects Success). - `live_get_execution_terminal_fields`: validates terminal execution has agent_id and appropriate timestamp (completed_at, failed_at, or cancelled_at) (expects Success). - `live_delete_seed`: deletes seed deployment agent (expects Success). - `live_get_after_delete_not_found`: confirms deleted deployment is no longer fetchable (expects DeploymentNotFoundError). - `live_status_after_delete_not_found_state`: confirms status on deleted deployment returns not found (expects DeploymentNotFoundError). 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; 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). - `live_create_execution_rejects_empty_input`: rejects empty execution input payload (expects InvalidContentError). - `live_create_execution_missing_deployment`: rejects execution for non-existent deployment (expects DeploymentNotFoundError). - `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 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). - `upd_config_only_existing_tools_with_config_id`: rebinds existing attached snapshots to an explicit existing app id via provider_data (expects Success). - `upd_snapshot_add_ids_with_config_id`: binds existing snapshot ids using provider_data operations (expects Success). - `upd_snapshot_add_raw_with_config_id`: creates and binds raw snapshot payloads using provider_data tools/connections/operations (expects Success). - `upd_mixed_add_remove_raw_with_config`: mixed add/remove/raw snapshot update with provider_data operations (expects Success). - `upd_reject_bind_with_undeclared_app_id`: rejects bind operation with undeclared app id in provider_data (expects InvalidContentError). - `upd_reject_raw_bind_with_undeclared_app_id`: rejects raw-tool bind operation with undeclared app id in provider_data (expects InvalidContentError). - `upd_reject_unbind_with_undeclared_app_id`: rejects unbind operation with undeclared app id in provider_data (expects InvalidContentError). - `upd_reject_unbind_unknown_tool_id`: rejects unbind on unknown tool id in provider_data (expects InvalidContentError). - `upd_missing_add_id_fails`: rejects unknown bind.tool.tool_id_with_ref in provider_data (expects InvalidContentError). - `upd_config_raw_payload_conflict`: detects conflict when creating duplicate provider_data raw connection app id (expects ResourceConflictError). - `upd_not_found_deployment`: update unknown deployment id returns not found (expects DeploymentNotFoundError). - `upd_put_tools_replaces_tool_list`: uses put_tools to declaratively replace the agent's tool list with a subset (expects Success). - `upd_put_tools_empty_clears_all_tools`: passes an empty put_tools list to remove all tools from the agent (expects Success). - `upd_put_tools_deduplicates`: passes duplicate tool ids in put_tools and confirms deduplication produces a single binding (expects Success). Failpoint scenarios: - `fp_retry_create_config_then_success`: injects transient config-create failures; retries then succeeds (expects Success). - `fp_non_retryable_create_agent_conflict`: injects non-retryable agent conflict (expects ResourceConflictError). - `fp_create_agent_failure_triggers_rollback`: injects repeated agent-create failure and checks rollback (expects DeploymentError). - `fp_update_bindings_failure_triggers_rollback`: injects update-stage binding failure and validates rollback path (expects DeploymentError). - `fp_update_bindings_failure_with_rollback_failure`: injects update failure + rollback failure and expects terminal error (expects DeploymentError). - `fp_update_failure_then_put_tools_restore`: injects a binding-stage failure to corrupt the tool list, then uses put_tools to restore the original snapshot set and confirms the tool list matches the pre-failure state (expects Success after restore). """ from __future__ import annotations import argparse import asyncio import os import re import textwrap import types from contextlib import suppress from dataclasses import dataclass from datetime import datetime, timezone from types import MethodType, SimpleNamespace from typing import TYPE_CHECKING, Any from uuid import uuid4 import langflow.services.adapters.deployment.watsonx_orchestrate.core.create as create_core_module import langflow.services.adapters.deployment.watsonx_orchestrate.core.retry as retry_module import langflow.services.adapters.deployment.watsonx_orchestrate.core.shared as shared_core_module import langflow.services.adapters.deployment.watsonx_orchestrate.core.update as update_core_module from dotenv import load_dotenv from fastapi import HTTPException 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 ( WatsonxOrchestrateDeploymentService, WxOCredentials, ) from langflow.services.adapters.deployment.watsonx_orchestrate.payloads import ( WatsonxDeploymentCreateResultData, WatsonxFlowArtifactProviderData, ) from lfx.services.adapters.deployment.exceptions import ( DeploymentError, DeploymentNotFoundError, InvalidContentError, InvalidDeploymentOperationError, InvalidDeploymentTypeError, OperationNotSupportedError, ResourceConflictError, ) from lfx.services.adapters.deployment.schema import ( BaseDeploymentData, BaseDeploymentDataUpdate, BaseFlowArtifact, ConfigListParams, DeploymentConfig, DeploymentCreate, DeploymentListParams, DeploymentType, DeploymentUpdate, ExecutionCreate, SnapshotListParams, VerifyCredentials, ) OUTCOME_SUCCESS = "Success" OUTCOME_INVALID_OPERATION = "InvalidDeploymentOperationError" OUTCOME_CONFLICT = "ResourceConflictError" OUTCOME_INVALID_CONTENT = "InvalidContentError" OUTCOME_FAILURE = "DeploymentError" OUTCOME_NOT_FOUND = "DeploymentNotFoundError" HTTP_STATUS_NOT_FOUND = 404 HTTP_STATUS_CONFLICT = 409 MIN_MIXED_SNAPSHOT_IDS = 2 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_]") if TYPE_CHECKING: from collections.abc import Awaitable, Callable class DummySettingsService: def __init__(self) -> None: self.settings = SimpleNamespace() @dataclass(slots=True) class ScenarioResult: name: str expected_outcomes: set[str] actual_outcome: str ok: bool detail: str @dataclass(slots=True) class WxoCreatedDeploymentResult: deployment_id: str provider_result: WatsonxDeploymentCreateResultData class WatsonxAdapterDirectE2E: def __init__( self, *, provider_backend_url: str, provider_api_key: str, 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()) self.db = object() self.provider_id = uuid4() self.service = WatsonxOrchestrateDeploymentService(DummySettingsService()) import langflow.services.adapters.deployment.watsonx_orchestrate.client as _client_mod self._client_mod = _client_mod deployment_context = DeploymentAdapterContext(provider_id=self.provider_id) self._deployment_context_token = DeploymentProviderIDContext.set_current(deployment_context) self._original_resolve_wxo_client_credentials = _client_mod.resolve_wxo_client_credentials async def _resolve_credentials(*, user_id, db, provider_id): # noqa: ARG001 authenticator = _client_mod.get_authenticator( instance_url=self.provider_backend_url, api_key=self.provider_api_key, ) return WxOCredentials(instance_url=self.provider_backend_url, authenticator=authenticator) _client_mod.resolve_wxo_client_credentials = _resolve_credentials # type: ignore[assignment] self.created_deployment_ids: set[str] = set() self.created_snapshot_ids: set[str] = set() self.created_config_ids: set[str] = set() self.tool_source_refs: dict[str, str] = {} 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} llm={self.llm}") try: results: list[ScenarioResult] = [] 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()) self._print_summary(results) if not self.keep_resources: await self._cleanup_resources() return 1 if any(not result.ok for result in results) else 0 finally: self._client_mod.resolve_wxo_client_credentials = self._original_resolve_wxo_client_credentials self._client_mod.clear_provider_clients_request_context() DeploymentProviderIDContext.reset_current(self._deployment_context_token) async def _run_live_scenarios(self) -> list[ScenarioResult]: duplicate_name = self._mk_name("dup_snapshot") scenarios = [ { "name": "live_create_success", "expected": {OUTCOME_SUCCESS}, "payload": self._build_create_payload( tool_payloads=[self._build_flow_payload(label="snap_live_success")], raw_connection=self._build_config_payload(label="cfg_live_success"), ), }, { "name": "live_invalid_config_reference", "expected": {OUTCOME_INVALID_CONTENT}, "payload": self._build_create_payload( tool_payloads=[self._build_flow_payload(label="snap_live_invalid_ref")], existing_connection_app_id="cfg_ref_not_supported", ), }, { "name": "live_duplicate_snapshot_names_conflict", "expected": {OUTCOME_SUCCESS}, "payload": self._build_create_payload( tool_payloads=[ self._build_flow_payload(label="snap_dup_a", name_override=duplicate_name), self._build_flow_payload(label="snap_dup_b", name_override=duplicate_name), ], raw_connection=self._build_config_payload(label="cfg_live_dup"), ), "assert_dedupe_snapshot_count": 1, }, ] 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()) return results async def _run_failpoint_scenarios(self) -> list[ScenarioResult]: scenarios = [ { "name": "fp_retry_create_config_then_success", "expected": {OUTCOME_SUCCESS}, "payload": self._build_create_payload( tool_payloads=[self._build_flow_payload(label="snap_fp_retry")], raw_connection=self._build_config_payload(label="cfg_fp_retry"), ), "inject": { "create_config": {"fail_first_n": 2, "error_type": "runtime", "message": "fp_create_config_retry"} }, }, { "name": "fp_create_config_wrapper_conflict_mapping", "expected": {OUTCOME_CONFLICT}, "payload": self._build_create_payload( tool_payloads=[self._build_flow_payload(label="snap_fp_cfg_wrapper_conflict")], raw_connection=self._build_config_payload(label="cfg_fp_cfg_wrapper_conflict"), ), "inject": { "create_config_wrapper": { "fail_first_n": 1, "error_type": "domain_conflict", "message": "fp_create_config_wrapper_conflict", } }, }, { "name": "fp_non_retryable_create_agent_conflict", "expected": {OUTCOME_CONFLICT}, "payload": self._build_create_payload( tool_payloads=[self._build_flow_payload(label="snap_fp_conflict")], raw_connection=self._build_config_payload(label="cfg_fp_conflict"), ), "inject": { "create_agent": { "fail_first_n": 1, "error_type": "domain_conflict", "message": "fp_create_agent_conflict", } }, }, { "name": "fp_create_agent_failure_triggers_rollback", "expected": {OUTCOME_FAILURE}, "detail_contains": "Please check server logs for details", "payload": self._build_create_payload( tool_payloads=[self._build_flow_payload(label="snap_fp_rollback")], raw_connection=self._build_config_payload(label="cfg_fp_rollback"), ), "inject": { "create_agent": { "fail_first_n": 3, "error_type": "runtime", "message": "fp_create_agent_final", }, "rollback_delete_config": { "fail_first_n": 2, "error_type": "runtime", "message": "fp_rollback_delete_config", }, }, }, ] results = await self._run_scenarios(scenarios) results.extend(await self._run_update_failpoint_scenarios()) return results async def _run_scenarios(self, scenarios: list[dict[str, Any]]) -> list[ScenarioResult]: results: list[ScenarioResult] = [] for index, scenario in enumerate(scenarios, start=1): print(f"\n[{index}/{len(scenarios)}] {scenario['name']}") status_code, detail, created = await self._run_create(scenario["payload"], inject=scenario.get("inject")) detail_contains = str(scenario.get("detail_contains") or "").strip() detail_ok = not detail_contains or detail_contains in detail ok = status_code in scenario["expected"] and detail_ok dedupe_expected = scenario.get("assert_dedupe_snapshot_count") if dedupe_expected is not None: created_snapshot_ids = self._extract_create_snapshot_ids(created.provider_result) if created else set() created_snapshot_count = len(created_snapshot_ids) dedupe_ok = ( status_code == OUTCOME_SUCCESS and created is not None and created_snapshot_count == int(dedupe_expected) and self._has_unique_snapshot_ids(created_snapshot_ids) ) ok = ok and dedupe_ok if not dedupe_ok: detail = ( f"{detail} | dedupe_check failed: expected_snapshot_count={dedupe_expected} " f"got={created_snapshot_count}" ) if created: self.created_deployment_ids.add(created.deployment_id) self.created_snapshot_ids.update(self._extract_create_snapshot_ids(created.provider_result)) self.created_config_ids.update(self._extract_create_app_ids(created.provider_result)) results.append( ScenarioResult( name=scenario["name"], expected_outcomes=set(scenario["expected"]), actual_outcome=status_code, ok=ok, detail=detail[:600], ) ) return results async def _run_create( self, payload: DeploymentCreate, *, inject: dict[str, dict[str, Any]] | None = None, ) -> tuple[str, str, WxoCreatedDeploymentResult | None]: originals: list[tuple[Any, str, Any]] = [] try: if inject: self._apply_injections(inject, originals) result = await self.service.create(user_id=self.user_id, payload=payload, db=self.db) 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: create_provider_result = self._parse_create_provider_result_payload(result.provider_result) created = WxoCreatedDeploymentResult( deployment_id=str(result.id), provider_result=create_provider_result, ) return OUTCOME_SUCCESS, "created", created finally: for target, attr_name, original in originals: setattr(target, attr_name, original) async def _run_list(self, *, params: DeploymentListParams | None = None) -> tuple[str, str, Any | None]: try: result = await self.service.list(user_id=self.user_id, db=self.db, params=params) 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, "listed", result async def _run_get(self, deployment_id: str) -> tuple[str, str, Any | None]: try: result = await self.service.get(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) 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, "fetched", result async def _run_update( self, deployment_id: str, payload: DeploymentUpdate, *, inject: dict[str, dict[str, Any]] | None = None, ) -> tuple[str, str, Any | None]: originals: list[tuple[Any, str, Any]] = [] try: if inject: self._apply_injections(inject, originals) result = await self.service.update( user_id=self.user_id, deployment_id=deployment_id, payload=payload, 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: self.created_config_ids.update(self._extract_update_created_app_ids(result)) return OUTCOME_SUCCESS, "updated", result finally: for target, attr_name, original in originals: setattr(target, attr_name, original) 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=params, 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, 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]: return await self._run_list_snapshots_with_params( params=SnapshotListParams(snapshot_ids=snapshot_ids), detail_label="snapshots_by_ids_listed", ) 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=params, 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, 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: result = await self.service.get_status(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) 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, "status", result async def _run_create_execution( self, deployment_id: str, *, provider_data: dict[str, Any], ) -> tuple[str, str, Any | None]: try: result = await self.service.create_execution( user_id=self.user_id, payload=ExecutionCreate(deployment_id=deployment_id, provider_data=provider_data), 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, "execution_created", result async def _run_get_execution(self, execution_id: str) -> tuple[str, str, Any | None]: try: result = await self.service.get_execution(user_id=self.user_id, execution_id=execution_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) 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, "execution_fetched", result async def _poll_execution_terminal(self, execution_id: str) -> tuple[str, str, Any | None]: """Poll get_execution until a terminal status is reached or max attempts exceeded.""" result = None for attempt in range(EXECUTION_POLL_MAX_ATTEMPTS): status_code, detail, result = await self._run_get_execution(execution_id) pr = getattr(result, "provider_result", None) or {} current_status = pr.get("status") if isinstance(pr, dict) else None print( f" [poll {attempt + 1}/{EXECUTION_POLL_MAX_ATTEMPTS}] " f"outcome={status_code} status={current_status} " f"provider_result={pr}" ) if status_code != OUTCOME_SUCCESS: return status_code, f"poll attempt {attempt + 1}: {detail}", result if current_status in EXECUTION_TERMINAL_STATUSES: return status_code, f"terminal after {attempt + 1} polls: {current_status}", result await asyncio.sleep(EXECUTION_POLL_INTERVAL_SECS) msg = f"execution did not reach terminal status after {EXECUTION_POLL_MAX_ATTEMPTS} polls" return OUTCOME_FAILURE, msg, result async def _run_delete(self, deployment_id: str) -> tuple[str, str, Any | None]: try: result = await self.service.delete(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) 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, "deleted", result def _build_result( self, *, name: str, expected: set[str], actual_outcome: str, detail: str, ok: bool, ) -> ScenarioResult: return ScenarioResult( name=name, expected_outcomes=expected, actual_outcome=actual_outcome, ok=ok, detail=detail[:600], ) def _outcome_from_http_exception(self, exc: HTTPException) -> str: status_code = int(exc.status_code) if status_code == HTTP_STATUS_NOT_FOUND: return OUTCOME_NOT_FOUND if status_code == HTTP_STATUS_CONFLICT: return OUTCOME_CONFLICT if status_code in {400, 405}: return OUTCOME_INVALID_OPERATION if status_code in {413, 415, 422}: return OUTCOME_INVALID_CONTENT return OUTCOME_FAILURE async def _run_live_lifecycle_scenarios(self) -> list[ScenarioResult]: results: list[ScenarioResult] = [] print("\n[life/1] live_lifecycle_create_seed") status_code, detail, created = await self._run_create( self._build_create_payload( tool_payloads=[self._build_flow_payload(label="snap_live_lifecycle_seed")], raw_connection=self._build_config_payload(label="cfg_live_lifecycle_seed"), ) ) create_ok = status_code == OUTCOME_SUCCESS and created is not None results.append( self._build_result( name="live_lifecycle_create_seed", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=detail, ok=create_ok, ) ) if not create_ok or created is None: return results deployment_id = created.deployment_id self.created_deployment_ids.add(deployment_id) self.created_snapshot_ids.update(self._extract_create_snapshot_ids(created.provider_result)) self.created_config_ids.update(self._extract_create_app_ids(created.provider_result)) print("[life/2] live_list_contains_seed") status_code, detail, list_result = await self._run_list( params=DeploymentListParams(deployment_ids=[deployment_id]) ) list_contains_seed = bool( list_result and any(str(deployment.id) == deployment_id for deployment in getattr(list_result, "deployments", [])) ) results.append( self._build_result( name="live_list_contains_seed", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=detail, ok=status_code == OUTCOME_SUCCESS and list_contains_seed, ) ) print("[life/3] live_get_seed") status_code, detail, get_result = await self._run_get(deployment_id) got_seed = bool(get_result and str(get_result.id) == deployment_id) results.append( self._build_result( name="live_get_seed", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=detail, ok=status_code == OUTCOME_SUCCESS and got_seed, ) ) updated_name = self._mk_name("dep_agent_updated") print("[life/4] live_update_seed_name_description") status_code, detail, _ = await self._run_update( deployment_id, DeploymentUpdate( spec=BaseDeploymentDataUpdate( name=updated_name, description="updated by direct adapter e2e", ) ), ) results.append( self._build_result( name="live_update_seed_name_description", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=detail, ok=status_code == OUTCOME_SUCCESS, ) ) print("[life/5] live_get_after_update_reflects_name") status_code, detail, get_after_update = await self._run_get(deployment_id) updated_name_ok = bool(get_after_update and getattr(get_after_update, "name", None) == updated_name) results.append( self._build_result( name="live_get_after_update_reflects_name", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=detail, ok=status_code == OUTCOME_SUCCESS and updated_name_ok, ) ) print("[life/6] live_get_status_connected") status_code, detail, status_result = await self._run_status(deployment_id) connected_ok = bool(status_result and getattr(status_result, "provider_data", {}).get("status") == "connected") results.append( self._build_result( name="live_get_status_connected", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=detail, ok=status_code == OUTCOME_SUCCESS and connected_ok, ) ) print("[life/7] live_create_execution_success") status_code, detail, execution_create_result = await self._run_create_execution( deployment_id, provider_data={"message": {"role": "user", "content": "hi"}}, ) has_execution_id = bool(execution_create_result and getattr(execution_create_result, "execution_id", None)) create_pr = getattr(execution_create_result, "provider_result", None) or {} create_pr_ok = ( has_execution_id and isinstance(create_pr, dict) and "execution_id" in create_pr and "run_id" not in create_pr and create_pr.get("status") is not None ) results.append( self._build_result( name="live_create_execution_success", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=detail, ok=status_code == OUTCOME_SUCCESS and create_pr_ok, ) ) execution_id_value = ( execution_create_result.execution_id if execution_create_result and getattr(execution_create_result, "execution_id", None) else None ) execution_id = str(execution_id_value) if execution_id_value else None print("[life/7b] live_create_execution_input_string") status_code, detail, exec_str_result = await self._run_create_execution( deployment_id, provider_data={"input": "hi again"}, ) str_ok = bool( status_code == OUTCOME_SUCCESS and exec_str_result and getattr(exec_str_result, "execution_id", None) ) results.append( self._build_result( name="live_create_execution_input_string", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=detail, ok=str_ok, ) ) print("[life/7c] live_create_execution_input_dict_content") status_code, detail, exec_dict_result = await self._run_create_execution( deployment_id, provider_data={"input": {"content": "haha"}}, ) dict_ok = bool( status_code == OUTCOME_SUCCESS and exec_dict_result and getattr(exec_dict_result, "execution_id", None) ) results.append( self._build_result( name="live_create_execution_input_dict_content", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=detail, ok=dict_ok, ) ) if execution_id: print("[life/8] live_get_execution_poll_terminal") terminal_result = await self._poll_execution_terminal(execution_id) poll_status_code = terminal_result[0] poll_detail = terminal_result[1] poll_result = terminal_result[2] poll_pr = getattr(poll_result, "provider_result", None) or {} got_terminal = isinstance(poll_pr, dict) and poll_pr.get("status") in EXECUTION_TERMINAL_STATUSES pr_has_execution_id = isinstance(poll_pr, dict) and "execution_id" in poll_pr pr_no_run_id = isinstance(poll_pr, dict) and "run_id" not in poll_pr results.append( self._build_result( name="live_get_execution_poll_terminal", expected={OUTCOME_SUCCESS}, actual_outcome=poll_status_code, detail=poll_detail, ok=poll_status_code == OUTCOME_SUCCESS and got_terminal and pr_has_execution_id and pr_no_run_id, ) ) print("[life/8b] live_get_execution_terminal_fields") terminal_fields_ok = ( got_terminal and isinstance(poll_pr, dict) and poll_pr.get("agent_id") is not None and ( poll_pr.get("completed_at") is not None or poll_pr.get("failed_at") is not None or poll_pr.get("cancelled_at") is not None ) ) results.append( self._build_result( name="live_get_execution_terminal_fields", expected={OUTCOME_SUCCESS}, actual_outcome=poll_status_code, detail=f"status={poll_pr.get('status')} has_timestamps={terminal_fields_ok}", ok=poll_status_code == OUTCOME_SUCCESS and terminal_fields_ok, ) ) print("[life/9] live_delete_seed") status_code, detail, _ = await self._run_delete(deployment_id) delete_ok = status_code == OUTCOME_SUCCESS if delete_ok: self.created_deployment_ids.discard(deployment_id) results.append( self._build_result( name="live_delete_seed", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=detail, ok=delete_ok, ) ) print("[life/10] live_get_after_delete_not_found") status_code, detail, _ = await self._run_get(deployment_id) results.append( self._build_result( name="live_get_after_delete_not_found", expected={OUTCOME_NOT_FOUND}, actual_outcome=status_code, detail=detail, ok=status_code == OUTCOME_NOT_FOUND, ) ) print("[life/11] live_status_after_delete_not_found_state") status_code, detail, _ = await self._run_status(deployment_id) results.append( self._build_result( name="live_status_after_delete_not_found_state", expected={OUTCOME_NOT_FOUND}, actual_outcome=status_code, detail=detail, ok=status_code == OUTCOME_NOT_FOUND, ) ) return results async def _run_live_list_snapshots_by_ids_scenarios(self) -> list[ScenarioResult]: results: list[ScenarioResult] = [] print("\n[snap-ids/1] creating seed for list_snapshots by-ids mode") _deployment_id, _config_id, seed_snapshot_ids, _ = await self._create_update_seed( label="snap_ids_seed", snapshot_count=2, ) known_ids = sorted(seed_snapshot_ids) if len(known_ids) < MIN_MIXED_SNAPSHOT_IDS: results.append( self._build_result( name="live_list_snapshots_by_ids_seed_insufficient", expected={OUTCOME_SUCCESS}, actual_outcome=OUTCOME_FAILURE, detail=f"need >= {MIN_MIXED_SNAPSHOT_IDS} snapshot ids, got {len(known_ids)}", ok=False, ) ) return results print("[snap-ids/2] live_list_snapshots_by_ids_returns_known") status_code, detail, snap_result = await self._run_list_snapshots_by_ids(known_ids) returned_ids = self._extract_snapshot_ids(snap_result) ids_match = set(known_ids) == returned_ids results.append( self._build_result( name="live_list_snapshots_by_ids_returns_known", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=f"{detail} | returned={sorted(returned_ids)} expected={known_ids}", ok=status_code == OUTCOME_SUCCESS and ids_match, ) ) print("[snap-ids/3] live_list_snapshots_by_ids_filters_unknown") bogus_id = str(uuid4()) mixed_ids = [known_ids[0], bogus_id] status_code, detail, snap_result = await self._run_list_snapshots_by_ids(mixed_ids) returned_ids = self._extract_snapshot_ids(snap_result) has_known = known_ids[0] in returned_ids no_bogus = bogus_id not in returned_ids results.append( self._build_result( name="live_list_snapshots_by_ids_filters_unknown", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=f"{detail} | returned={sorted(returned_ids)} has_known={has_known} no_bogus={no_bogus}", ok=status_code == OUTCOME_SUCCESS and has_known and no_bogus, ) ) # 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_SUCCESS}, actual_outcome=status_code, 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, ) ) return results async def _run_live_negative_scenarios(self) -> list[ScenarioResult]: results: list[ScenarioResult] = [] print("\n[neg/1] live_negative_create_seed") status_code, detail, created = await self._run_create( self._build_create_payload( tool_payloads=[self._build_flow_payload(label="snap_live_negative_seed")], raw_connection=self._build_config_payload(label="cfg_live_negative_seed"), ) ) create_ok = status_code == OUTCOME_SUCCESS and created is not None results.append( self._build_result( name="live_negative_create_seed", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=detail, ok=create_ok, ) ) if not create_ok or created is None: return results deployment_id = created.deployment_id self.created_deployment_ids.add(deployment_id) self.created_snapshot_ids.update(self._extract_create_snapshot_ids(created.provider_result)) self.created_config_ids.update(self._extract_create_app_ids(created.provider_result)) print("[neg/2] live_create_execution_rejects_empty_input") status_code, detail, _ = await self._run_create_execution( deployment_id, provider_data={"input": " "}, ) results.append( self._build_result( name="live_create_execution_rejects_empty_input", expected={OUTCOME_INVALID_CONTENT}, actual_outcome=status_code, detail=detail, ok=status_code == OUTCOME_INVALID_CONTENT, ) ) print("[neg/2b] live_create_execution_missing_deployment") status_code, detail, _ = await self._run_create_execution( str(uuid4()), provider_data={"input": "uh oh"}, ) results.append( self._build_result( name="live_create_execution_missing_deployment", expected={OUTCOME_NOT_FOUND}, actual_outcome=status_code, detail=detail, ok=status_code == OUTCOME_NOT_FOUND, ) ) print("[neg/3] live_delete_missing_not_found") status_code, detail, _ = await self._run_delete(str(uuid4())) results.append( self._build_result( name="live_delete_missing_not_found", expected={OUTCOME_NOT_FOUND}, actual_outcome=status_code, detail=detail, ok=status_code == OUTCOME_NOT_FOUND, ) ) print("[neg/4] live_negative_delete_seed") status_code, detail, _ = await self._run_delete(deployment_id) delete_ok = status_code == OUTCOME_SUCCESS if delete_ok: self.created_deployment_ids.discard(deployment_id) results.append( self._build_result( name="live_negative_delete_seed", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=detail, ok=delete_ok, ) ) return results def _extract_snapshot_ids(self, snapshot_result: Any) -> set[str]: snapshots = getattr(snapshot_result, "snapshots", []) if snapshot_result else [] return {str(snapshot.id) for snapshot in snapshots if snapshot and snapshot.id} def _parse_create_provider_result_payload(self, provider_result: Any) -> WatsonxDeploymentCreateResultData: create_result_slot = self.service.payload_schemas.deployment_create_result if create_result_slot is None: return WatsonxDeploymentCreateResultData() try: parsed_result = create_result_slot.parse(provider_result) except Exception: # noqa: BLE001 return WatsonxDeploymentCreateResultData() if isinstance(parsed_result, WatsonxDeploymentCreateResultData): return parsed_result return WatsonxDeploymentCreateResultData.model_validate(parsed_result) def _extract_create_app_ids(self, create_provider_result: WatsonxDeploymentCreateResultData) -> list[str]: app_ids = [str(app_id).strip() for app_id in create_provider_result.app_ids if str(app_id).strip()] return list(dict.fromkeys(app_ids)) def _extract_create_snapshot_ids(self, create_provider_result: WatsonxDeploymentCreateResultData) -> set[str]: bindings = create_provider_result.tools_with_refs snapshot_ids: set[str] = set() for binding in bindings: tool_id = str(binding.tool_id or "").strip() if tool_id: snapshot_ids.add(tool_id) return snapshot_ids def _extract_create_tool_ref_map( self, create_provider_result: WatsonxDeploymentCreateResultData, ) -> dict[str, str]: """Return tool_id → source_ref mapping from create result and cache it.""" ref_map: dict[str, str] = {} for binding in create_provider_result.tools_with_refs: tool_id = str(binding.tool_id or "").strip() source_ref = str(binding.source_ref or "").strip() if tool_id and source_ref: ref_map[tool_id] = source_ref self.tool_source_refs.update(ref_map) return ref_map def _make_tool_ref(self, tool_id: str) -> dict[str, str]: """Build a WatsonxToolRefBinding dict using the cached source_ref mapping.""" source_ref = self.tool_source_refs.get(tool_id) if source_ref is None: msg = f"tool_id={tool_id!r} has no cached source_ref; was it created via _create_update_seed?" raise ValueError(msg) return {"source_ref": source_ref, "tool_id": tool_id} def _make_tool_id_with_ref(self, tool_id: str) -> dict[str, Any]: """Build a bind-operation tool selector dict using the cached source_ref mapping.""" return {"tool_id_with_ref": self._make_tool_ref(tool_id)} def _extract_update_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("created_snapshot_ids", []) else: 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() provider_result = getattr(update_result, "provider_result", None) if isinstance(provider_result, dict): app_ids = provider_result.get("created_app_ids", []) else: app_ids = getattr(provider_result, "created_app_ids", []) if provider_result else [] return {str(app_id).strip() for app_id in app_ids if str(app_id).strip()} async def _create_update_seed( self, *, label: str, snapshot_count: int = 2, ) -> tuple[str, str | None, set[str], str]: snapshots = [self._build_flow_payload(label=f"{label}_snap_{idx}") for idx in range(snapshot_count)] status_code, detail, created = await self._run_create( self._build_create_payload( tool_payloads=snapshots, raw_connection=self._build_config_payload(label=f"{label}_cfg"), ) ) if status_code != OUTCOME_SUCCESS or not created: msg = f"Failed to create seed deployment '{label}': status={status_code}, detail={detail}" raise RuntimeError(msg) deployment_id = created.deployment_id self.created_deployment_ids.add(deployment_id) created_app_ids = self._extract_create_app_ids(created.provider_result) created_snapshot_ids = self._extract_create_snapshot_ids(created.provider_result) self._extract_create_tool_ref_map(created.provider_result) self.created_snapshot_ids.update(created_snapshot_ids) self.created_config_ids.update(created_app_ids) config_id = created_app_ids[0] if created_app_ids else None return deployment_id, config_id, set(created_snapshot_ids), detail async def _run_live_update_matrix_scenarios(self) -> list[ScenarioResult]: results: list[ScenarioResult] = [] print("\n[upd] building update matrix seed resources") ( primary_deployment_id, _primary_config_id, primary_snapshot_ids, _, ) = await self._create_update_seed(label="upd_primary", snapshot_count=2) donor_deployment_id, donor_config_id, donor_snapshot_ids, _ = await self._create_update_seed( label="upd_donor", snapshot_count=1, ) mixed_donor_deployment_id, _mixed_donor_cfg, mixed_donor_snapshot_ids, _ = await self._create_update_seed( label="upd_mixed_donor", snapshot_count=1, ) donor_snapshot_id = next(iter(donor_snapshot_ids), "") mixed_donor_snapshot_id = next(iter(mixed_donor_snapshot_ids), "") removable_snapshot_id = next(iter(primary_snapshot_ids), "") retained_snapshot_ids = set(primary_snapshot_ids) retained_snapshot_ids.discard(removable_snapshot_id) if not donor_config_id: results.append( self._build_result( name="upd_seed_missing_donor_config", expected={OUTCOME_SUCCESS}, actual_outcome=OUTCOME_FAILURE, detail="donor deployment config id is missing", ok=False, ) ) return results print("[upd/1] upd_spec_only_name_desc") updated_name = self._mk_name("dep_upd_spec_only") status_code, detail, update_result = await self._run_update( primary_deployment_id, DeploymentUpdate( spec=BaseDeploymentDataUpdate( name=updated_name, description="updated by update matrix spec-only", ) ), ) get_status, _get_detail, get_after_update = await self._run_get(primary_deployment_id) spec_ok = bool(get_after_update and getattr(get_after_update, "name", None) == updated_name) spec_snapshot_ids_ok = bool(update_result and not getattr(update_result, "snapshot_ids", [])) results.append( self._build_result( name="upd_spec_only_name_desc", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=detail, ok=( status_code == OUTCOME_SUCCESS and get_status == OUTCOME_SUCCESS and spec_ok and spec_snapshot_ids_ok ), ) ) print("[upd/2] upd_snapshot_remove_only_no_config") status_code, detail, remove_result = await self._run_update( primary_deployment_id, DeploymentUpdate( provider_data={ "tools": {}, "operations": [{"op": "remove_tool", "tool": self._make_tool_ref(removable_snapshot_id)}], } ), ) list_status, _list_detail, list_after_remove = await self._run_list_snapshots(primary_deployment_id) attached_after_remove = self._extract_snapshot_ids(list_after_remove) remove_snapshot_ids_ok = bool(remove_result and not getattr(remove_result, "snapshot_ids", [])) results.append( self._build_result( name="upd_snapshot_remove_only_no_config", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=detail, ok=( status_code == OUTCOME_SUCCESS and list_status == OUTCOME_SUCCESS and removable_snapshot_id not in attached_after_remove and retained_snapshot_ids.issubset(attached_after_remove) and remove_snapshot_ids_ok ), ) ) print("[upd/3] upd_config_only_existing_tools_with_config_id") retained_snapshot_ids_sorted = sorted(retained_snapshot_ids) status_code, detail, config_only_result = await self._run_update( primary_deployment_id, DeploymentUpdate( provider_data={ "tools": {}, "connections": {}, "operations": [ { "op": "bind", "tool": self._make_tool_id_with_ref(tool_id), "app_ids": [str(donor_config_id)], } for tool_id in retained_snapshot_ids_sorted ], } ), ) 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_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", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=detail, ok=( status_code == OUTCOME_SUCCESS and list_status == OUTCOME_SUCCESS and retained_snapshot_ids.issubset(attached_after_config_only) and config_only_created_ok ), ) ) print("[upd/4] upd_snapshot_add_ids_with_config_id") status_code, detail, add_id_result = await self._run_update( primary_deployment_id, DeploymentUpdate( provider_data={ "tools": {}, "connections": {}, "operations": [ { "op": "bind", "tool": self._make_tool_id_with_ref(donor_snapshot_id), "app_ids": [str(donor_config_id)], } ], } ), ) 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_added_snapshot_ids(add_id_result) results.append( self._build_result( name="upd_snapshot_add_ids_with_config_id", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=detail, ok=( status_code == OUTCOME_SUCCESS and list_status == OUTCOME_SUCCESS and donor_snapshot_id in attached_after_add_id and donor_snapshot_id in add_id_snapshot_ids ), ) ) print("[upd/5] upd_snapshot_add_raw_with_config_id") raw_payload = self._build_flow_payload(label="upd_add_raw") status_code, detail, add_raw_result = await self._run_update( primary_deployment_id, DeploymentUpdate( provider_data={ "tools": {"raw_payloads": [raw_payload.model_dump(mode="json")]}, "connections": {}, "operations": [ { "op": "bind", "tool": {"name_of_raw": raw_payload.name}, "app_ids": [str(donor_config_id)], } ], } ), ) add_raw_snapshot_ids = self._extract_update_snapshot_ids(add_raw_result) self.created_snapshot_ids.update(add_raw_snapshot_ids) list_status, _list_detail, list_after_add_raw = await self._run_list_snapshots(primary_deployment_id) attached_after_add_raw = self._extract_snapshot_ids(list_after_add_raw) add_raw_created_ids = add_raw_snapshot_ids results.append( self._build_result( name="upd_snapshot_add_raw_with_config_id", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=detail, ok=( status_code == OUTCOME_SUCCESS and list_status == OUTCOME_SUCCESS and bool(add_raw_created_ids) and add_raw_created_ids.issubset(attached_after_add_raw) ), ) ) print("[upd/6] upd_mixed_add_remove_raw_with_config") mixed_raw_payload = self._build_flow_payload(label="upd_mixed_raw") mixed_remove_id = next(iter(retained_snapshot_ids), "") status_code, detail, mixed_result = await self._run_update( primary_deployment_id, DeploymentUpdate( provider_data={ "tools": { "raw_payloads": [mixed_raw_payload.model_dump(mode="json")], }, "connections": {}, "operations": [ { "op": "bind", "tool": self._make_tool_id_with_ref(mixed_donor_snapshot_id), "app_ids": [str(donor_config_id)], }, {"op": "remove_tool", "tool": self._make_tool_ref(mixed_remove_id)}, { "op": "bind", "tool": {"name_of_raw": mixed_raw_payload.name}, "app_ids": [str(donor_config_id)], }, ], } ), ) 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) results.append( self._build_result( name="upd_mixed_add_remove_raw_with_config", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=detail, ok=( status_code == OUTCOME_SUCCESS and list_status == OUTCOME_SUCCESS and mixed_remove_id not in attached_after_mixed and mixed_donor_snapshot_id in attached_after_mixed and mixed_donor_snapshot_id in mixed_snapshot_ids and len(mixed_snapshot_ids) >= MIN_MIXED_SNAPSHOT_IDS ), ) ) print("[upd/7] upd_reject_bind_with_undeclared_app_id") status_code, detail, _ = await self._run_update( primary_deployment_id, DeploymentUpdate( provider_data={ "tools": {}, "operations": [ { "op": "bind", "tool": self._make_tool_id_with_ref(donor_snapshot_id), "app_ids": ["undeclared_app_for_bind"], } ], } ), ) results.append( self._build_result( name="upd_reject_bind_with_undeclared_app_id", expected={OUTCOME_INVALID_CONTENT}, actual_outcome=status_code, detail=detail, ok=status_code == OUTCOME_INVALID_CONTENT, ) ) print("[upd/8] upd_reject_raw_bind_with_undeclared_app_id") missing_cfg_raw_payload = self._build_flow_payload(label="upd_no_cfg_raw") status_code, detail, _ = await self._run_update( primary_deployment_id, DeploymentUpdate( provider_data={ "tools": {"raw_payloads": [missing_cfg_raw_payload.model_dump(mode="json")]}, "operations": [ { "op": "bind", "tool": {"name_of_raw": missing_cfg_raw_payload.name}, "app_ids": ["undeclared_app_for_raw_bind"], } ], } ), ) results.append( self._build_result( name="upd_reject_raw_bind_with_undeclared_app_id", expected={OUTCOME_INVALID_CONTENT}, actual_outcome=status_code, detail=detail, ok=status_code == OUTCOME_INVALID_CONTENT, ) ) print("[upd/9] upd_reject_unbind_with_undeclared_app_id") status_code, detail, _ = await self._run_update( primary_deployment_id, DeploymentUpdate( provider_data={ "tools": {}, "operations": [ { "op": "unbind", "tool": self._make_tool_ref(donor_snapshot_id), "app_ids": ["undeclared_app_for_unbind"], } ], } ), ) results.append( self._build_result( name="upd_reject_unbind_with_undeclared_app_id", expected={OUTCOME_INVALID_CONTENT}, actual_outcome=status_code, detail=detail, ok=status_code == OUTCOME_INVALID_CONTENT, ) ) print("[upd/10] upd_reject_unbind_unknown_tool_id") status_code, detail, _ = await self._run_update( primary_deployment_id, DeploymentUpdate( provider_data={ "connections": {}, "operations": [ { "op": "unbind", "tool": {"source_ref": str(uuid4()), "tool_id": str(uuid4())}, "app_ids": [str(donor_config_id)], } ], } ), ) results.append( self._build_result( name="upd_reject_unbind_unknown_tool_id", expected={OUTCOME_INVALID_CONTENT}, actual_outcome=status_code, detail=detail, ok=status_code == OUTCOME_INVALID_CONTENT, ) ) print("[upd/11] upd_missing_add_id_fails") status_code, detail, _ = await self._run_update( primary_deployment_id, DeploymentUpdate( provider_data={ "tools": {}, "connections": {}, "operations": [ { "op": "bind", "tool": {"tool_id_with_ref": {"source_ref": str(uuid4()), "tool_id": str(uuid4())}}, "app_ids": [str(donor_config_id)], } ], } ), ) results.append( self._build_result( name="upd_missing_add_id_fails", expected={OUTCOME_INVALID_CONTENT}, actual_outcome=status_code, detail=detail, ok=status_code == OUTCOME_INVALID_CONTENT, ) ) print("[upd/12] upd_config_raw_payload_conflict") conflict_seed_deployment_id, _conflict_cfg_id, _conflict_snapshot_ids, _ = await self._create_update_seed( label="upd_conflict_seed", snapshot_count=1, ) conflict_suffix = uuid4().hex[:8] conflict_name = f"dup_cfg_{conflict_suffix}" conflict_tool_id = next(iter(_conflict_snapshot_ids), "") if not conflict_tool_id: results.append( self._build_result( name="upd_config_raw_payload_conflict", expected={OUTCOME_CONFLICT}, actual_outcome=OUTCOME_FAILURE, detail="conflict seed snapshot id is missing", ok=False, ) ) return results conflict_payload = DeploymentUpdate( provider_data={ "tools": {}, "connections": {"raw_payloads": [{"app_id": conflict_name, "environment_variables": {}}]}, "operations": [ { "op": "bind", "tool": self._make_tool_id_with_ref(conflict_tool_id), "app_ids": [conflict_name], } ], } ) setup_status, _setup_detail, setup_result = await self._run_update( conflict_seed_deployment_id, conflict_payload ) setup_created_app_ids = self._extract_update_created_app_ids(setup_result) status_code, detail, _ = await self._run_update(conflict_seed_deployment_id, conflict_payload) results.append( self._build_result( name="upd_config_raw_payload_conflict", expected={OUTCOME_CONFLICT}, actual_outcome=status_code, detail=( f"setup={setup_status}:{_setup_detail} setup_created_app_ids={sorted(setup_created_app_ids)} " f"conflict={status_code}:{detail}" ), ok=( setup_status == OUTCOME_SUCCESS and bool(setup_created_app_ids) and status_code == OUTCOME_CONFLICT ), ) ) print("[upd/13] upd_not_found_deployment") status_code, detail, _ = await self._run_update( str(uuid4()), DeploymentUpdate(spec=BaseDeploymentDataUpdate(description="not found update")), ) results.append( self._build_result( name="upd_not_found_deployment", expected={OUTCOME_NOT_FOUND}, actual_outcome=status_code, detail=detail, ok=status_code == OUTCOME_NOT_FOUND, ) ) print("[upd/14] upd_put_tools_replaces_tool_list") put_tools_id, _put_tools_cfg, put_tools_snaps, _ = await self._create_update_seed( label="upd_put_tools", snapshot_count=2 ) put_tools_sorted = sorted(put_tools_snaps) keep_only = [put_tools_sorted[0]] status_code, detail, _ = await self._run_update( put_tools_id, DeploymentUpdate(provider_data={"put_tools": keep_only}), ) list_status, _list_detail, snap_after = await self._run_list_snapshots(put_tools_id) attached_after = self._extract_snapshot_ids(snap_after) results.append( self._build_result( name="upd_put_tools_replaces_tool_list", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=f"{detail} | attached_after={sorted(attached_after)} keep_only={keep_only}", ok=( status_code == OUTCOME_SUCCESS and list_status == OUTCOME_SUCCESS and attached_after == set(keep_only) ), ) ) print("[upd/15] upd_put_tools_empty_clears_all_tools") status_code, detail, _ = await self._run_update( put_tools_id, DeploymentUpdate(provider_data={"put_tools": []}), ) list_status, _list_detail, snap_after_clear = await self._run_list_snapshots(put_tools_id) attached_after_clear = self._extract_snapshot_ids(snap_after_clear) results.append( self._build_result( name="upd_put_tools_empty_clears_all_tools", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=f"{detail} | attached_after={sorted(attached_after_clear)}", ok=( status_code == OUTCOME_SUCCESS and list_status == OUTCOME_SUCCESS and len(attached_after_clear) == 0 ), ) ) print("[upd/16] upd_put_tools_deduplicates") dup_id = put_tools_sorted[0] status_code, detail, _ = await self._run_update( put_tools_id, DeploymentUpdate(provider_data={"put_tools": [dup_id, dup_id, dup_id]}), ) list_status, _list_detail, snap_after_dedup = await self._run_list_snapshots(put_tools_id) attached_after_dedup = self._extract_snapshot_ids(snap_after_dedup) results.append( self._build_result( name="upd_put_tools_deduplicates", expected={OUTCOME_SUCCESS}, actual_outcome=status_code, detail=f"{detail} | attached_after={sorted(attached_after_dedup)}", ok=( status_code == OUTCOME_SUCCESS and list_status == OUTCOME_SUCCESS and attached_after_dedup == {dup_id} ), ) ) # keep seed deployments tracked for shared final cleanup self.created_deployment_ids.add(primary_deployment_id) self.created_deployment_ids.add(donor_deployment_id) self.created_deployment_ids.add(mixed_donor_deployment_id) self.created_deployment_ids.add(put_tools_id) return results async def _run_live_concurrency_scenarios(self) -> list[ScenarioResult]: results: list[ScenarioResult] = [] iterations_raw = os.getenv("WXO_CONCURRENCY_REPEAT", str(DEFAULT_CONCURRENCY_ITERATIONS)).strip() try: iterations = max(1, int(iterations_raw)) except ValueError: iterations = DEFAULT_CONCURRENCY_ITERATIONS print(f"\n[concurrency] running iterations={iterations}") for iteration in range(1, iterations + 1): print(f"\n[concurrency] iteration {iteration}/{iterations}") results.extend(await self._run_live_concurrency_iteration(iteration=iteration)) return results async def _run_live_concurrency_iteration(self, *, iteration: int) -> list[ScenarioResult]: results: list[ScenarioResult] = [] print(f"[cc/{iteration}.1] cc_create_same_prefix_race") 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") shared_payload = self._build_create_payload( tool_payloads=[self._build_flow_payload(label="cc_shared_snap", name_override=shared_snap_name)], raw_connection=DeploymentConfig( name=shared_cfg_name, description="concurrency create collision", environment_variables={}, ), ) shared_payload.spec = shared_payload.spec.model_copy(update={"name": shared_dep_name}, deep=True) create_race = await self._run_parallel_calls( { "left": lambda: self._run_create(shared_payload.model_copy(deep=True)), "right": lambda: self._run_create(shared_payload.model_copy(deep=True)), } ) left_status, left_detail, left_created = create_race["left"] right_status, right_detail, right_created = create_race["right"] self._track_created_result(left_created) self._track_created_result(right_created) create_pair = (left_status, right_status) create_pair_ok = create_pair in { (OUTCOME_SUCCESS, OUTCOME_CONFLICT), (OUTCOME_CONFLICT, OUTCOME_SUCCESS), (OUTCOME_SUCCESS, OUTCOME_SUCCESS), } create_no_internal = OUTCOME_FAILURE not in {left_status, right_status} results.append( self._build_result( name="cc_create_same_prefix_race", expected={OUTCOME_SUCCESS, OUTCOME_CONFLICT}, actual_outcome=max(left_status, right_status), detail=f"left={left_status}:{left_detail} right={right_status}:{right_detail}", ok=create_pair_ok and create_no_internal, ) ) print(f"[cc/{iteration}.2] cc_update_spec_vs_snapshot_race") primary_id, _primary_cfg_id, _primary_snaps, _ = await self._create_update_seed( label=f"cc_upd_primary_{iteration}", snapshot_count=2, ) _donor_id, donor_cfg_id, donor_snapshot_ids, _ = await self._create_update_seed( label=f"cc_upd_donor_{iteration}", snapshot_count=1, ) donor_snapshot_id = next(iter(donor_snapshot_ids), "") update_race = await self._run_parallel_calls( { "spec": lambda: self._run_update( primary_id, DeploymentUpdate(spec=BaseDeploymentDataUpdate(description="cc concurrent spec update")), ), "snapshot": lambda: self._run_update( primary_id, DeploymentUpdate( provider_data={ "tools": {}, "connections": {}, "operations": [ { "op": "bind", "tool": self._make_tool_id_with_ref(donor_snapshot_id), "app_ids": [str(donor_cfg_id)], } ], } ), ), } ) spec_status, spec_detail, _ = update_race["spec"] snapshot_status, snapshot_detail, _ = update_race["snapshot"] list_status, _list_detail, list_after = await self._run_list_snapshots(primary_id) attached_after = self._extract_snapshot_ids(list_after) no_internal_race = OUTCOME_FAILURE not in {spec_status, snapshot_status} results.append( self._build_result( name="cc_update_spec_vs_snapshot_race", expected={ OUTCOME_SUCCESS, OUTCOME_INVALID_OPERATION, OUTCOME_CONFLICT, OUTCOME_INVALID_CONTENT, OUTCOME_NOT_FOUND, }, actual_outcome=max(spec_status, snapshot_status), detail=f"spec={spec_status}:{spec_detail} snapshot={snapshot_status}:{snapshot_detail}", ok=( no_internal_race and list_status == OUTCOME_SUCCESS and self._has_unique_snapshot_ids(attached_after) ), ) ) print(f"[cc/{iteration}.3] cc_update_vs_delete_deployment_race") race_delete_id, _race_delete_cfg, _race_delete_snaps, _ = await self._create_update_seed( label=f"cc_upd_del_{iteration}", snapshot_count=1, ) update_delete_race = await self._run_parallel_calls( { "update": lambda: self._run_update( race_delete_id, DeploymentUpdate(spec=BaseDeploymentDataUpdate(description="cc update while delete")), ), "delete": lambda: self._run_delete(race_delete_id), } ) upd_status, upd_detail, _ = update_delete_race["update"] del_status, del_detail, _ = update_delete_race["delete"] status_after_delete_race, _status_detail, _status_payload = await self._run_status(race_delete_id) allowed_pairs = { (OUTCOME_SUCCESS, OUTCOME_SUCCESS), (OUTCOME_NOT_FOUND, OUTCOME_SUCCESS), (OUTCOME_SUCCESS, OUTCOME_NOT_FOUND), (OUTCOME_NOT_FOUND, OUTCOME_NOT_FOUND), } results.append( self._build_result( name="cc_update_vs_delete_deployment_race", expected={OUTCOME_SUCCESS, OUTCOME_NOT_FOUND, OUTCOME_FAILURE}, actual_outcome=max(upd_status, del_status), detail=( f"update={upd_status}:{upd_detail} " f"delete={del_status}:{del_detail} status={status_after_delete_race}" ), ok=(upd_status, del_status) in allowed_pairs or ( OUTCOME_FAILURE in {upd_status, del_status} and "not found" in f"{upd_detail} {del_detail}".lower() ), ) ) self.created_deployment_ids.discard(race_delete_id) print(f"[cc/{iteration}.4] cc_execution_vs_delete_deployment_race") exec_delete_id, _exec_delete_cfg, _exec_delete_snapshots, _ = await self._create_update_seed( label=f"cc_exec_del_{iteration}", snapshot_count=1, ) exec_delete_race = await self._run_parallel_calls( { "execution": lambda: self._run_create_execution( exec_delete_id, provider_data={"message": {"role": "user", "content": "woah"}}, ), "delete": lambda: self._run_delete(exec_delete_id), } ) exec_status, exec_detail, _ = exec_delete_race["execution"] del_exec_status, del_exec_detail, _ = exec_delete_race["delete"] results.append( self._build_result( name="cc_execution_vs_delete_deployment_race", expected={OUTCOME_SUCCESS, OUTCOME_NOT_FOUND, OUTCOME_SUCCESS, OUTCOME_FAILURE}, actual_outcome=max(exec_status, del_exec_status), detail=f"execution={exec_status}:{exec_detail} delete={del_exec_status}:{del_exec_detail}", ok=OUTCOME_FAILURE not in {exec_status, del_exec_status} or "not found" in f"{exec_detail} {del_exec_detail}".lower(), ) ) self.created_deployment_ids.discard(exec_delete_id) print(f"[cc/{iteration}.5] cc_delete_snapshot_during_update_bindings") delete_bind_id, bind_cfg_id, bind_snapshot_ids, _ = await self._create_update_seed( label=f"cc_del_bind_{iteration}", snapshot_count=2, ) delete_target_snapshot_id = next(iter(bind_snapshot_ids), "") async def _delete_target_before_bind(*args: Any, **kwargs: Any) -> None: # noqa: ARG001 clients = kwargs.get("clients") existing_tool_deltas = kwargs.get("existing_tool_deltas") or {} target_ids = list(existing_tool_deltas.keys()) tool_id = str(target_ids[0]) if target_ids else delete_target_snapshot_id if clients and tool_id: await self._safe_delete_snapshot(clients=clients, snapshot_id=tool_id) del_bind_status, del_bind_detail, _ = await self._run_with_stage_hook( stage="update_bindings", operation=lambda: self._run_update( delete_bind_id, DeploymentUpdate( provider_data={ "tools": {}, "connections": {}, "operations": [ { "op": "unbind", "tool": self._make_tool_ref(tool_id), "app_ids": [str(bind_cfg_id)], } for tool_id in sorted(bind_snapshot_ids) ], } ), ), hook_before=_delete_target_before_bind, ) results.append( self._build_result( name="cc_delete_snapshot_during_update_bindings", expected={OUTCOME_INVALID_CONTENT, OUTCOME_INVALID_OPERATION, OUTCOME_SUCCESS}, actual_outcome=del_bind_status, detail=del_bind_detail, ok=del_bind_status in {OUTCOME_SUCCESS, OUTCOME_INVALID_OPERATION, OUTCOME_INVALID_CONTENT}, ) ) print(f"[cc/{iteration}.6] cc_delete_config_after_update_raw_create") delete_cfg_id, _delete_cfg_base_id, delete_cfg_snapshots, _ = await self._create_update_seed( label=f"cc_del_cfg_{iteration}", snapshot_count=1, ) target_tool_id = next(iter(delete_cfg_snapshots), "") if not target_tool_id: results.append( self._build_result( name="cc_delete_config_after_update_raw_create", expected={OUTCOME_INVALID_OPERATION, OUTCOME_INVALID_CONTENT, OUTCOME_CONFLICT, OUTCOME_FAILURE}, actual_outcome=OUTCOME_FAILURE, detail="seed snapshot id missing for update raw config stage", ok=False, ) ) return results raw_cfg_name = self._mk_name("cc_raw_cfg") async def _delete_created_app_after_config_create(created_app_id: Any, **kwargs: Any) -> None: if not created_app_id: created_app_id = kwargs.get("app_id") app_id = str(created_app_id or "").strip() clients = await self.service._get_provider_clients(user_id=self.user_id, db=self.db) # noqa: SLF001 if app_id: await self._safe_delete_config(clients=clients, config_id=app_id) del_cfg_status, del_cfg_detail, _ = await self._run_with_stage_hook( stage="update_create_config", operation=lambda: self._run_update( delete_cfg_id, DeploymentUpdate( provider_data={ "tools": {}, "connections": {"raw_payloads": [{"app_id": raw_cfg_name, "environment_variables": {}}]}, "operations": [ { "op": "bind", "tool": self._make_tool_id_with_ref(target_tool_id), "app_ids": [raw_cfg_name], } ], } ), ), hook_after=_delete_created_app_after_config_create, ) results.append( self._build_result( name="cc_delete_config_after_update_raw_create", expected={OUTCOME_INVALID_OPERATION, OUTCOME_INVALID_CONTENT, OUTCOME_CONFLICT, OUTCOME_FAILURE}, actual_outcome=del_cfg_status, detail=del_cfg_detail, ok=del_cfg_status in { OUTCOME_INVALID_OPERATION, OUTCOME_INVALID_CONTENT, OUTCOME_CONFLICT, OUTCOME_FAILURE, }, ) ) print(f"[cc/{iteration}.7] cc_create_during_create_snapshots_stage") 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") race_payload = self._build_create_payload( tool_payloads=[self._build_flow_payload(label="cc_stage_snap", name_override=create_race_snap)], raw_connection=DeploymentConfig( name=create_race_cfg, description="cc competing create", environment_variables={}, ), ) 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 async def _launch_competing_create(*args: Any, **kwargs: Any) -> None: # noqa: ARG001 nonlocal competing_create_task if competing_create_task is None: competing_create_task = asyncio.create_task(self._run_create(race_payload.model_copy(deep=True))) await asyncio.sleep(0) staged_create_status, staged_create_detail, staged_create_created = await self._run_with_stage_hook( stage="create_snapshots", operation=lambda: self._run_create(race_payload.model_copy(deep=True)), hook_before=_launch_competing_create, ) competing_create_result = (OUTCOME_FAILURE, "competing create did not start", None) if competing_create_task is not None: competing_create_result = await competing_create_task comp_status, comp_detail, comp_created = competing_create_result self._track_created_result(staged_create_created) self._track_created_result(comp_created) staged_pair = (staged_create_status, comp_status) results.append( self._build_result( name="cc_create_during_create_snapshots_stage", expected={OUTCOME_SUCCESS, OUTCOME_CONFLICT}, actual_outcome=max(staged_create_status, comp_status), detail=f"main={staged_create_status}:{staged_create_detail} competing={comp_status}:{comp_detail}", ok=( staged_pair in { (OUTCOME_SUCCESS, OUTCOME_CONFLICT), (OUTCOME_CONFLICT, OUTCOME_SUCCESS), (OUTCOME_SUCCESS, OUTCOME_SUCCESS), } and OUTCOME_FAILURE not in staged_pair ), ) ) print(f"[cc/{iteration}.8] cc_create_during_update_raw_config_stage") update_create_id, _update_create_cfg, update_create_snaps, _ = await self._create_update_seed( label=f"cc_create_update_cfg_{iteration}", snapshot_count=1, ) update_target_tool_id = next(iter(update_create_snaps), "") if not update_target_tool_id: results.append( self._build_result( name="cc_create_during_update_raw_config_stage", expected={OUTCOME_SUCCESS, OUTCOME_CONFLICT}, actual_outcome=OUTCOME_FAILURE, detail="seed snapshot id missing for competing update", ok=False, ) ) return results 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={ "tools": {}, "connections": {"raw_payloads": [{"app_id": update_cfg_name, "environment_variables": {}}]}, "operations": [ { "op": "bind", "tool": self._make_tool_id_with_ref(update_target_tool_id), "app_ids": [update_cfg_name], } ], }, ) async def _launch_competing_update_create(*args: Any, **kwargs: Any) -> None: # noqa: ARG001 nonlocal competing_update_task if competing_update_task is None: competing_update_task = asyncio.create_task( self._run_update(update_create_id, competing_update_payload.model_copy(deep=True)) ) await asyncio.sleep(0) update_cfg_status, update_cfg_detail, _ = await self._run_with_stage_hook( stage="update_create_config", operation=lambda: self._run_update(update_create_id, competing_update_payload.model_copy(deep=True)), hook_before=_launch_competing_update_create, ) competing_update_result = (OUTCOME_FAILURE, "competing update did not start", None) if competing_update_task is not None: competing_update_result = await competing_update_task competing_upd_status, competing_upd_detail, _ = competing_update_result results.append( self._build_result( name="cc_create_during_update_raw_config_stage", expected={OUTCOME_SUCCESS, OUTCOME_CONFLICT}, actual_outcome=max(update_cfg_status, competing_upd_status), detail=( f"main={update_cfg_status}:{update_cfg_detail} " f"competing={competing_upd_status}:{competing_upd_detail}" ), ok=OUTCOME_FAILURE not in {update_cfg_status, competing_upd_status}, ) ) print(f"[cc/{iteration}.9] cc_delete_resources_during_update_rollback") rollback_id, rollback_cfg_id, rollback_seed_snapshots, _ = await self._create_update_seed( label=f"cc_rollback_delete_{iteration}", snapshot_count=1, ) if not rollback_cfg_id: results.append( self._build_result( name="cc_delete_resources_during_update_rollback", expected={OUTCOME_SUCCESS}, actual_outcome=OUTCOME_FAILURE, detail="seed rollback config id missing", ok=False, ) ) return results rollback_seed_tool_id = next(iter(rollback_seed_snapshots), "") if not rollback_seed_tool_id: results.append( self._build_result( name="cc_delete_resources_during_update_rollback", expected={OUTCOME_FAILURE}, actual_outcome=OUTCOME_FAILURE, detail="seed snapshot id missing for rollback race", ok=False, ) ) return results 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( stage="update_rollback_resources", operation=lambda: self._run_update( rollback_id, DeploymentUpdate( spec=BaseDeploymentDataUpdate(description="cc rollback delete race"), provider_data={ "tools": { "raw_payloads": [rollback_raw_flow.model_dump(mode="json")], }, "connections": { "raw_payloads": [{"app_id": rollback_raw_cfg_name, "environment_variables": {}}], }, "operations": [ { "op": "unbind", "tool": self._make_tool_ref(rollback_seed_tool_id), "app_ids": [str(rollback_cfg_id)], }, { "op": "bind", "tool": {"name_of_raw": rollback_raw_flow.name}, "app_ids": [rollback_raw_cfg_name], }, ], }, ), inject={ "update_bindings": { "fail_first_n": 1, "error_type": "runtime", "message": "cc_rollback_trigger", } }, ), hook_before=self._delete_resources_before_rollback_hook, ) results.append( self._build_result( name="cc_delete_resources_during_update_rollback", expected={OUTCOME_FAILURE}, actual_outcome=rollback_status, detail=rollback_detail, ok=rollback_status == OUTCOME_FAILURE, ) ) print(f"[cc/{iteration}.10] cc_create_during_update_rollback") rollback_create_id, rollback_create_cfg_id, rollback_create_snaps, _ = await self._create_update_seed( label=f"cc_rollback_create_{iteration}", snapshot_count=1, ) if not rollback_create_cfg_id: results.append( self._build_result( name="cc_create_during_update_rollback", expected={OUTCOME_SUCCESS}, actual_outcome=OUTCOME_FAILURE, detail="seed rollback-create config id missing", ok=False, ) ) return results rollback_create_seed_tool_id = next(iter(rollback_create_snaps), "") if not rollback_create_seed_tool_id: results.append( self._build_result( name="cc_create_during_update_rollback", expected={OUTCOME_FAILURE, OUTCOME_SUCCESS, OUTCOME_CONFLICT}, actual_outcome=OUTCOME_FAILURE, detail="seed snapshot id missing for rollback create race", ok=False, ) ) return results rollback_create_cfg_name = self._mk_name("cc_rb_create_cfg") competing_rollback_create_task: asyncio.Task[tuple[str, str, Any | None]] | None = None async def _launch_competing_create_before_rollback(*args: Any, **kwargs: Any) -> None: # noqa: ARG001 nonlocal competing_rollback_create_task if competing_rollback_create_task is not None: return competing_rollback_create_task = asyncio.create_task( self._run_update( rollback_create_id, DeploymentUpdate( provider_data={ "tools": {}, "connections": { "raw_payloads": [{"app_id": rollback_create_cfg_name, "environment_variables": {}}] }, "operations": [ { "op": "bind", "tool": self._make_tool_id_with_ref(rollback_create_seed_tool_id), "app_ids": [rollback_create_cfg_name], } ], }, ), ) ) await asyncio.sleep(0) rollback_create_raw_flow = self._build_flow_payload(label=f"cc_rb_create_raw_{iteration}") rollback_create_status, rollback_create_detail, _ = await self._run_with_stage_hook( stage="update_rollback_resources", operation=lambda: self._run_update( rollback_create_id, DeploymentUpdate( spec=BaseDeploymentDataUpdate(description="cc rollback create race"), provider_data={ "tools": { "raw_payloads": [rollback_create_raw_flow.model_dump(mode="json")], }, "connections": { "raw_payloads": [{"app_id": rollback_create_cfg_name, "environment_variables": {}}], }, "operations": [ { "op": "unbind", "tool": self._make_tool_ref(rollback_create_seed_tool_id), "app_ids": [str(rollback_create_cfg_id)], }, { "op": "bind", "tool": {"name_of_raw": rollback_create_raw_flow.name}, "app_ids": [rollback_create_cfg_name], }, ], }, ), inject={ "update_bindings": { "fail_first_n": 1, "error_type": "runtime", "message": "cc_rollback_create_trigger", } }, ), hook_before=_launch_competing_create_before_rollback, ) competing_rollback_create_result = (OUTCOME_FAILURE, "competing rollback create missing", None) if competing_rollback_create_task is not None: competing_rollback_create_result = await competing_rollback_create_task comp_rb_status, comp_rb_detail, _ = competing_rollback_create_result results.append( self._build_result( name="cc_create_during_update_rollback", expected={OUTCOME_FAILURE, OUTCOME_SUCCESS, OUTCOME_CONFLICT}, actual_outcome=max(rollback_create_status, comp_rb_status), detail=( f"main={rollback_create_status}:{rollback_create_detail} " f"competing={comp_rb_status}:{comp_rb_detail}" ), ok=rollback_create_status == OUTCOME_FAILURE and comp_rb_status in {OUTCOME_SUCCESS, OUTCOME_CONFLICT, OUTCOME_FAILURE}, ) ) print(f"[cc/{iteration}.11] cc_parallel_updates_isolation") dep_a, cfg_a, _snap_a, _ = await self._create_update_seed(label=f"cc_iso_a_{iteration}", snapshot_count=1) dep_b, cfg_b, _snap_b, _ = await self._create_update_seed(label=f"cc_iso_b_{iteration}", snapshot_count=1) isolation_race = await self._run_parallel_calls( { "a": lambda: self._run_update( dep_a, DeploymentUpdate(spec=BaseDeploymentDataUpdate(description="isolation-a")), ), "b": lambda: self._run_update( dep_b, DeploymentUpdate(spec=BaseDeploymentDataUpdate(description="isolation-b")), ), } ) iso_a_status, iso_a_detail, _ = isolation_race["a"] iso_b_status, iso_b_detail, _ = isolation_race["b"] cfg_list_a_status, _cfg_a_detail, cfg_list_a = await self._run_list_configs(dep_a) cfg_list_b_status, _cfg_b_detail, cfg_list_b = await self._run_list_configs(dep_b) cfg_ids_a = self._extract_config_ids(cfg_list_a) cfg_ids_b = self._extract_config_ids(cfg_list_b) print( "[cc/debug] parallel_updates_isolation " f"dep_a={dep_a} cfg_a={cfg_a} update_a={iso_a_status}:{iso_a_detail} " f"list_a={cfg_list_a_status}:{_cfg_a_detail} cfg_ids_a={sorted(cfg_ids_a)} " f"dep_b={dep_b} cfg_b={cfg_b} update_b={iso_b_status}:{iso_b_detail} " f"list_b={cfg_list_b_status}:{_cfg_b_detail} cfg_ids_b={sorted(cfg_ids_b)}" ) isolation_ok = ( cfg_list_a_status == OUTCOME_SUCCESS and cfg_list_b_status == OUTCOME_SUCCESS and cfg_ids_a and cfg_ids_b ) if cfg_a: isolation_ok = isolation_ok and str(cfg_a) in cfg_ids_a if cfg_b: isolation_ok = isolation_ok and str(cfg_b) in cfg_ids_b results.append( self._build_result( name="cc_parallel_updates_isolation", expected={OUTCOME_SUCCESS}, actual_outcome=max(iso_a_status, iso_b_status), detail=f"a={iso_a_status}:{iso_a_detail} b={iso_b_status}:{iso_b_detail}", ok=isolation_ok and iso_a_status == OUTCOME_SUCCESS and iso_b_status == OUTCOME_SUCCESS, ) ) return results async def _run_parallel_calls( self, calls: dict[str, Callable[[], Awaitable[tuple[str, str, Any | None]]]], ) -> dict[str, tuple[str, str, Any | None]]: tasks = {name: asyncio.create_task(call()) for name, call in calls.items()} gathered = await asyncio.gather(*tasks.values(), return_exceptions=True) results: dict[str, tuple[str, str, Any | None]] = {} for name, outcome in zip(tasks, gathered, strict=False): if isinstance(outcome, Exception): results[name] = (OUTCOME_FAILURE, str(outcome), None) else: results[name] = outcome return results def _track_created_result(self, created: WxoCreatedDeploymentResult | None) -> None: if not created: return self.created_deployment_ids.add(created.deployment_id) self.created_config_ids.update(self._extract_create_app_ids(created.provider_result)) self.created_snapshot_ids.update(self._extract_create_snapshot_ids(created.provider_result)) def _has_unique_snapshot_ids(self, snapshot_ids: set[str]) -> bool: snapshot_list = [str(item) for item in snapshot_ids] return len(snapshot_list) == len(set(snapshot_list)) def _extract_config_ids(self, config_result: Any) -> set[str]: configs = getattr(config_result, "configs", []) if config_result else [] 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 { "create_config": (create_core_module, "create_connection_with_conflict_mapping"), "create_snapshots": (create_core_module, "create_and_upload_wxo_flow_tools_with_bindings"), "create_agent": (create_core_module, "create_agent_deployment"), "update_create_config": (update_core_module, "create_connection_with_conflict_mapping"), "update_create_tools": (update_core_module, "create_and_upload_wxo_flow_tools_with_bindings"), "update_bindings": (update_core_module, "_update_existing_tool_connection_deltas"), "update_rollback_resources": (update_core_module, "rollback_update_resources"), } async def _run_with_stage_hook( self, *, stage: str, operation: Callable[[], Awaitable[tuple[str, str, Any | None]]], hook_before: Callable[..., Awaitable[None]] | None = None, hook_after: Callable[..., Awaitable[None]] | None = None, ) -> tuple[str, str, Any | None]: originals: list[tuple[Any, str, Any]] = [] self._apply_stage_hook( stage=stage, originals=originals, hook_before=hook_before, hook_after=hook_after, ) try: return await operation() finally: for target, attr_name, original in originals: setattr(target, attr_name, original) def _apply_stage_hook( self, *, stage: str, originals: list[tuple[Any, str, Any]], hook_before: Callable[..., Awaitable[None]] | None = None, hook_after: Callable[..., Awaitable[None]] | None = None, ) -> None: mapping = self._stage_hook_mapping() target_and_method = mapping.get(stage) if target_and_method is None: return target, method_name = target_and_method original = getattr(target, method_name) originals.append((target, method_name, original)) before_called = {"value": False} after_called = {"value": False} if isinstance(target, types.ModuleType): async def _wrapped_module( *args, __orig=original, __before=hook_before, __after=hook_after, __before_called=before_called, __after_called=after_called, **kwargs, ): if __before is not None and not __before_called["value"]: __before_called["value"] = True await __before(*args, **kwargs) result = await __orig(*args, **kwargs) if __after is not None and not __after_called["value"]: __after_called["value"] = True await __after(result, *args, **kwargs) return result setattr(target, method_name, _wrapped_module) return async def _wrapped_method( _self, *args, __orig=original, __before=hook_before, __after=hook_after, __before_called=before_called, __after_called=after_called, **kwargs, ): if __before is not None and not __before_called["value"]: __before_called["value"] = True await __before(*args, **kwargs) result = await __orig(*args, **kwargs) if __after is not None and not __after_called["value"]: __after_called["value"] = True await __after(result, *args, **kwargs) return result setattr(target, method_name, MethodType(_wrapped_method, target)) async def _safe_delete_snapshot(self, *, clients: Any, snapshot_id: str) -> None: with suppress(ClientAPIException): await asyncio.to_thread(clients.tool.delete, snapshot_id) async def _safe_delete_config(self, *, clients: Any, config_id: str) -> None: with suppress(ClientAPIException): await asyncio.to_thread(clients.connections.delete, config_id) async def _delete_resources_before_rollback_hook(self, *args: Any, **kwargs: Any) -> None: # noqa: ARG002 clients = kwargs.get("clients") if clients is None: return for tool_id in kwargs.get("created_tool_ids") or []: await self._safe_delete_snapshot(clients=clients, snapshot_id=str(tool_id)) app_id = kwargs.get("created_app_id") if app_id: await self._safe_delete_config(clients=clients, config_id=str(app_id)) for created_app_id in kwargs.get("created_app_ids") or []: await self._safe_delete_config(clients=clients, config_id=str(created_app_id)) async def _run_update_failpoint_scenarios(self) -> list[ScenarioResult]: results: list[ScenarioResult] = [] print("\n[fp-upd] creating seed deployment") deployment_id, config_id, snapshot_ids, _ = await self._create_update_seed( label="fp_upd_seed", snapshot_count=1, ) if not config_id: results.append( self._build_result( name="fp_update_seed_missing_config", expected={OUTCOME_SUCCESS}, actual_outcome=OUTCOME_FAILURE, detail="seed deployment config id is missing", ok=False, ) ) return results seed_tool_id = next(iter(snapshot_ids), "") if not seed_tool_id: results.append( self._build_result( name="fp_update_seed_missing_snapshot", expected={OUTCOME_SUCCESS}, actual_outcome=OUTCOME_FAILURE, detail="seed deployment snapshot id is missing", ok=False, ) ) return results failpoint_raw_app_id = self._mk_name("fp_upd_cfg") update_payload = DeploymentUpdate( spec=BaseDeploymentDataUpdate(description="trigger update failpoint"), provider_data={ "tools": {}, "connections": {"raw_payloads": [{"app_id": failpoint_raw_app_id, "environment_variables": {}}]}, "operations": [ { "op": "bind", "tool": self._make_tool_id_with_ref(seed_tool_id), "app_ids": [failpoint_raw_app_id], } ], }, ) print("[fp-upd/1] fp_update_bindings_failure_triggers_rollback") status_code, detail, _ = await self._run_update( deployment_id, update_payload, inject={ "update_bindings": { "fail_first_n": 1, "error_type": "runtime", "message": "fp_update_bindings_failure", } }, ) results.append( self._build_result( name="fp_update_bindings_failure_triggers_rollback", expected={OUTCOME_FAILURE}, actual_outcome=status_code, detail=detail, ok=status_code == OUTCOME_FAILURE, ) ) print("[fp-upd/2] fp_update_bindings_failure_with_rollback_failure") status_code, detail, _ = await self._run_update( deployment_id, update_payload, inject={ "update_bindings": { "fail_first_n": 1, "error_type": "runtime", "message": "fp_update_bindings_failure_again", }, "update_rollback_resources": { "fail_first_n": 1, "error_type": "runtime", "message": "fp_update_rollback_failure", }, }, ) results.append( self._build_result( name="fp_update_bindings_failure_with_rollback_failure", expected={OUTCOME_FAILURE}, actual_outcome=status_code, detail=detail, ok=status_code == OUTCOME_FAILURE, ) ) print("[fp-upd/3] fp_update_failure_then_put_tools_restore") restore_id, restore_cfg_id, restore_snaps, _ = await self._create_update_seed( label="fp_put_tools_restore", snapshot_count=2, ) original_snap_ids = sorted(restore_snaps) if not restore_cfg_id or len(original_snap_ids) < MIN_MIXED_SNAPSHOT_IDS: results.append( self._build_result( name="fp_update_failure_then_put_tools_restore", expected={OUTCOME_SUCCESS}, actual_outcome=OUTCOME_FAILURE, detail=f"seed insufficient: cfg={restore_cfg_id} snaps={len(original_snap_ids)}", ok=False, ) ) return results 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={ "tools": {}, "connections": {"raw_payloads": [{"app_id": restore_raw_cfg, "environment_variables": {}}]}, "operations": [ { "op": "bind", "tool": self._make_tool_id_with_ref(original_snap_ids[0]), "app_ids": [restore_raw_cfg], } ], } ), inject={ "update_bindings": { "fail_first_n": 1, "error_type": "runtime", "message": "fp_put_tools_restore_trigger", } }, ) failure_triggered = inject_status == OUTCOME_FAILURE print("[fp-upd/3b] restoring via put_tools") restore_status, restore_detail, _ = await self._run_update( restore_id, DeploymentUpdate(provider_data={"put_tools": original_snap_ids}), ) list_status, _list_detail, snap_after_restore = await self._run_list_snapshots(restore_id) attached_after_restore = self._extract_snapshot_ids(snap_after_restore) restored_ok = set(original_snap_ids) == attached_after_restore results.append( self._build_result( name="fp_update_failure_then_put_tools_restore", expected={OUTCOME_SUCCESS}, actual_outcome=restore_status, detail=( f"failure_triggered={failure_triggered} inject={inject_status}:{inject_detail} " f"restore={restore_status}:{restore_detail} " f"attached_after={sorted(attached_after_restore)} expected={original_snap_ids}" ), ok=( failure_triggered and restore_status == OUTCOME_SUCCESS and list_status == OUTCOME_SUCCESS and restored_ok ), ) ) return results def _apply_injections(self, inject: dict[str, dict[str, Any]], originals: list[tuple[Any, str, Any]]) -> None: mapping = { "create_config": (shared_core_module, "create_config"), "create_config_wrapper": (create_core_module, "create_connection_with_conflict_mapping"), "create_snapshots": (create_core_module, "create_and_upload_wxo_flow_tools_with_bindings"), "create_agent": (create_core_module, "create_agent_deployment"), "rollback_delete_agent": (retry_module, "delete_agent_if_exists"), "rollback_delete_tool": (retry_module, "delete_tool_if_exists"), "rollback_delete_config": (retry_module, "delete_config_if_exists"), "update_bindings": (update_core_module, "_update_existing_tool_connection_deltas"), "update_rollback_resources": (update_core_module, "rollback_update_resources"), } for stage, config in inject.items(): target_and_method = mapping.get(stage) if target_and_method is None: continue target, method_name = target_and_method original = getattr(target, method_name) originals.append((target, method_name, original)) counter = {"value": 0} fail_first_n = int(config.get("fail_first_n", 0)) error_type = str(config.get("error_type", "runtime")).strip().lower() message = str(config.get("message") or f"injected failure: {stage}") if isinstance(target, types.ModuleType): async def _wrapped_module( *args, __orig=original, __ctr=counter, __n=fail_first_n, __type=error_type, __msg=message, **kwargs, ): __ctr["value"] += 1 if __ctr["value"] <= __n: if __type == "domain_conflict": raise ResourceConflictError(message=__msg) if __type == "domain_not_found": raise DeploymentNotFoundError(message=__msg) if __type == "domain_invalid_content": raise InvalidContentError(message=__msg) if __type == "domain_invalid_operation": raise InvalidDeploymentOperationError(message=__msg) if __type == "domain_failure": raise DeploymentError(message=__msg, error_code="deployment_error") raise RuntimeError(__msg) return await __orig(*args, **kwargs) setattr(target, method_name, _wrapped_module) continue async def _wrapped_method( _self, *args, __orig=original, __ctr=counter, __n=fail_first_n, __type=error_type, __msg=message, **kwargs, ): __ctr["value"] += 1 if __ctr["value"] <= __n: if __type == "domain_conflict": raise ResourceConflictError(message=__msg) if __type == "domain_not_found": raise DeploymentNotFoundError(message=__msg) if __type == "domain_invalid_content": raise InvalidContentError(message=__msg) if __type == "domain_invalid_operation": raise InvalidDeploymentOperationError(message=__msg) if __type == "domain_failure": raise DeploymentError(message=__msg, error_code="deployment_error") raise RuntimeError(__msg) return await __orig(*args, **kwargs) setattr(target, method_name, MethodType(_wrapped_method, target)) def _build_create_payload( self, *, tool_payloads: list[BaseFlowArtifact[WatsonxFlowArtifactProviderData]], raw_connection: DeploymentConfig | None = None, existing_connection_app_id: str | None = None, ) -> DeploymentCreate: spec = BaseDeploymentData( name=self._mk_name("dep_agent"), description="direct adapter scenario", type=DeploymentType.AGENT, ) raw_tool_payloads = [snapshot.model_copy(deep=True) for snapshot in tool_payloads] connections: dict[str, Any] = {} if raw_connection is not None: operation_app_id = str(raw_connection.name).strip() connections["raw_payloads"] = [ { "app_id": operation_app_id, "environment_variables": raw_connection.environment_variables, "provider_config": raw_connection.provider_config, } ] elif existing_connection_app_id: operation_app_id = str(existing_connection_app_id).strip() else: operation_app_id = self._mk_name("cfg_default_app") connections["raw_payloads"] = [{"app_id": operation_app_id, "environment_variables": {}}] operations = [ { "op": "bind", "tool": {"name_of_raw": str(flow_payload.name).strip()}, "app_ids": [operation_app_id], } for flow_payload in raw_tool_payloads ] provider_data = { "tools": {"raw_payloads": raw_tool_payloads}, "connections": connections, "operations": operations, "llm": self.llm, } return DeploymentCreate(spec=spec, provider_data=provider_data) def _build_flow_payload( self, *, label: str, name_override: str | None = None, ) -> BaseFlowArtifact[WatsonxFlowArtifactProviderData]: flow_id = uuid4() return BaseFlowArtifact[WatsonxFlowArtifactProviderData]( id=flow_id, name=name_override or self._mk_name(label), description="direct adapter flow payload", data=self._build_flow_data_payload(), tags=["e2e", "watsonx-direct-adapter"], provider_data=WatsonxFlowArtifactProviderData( project_id=self.project_id, source_ref=str(flow_id), ), ) def _build_config_payload(self, *, label: str) -> DeploymentConfig: return DeploymentConfig( name=self._mk_name(label), description="direct adapter config payload", environment_variables={}, ) def _build_flow_data_payload(self) -> dict[str, Any]: chat_input_node_id = f"ChatInput-{uuid4().hex[:8]}" chat_output_node_id = f"ChatOutput-{uuid4().hex[:8]}" return { "nodes": [ { "id": chat_input_node_id, "type": "genericNode", "position": {"x": 100, "y": 100}, "data": { "type": "ChatInput", "id": chat_input_node_id, "node": {"template": {"_type": "CustomComponent"}}, }, }, { "id": chat_output_node_id, "type": "genericNode", "position": {"x": 400, "y": 100}, "data": { "type": "ChatOutput", "id": chat_output_node_id, "node": {"template": {"_type": "CustomComponent"}}, }, }, ], "edges": [], "viewport": {"x": 0, "y": 0, "zoom": 1}, } async def _cleanup_resources(self) -> None: clients = await self.service._get_provider_clients(user_id=self.user_id, db=self.db) # noqa: SLF001 print("\nCleanup Resources") print("-" * 90) print( f"cleanup targets: deployments={len(self.created_deployment_ids)} " f"snapshots={len(self.created_snapshot_ids)} " f"configs={len(self.created_config_ids)}" ) deleted_deployments = 0 deleted_snapshots = 0 deleted_configs = 0 for deployment_id in sorted(self.created_deployment_ids): print(f"[cleanup] deleting deployment {deployment_id}...") with suppress(Exception): await self.service.delete(user_id=self.user_id, deployment_id=deployment_id, db=self.db) deleted_deployments += 1 print(f"[cleanup] deleted deployment {deployment_id}") for snapshot_id in sorted(self.created_snapshot_ids): print(f"[cleanup] deleting snapshot {snapshot_id}...") try: await asyncio.to_thread(clients.tool.delete, snapshot_id) deleted_snapshots += 1 print(f"[cleanup] deleted snapshot {snapshot_id}") except ClientAPIException as exc: if exc.response.status_code != HTTP_STATUS_NOT_FOUND: print(f"[cleanup-warning] snapshot {snapshot_id}: {exc}") else: print(f"[cleanup] snapshot {snapshot_id} already deleted (404)") for config_id in sorted(self.created_config_ids): print(f"[cleanup] deleting config {config_id}...") try: await asyncio.to_thread(clients.connections.delete, config_id) deleted_configs += 1 print(f"[cleanup] deleted config {config_id}") except ClientAPIException as exc: if exc.response.status_code != HTTP_STATUS_NOT_FOUND: print(f"[cleanup-warning] config {config_id}: {exc}") else: print(f"[cleanup] config {config_id} already deleted (404)") print( f"cleanup completed: deployments_deleted={deleted_deployments} " f"snapshots_deleted={deleted_snapshots} " f"configs_deleted={deleted_configs}" ) print("-" * 90) def _mk_name(self, prefix: str) -> str: raw = f"{prefix}_{self.run_suffix}_{uuid4().hex[:6]}" normalized = _INVALID_WXO_NAME_CHARS.sub("_", raw) if not normalized or not normalized[0].isalpha(): normalized = f"n_{normalized}" return normalized def _scenario_group(self, name: str) -> str: prefix = name.split("_", 1)[0] if prefix in {"live", "upd", "cc", "fp"}: return prefix return "other" def _scenario_label(self, name: str) -> str: if "_" not in name: return name return name.split("_", 1)[1] def _format_expected_outcomes(self, outcomes: set[str]) -> str: ordered = sorted(outcomes, key=lambda item: (item != OUTCOME_SUCCESS, item)) return ", ".join(ordered) def _print_result_row(self, result: ScenarioResult) -> None: verdict = "PASS" if result.ok else "FAIL" scenario_label = self._scenario_label(result.name) expected = self._format_expected_outcomes(result.expected_outcomes) print(f"{verdict:<5} | {scenario_label:<44} | got: {result.actual_outcome}") print(f"{'':5} | {'':44} | expected: {expected}") if not result.ok: wrapped_detail = textwrap.wrap(result.detail, width=108) or [result.detail] for index, line in enumerate(wrapped_detail): prefix = "detail: " if index == 0 else " " print(f"{'':5} | {'':44} | {prefix}{line}") def _print_summary(self, results: list[ScenarioResult]) -> None: print("\nScenario Summary") print("-" * 90) total = len(results) passed = sum(1 for result in results if result.ok) failed = total - passed print(f"Total={total} Passed={passed} Failed={failed}") print("-" * 90) grouped_results: dict[str, list[ScenarioResult]] = {"live": [], "upd": [], "cc": [], "fp": [], "other": []} for result in results: grouped_results[self._scenario_group(result.name)].append(result) for group in ("live", "upd", "cc", "fp", "other"): group_items = grouped_results[group] if not group_items: continue print(f"\n[{group}] ({len(group_items)})") for result in group_items: self._print_result_row(result) print("-" * 90) def _get_required_env(name: str) -> str: value = os.getenv(name, "").strip() if not value: msg = f"Environment variable '{name}' is required." raise RuntimeError(msg) return value 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() async def _main() -> int: load_dotenv() args = _parse_args() runner = WatsonxAdapterDirectE2E( provider_backend_url=_get_required_env("WXO_INSTANCE_URL"), provider_api_key=_get_required_env("WXO_API_KEY"), project_id=args.project_id, mode=args.mode, keep_resources=args.keep_resources, llm=args.llm, ) return await runner.run() if __name__ == "__main__": raise SystemExit(asyncio.run(_main()))