Files
Hamza Rashid c77c473bb2 feat: e2e tests for deployments api (#12767)
* update e2e tests

* fix(watsonx): harden existing-resource create flow and rollback journaling
- move the direct Watsonx adapter E2E runner to `scripts/e2e_deployment_tests/watsonx_orchestrate/adapter.py` (rename-only) so deployment E2E assets are consolidated under one folder
- add `scripts/e2e_deployment_tests/watsonx_orchestrate/api.py` with a full `/api/v1/deployments` matrix covering create/update happy paths, validation rejections, attachment patching, rollback/error paths, concurrency races, large payload tiers, and failpoint scenarios with owned-resource cleanup
- clarify existing-resource create behavior in `src/backend/base/langflow/api/v1/deployments.py`: DB-only onboarding keeps `created_*` fields empty unless provider mutation operations are requested
- add `util_create_result_from_existing_resource` in `src/backend/base/langflow/api/v1/mappers/deployments/watsonx_orchestrate/mapper.py` to normalize non-mutating onboard responses into a create-style result with empty `app_ids` and `tools_with_refs`
- extend `src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/config.py` to accept `created_app_ids_journal` and append app ids immediately after successful provider connection creation for rollback safety
- update shared connection orchestration in `src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/shared.py` to normalize provider app ids via `RawConnectionCreatePlan.__post_init__`, propagate `created_app_ids_journal`, dedupe rollback ids, and wrap validation-stage failures as `ConnectionCreateBatchError` with rollback metadata
- wire journaling into create/update rollback flows in `src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/create.py` and `src/backend/base/langflow/services/adapters/deployment/watsonx_orchestrate/core/update.py` so partial provider-side connection creation is always captured for cleanup
- add mapper coverage in `src/backend/tests/unit/api/v1/test_deployment_mapper_watsonx.py` for existing-resource create-result normalization
- expand `src/backend/tests/unit/services/deployment/test_watsonx_orchestrate.py` with coverage for provider app-id normalization, validation-failure rollback metadata, and create/update rollback when failures occur after provider connection creation; update mocks/monkeypatch targets to match the shared connection entrypoint signature

* simplify create result logic (improves error friendliness

(cherry picked from commit 9f42c9e707)
2026-04-23 17:49:51 -07:00

3604 lines
158 KiB
Python

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