from __future__ import annotations import importlib import io import zipfile from types import SimpleNamespace from typing import Any from uuid import UUID import pytest from fastapi import HTTPException, status from lfx.services.adapters.deployment.exceptions import ( AuthorizationError, CredentialResolutionError, DeploymentError, DeploymentNotConfiguredError, DeploymentNotFoundError, DeploymentSupportError, InvalidContentError, InvalidDeploymentOperationError, OperationNotSupportedError, ResourceConflictError, ResourceNotFoundError, ) from lfx.services.adapters.deployment.schema import ( BaseDeploymentData, BaseDeploymentDataUpdate, BaseFlowArtifact, ConfigItem, ConfigListParams, DeploymentConfig, DeploymentCreate, DeploymentListParams, DeploymentType, DeploymentUpdate, EnvVarValueSpec, ExecutionCreate, ExecutionCreateResult, ExecutionStatusResult, SnapshotItem, SnapshotListParams, SnapshotListResult, ) from pydantic import ValidationError try: from langflow.services.adapters.deployment.watsonx_orchestrate import WatsonxOrchestrateDeploymentService except ModuleNotFoundError: pytest.skip( "Skipping Watsonx deployment tests: optional IBM SDK dependencies not available.", allow_module_level=True, ) tools_module = importlib.import_module("langflow.services.adapters.deployment.watsonx_orchestrate.core.tools") service_module = importlib.import_module("langflow.services.adapters.deployment.watsonx_orchestrate.service") update_core_module = importlib.import_module("langflow.services.adapters.deployment.watsonx_orchestrate.core.update") create_core_module = importlib.import_module("langflow.services.adapters.deployment.watsonx_orchestrate.core.create") shared_core_module = importlib.import_module("langflow.services.adapters.deployment.watsonx_orchestrate.core.shared") payloads_module = importlib.import_module("langflow.services.adapters.deployment.watsonx_orchestrate.payloads") client_module = importlib.import_module("langflow.services.adapters.deployment.watsonx_orchestrate.client") types_module = importlib.import_module("langflow.services.adapters.deployment.watsonx_orchestrate.types") deployment_context_module = importlib.import_module("langflow.services.adapters.deployment.context") WxOCredentials = importlib.import_module( "langflow.services.adapters.deployment.watsonx_orchestrate.types" ).WxOCredentials # Aliases for classes used in tests (module-level to satisfy N806). ToolConnectionOps = update_core_module.ToolConnectionOps OrderedUniqueStrs = shared_core_module.OrderedUniqueStrs WatsonxDeploymentUpdatePayload = payloads_module.WatsonxDeploymentUpdatePayload WatsonxRenameToolOperation = payloads_module.WatsonxRenameToolOperation ListConfigsResponse = importlib.import_module( "ibm_watsonx_orchestrate_clients.connections.connections_client" ).ListConfigsResponse TEST_WXO_LLM = "ibm/granite-3.3-8b" def _normalized_provider_app_id(app_id: str) -> str: return payloads_module.validate_wxo_name(app_id, field_label="Connection app id") def _assert_langflow_agent_name(agent_name: str, *, display_name: str | None = None) -> None: prefix = "langflow_" assert agent_name.startswith(prefix) if display_name is not None: normalized_display_name = payloads_module.normalize_wxo_name(display_name).strip("_") assert normalized_display_name assert agent_name.startswith(f"{prefix}{normalized_display_name}_") uuid_segment = agent_name.rsplit("_", maxsplit=1)[-1] assert len(uuid_segment) == 8 assert all(char in "0123456789abcdef" for char in uuid_segment) assert payloads_module.validate_wxo_name(agent_name, field_label="Agent name") == agent_name def _agent_technical_name(display_name: str = "my deployment") -> str: normalized_display_name = payloads_module.normalize_wxo_name(display_name).strip("_") return f"langflow_{normalized_display_name}_1234abcd" def _reload_wxo_auth_modules(): constants_module = importlib.import_module("langflow.services.adapters.deployment.watsonx_orchestrate.constants") importlib.reload(constants_module) return importlib.reload(client_module) def _tool_refs(*tool_ids: str) -> list[dict[str, str]]: """Build WatsonxToolRefBinding dicts for test data.""" return [{"source_ref": f"fv-{tid}", "tool_id": tid} for tid in tool_ids] def _tool_ref(tool_id: str) -> dict[str, str]: """Build a single WatsonxToolRefBinding dict for test data.""" return {"source_ref": f"fv-{tool_id}", "tool_id": tool_id} class DummySettingsService: def __init__(self): self.settings = SimpleNamespace() class FakeAgentClient: def __init__( self, deployment: dict | list | None, listed_agents: list[dict] | None = None, get_payloads: dict[str, dict | list[dict]] | None = None, create_response: object | None = None, create_exception: Exception | None = None, *, fill_missing_metadata: bool = True, ): self._deployment = {**deployment} if isinstance(deployment, dict) else deployment if isinstance(self._deployment, dict) and self._deployment.get("id") and not self._deployment.get("name"): self._deployment["name"] = f"agent-{self._deployment['id']}" if isinstance(self._deployment, dict) and fill_missing_metadata: self._deployment.setdefault("display_name", self._deployment.get("name", "Test Agent")) self._deployment.setdefault("description", "desc") self._deployment.setdefault("llm", TEST_WXO_LLM) self._listed_agents = [dict(agent) for agent in listed_agents] if listed_agents else [] if fill_missing_metadata: for agent in self._listed_agents: if agent.get("id") and not agent.get("name"): agent["name"] = f"agent-{agent['id']}" agent.setdefault("display_name", agent.get("name", "Test Agent")) agent.setdefault("description", "desc") agent.setdefault("llm", TEST_WXO_LLM) self._get_payloads = get_payloads or {} self._create_response = create_response or SimpleNamespace(id="dep-created") self._create_exception = create_exception self.update_calls: list[tuple[str, dict]] = [] self.post_calls: list[tuple[str, dict]] = [] self.create_calls: list[dict] = [] self.delete_calls: list[str] = [] self.last_list_params = None def get_draft_by_id(self, deployment_id: str): # noqa: ARG002 return self._deployment def get_drafts_by_ids(self, deployment_ids: list[str]): return [agent for agent in self._listed_agents if str(agent.get("id") or "").strip() in set(deployment_ids)] def get_drafts_by_names(self, agent_names: list[str]): return [agent for agent in self._listed_agents if str(agent.get("name") or "").strip() in set(agent_names)] def get_draft_by_name(self, agent_name: str): return [agent for agent in self._listed_agents if agent.get("name") == agent_name] def get(self): return self._listed_agents def update(self, deployment_id: str, payload: dict): self.update_calls.append((deployment_id, payload)) def create(self, payload: dict): self.create_calls.append(payload) if self._create_exception is not None: raise self._create_exception return self._create_response def delete(self, deployment_id: str): self.delete_calls.append(deployment_id) def _post(self, path: str, data: dict): self.post_calls.append((path, data)) return { "thread_id": "thread-1", "run_id": "run-1", "task_id": "task-1", "message_id": "message-1", } def _get(self, path: str, params: dict | None = None): if path == "/agents": self.last_list_params = params ids = params.get("ids", []) if isinstance(params, dict) else [] names = params.get("names", []) if isinstance(params, dict) else [] id_set = {str(item) for item in ids} name_set = {str(item) for item in names} if id_set or name_set: return [ agent for agent in self._listed_agents if str(agent.get("id") or "").strip() in id_set or str(agent.get("name") or "").strip() in name_set ] return self._listed_agents return self._get_payloads.get(path, {}) class FakeToolClient: def __init__(self, tools: list[dict], existing_names: set[str] | None = None): self._tools_by_id = {str(tool.get("id")): dict(tool) for tool in tools if tool.get("id")} self._existing_names = existing_names or set() self.delete_calls: list[str] = [] self.update_calls: list[tuple[str, dict]] = [] self.create_calls: list[dict] = [] self.get_drafts_by_ids_calls: list[list[str]] = [] def get_drafts_by_ids(self, tool_ids: list[str]): self.get_drafts_by_ids_calls.append(list(tool_ids)) return [dict(self._tools_by_id[tool_id]) for tool_id in tool_ids if tool_id in self._tools_by_id] def get_draft_by_name(self, tool_name: str): if tool_name in self._existing_names: return [{"name": tool_name}] return [] def get_drafts_by_names(self, names: list[str]): return [dict(tool) for tool in self._tools_by_id.values() if tool.get("name") in names] def update(self, tool_id: str, payload: dict): self.update_calls.append((tool_id, payload)) current = self._tools_by_id.get(tool_id, {"id": tool_id}) merged = dict(current) merged.update(payload) self._tools_by_id[tool_id] = merged def create(self, payload: dict): self.create_calls.append(payload) tool_id = f"created-tool-{len(self.create_calls)}" self._tools_by_id[tool_id] = {"id": tool_id, **payload} return {"id": tool_id} def delete(self, tool_id: str): self.delete_calls.append(tool_id) class FakeConnectionsClient: def __init__(self, existing_app_id: str | None = None): self._connections_by_app_id: dict[str, str] = {} if existing_app_id: self._connections_by_app_id[existing_app_id] = "conn-1" self.delete_calls: list[str] = [] self.delete_credentials_calls: list[tuple[str, object, bool]] = [] self._list_entries: list[dict] = [] self._draft_entries_by_id: list[object] = [] self.create_calls: list[dict] = [] self.create_config_calls: list[tuple[str, dict]] = [] self.create_credentials_calls: list[tuple[str, object, bool, dict]] = [] def get_draft_by_app_id(self, app_id: str): if app_id in self._connections_by_app_id: return SimpleNamespace(connection_id=self._connections_by_app_id[app_id]) return None def get_config(self, app_id: str, env): # noqa: ARG002 from ibm_watsonx_orchestrate_core.types.connections import ConnectionSecurityScheme return SimpleNamespace(security_scheme=ConnectionSecurityScheme.KEY_VALUE) def get_credentials(self, app_id: str, env, *, use_app_credentials: bool): # noqa: ARG002 return {"runtime_credentials": {"TOKEN": "value"}} def delete_credentials(self, app_id: str, env, *, use_app_credentials: bool): self.delete_credentials_calls.append((app_id, env, use_app_credentials)) def delete(self, app_id: str): self.delete_calls.append(app_id) self._connections_by_app_id.pop(app_id, None) def create(self, payload: dict): self.create_calls.append(payload) app_id = str(payload.get("app_id")) self._connections_by_app_id[app_id] = f"conn-{app_id}" def create_config(self, app_id: str, payload: dict): self.create_config_calls.append((app_id, payload)) def create_credentials(self, app_id: str, env, *, use_app_credentials: bool, payload: dict): self.create_credentials_calls.append((app_id, env, use_app_credentials, payload)) def list(self): return self._list_entries def get_drafts_by_ids(self, conn_ids: list[str]): requested = set(conn_ids) entries = [] for entry in self._draft_entries_by_id: connection_id = str(getattr(entry, "connection_id", "") or "") if connection_id in requested: entries.append(entry) return entries class FakeBaseClient: def __init__( self, *, post_response: dict | None = None, get_payloads: dict[str, dict] | None = None, ): self.post_response = post_response or { "thread_id": "thread-1", "run_id": "run-1", "task_id": "task-1", "message_id": "message-1", } self._get_payloads = get_payloads or {} self.post_calls: list[tuple[str, dict]] = [] self.get_calls: list[str] = [] def _post(self, path: str, data: dict): self.post_calls.append((path, data)) return self.post_response def _get(self, path: str, params: dict | None = None): # noqa: ARG002 self.get_calls.append(path) return self._get_payloads.get(path, {}) class FakeWXOClients(SimpleNamespace): def __init__(self, *, agent=None, tool=None, connections=None): super().__init__( agent=agent or FakeAgentClient({"id": "dep-1", "tools": []}), tool=tool or FakeToolClient([]), connections=connections or FakeConnectionsClient(), ) self.upload_tool_artifact_calls: list[str] = [] def upload_tool_artifact(self, tool_id: str, files: dict[str, tuple[str, io.BytesIO, str, dict[str, str]]]): file_obj = files["file"][1] file_obj.read() self.upload_tool_artifact_calls.append(tool_id) return {"id": tool_id} def _with_wxo_wrappers(ns): """Attach WxOClient SDK wrapper methods to a SimpleNamespace test double.""" if hasattr(ns, "_base") and ns._base is not None: ns.get_agents_raw = lambda params=None: ns._base._get("/agents", params=params) ns.get_models_raw = lambda params=None: ns._base._get("/models", params=params) ns.get_tools_raw = lambda params=None: ns._base._get("/tools", params=params) ns.post_run = lambda *, data: ns._base._post("/runs", data) ns.get_run = lambda run_id: ns._base._get(f"/runs/{run_id}") return ns def _attach_provider_clients(service: WatsonxOrchestrateDeploymentService, clients: object) -> None: async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return clients service._get_provider_clients = mock_get_provider_clients def _create_provider_spec( *, display_name: str = "my deployment", existing_tool_ids: list[str] | None = None, existing_app_ids: list[str] | None = None, ) -> dict: tool_ids = existing_tool_ids or ["tool-existing-1"] app_ids = existing_app_ids or ["app-existing-1"] return { "display_name": display_name, "tools": {}, "connections": {}, "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"tool_id_with_ref": _tool_ref(tool_ids[0])}, "app_ids": [app_ids[0]], } ], } @pytest.mark.anyio async def test_process_config_uses_raw_payload_but_overrides_name(monkeypatch): from langflow.services.adapters.deployment.watsonx_orchestrate.core.config import process_config captured = {} async def mock_create_config(*, clients, config, user_id, db, created_app_ids_journal=None): # noqa: ARG001 captured["name"] = config.name captured["env_vars"] = config.environment_variables return config.name monkeypatch.setattr( "langflow.services.adapters.deployment.watsonx_orchestrate.core.config.create_config", mock_create_config, ) app_id = await process_config( clients=SimpleNamespace(), user_id="user-1", db=object(), deployment_name="my_deployment", config=ConfigItem( raw_payload=DeploymentConfig( name="caller_supplied_name", description="from payload", environment_variables=None, ) ), ) assert app_id == "my_deployment" assert captured["name"] == "my_deployment" assert captured["env_vars"] is None @pytest.mark.anyio async def test_process_config_rejects_reference_id(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.config import process_config with pytest.raises(InvalidDeploymentOperationError, match="Config reference binding is not supported"): await process_config( clients=SimpleNamespace(), user_id="user-1", db=object(), deployment_name="my_deployment", config=ConfigItem(reference_id="existing-config"), ) @pytest.mark.anyio async def test_create_rejects_legacy_top_level_config_section(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": []}), tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(InvalidDeploymentOperationError, match="Top-level 'snapshot' and 'config' create sections"): await service.create( user_id="user-1", db=object(), payload=DeploymentCreate( spec=BaseDeploymentData( name=_agent_technical_name(), description="desc", type=DeploymentType.AGENT, ), provider_data=_create_provider_spec(), config=ConfigItem(reference_id="existing-config"), ), ) @pytest.mark.anyio async def test_create_rejects_missing_llm(): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) provider_data = _create_provider_spec() provider_data.pop("llm", None) with pytest.raises(InvalidContentError, match=r"Missing required field 'llm'"): await service.create( user_id="user-1", db=object(), payload=DeploymentCreate( spec=BaseDeploymentData( name=_agent_technical_name(), description="desc", type=DeploymentType.AGENT, ), provider_data=provider_data, ), ) def test_create_payload_rejects_empty_work(): with pytest.raises( ValidationError, match=r"At least one bind/attach_tool operation or tools\.raw_payloads entry must be provided for create", ): payloads_module.WatsonxDeploymentCreatePayload.model_validate( { "display_name": "my deployment", "llm": TEST_WXO_LLM, } ) @pytest.mark.anyio async def test_resolve_runtime_credentials_supports_variable_and_raw_sources(monkeypatch): async def mock_resolve_variable_value(variable_name: str, *, user_id, db, **kwargs): # noqa: ARG001 return f"resolved::{variable_name}" monkeypatch.setattr( "langflow.services.adapters.deployment.watsonx_orchestrate.client.resolve_variable_value", mock_resolve_variable_value, ) from langflow.services.adapters.deployment.watsonx_orchestrate.client import resolve_runtime_credentials runtime_credentials = await resolve_runtime_credentials( user_id="user-1", db=object(), environment_variables={ "FROM_RAW": EnvVarValueSpec(source="raw", value="raw-token"), "FROM_VAR": EnvVarValueSpec(source="variable", value="OPENAI_API_KEY"), }, ) assert runtime_credentials.model_dump() == { "FROM_RAW": "raw-token", "FROM_VAR": "resolved::OPENAI_API_KEY", } @pytest.mark.anyio async def test_update_rejects_legacy_top_level_snapshot_or_config(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}), tool=FakeToolClient([{"id": "tool-1", "binding": {"langflow": {"connections": {}}}}]), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(InvalidDeploymentOperationError, match="no longer supported"): await service.update( user_id="user-1", deployment_id="dep-1", payload=DeploymentUpdate(snapshot={"add_ids": ["tool-1"]}), db=object(), ) @pytest.mark.anyio async def test_update_rejects_legacy_top_level_config_section(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}), tool=FakeToolClient([{"id": "tool-1", "binding": {"langflow": {"connections": {}}}}]), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(InvalidDeploymentOperationError, match="no longer supported"): await service.update( user_id="user-1", deployment_id="dep-1", payload=DeploymentUpdate(config={"config_id": "cfg-legacy"}), db=object(), ) @pytest.mark.anyio async def test_update_provider_data_binds_existing_tool_and_updates_agent_tools(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "display_name": "Existing Agent", "tools": ["tool-1"]}) fake_tool = FakeToolClient( [ {"id": "tool-1", "name": "tool-1", "binding": {"langflow": {"connections": {}}}}, {"id": "tool-3", "name": "tool-3", "binding": {"langflow": {"connections": {}}}}, ] ) fake_clients = SimpleNamespace( agent=fake_agent, tool=fake_tool, connections=FakeConnectionsClient(existing_app_id="cfg-new"), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients async def mock_validate_connection(connections_client, *, app_id): # noqa: ARG001 assert app_id == "cfg-new" return SimpleNamespace(connection_id="conn-new") monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) monkeypatch.setattr(update_core_module, "validate_connection", mock_validate_connection) result = await service.update( user_id="user-1", deployment_id="dep-1", payload=DeploymentUpdate( provider_data={ "tools": {}, "connections": {}, "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-3")}, "app_ids": ["cfg-new"], } ], } ), db=object(), ) assert result.provider_result is not None assert result.provider_result.created_app_ids == [] assert result.provider_result.created_snapshot_ids == [] assert result.provider_result.added_snapshot_ids == ["tool-3"] assert result.provider_result.display_name == "Existing Agent" assert [tool_id for tool_id, _payload in fake_tool.update_calls] == ["tool-3"] _, updated_tool_payload = fake_tool.update_calls[0] assert updated_tool_payload["binding"]["langflow"]["connections"]["cfg-new"] == "conn-new" _, agent_payload = fake_agent.update_calls[0] assert agent_payload["tools"] == ["tool-1", "tool-3"] assert agent_payload["llm"] == TEST_WXO_LLM @pytest.mark.anyio async def test_update_provider_data_bind_unbind_and_rename_preserves_connection_deltas(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}) fake_tool = FakeToolClient( [ { "id": "tool-1", "name": "tool-1", "display_name": "tool-1", "binding": { "langflow": { "connections": {"cfg-keep": "conn-keep", "cfg-remove": "conn-remove"}, } }, }, ] ) fake_clients = SimpleNamespace( agent=fake_agent, tool=fake_tool, connections=FakeConnectionsClient(existing_app_id="cfg-add"), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients async def mock_validate_connection(connections_client, *, app_id): # noqa: ARG001 connection_id_by_app_id = { "cfg-add": "conn-add", "cfg-remove": "conn-remove", } assert app_id in connection_id_by_app_id return SimpleNamespace(connection_id=connection_id_by_app_id[app_id]) monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) monkeypatch.setattr(update_core_module, "validate_connection", mock_validate_connection) result = await service.update( user_id="user-1", deployment_id="dep-1", payload=DeploymentUpdate( provider_data={ "tools": {}, "connections": {}, "llm": TEST_WXO_LLM, "operations": [ {"op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-1")}, "app_ids": ["cfg-add"]}, {"op": "unbind", "tool": _tool_ref("tool-1"), "app_ids": ["cfg-remove"]}, {"op": "rename_tool", "tool": _tool_ref("tool-1"), "tool_display_name": "Renamed Tool"}, ], } ), db=object(), ) assert result.provider_result is not None assert [tool_id for tool_id, _payload in fake_tool.update_calls] == ["tool-1"] _, rename_payload = fake_tool.update_calls[0] _assert_langflow_agent_name(rename_payload["name"], display_name="Renamed Tool") assert rename_payload["display_name"] == "Renamed Tool" assert rename_payload["binding"]["langflow"]["connections"] == { "cfg-keep": "conn-keep", "cfg-add": "conn-add", } _, agent_payload = fake_agent.update_calls[0] assert agent_payload["tools"] == ["tool-1"] assert agent_payload["llm"] == TEST_WXO_LLM @pytest.mark.anyio async def test_update_provider_data_llm_only_updates_agent(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return SimpleNamespace(agent=fake_agent) monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.update( user_id="user-1", deployment_id="dep-1", payload=DeploymentUpdate(provider_data={"llm": TEST_WXO_LLM}), db=object(), ) assert result.id == "dep-1" assert len(fake_agent.update_calls) == 1 _, agent_payload = fake_agent.update_calls[0] assert agent_payload == {"llm": TEST_WXO_LLM} @pytest.mark.anyio async def test_update_provider_data_accepts_missing_llm(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}) fake_tool = FakeToolClient([{"id": "tool-1", "name": "tool-1", "binding": {"langflow": {"connections": {}}}}]) fake_connections = FakeConnectionsClient(existing_app_id="cfg-1") async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return SimpleNamespace(agent=fake_agent, tool=fake_tool, connections=fake_connections) monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.update( user_id="user-1", deployment_id="dep-1", payload=DeploymentUpdate( provider_data={ "connections": {}, "operations": [{"op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-1")}, "app_ids": ["cfg-1"]}], } ), db=object(), ) assert result.id == "dep-1" assert fake_agent.update_calls assert all("llm" not in payload for _, payload in fake_agent.update_calls) def test_update_payload_rejects_connections_without_bind_or_unbind_operations(): with pytest.raises( ValidationError, match="connections require at least one bind/unbind operation that references app_ids", ): payloads_module.WatsonxDeploymentUpdatePayload.model_validate( { "llm": TEST_WXO_LLM, "connections": { "raw_payloads": [ {"app_id": "cfg-1", "environment_variables": {"K": {"source": "raw", "value": "v"}}} ] }, "operations": [], } ) def test_update_payload_rejects_existing_tool_bind_with_empty_app_ids(): with pytest.raises(ValidationError): payloads_module.WatsonxDeploymentUpdatePayload.model_validate( { "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-existing")}, "app_ids": [], } ], } ) def test_update_payload_accepts_attach_tool_operation(): payload = payloads_module.WatsonxDeploymentUpdatePayload.model_validate( { "llm": TEST_WXO_LLM, "operations": [ { "op": "attach_tool", "tool": _tool_ref("tool-existing"), } ], } ) attach_op = payload.operations[0] assert isinstance(attach_op, payloads_module.WatsonxAttachToolOperation) assert attach_op.tool.tool_id == "tool-existing" def test_update_payload_rejects_attach_and_bind_for_same_existing_tool(): with pytest.raises( ValidationError, match="attach_tool cannot be combined with bind\\.tool\\.tool_id_with_ref", ): payloads_module.WatsonxDeploymentUpdatePayload.model_validate( { "llm": TEST_WXO_LLM, "operations": [ {"op": "attach_tool", "tool": _tool_ref("tool-existing")}, { "op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-existing")}, "app_ids": ["cfg-1"], }, ], } ) def test_update_payload_rejects_remove_with_other_ops_for_same_tool(): with pytest.raises( ValidationError, match="remove_tool cannot be combined with bind/attach_tool/unbind for the same tool_id", ): payloads_module.WatsonxDeploymentUpdatePayload.model_validate( { "llm": TEST_WXO_LLM, "operations": [ {"op": "remove_tool", "tool": _tool_ref("tool-existing")}, { "op": "unbind", "tool": _tool_ref("tool-existing"), "app_ids": ["cfg-1"], }, ], } ) def test_update_payload_rejects_bind_unbind_overlapping_app_ids_for_same_tool(): with pytest.raises( ValidationError, match="bind and unbind app_ids overlap for the same tool_id", ): payloads_module.WatsonxDeploymentUpdatePayload.model_validate( { "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-existing")}, "app_ids": ["cfg-1"], }, { "op": "unbind", "tool": _tool_ref("tool-existing"), "app_ids": ["cfg-1"], }, ], } ) def test_update_payload_rejects_duplicate_attach_tool_for_same_tool(): with pytest.raises( ValidationError, match="Duplicate attach_tool operation for tool_id", ): payloads_module.WatsonxDeploymentUpdatePayload.model_validate( { "llm": TEST_WXO_LLM, "operations": [ {"op": "attach_tool", "tool": _tool_ref("tool-existing")}, {"op": "attach_tool", "tool": _tool_ref("tool-existing")}, ], } ) def test_update_payload_rejects_duplicate_remove_tool_for_same_tool(): with pytest.raises( ValidationError, match="Duplicate remove_tool operation for tool_id", ): payloads_module.WatsonxDeploymentUpdatePayload.model_validate( { "llm": TEST_WXO_LLM, "operations": [ {"op": "remove_tool", "tool": _tool_ref("tool-a")}, {"op": "remove_tool", "tool": _tool_ref("tool-a")}, ], } ) def test_create_payload_rejects_existing_tool_bind_with_empty_app_ids(): with pytest.raises(ValidationError): payloads_module.WatsonxDeploymentCreatePayload.model_validate( { "display_name": "my deployment", "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-existing")}, "app_ids": [], } ], } ) def test_create_payload_accepts_attach_tool_operation(): payload = payloads_module.WatsonxDeploymentCreatePayload.model_validate( { "display_name": "my deployment", "llm": TEST_WXO_LLM, "operations": [ { "op": "attach_tool", "tool": _tool_ref("tool-existing"), } ], } ) attach_op = payload.operations[0] assert isinstance(attach_op, payloads_module.WatsonxAttachToolOperation) assert attach_op.tool.tool_id == "tool-existing" def test_create_payload_rejects_attach_and_bind_for_same_existing_tool(): with pytest.raises( ValidationError, match="attach_tool cannot be combined with bind\\.tool\\.tool_id_with_ref", ): payloads_module.WatsonxDeploymentCreatePayload.model_validate( { "display_name": "my deployment", "llm": TEST_WXO_LLM, "operations": [ {"op": "attach_tool", "tool": _tool_ref("tool-existing")}, { "op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-existing")}, "app_ids": ["cfg-1"], }, ], } ) @pytest.mark.anyio async def test_update_provider_data_put_tools_with_llm_updates_agent(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "tools": ["tool-1", "tool-2"]}) fake_clients = SimpleNamespace( agent=fake_agent, tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.update( user_id="user-1", deployment_id="dep-1", payload=DeploymentUpdate( provider_data={ "put_tools": ["tool-a", "tool-b"], "llm": TEST_WXO_LLM, } ), db=object(), ) assert result.id == "dep-1" assert len(fake_agent.update_calls) == 1 _, agent_payload = fake_agent.update_calls[0] assert agent_payload["tools"] == ["tool-a", "tool-b"] assert agent_payload["llm"] == TEST_WXO_LLM @pytest.mark.anyio async def test_update_provider_data_creates_raw_tools_without_operations(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "display_name": "Existing Agent", "tools": ["tool-1"]}) fake_tool = FakeToolClient([{"id": "tool-1", "name": "tool-1", "binding": {"langflow": {"connections": {}}}}]) fake_connections = FakeConnectionsClient() fake_clients = SimpleNamespace( agent=fake_agent, tool=fake_tool, connections=fake_connections, ) captured: dict[str, object] = {} async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients async def mock_create_and_upload(*, clients, tool_bindings): _ = clients captured["tool_bindings"] = tool_bindings return ["new-tool-raw-1"] monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) monkeypatch.setattr( update_core_module, "create_and_upload_wxo_flow_tools_with_bindings", mock_create_and_upload, ) result = await service.update( user_id="user-1", deployment_id="dep-1", payload=DeploymentUpdate( provider_data={ "llm": TEST_WXO_LLM, "tools": { "raw_payloads": [ { "id": str(UUID("00000000-0000-0000-0000-000000000071")), "name": "snapshot-new-raw-only", "description": "desc", "data": {"nodes": [], "edges": []}, "tags": [], "provider_data": { "project_id": "project-1", "source_ref": "fv-raw-only-1", "tool_name": "snapshot_new_raw_only", "tool_display_name": "Snapshot New Raw Only", }, } ] }, } ), db=object(), ) assert captured["tool_bindings"][0].connections == {} assert result.provider_result is not None assert result.provider_result.display_name == "Existing Agent" assert result.provider_result.created_snapshot_ids == ["new-tool-raw-1"] assert result.provider_result.added_snapshot_ids == ["new-tool-raw-1"] assert fake_connections.create_calls == [] _, agent_payload = fake_agent.update_calls[0] assert agent_payload["tools"] == ["tool-1", "new-tool-raw-1"] assert agent_payload["llm"] == TEST_WXO_LLM @pytest.mark.anyio async def test_update_provider_data_creates_raw_connection_and_raw_tool(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}) fake_tool = FakeToolClient([{"id": "tool-1", "name": "tool-1", "binding": {"langflow": {"connections": {}}}}]) fake_connections = FakeConnectionsClient() fake_clients = SimpleNamespace( agent=fake_agent, tool=fake_tool, connections=fake_connections, ) captured: dict[str, object] = {} async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients async def mock_create_config(*, clients, config, user_id, db, created_app_ids_journal=None): # noqa: ARG001 captured["created_app_id"] = config.name fake_connections._connections_by_app_id[config.name] = f"conn-{config.name}" return config.name async def mock_validate_connection(connections_client, *, app_id): # noqa: ARG001 return SimpleNamespace(connection_id=f"conn-{app_id}") async def mock_create_and_upload(*, clients, tool_bindings): _ = clients first_binding = tool_bindings[0] captured["connections"] = first_binding.connections return ["new-tool-1"] monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) monkeypatch.setattr(shared_core_module, "create_config", mock_create_config) monkeypatch.setattr(update_core_module, "validate_connection", mock_validate_connection) monkeypatch.setattr( update_core_module, "create_and_upload_wxo_flow_tools_with_bindings", mock_create_and_upload, ) result = await service.update( user_id="user-1", deployment_id="dep-1", payload=DeploymentUpdate( provider_data={ "tools": { "raw_payloads": [ { "id": str(UUID("00000000-0000-0000-0000-000000000011")), "name": "snapshot-new-1", "description": "desc", "data": {"nodes": [], "edges": []}, "tags": [], "provider_data": { "project_id": "project-1", "source_ref": "fv-update-1", "tool_name": "snapshot_new_1", "tool_display_name": "Snapshot New 1", }, } ] }, "connections": { "raw_payloads": [ { "app_id": "cfg", "environment_variables": {"API_KEY": {"source": "raw", "value": "secret"}}, } ] }, "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"name_of_raw": "snapshot_new_1"}, "app_ids": ["cfg"], } ], } ), db=object(), ) assert captured["created_app_id"] == "cfg" assert captured["connections"] == {"cfg": "conn-cfg"} assert result.provider_result is not None assert result.provider_result.created_app_ids == ["cfg"] assert result.provider_result.created_snapshot_ids == ["new-tool-1"] assert result.provider_result.added_snapshot_ids == ["new-tool-1"] _, agent_payload = fake_agent.update_calls[0] assert agent_payload["tools"] == ["tool-1", "new-tool-1"] assert agent_payload["llm"] == TEST_WXO_LLM @pytest.mark.anyio async def test_update_provider_data_binds_existing_tool_using_provider_app_id_for_raw_connection(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}) fake_tool = FakeToolClient([{"id": "tool-1", "name": "tool-1", "binding": {"langflow": {"connections": {}}}}]) fake_connections = FakeConnectionsClient() fake_clients = SimpleNamespace( agent=fake_agent, tool=fake_tool, connections=fake_connections, ) captured: dict[str, str] = {} async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients async def mock_create_config(*, clients, config, user_id, db, created_app_ids_journal=None): # noqa: ARG001 captured["created_app_id"] = config.name fake_connections._connections_by_app_id[config.name] = f"conn-{config.name}" return config.name async def mock_validate_connection(connections_client, *, app_id): # noqa: ARG001 return SimpleNamespace(connection_id=f"conn-{app_id}") monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) monkeypatch.setattr(shared_core_module, "create_config", mock_create_config) monkeypatch.setattr(update_core_module, "validate_connection", mock_validate_connection) await service.update( user_id="user-1", deployment_id="dep-1", payload=DeploymentUpdate( provider_data={ "tools": {}, "connections": { "raw_payloads": [ { "app_id": "cfg", "environment_variables": {"API_KEY": {"source": "raw", "value": "secret"}}, } ] }, "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-1")}, "app_ids": ["cfg"], } ], } ), db=object(), ) assert [tool_id for tool_id, _payload in fake_tool.update_calls] == ["tool-1"] _, updated_tool_payload = fake_tool.update_calls[0] assert updated_tool_payload["binding"]["langflow"]["connections"] == {"cfg": "conn-cfg"} assert captured["created_app_id"] == "cfg" @pytest.mark.anyio async def test_update_provider_data_mixed_operations_preserve_encounter_order(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "tools": ["tool-1", "tool-2"]}) fake_tool = FakeToolClient( [ { "id": "tool-1", "name": "tool-1", "binding": {"langflow": {"connections": {"cfg-1": "conn-old-1", "cfg-2": "conn-old-2"}}}, }, {"id": "tool-2", "name": "tool-2", "binding": {"langflow": {"connections": {}}}}, {"id": "tool-3", "name": "tool-3", "binding": {"langflow": {"connections": {}}}}, ] ) fake_clients = SimpleNamespace( agent=fake_agent, tool=fake_tool, connections=FakeConnectionsClient(existing_app_id="cfg-1"), ) validate_calls: list[str] = [] async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients async def mock_validate_connection(connections_client, *, app_id): # noqa: ARG001 validate_calls.append(app_id) return SimpleNamespace(connection_id=f"conn-{app_id}") monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) monkeypatch.setattr(update_core_module, "validate_connection", mock_validate_connection) result = await service.update( user_id="user-1", deployment_id="dep-1", payload=DeploymentUpdate( provider_data={ "tools": {}, "connections": {}, "llm": TEST_WXO_LLM, "operations": [ {"op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-3")}, "app_ids": ["cfg-2", "cfg-1"]}, {"op": "unbind", "tool": _tool_ref("tool-1"), "app_ids": ["cfg-1", "cfg-2"]}, {"op": "remove_tool", "tool": _tool_ref("tool-2")}, ], } ), db=object(), ) assert validate_calls == ["cfg-2", "cfg-1"] assert result.provider_result is not None assert result.provider_result.created_app_ids == [] assert result.provider_result.created_snapshot_ids == [] assert result.provider_result.added_snapshot_ids == ["tool-3"] # Existing tool updates are dispatched concurrently via asyncio.gather, so # completion order is non-deterministic. Assert the set of updated tool ids # and look up each payload by tool_id. update_calls_by_id = dict(fake_tool.update_calls) assert set(update_calls_by_id) == {"tool-3", "tool-1"} tool3_payload = update_calls_by_id["tool-3"] assert list(tool3_payload["binding"]["langflow"]["connections"]) == ["cfg-2", "cfg-1"] assert tool3_payload["binding"]["langflow"]["connections"] == {"cfg-2": "conn-cfg-2", "cfg-1": "conn-cfg-1"} tool1_payload = update_calls_by_id["tool-1"] assert tool1_payload["binding"]["langflow"]["connections"] == {} _, agent_payload = fake_agent.update_calls[0] assert agent_payload["tools"] == ["tool-1", "tool-3"] assert agent_payload["llm"] == TEST_WXO_LLM def test_ordered_unique_strs_preserves_encounter_order_and_safe_discard(): ordered = update_core_module.OrderedUniqueStrs() ordered.extend(["b", "a", "b", "c"]) ordered.add("a") ordered.add("d") ordered.discard("c") ordered.discard("missing") assert ordered.to_list() == ["b", "a", "d"] def test_build_provider_update_plan_preserves_operation_encounter_order(): provider_update = payloads_module.WatsonxDeploymentUpdatePayload.model_validate( { "tools": { "raw_payloads": [ { "id": str(UUID("00000000-0000-0000-0000-000000000041")), "name": "snapshot-raw-1", "description": "desc", "data": {"nodes": [], "edges": []}, "tags": [], "provider_data": { "project_id": "project-1", "source_ref": "fv-plan-1", "tool_name": "snapshot_raw_1", "tool_display_name": "Snapshot Raw 1", }, } ], }, "connections": { "raw_payloads": [ {"app_id": "cfg-raw-1", "environment_variables": {"API_KEY": {"source": "raw", "value": "x"}}}, {"app_id": "cfg-raw-2", "environment_variables": {"API_KEY": {"source": "raw", "value": "y"}}}, ], }, "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-c")}, "app_ids": ["cfg-2", "cfg-1", "cfg-2"], }, {"op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-a")}, "app_ids": ["cfg-1"]}, {"op": "unbind", "tool": _tool_ref("tool-c"), "app_ids": ["cfg-3", "cfg-3"]}, {"op": "remove_tool", "tool": _tool_ref("tool-b")}, {"op": "bind", "tool": {"name_of_raw": "snapshot_raw_1"}, "app_ids": ["cfg-raw-2", "cfg-raw-1"]}, ], } ) plan = update_core_module.build_provider_update_plan( agent={"id": "dep-1", "tools": ["tool-a", "tool-b"]}, provider_update=provider_update, ) assert [ref.tool_id for ref in plan.added_existing_tool_refs] == ["tool-c"] assert plan.final_existing_tool_ids == ["tool-a", "tool-c"] assert plan.existing_app_ids == ["cfg-2", "cfg-1", "cfg-3"] assert [item.operation_app_id for item in plan.raw_connections_to_create] == ["cfg-raw-1", "cfg-raw-2"] assert [item.provider_app_id for item in plan.raw_connections_to_create] == [ _normalized_provider_app_id("cfg-raw-1"), _normalized_provider_app_id("cfg-raw-2"), ] assert len(plan.raw_tools_to_create) == 1 assert plan.raw_tools_to_create[0].app_ids == ["cfg-raw-2", "cfg-raw-1"] delta = plan.existing_tool_deltas["tool-c"] assert delta.bind.to_list() == ["cfg-2", "cfg-1"] assert delta.unbind.to_list() == ["cfg-3"] assert [ref.tool_id for ref in plan.removed_existing_tool_refs] == ["tool-b"] def test_build_provider_update_plan_creates_unbound_raw_tools_alongside_bound_raw_tools(): provider_update = payloads_module.WatsonxDeploymentUpdatePayload.model_validate( { "tools": { "raw_payloads": [ { "id": str(UUID("00000000-0000-0000-0000-000000000051")), "name": "snapshot-bound", "description": "desc", "data": {"nodes": [], "edges": []}, "tags": [], "provider_data": { "project_id": "project-1", "source_ref": "fv-bound", "tool_name": "snapshot_bound", "tool_display_name": "Snapshot Bound", }, }, { "id": str(UUID("00000000-0000-0000-0000-000000000052")), "name": "snapshot-unbound", "description": "desc", "data": {"nodes": [], "edges": []}, "tags": [], "provider_data": { "project_id": "project-1", "source_ref": "fv-unbound", "tool_name": "snapshot_unbound", "tool_display_name": "Snapshot Unbound", }, }, ], }, "connections": { "raw_payloads": [ {"app_id": "cfg", "environment_variables": {"API_KEY": {"source": "raw", "value": "x"}}}, ], }, "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"name_of_raw": "snapshot_bound"}, "app_ids": ["cfg"], } ], } ) plan = update_core_module.build_provider_update_plan( agent={"id": "dep-1", "tools": []}, provider_update=provider_update, ) assert [item.raw_name for item in plan.raw_tools_to_create] == ["snapshot_bound", "snapshot_unbound"] assert plan.raw_tools_to_create[0].app_ids == ["cfg"] assert plan.raw_tools_to_create[1].app_ids == [] def test_build_provider_update_plan_put_tools_replaces_agent_tool_list(): """put_tools standalone path seeds final_existing_tool_ids from the payload, not the agent.""" provider_update = payloads_module.WatsonxDeploymentUpdatePayload.model_validate( {"put_tools": ["tool-x", "tool-y", "tool-z"]} ) plan = update_core_module.build_provider_update_plan( agent={"id": "dep-1", "tools": ["tool-a", "tool-b"]}, provider_update=provider_update, ) assert plan.final_existing_tool_ids == ["tool-x", "tool-y", "tool-z"] assert plan.added_existing_tool_refs == [] assert plan.raw_tools_to_create == [] assert plan.existing_tool_deltas == {} def test_build_provider_update_plan_put_tools_deduplicates(): """Duplicate IDs in put_tools are collapsed (validator + OrderedUniqueStrs).""" provider_update = payloads_module.WatsonxDeploymentUpdatePayload.model_validate( {"put_tools": ["tool-a", "tool-b", "tool-a"]} ) plan = update_core_module.build_provider_update_plan( agent={"id": "dep-1", "tools": []}, provider_update=provider_update, ) assert plan.final_existing_tool_ids == ["tool-a", "tool-b"] def test_build_provider_update_plan_put_tools_empty_clears_all(): """An empty put_tools list clears the agent's tool list entirely.""" provider_update = payloads_module.WatsonxDeploymentUpdatePayload.model_validate({"put_tools": []}) plan = update_core_module.build_provider_update_plan( agent={"id": "dep-1", "tools": ["tool-a", "tool-b"]}, provider_update=provider_update, ) assert plan.final_existing_tool_ids == [] def test_build_provider_update_plan_attaches_existing_tool_without_connection_deltas(): provider_update = payloads_module.WatsonxDeploymentUpdatePayload.model_validate( { "llm": TEST_WXO_LLM, "operations": [ { "op": "attach_tool", "tool": _tool_ref("tool-existing"), } ], } ) plan = update_core_module.build_provider_update_plan( agent={"id": "dep-1", "tools": ["tool-a"]}, provider_update=provider_update, ) assert plan.final_existing_tool_ids == ["tool-a", "tool-existing"] assert [ref.tool_id for ref in plan.added_existing_tool_refs] == ["tool-existing"] assert plan.existing_tool_deltas == {} def test_build_provider_create_plan_creates_unbound_raw_tools_without_bind_operations(): provider_create = payloads_module.WatsonxDeploymentCreatePayload.model_validate( { "display_name": "my deployment", "tools": { "raw_payloads": [ { "id": str(UUID("00000000-0000-0000-0000-000000000071")), "name": "snapshot-unbound", "description": "desc", "data": {"nodes": [], "edges": []}, "tags": [], "provider_data": { "project_id": "project-1", "source_ref": "fv-create-unbound", "tool_name": "snapshot_unbound", "tool_display_name": "Snapshot Unbound", }, } ] }, "llm": TEST_WXO_LLM, "operations": [], } ) plan = create_core_module.build_provider_create_plan( deployment_name=None, provider_create=provider_create, ) assert [item.raw_name for item in plan.raw_tools_to_create] == ["snapshot_unbound"] assert plan.raw_tools_to_create[0].app_ids == [] assert plan.selected_operation_app_ids == [] assert plan.existing_tool_ids == [] _assert_langflow_agent_name(plan.deployment_name, display_name="my deployment") @pytest.mark.parametrize("display_name", ["123 starts with digits", "!!!"]) def test_build_provider_create_plan_keeps_technical_name_separate_from_display_name(display_name): provider_create = payloads_module.WatsonxDeploymentCreatePayload.model_validate( { "display_name": display_name, "tools": { "raw_payloads": [ { "id": str(UUID("00000000-0000-0000-0000-000000000072")), "name": "snapshot-edge", "description": "desc", "data": {"nodes": [], "edges": []}, "tags": [], "provider_data": { "project_id": "project-1", "source_ref": "fv-create-edge", "tool_name": "snapshot_edge", "tool_display_name": "Snapshot Edge", }, } ] }, "llm": TEST_WXO_LLM, "operations": [], } ) technical_name = _agent_technical_name("stable agent") plan = create_core_module.build_provider_create_plan( deployment_name=technical_name, provider_create=provider_create, ) assert plan.deployment_name == technical_name assert plan.display_name == display_name def test_build_provider_create_plan_rejects_invalid_technical_name(): provider_create = payloads_module.WatsonxDeploymentCreatePayload.model_validate( { "display_name": "my deployment", "tools": { "raw_payloads": [ { "id": str(UUID("00000000-0000-0000-0000-000000000072")), "name": "snapshot-edge", "description": "desc", "data": {"nodes": [], "edges": []}, "tags": [], "provider_data": { "project_id": "project-1", "source_ref": "fv-create-edge", "tool_name": "snapshot_edge", "tool_display_name": "Snapshot Edge", }, } ] }, "llm": TEST_WXO_LLM, "operations": [], } ) with pytest.raises( InvalidContentError, match="Agent name must start with a letter", ): create_core_module.build_provider_create_plan( deployment_name="123-invalid-technical-name", provider_create=provider_create, ) def test_build_provider_create_plan_rejects_unnormalized_explicit_technical_name(): provider_create = payloads_module.WatsonxDeploymentCreatePayload.model_validate( { "display_name": "my deployment", "tools": { "raw_payloads": [ { "id": str(UUID("00000000-0000-0000-0000-000000000073")), "name": "snapshot-edge", "description": "desc", "data": {"nodes": [], "edges": []}, "tags": [], "provider_data": { "project_id": "project-1", "source_ref": "fv-create-edge", "tool_name": "snapshot_edge", "tool_display_name": "Snapshot Edge", }, } ] }, "llm": TEST_WXO_LLM, "operations": [], } ) with pytest.raises( InvalidContentError, match="Agent name must only contain letters, numbers, and underscores", ): create_core_module.build_provider_create_plan( deployment_name="agent technical name", provider_create=provider_create, ) @pytest.mark.parametrize("display_name", ["123 starts with digits", "a" * 500]) def test_build_provider_create_plan_generates_technical_name_from_display_name(display_name): provider_create = payloads_module.WatsonxDeploymentCreatePayload.model_validate( { "display_name": display_name, "tools": { "raw_payloads": [ { "id": str(UUID("00000000-0000-0000-0000-000000000073")), "name": "snapshot-edge", "description": "desc", "data": {"nodes": [], "edges": []}, "tags": [], "provider_data": { "project_id": "project-1", "source_ref": "fv-create-edge", "tool_name": "snapshot_edge", "tool_display_name": "Snapshot Edge", }, } ] }, "llm": TEST_WXO_LLM, "operations": [], } ) plan = create_core_module.build_provider_create_plan( deployment_name=None, provider_create=provider_create, ) _assert_langflow_agent_name(plan.deployment_name, display_name=display_name) assert plan.display_name == display_name @pytest.mark.parametrize("display_name", ["!!!", "🔥🔥"]) def test_build_provider_create_plan_uses_resource_fallback_for_symbol_only_display_name(display_name): provider_create = payloads_module.WatsonxDeploymentCreatePayload.model_validate( { "display_name": display_name, "tools": { "raw_payloads": [ { "id": str(UUID("00000000-0000-0000-0000-000000000074")), "name": "snapshot-edge", "description": "desc", "data": {"nodes": [], "edges": []}, "tags": [], "provider_data": { "project_id": "project-1", "source_ref": "fv-create-edge", "tool_name": "snapshot_edge", "tool_display_name": "Snapshot Edge", }, } ] }, "llm": TEST_WXO_LLM, "operations": [], } ) plan = create_core_module.build_provider_create_plan( deployment_name=None, provider_create=provider_create, ) assert plan.deployment_name.startswith("langflow_agent_") assert plan.display_name == display_name def test_build_provider_create_plan_attaches_existing_tool_without_connection_updates(): provider_create = payloads_module.WatsonxDeploymentCreatePayload.model_validate( { "display_name": "my deployment", "llm": TEST_WXO_LLM, "operations": [ { "op": "attach_tool", "tool": _tool_ref("tool-existing"), } ], } ) plan = create_core_module.build_provider_create_plan( deployment_name=_agent_technical_name(), provider_create=provider_create, ) assert plan.existing_tool_ids == ["tool-existing"] assert plan.existing_tool_bindings == {} @pytest.mark.anyio async def test_update_existing_tool_connection_deltas_uses_bind_order_in_errors(): fake_tool = FakeToolClient([{"id": "tool-c", "name": "tool-c", "binding": {"langflow": {"connections": {}}}}]) clients = SimpleNamespace(tool=fake_tool) delta = update_core_module.ToolConnectionOps() delta.bind.extend(["cfg-missing-first", "cfg-present"]) with pytest.raises(InvalidContentError, match="cfg-missing-first"): await update_core_module._update_existing_tools( clients=clients, existing_tool_deltas={"tool-c": delta}, tool_renames={}, resolved_connections={"cfg-present": "conn-present"}, operation_to_provider_app_id={}, original_tools={}, tool_by_id={"tool-c": fake_tool._tools_by_id["tool-c"]}, ) @pytest.mark.anyio async def test_apply_provider_create_plan_binds_raw_tools_with_provider_app_ids(monkeypatch): provider_create = payloads_module.WatsonxDeploymentCreatePayload.model_validate( { "display_name": "my deployment", "tools": { "raw_payloads": [ { "id": str(UUID("00000000-0000-0000-0000-000000000081")), "name": "snapshot-raw-1", "description": "desc", "data": {"nodes": [], "edges": []}, "tags": [], "provider_data": { "project_id": "project-1", "source_ref": "fv-create-1", "tool_name": "snapshot_raw_1", "tool_display_name": "Snapshot Raw 1", }, } ] }, "connections": { "raw_payloads": [ {"app_id": "cfg", "environment_variables": {"API_KEY": {"source": "raw", "value": "x"}}} ] }, "llm": TEST_WXO_LLM, "operations": [ {"op": "bind", "tool": {"name_of_raw": "snapshot_raw_1"}, "app_ids": ["cfg"]}, ], } ) plan = create_core_module.build_provider_create_plan( deployment_name=_agent_technical_name(), provider_create=provider_create, ) fake_clients = FakeWXOClients( agent=FakeAgentClient({"id": "dep-1", "tools": []}), tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) captured: dict[str, Any] = {} async def mock_create_and_upload(*, clients, tool_bindings): _ = clients first_binding = tool_bindings[0] captured["connections"] = first_binding.connections return ["created-tool-1"] monkeypatch.setattr( create_core_module, "create_and_upload_wxo_flow_tools_with_bindings", mock_create_and_upload, ) result = await create_core_module.apply_provider_create_plan_with_rollback( clients=fake_clients, user_id="user-1", db=object(), deployment_spec=BaseDeploymentData( name=_agent_technical_name(), description=" ", type=DeploymentType.AGENT, ), plan=plan, ) assert fake_clients.connections.create_calls == [{"app_id": "cfg"}] assert captured["connections"] == {"cfg": "conn-cfg"} assert fake_clients.agent.create_calls assert fake_clients.agent.create_calls[0]["name"] == _agent_technical_name() assert fake_clients.agent.create_calls[0]["display_name"] == "my deployment" assert fake_clients.agent.create_calls[0]["description"] == "Langflow deployment my deployment" assert fake_clients.agent.create_calls[0]["tools"] == ["created-tool-1"] assert fake_clients.agent.create_calls[0]["llm"] == TEST_WXO_LLM assert result.agent_id == "dep-created" assert result.description == "Langflow deployment my deployment" assert result.app_ids == ["cfg"] assert [(binding.tool_id, binding.app_ids) for binding in result.tool_app_bindings] == [("created-tool-1", ["cfg"])] assert [(binding.source_ref, binding.tool_id) for binding in result.tools_with_refs] == [ ("fv-create-1", "created-tool-1") ] @pytest.mark.anyio async def test_apply_provider_create_plan_rolls_back_mutated_existing_tools_with_writable_payload(monkeypatch): provider_create = payloads_module.WatsonxDeploymentCreatePayload.model_validate( { "display_name": "my deployment", "tools": {}, "connections": {}, "llm": TEST_WXO_LLM, "operations": [{"op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-1")}, "app_ids": ["cfg-1"]}], } ) plan = create_core_module.build_provider_create_plan( deployment_name=_agent_technical_name(), provider_create=provider_create, ) fake_tool = FakeToolClient( [ { "id": "tool-1", "name": "tool-1", "display_name": "Tool 1", "description": "desc", "binding": {"langflow": {"connections": {"old": "conn-old"}}}, "created_at": "read-only-field", } ] ) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}), tool=fake_tool, connections=FakeConnectionsClient(existing_app_id="cfg-1"), ) async def mock_validate_connection(connections_client, *, app_id): # noqa: ARG001 return SimpleNamespace(connection_id="conn-new") async def mock_create_agent_deployment( *, clients, # noqa: ARG001 payload, # noqa: ARG001 ): msg = "create failed" raise RuntimeError(msg) monkeypatch.setattr(create_core_module, "validate_connection", mock_validate_connection) monkeypatch.setattr(create_core_module, "create_agent_deployment", mock_create_agent_deployment) with pytest.raises(RuntimeError, match="create failed"): await create_core_module.apply_provider_create_plan_with_rollback( clients=fake_clients, user_id="user-1", db=object(), deployment_spec=BaseDeploymentData( name=_agent_technical_name(), description="desc", type=DeploymentType.AGENT, ), plan=plan, ) assert len(fake_tool.update_calls) == 2 first_payload = fake_tool.update_calls[0][1] rollback_payload = fake_tool.update_calls[1][1] assert "id" not in first_payload assert "created_at" not in first_payload assert first_payload["binding"]["langflow"]["connections"]["cfg-1"] == "conn-new" assert rollback_payload["binding"]["langflow"]["connections"] == {"old": "conn-old"} @pytest.mark.anyio async def test_resolve_connections_for_operations_normalizes_provider_app_id_and_uses_created_id_for_validation( monkeypatch, ): raw_payload = payloads_module.WatsonxConnectionRawPayload.model_validate( {"app_id": "cfg-s-003", "environment_variables": {"API_KEY": {"source": "raw", "value": "x"}}} ) raw_connections_to_create = [ shared_core_module.RawConnectionCreatePlan( operation_app_id="cfg-s-003", provider_app_id="cfg-s-003", payload=raw_payload, ) ] fake_clients = SimpleNamespace(connections=FakeConnectionsClient()) validated_app_ids: list[str] = [] async def mock_create_connection_with_conflict_mapping( *, clients, app_id, payload, user_id, db, error_prefix, created_app_ids_journal=None, ): _ = clients, app_id, payload, user_id, db, error_prefix created_id = "cfg_s_003" if created_app_ids_journal is not None: created_app_ids_journal.append(created_id) return created_id async def mock_validate_connection(connections_client, *, app_id): # noqa: ARG001 validated_app_ids.append(app_id) return SimpleNamespace(connection_id=f"conn-{app_id}") monkeypatch.setattr( shared_core_module, "create_connection_with_conflict_mapping", mock_create_connection_with_conflict_mapping, ) result = await shared_core_module.resolve_connections_for_operations( clients=fake_clients, user_id="user-1", db=object(), existing_app_ids=[], raw_connections_to_create=raw_connections_to_create, error_prefix="CREATE", validate_connection_fn=mock_validate_connection, ) assert raw_connections_to_create[0].provider_app_id == "cfg_s_003" assert validated_app_ids == ["cfg_s_003"] assert result.operation_to_provider_app_id == {"cfg-s-003": "cfg_s_003"} assert result.resolved_connections == {"cfg_s_003": "conn-cfg_s_003"} assert result.created_app_ids == ["cfg_s_003"] @pytest.mark.anyio async def test_resolve_connections_for_operations_wraps_validation_failure_with_rollback_metadata(monkeypatch): raw_payload = payloads_module.WatsonxConnectionRawPayload.model_validate( {"app_id": "cfg-s-003", "environment_variables": {"API_KEY": {"source": "raw", "value": "x"}}} ) raw_connections_to_create = [ shared_core_module.RawConnectionCreatePlan( operation_app_id="cfg-s-003", provider_app_id="cfg-s-003", payload=raw_payload, ) ] fake_clients = SimpleNamespace(connections=FakeConnectionsClient()) created_app_ids_journal: list[str] = [] async def mock_create_connection_with_conflict_mapping( *, clients, app_id, payload, user_id, db, error_prefix, created_app_ids_journal=None, ): _ = clients, app_id, payload, user_id, db, error_prefix created_id = "cfg_s_003" if created_app_ids_journal is not None: created_app_ids_journal.append(created_id) return created_id async def mock_validate_connection(connections_client, *, app_id): # noqa: ARG001 msg = f"Connection '{app_id}' not found. Ensure the connection exists with a draft configuration." raise InvalidContentError(message=msg) monkeypatch.setattr( shared_core_module, "create_connection_with_conflict_mapping", mock_create_connection_with_conflict_mapping, ) with pytest.raises(shared_core_module.ConnectionCreateBatchError) as exc_info: await shared_core_module.resolve_connections_for_operations( clients=fake_clients, user_id="user-1", db=object(), existing_app_ids=[], raw_connections_to_create=raw_connections_to_create, error_prefix="CREATE", validate_connection_fn=mock_validate_connection, created_app_ids_journal=created_app_ids_journal, ) assert created_app_ids_journal == ["cfg_s_003"] assert exc_info.value.created_app_ids == ["cfg_s_003"] assert len(exc_info.value.errors) == 1 assert isinstance(exc_info.value.errors[0], InvalidContentError) @pytest.mark.anyio async def test_apply_provider_create_plan_rolls_back_successfully_created_raw_connections_on_partial_batch_failure( monkeypatch, ): provider_create = payloads_module.WatsonxDeploymentCreatePayload.model_validate( { "display_name": "my deployment", "tools": {}, "connections": { "raw_payloads": [ {"app_id": "cfg-a", "environment_variables": {"API_KEY": {"source": "raw", "value": "x"}}}, {"app_id": "cfg-b", "environment_variables": {"API_KEY": {"source": "raw", "value": "y"}}}, ] }, "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-existing-1")}, "app_ids": ["cfg-a", "cfg-b"], }, ], } ) plan = create_core_module.build_provider_create_plan( deployment_name=_agent_technical_name(), provider_create=provider_create, ) fake_clients = SimpleNamespace(connections=FakeConnectionsClient()) captured: dict[str, Any] = {} async def mock_create_connection_with_conflict_mapping( *, clients, app_id, payload, user_id, db, error_prefix, created_app_ids_journal=None, ): _ = clients, payload, user_id, db, error_prefix if app_id == _normalized_provider_app_id("cfg-a"): if created_app_ids_journal is not None: created_app_ids_journal.append(app_id) return app_id msg = "boom-create-connection" raise RuntimeError(msg) async def mock_rollback_created_resources(*, clients, agent_id, tool_ids, app_ids=None): # noqa: ARG001 captured["rollback_app_ids"] = list(app_ids or []) monkeypatch.setattr( shared_core_module, "create_connection_with_conflict_mapping", mock_create_connection_with_conflict_mapping, ) monkeypatch.setattr(create_core_module, "rollback_created_resources", mock_rollback_created_resources) with pytest.raises(RuntimeError, match="boom-create-connection"): await create_core_module.apply_provider_create_plan_with_rollback( clients=fake_clients, user_id="user-1", db=object(), deployment_spec=BaseDeploymentData( name=_agent_technical_name(), description="desc", type=DeploymentType.AGENT, ), plan=plan, ) assert captured.get("rollback_app_ids") == [_normalized_provider_app_id("cfg-a")] @pytest.mark.anyio async def test_apply_provider_create_plan_rolls_back_all_journaled_raw_connections_when_multiple_succeed_then_fail( monkeypatch, ): provider_create = payloads_module.WatsonxDeploymentCreatePayload.model_validate( { "display_name": "my deployment", "tools": {}, "connections": { "raw_payloads": [ {"app_id": "cfg-a", "environment_variables": {"API_KEY": {"source": "raw", "value": "x"}}}, {"app_id": "cfg-b", "environment_variables": {"API_KEY": {"source": "raw", "value": "y"}}}, {"app_id": "cfg-c", "environment_variables": {"API_KEY": {"source": "raw", "value": "z"}}}, ] }, "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-existing-1")}, "app_ids": ["cfg-a", "cfg-b", "cfg-c"], }, ], } ) plan = create_core_module.build_provider_create_plan( deployment_name=_agent_technical_name(), provider_create=provider_create, ) fake_clients = SimpleNamespace(connections=FakeConnectionsClient()) captured: dict[str, Any] = {} async def mock_create_connection_with_conflict_mapping( *, clients, app_id, payload, user_id, db, error_prefix, created_app_ids_journal=None, ): _ = clients, payload, user_id, db, error_prefix if app_id == _normalized_provider_app_id("cfg-c"): msg = "boom-create-connection" raise RuntimeError(msg) if created_app_ids_journal is not None: created_app_ids_journal.append(app_id) return app_id async def mock_rollback_created_resources(*, clients, agent_id, tool_ids, app_ids=None): # noqa: ARG001 captured["rollback_app_ids"] = list(app_ids or []) monkeypatch.setattr( shared_core_module, "create_connection_with_conflict_mapping", mock_create_connection_with_conflict_mapping, ) monkeypatch.setattr(create_core_module, "rollback_created_resources", mock_rollback_created_resources) with pytest.raises(RuntimeError, match="boom-create-connection"): await create_core_module.apply_provider_create_plan_with_rollback( clients=fake_clients, user_id="user-1", db=object(), deployment_spec=BaseDeploymentData( name=_agent_technical_name(), description="desc", type=DeploymentType.AGENT, ), plan=plan, ) assert captured["rollback_app_ids"] == [ _normalized_provider_app_id("cfg-a"), _normalized_provider_app_id("cfg-b"), ] @pytest.mark.anyio async def test_apply_provider_create_plan_rolls_back_journaled_app_ids_when_create_fails_after_provider_create( monkeypatch, ): provider_create = payloads_module.WatsonxDeploymentCreatePayload.model_validate( { "display_name": "my deployment", "tools": {}, "connections": { "raw_payloads": [ {"app_id": "cfg-a", "environment_variables": {"API_KEY": {"source": "raw", "value": "x"}}}, ] }, "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-existing-1")}, "app_ids": ["cfg-a"], }, ], } ) plan = create_core_module.build_provider_create_plan( deployment_name=_agent_technical_name(), provider_create=provider_create, ) fake_clients = SimpleNamespace(connections=FakeConnectionsClient()) captured: dict[str, Any] = {} async def mock_create_connection_with_conflict_mapping( *, clients, app_id, payload, user_id, db, error_prefix, created_app_ids_journal=None, ): _ = clients, payload, user_id, db, error_prefix if created_app_ids_journal is not None: created_app_ids_journal.append(app_id) msg = "boom-after-provider-create" raise RuntimeError(msg) async def mock_rollback_created_resources(*, clients, agent_id, tool_ids, app_ids=None): # noqa: ARG001 captured["rollback_app_ids"] = list(app_ids or []) monkeypatch.setattr( shared_core_module, "create_connection_with_conflict_mapping", mock_create_connection_with_conflict_mapping, ) monkeypatch.setattr(create_core_module, "rollback_created_resources", mock_rollback_created_resources) with pytest.raises(RuntimeError, match="boom-after-provider-create"): await create_core_module.apply_provider_create_plan_with_rollback( clients=fake_clients, user_id="user-1", db=object(), deployment_spec=BaseDeploymentData( name=_agent_technical_name(), description="desc", type=DeploymentType.AGENT, ), plan=plan, ) assert captured["rollback_app_ids"] == [_normalized_provider_app_id("cfg-a")] @pytest.mark.anyio async def test_apply_provider_update_plan_rolls_back_successfully_created_raw_connections_on_partial_batch_failure( monkeypatch, ): provider_update = payloads_module.WatsonxDeploymentUpdatePayload.model_validate( { "tools": {}, "connections": { "raw_payloads": [ {"app_id": "cfg-a", "environment_variables": {"API_KEY": {"source": "raw", "value": "x"}}}, {"app_id": "cfg-b", "environment_variables": {"API_KEY": {"source": "raw", "value": "y"}}}, ] }, "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-existing-1")}, "app_ids": ["cfg-a", "cfg-b"], }, ], } ) plan = update_core_module.build_provider_update_plan( agent={"id": "dep-1", "tools": ["tool-existing-1"]}, provider_update=provider_update, ) agent = { "id": "dep-1", "name": "agent_dep_1", "display_name": "Agent Dep 1", "description": "desc", "tools": ["tool-existing-1"], } fake_clients = SimpleNamespace( agent=FakeAgentClient(agent), tool=FakeToolClient([{"id": "tool-existing-1", "binding": {"langflow": {"connections": {}}}}]), connections=FakeConnectionsClient(), ) captured: dict[str, Any] = {} async def mock_create_connection_with_conflict_mapping( *, clients, app_id, payload, user_id, db, error_prefix, created_app_ids_journal=None, ): _ = clients, payload, user_id, db, error_prefix if app_id == _normalized_provider_app_id("cfg-a"): if created_app_ids_journal is not None: created_app_ids_journal.append(app_id) return app_id msg = "boom-update-connection" raise RuntimeError(msg) async def mock_rollback_update_resources(*, clients, created_tool_ids, created_app_id, original_tools): # noqa: ARG001 _ = (created_tool_ids, created_app_id, original_tools) async def mock_rollback_created_app_ids(*, clients, created_app_ids): # noqa: ARG001 captured["rolled_back_app_ids"] = list(created_app_ids) monkeypatch.setattr( shared_core_module, "create_connection_with_conflict_mapping", mock_create_connection_with_conflict_mapping, ) monkeypatch.setattr(update_core_module, "rollback_update_resources", mock_rollback_update_resources) monkeypatch.setattr(update_core_module, "rollback_created_app_ids", mock_rollback_created_app_ids) with pytest.raises(RuntimeError, match="boom-update-connection"): await update_core_module.apply_provider_update_plan_with_rollback( clients=fake_clients, user_id="user-1", db=object(), agent_id="dep-1", agent=agent, update_payload={}, plan=plan, ) assert captured["rolled_back_app_ids"] == [_normalized_provider_app_id("cfg-a")] @pytest.mark.anyio async def test_apply_provider_update_plan_rolls_back_all_journaled_raw_connections_when_multiple_succeed_then_fail( monkeypatch, ): provider_update = payloads_module.WatsonxDeploymentUpdatePayload.model_validate( { "tools": {}, "connections": { "raw_payloads": [ {"app_id": "cfg-a", "environment_variables": {"API_KEY": {"source": "raw", "value": "x"}}}, {"app_id": "cfg-b", "environment_variables": {"API_KEY": {"source": "raw", "value": "y"}}}, {"app_id": "cfg-c", "environment_variables": {"API_KEY": {"source": "raw", "value": "z"}}}, ] }, "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-existing-1")}, "app_ids": ["cfg-a", "cfg-b", "cfg-c"], }, ], } ) plan = update_core_module.build_provider_update_plan( agent={"id": "dep-1", "tools": ["tool-existing-1"]}, provider_update=provider_update, ) agent = { "id": "dep-1", "name": "agent_dep_1", "display_name": "Agent Dep 1", "description": "desc", "tools": ["tool-existing-1"], } fake_clients = SimpleNamespace( agent=FakeAgentClient(agent), tool=FakeToolClient([{"id": "tool-existing-1", "binding": {"langflow": {"connections": {}}}}]), connections=FakeConnectionsClient(), ) captured: dict[str, Any] = {} async def mock_create_connection_with_conflict_mapping( *, clients, app_id, payload, user_id, db, error_prefix, created_app_ids_journal=None, ): _ = clients, payload, user_id, db, error_prefix if app_id == _normalized_provider_app_id("cfg-c"): msg = "boom-update-connection" raise RuntimeError(msg) if created_app_ids_journal is not None: created_app_ids_journal.append(app_id) return app_id async def mock_rollback_update_resources(*, clients, created_tool_ids, created_app_id, original_tools): # noqa: ARG001 _ = (created_tool_ids, created_app_id, original_tools) async def mock_rollback_created_app_ids(*, clients, created_app_ids): # noqa: ARG001 captured["rolled_back_app_ids"] = list(created_app_ids) monkeypatch.setattr( shared_core_module, "create_connection_with_conflict_mapping", mock_create_connection_with_conflict_mapping, ) monkeypatch.setattr(update_core_module, "rollback_update_resources", mock_rollback_update_resources) monkeypatch.setattr(update_core_module, "rollback_created_app_ids", mock_rollback_created_app_ids) with pytest.raises(RuntimeError, match="boom-update-connection"): await update_core_module.apply_provider_update_plan_with_rollback( clients=fake_clients, user_id="user-1", db=object(), agent_id="dep-1", agent=agent, update_payload={}, plan=plan, ) assert captured["rolled_back_app_ids"] == [ _normalized_provider_app_id("cfg-a"), _normalized_provider_app_id("cfg-b"), ] @pytest.mark.anyio async def test_apply_provider_update_plan_rolls_back_journaled_app_ids_when_create_fails_after_provider_create( monkeypatch, ): provider_update = payloads_module.WatsonxDeploymentUpdatePayload.model_validate( { "tools": {}, "connections": { "raw_payloads": [ {"app_id": "cfg-a", "environment_variables": {"API_KEY": {"source": "raw", "value": "x"}}}, ] }, "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-existing-1")}, "app_ids": ["cfg-a"], }, ], } ) plan = update_core_module.build_provider_update_plan( agent={"id": "dep-1", "tools": ["tool-existing-1"]}, provider_update=provider_update, ) agent = { "id": "dep-1", "name": "agent_dep_1", "display_name": "Agent Dep 1", "description": "desc", "tools": ["tool-existing-1"], } fake_clients = SimpleNamespace( agent=FakeAgentClient(agent), tool=FakeToolClient([{"id": "tool-existing-1", "binding": {"langflow": {"connections": {}}}}]), connections=FakeConnectionsClient(), ) captured: dict[str, Any] = {} async def mock_create_connection_with_conflict_mapping( *, clients, app_id, payload, user_id, db, error_prefix, created_app_ids_journal=None, ): _ = clients, payload, user_id, db, error_prefix if created_app_ids_journal is not None: created_app_ids_journal.append(app_id) msg = "boom-after-provider-create" raise RuntimeError(msg) async def mock_rollback_update_resources(*, clients, created_tool_ids, created_app_id, original_tools): # noqa: ARG001 _ = (created_tool_ids, created_app_id, original_tools) async def mock_rollback_created_app_ids(*, clients, created_app_ids): # noqa: ARG001 captured["rolled_back_app_ids"] = list(created_app_ids) monkeypatch.setattr( shared_core_module, "create_connection_with_conflict_mapping", mock_create_connection_with_conflict_mapping, ) monkeypatch.setattr(update_core_module, "rollback_update_resources", mock_rollback_update_resources) monkeypatch.setattr(update_core_module, "rollback_created_app_ids", mock_rollback_created_app_ids) with pytest.raises(RuntimeError, match="boom-after-provider-create"): await update_core_module.apply_provider_update_plan_with_rollback( clients=fake_clients, user_id="user-1", db=object(), agent_id="dep-1", agent=agent, update_payload={}, plan=plan, ) assert captured["rolled_back_app_ids"] == [_normalized_provider_app_id("cfg-a")] @pytest.mark.anyio async def test_create_provider_data_prefixes_tool_and_deployment_names_but_not_connection_app_ids(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = FakeWXOClients( agent=FakeAgentClient({"id": "dep-1", "tools": []}), tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) captured: dict[str, Any] = {} async def mock_create_and_upload(*, clients, tool_bindings): _ = clients first_binding = tool_bindings[0] captured["connections"] = first_binding.connections return ["created-tool-1"] _attach_provider_clients(service, fake_clients) monkeypatch.setattr(payloads_module, "uuid4", lambda: SimpleNamespace(hex="abcdef1234567890")) monkeypatch.setattr( create_core_module, "create_and_upload_wxo_flow_tools_with_bindings", mock_create_and_upload, ) result = await service.create( user_id="user-1", payload=DeploymentCreate( spec=BaseDeploymentData( description="desc", type=DeploymentType.AGENT, ), provider_data={ "display_name": "my deployment", "tools": { "raw_payloads": [ { "id": str(UUID("00000000-0000-0000-0000-000000000091")), "name": "snapshot-new-1", "description": "desc", "data": {"nodes": [], "edges": []}, "tags": [], "provider_data": { "project_id": "project-1", "source_ref": "fv-create-service-1", "tool_name": "snapshot_new_1", "tool_display_name": "Snapshot New 1", }, } ] }, "connections": { "raw_payloads": [ {"app_id": "cfg", "environment_variables": {"API_KEY": {"source": "raw", "value": "x"}}} ] }, "llm": TEST_WXO_LLM, "operations": [ {"op": "bind", "tool": {"name_of_raw": "snapshot_new_1"}, "app_ids": ["cfg"]}, ], }, ), db=object(), ) assert fake_clients.connections.create_calls == [{"app_id": "cfg"}] assert captured["connections"] == {"cfg": "conn-cfg"} assert fake_clients.agent.create_calls assert fake_clients.agent.create_calls[0]["name"] == "langflow_my_deployment_abcdef12" assert fake_clients.agent.create_calls[0]["display_name"] == "my deployment" assert fake_clients.agent.create_calls[0]["description"] == "desc" assert fake_clients.agent.create_calls[0]["tools"] == ["created-tool-1"] assert fake_clients.agent.create_calls[0]["llm"] == TEST_WXO_LLM assert result.provider_result is not None provider_result = ( result.provider_result.model_dump() if hasattr(result.provider_result, "model_dump") else result.provider_result ) assert provider_result["app_ids"] == ["cfg"] assert "deployment_name" not in provider_result assert "description" not in provider_result assert provider_result["tool_app_bindings"] == [{"tool_id": "created-tool-1", "app_ids": ["cfg"]}] assert provider_result["tools_with_refs"] == [{"source_ref": "fv-create-service-1", "tool_id": "created-tool-1"}] assert result.type == DeploymentType.AGENT assert result.name == "langflow_my_deployment_abcdef12" assert result.description == "desc" @pytest.mark.anyio async def test_list_deployments_filters_with_provider_draft_filters(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient( {"id": "dep-1", "tools": []}, listed_agents=[ {"id": "dep-1", "name": "deployment-1", "tools": [], "environments": [{"name": "draft"}]}, {"id": "dep-2", "name": "deployment-2", "tools": [], "environments": [{"name": "prod"}]}, { "id": "dep-3", "name": "deployment-3", "display_name": "Deployment Three", "description": "Provider description", "tools": [], "environments": [{"name": "draft"}], }, { "id": "dep-4", "name": "deployment-4", "display_name": "Deployment Four", "description": "Provider description", "tools": [], "environments": [{"name": "draft"}, {"name": "live"}], }, ], ) fake_clients = _with_wxo_wrappers( SimpleNamespace( _base=fake_agent, agent=fake_agent, tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.list( user_id="user-1", db=object(), params=DeploymentListParams( deployment_types=[DeploymentType.AGENT], provider_params={"ids": ["dep-2"], "names": ["deployment-3", "deployment-4"], "environment": "draft"}, ), ) assert sorted(item.id for item in result.deployments) == ["dep-3"] assert fake_agent.last_list_params == {"ids": ["dep-2"], "names": ["deployment-3", "deployment-4"]} assert result.deployments[0].provider_data == { "display_name": "Deployment Three", "description": "Provider description", "tool_ids": [], "llm": TEST_WXO_LLM, "environments": ["draft"], } @pytest.mark.anyio async def test_update_provider_data_maps_raw_connection_conflict_to_deployment_conflict(monkeypatch): from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}), tool=FakeToolClient([{"id": "tool-1", "binding": {"langflow": {"connections": {}}}}]), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients async def mock_create_config(*, clients, config, user_id, db, created_app_ids_journal=None): # noqa: ARG001 response = SimpleNamespace(status_code=409, text='{"detail":"already exists"}') raise ClientAPIException(response=response) monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) monkeypatch.setattr(shared_core_module, "create_config", mock_create_config) with pytest.raises(ResourceConflictError, match="already exists in the provider") as exc_info: await service.update( user_id="user-1", deployment_id="dep-1", payload=DeploymentUpdate( provider_data={ "tools": { "raw_payloads": [ { "id": str(UUID("00000000-0000-0000-0000-000000000012")), "name": "snapshot-new-1", "description": "desc", "data": {"nodes": [], "edges": []}, "tags": [], "provider_data": { "project_id": "project-1", "source_ref": "fv-conflict-1", "tool_name": "snapshot_new_1", "tool_display_name": "Snapshot New 1", }, } ] }, "connections": { "raw_payloads": [ { "app_id": "cfg", "environment_variables": {"API_KEY": {"source": "raw", "value": "secret"}}, } ] }, "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"name_of_raw": "snapshot_new_1"}, "app_ids": ["cfg"], } ], } ), db=object(), ) assert exc_info.value.resource == "connection" @pytest.mark.anyio async def test_create_provider_data_maps_raw_connection_conflict_to_deployment_conflict(monkeypatch): from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = FakeWXOClients( agent=FakeAgentClient({"id": "dep-1", "tools": []}), tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) captured: dict[str, str] = {} async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients async def mock_create_config(*, clients, config, user_id, db, created_app_ids_journal=None): # noqa: ARG001 captured["attempted_app_id"] = config.name response = SimpleNamespace(status_code=409, text='{"detail":"already exists"}') raise ClientAPIException(response=response) monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) monkeypatch.setattr(shared_core_module, "create_config", mock_create_config) with pytest.raises(ResourceConflictError, match="already exists in the provider") as exc_info: await service.create( user_id="user-1", payload=DeploymentCreate( spec=BaseDeploymentData( name=_agent_technical_name(), description="desc", type=DeploymentType.AGENT, ), provider_data={ "display_name": "my deployment", "tools": { "raw_payloads": [ { "id": str(UUID("00000000-0000-0000-0000-000000000013")), "name": "snapshot-new-1", "description": "desc", "data": {"nodes": [], "edges": []}, "tags": [], "provider_data": { "project_id": "project-1", "source_ref": "fv-create-conflict-1", "tool_name": "snapshot_new_1", "tool_display_name": "Snapshot New 1", }, } ] }, "connections": { "raw_payloads": [ { "app_id": "cfg", "environment_variables": {"API_KEY": {"source": "raw", "value": "secret"}}, } ] }, "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"name_of_raw": "snapshot_new_1"}, "app_ids": ["cfg"], } ], }, ), db=object(), ) assert exc_info.value.resource == "connection" assert captured["attempted_app_id"] == "cfg" @pytest.mark.anyio @pytest.mark.parametrize( ("provider_data", "error_message"), [ ( { "tools": {}, "connections": { "raw_payloads": [ { "app_id": "app-in-use", "environment_variables": {"API_KEY": {"source": "raw", "value": "secret"}}, }, { "app_id": "app-unused", "environment_variables": {"API_KEY": {"source": "raw", "value": "secret"}}, }, ] }, "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-1")}, "app_ids": ["app-in-use"], } ], }, "connections\\.raw_payloads contains app_id values not referenced by operations", ), ], ) async def test_update_provider_data_validation_errors_raise_invalid_content( monkeypatch, provider_data: dict, error_message: str ): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}), tool=FakeToolClient([{"id": "tool-1", "binding": {"langflow": {"connections": {}}}}]), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(InvalidContentError, match=error_message): await service.update( user_id="user-1", deployment_id="dep-1", payload=DeploymentUpdate(provider_data=provider_data), db=object(), ) @pytest.mark.anyio async def test_update_provider_data_rolls_back_mutated_tools_with_writable_payload(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) class FailingAgentClient(FakeAgentClient): def update(self, deployment_id: str, payload: dict): self.update_calls.append((deployment_id, payload)) msg = "agent update failed" raise RuntimeError(msg) fake_tool = FakeToolClient( [ { "id": "tool-1", "name": "tool-1", "display_name": "Tool 1", "description": "desc", "binding": {"langflow": {"connections": {"old": "conn-old"}}}, "created_at": "read-only-field", } ] ) fake_clients = SimpleNamespace( agent=FailingAgentClient({"id": "dep-1", "tools": ["tool-1"]}), tool=fake_tool, connections=FakeConnectionsClient(existing_app_id="cfg-1"), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients async def mock_validate_connection(connections_client, *, app_id): # noqa: ARG001 return SimpleNamespace(connection_id="conn-new") monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) monkeypatch.setattr(update_core_module, "validate_connection", mock_validate_connection) with pytest.raises(DeploymentError, match="Please check server logs for details"): await service.update( user_id="user-1", deployment_id="dep-1", payload=DeploymentUpdate( spec=BaseDeploymentDataUpdate(description="trigger update"), provider_data={ "tools": {}, "connections": {}, "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"tool_id_with_ref": _tool_ref("tool-1")}, "app_ids": ["cfg-1"], } ], }, ), db=object(), ) assert len(fake_tool.update_calls) == 2 first_payload = fake_tool.update_calls[0][1] rollback_payload = fake_tool.update_calls[1][1] assert "id" not in first_payload assert "created_at" not in first_payload assert first_payload["binding"]["langflow"]["connections"]["cfg-1"] == "conn-new" assert rollback_payload["binding"]["langflow"]["connections"] == {"old": "conn-old"} @pytest.mark.anyio async def test_update_provider_data_rolls_back_partially_created_raw_tools(monkeypatch): core_tools_module = importlib.import_module("langflow.services.adapters.deployment.watsonx_orchestrate.core.tools") service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_connections = FakeConnectionsClient() fake_tool = FakeToolClient([]) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": []}), tool=fake_tool, connections=fake_connections, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients async def mock_create_config(*, clients, config, user_id, db, created_app_ids_journal=None): # noqa: ARG001 fake_connections._connections_by_app_id[config.name] = f"conn-{config.name}" return config.name async def mock_validate_connection(connections_client, *, app_id): # noqa: ARG001 return SimpleNamespace(connection_id=f"conn-{app_id}") async def mock_create_and_upload_with_bindings(*, clients, tool_bindings): _ = clients, tool_bindings raise core_tools_module.ToolUploadBatchError( created_tool_ids=["created-tool-1"], errors=[RuntimeError("upload failed")], ) monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) monkeypatch.setattr(shared_core_module, "create_config", mock_create_config) monkeypatch.setattr(update_core_module, "validate_connection", mock_validate_connection) monkeypatch.setattr( update_core_module, "create_and_upload_wxo_flow_tools_with_bindings", mock_create_and_upload_with_bindings, ) with pytest.raises(DeploymentError, match="Please check server logs for details"): await service.update( user_id="user-1", deployment_id="dep-1", payload=DeploymentUpdate( provider_data={ "tools": { "raw_payloads": [ { "id": str(UUID("00000000-0000-0000-0000-000000000021")), "name": "snapshot-new-1", "description": "desc", "data": {"nodes": [], "edges": []}, "tags": [], "provider_data": { "project_id": "project-1", "source_ref": "fv-rollback-1", "tool_name": "snapshot_new_1", "tool_display_name": "Snapshot New 1", }, } ] }, "connections": { "raw_payloads": [ { "app_id": "cfg", "environment_variables": {"API_KEY": {"source": "raw", "value": "secret"}}, } ] }, "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"name_of_raw": "snapshot_new_1"}, "app_ids": ["cfg"], } ], } ), db=object(), ) assert fake_tool.delete_calls == ["created-tool-1"] assert fake_connections.delete_calls == ["cfg"] @pytest.mark.anyio async def test_create_provider_data_rolls_back_partially_created_raw_tools(monkeypatch): core_tools_module = importlib.import_module("langflow.services.adapters.deployment.watsonx_orchestrate.core.tools") service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_connections = FakeConnectionsClient() fake_tool = FakeToolClient([]) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": []}), tool=fake_tool, connections=fake_connections, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients async def mock_create_connection_with_conflict_mapping( *, clients, app_id, payload, user_id, db, error_prefix, # noqa: ARG001 created_app_ids_journal=None, ): _ = clients, payload, user_id, db if created_app_ids_journal is not None: created_app_ids_journal.append(app_id) fake_connections._connections_by_app_id[app_id] = f"conn-{app_id}" return app_id async def mock_validate_connection(connections_client, *, app_id): # noqa: ARG001 return SimpleNamespace(connection_id=f"conn-{app_id}") async def mock_create_and_upload_with_bindings(*, clients, tool_bindings): _ = clients, tool_bindings msg = "upload failed" raise core_tools_module.ToolUploadBatchError( created_tool_ids=["created-tool-1"], errors=[RuntimeError(msg)], ) monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) monkeypatch.setattr( shared_core_module, "create_connection_with_conflict_mapping", mock_create_connection_with_conflict_mapping, ) monkeypatch.setattr(create_core_module, "validate_connection", mock_validate_connection) monkeypatch.setattr( create_core_module, "create_and_upload_wxo_flow_tools_with_bindings", mock_create_and_upload_with_bindings, ) with pytest.raises(DeploymentError, match="Please check server logs for details"): await service.create( user_id="user-1", payload=DeploymentCreate( spec=BaseDeploymentData( name=_agent_technical_name(), description="desc", type=DeploymentType.AGENT, ), provider_data={ "display_name": "my deployment", "tools": { "raw_payloads": [ { "id": str(UUID("00000000-0000-0000-0000-000000000023")), "name": "snapshot-new-1", "description": "desc", "data": {"nodes": [], "edges": []}, "tags": [], "provider_data": { "project_id": "project-1", "source_ref": "fv-create-rollback-1", "tool_name": "snapshot_new_1", "tool_display_name": "Snapshot New 1", }, } ] }, "connections": { "raw_payloads": [ { "app_id": "cfg", "environment_variables": {"API_KEY": {"source": "raw", "value": "secret"}}, } ] }, "llm": TEST_WXO_LLM, "operations": [ { "op": "bind", "tool": {"name_of_raw": "snapshot_new_1"}, "app_ids": ["cfg"], } ], }, ), db=object(), ) assert fake_tool.delete_calls == ["created-tool-1"] assert fake_connections.delete_calls == ["cfg"] @pytest.mark.anyio async def test_process_raw_flows_with_app_id_awaits_connection_validation(monkeypatch): from langflow.services.adapters.deployment.watsonx_orchestrate.core import tools as tools_core_module fake_clients = SimpleNamespace( tool=SimpleNamespace(), connections=SimpleNamespace(), ) captured: dict[str, object] = {} async def mock_validate_connection(connections_client, *, app_id): # noqa: ARG001 return SimpleNamespace(connection_id="conn-123") async def mock_create_and_upload_wxo_flow_tools( *, clients, flow_payloads, connections, ): captured["clients"] = clients captured["flow_payloads"] = flow_payloads captured["connections"] = connections return [] monkeypatch.setattr( "langflow.services.adapters.deployment.watsonx_orchestrate.core.config.validate_connection", mock_validate_connection, ) monkeypatch.setattr( tools_core_module, "create_and_upload_wxo_flow_tools", mock_create_and_upload_wxo_flow_tools, ) result = await tools_core_module.process_raw_flows_with_app_id( clients=fake_clients, app_id="app-1", flows=[], ) assert result == [] assert captured["connections"] == {"app-1": "conn-123"} @pytest.mark.anyio async def test_process_raw_flows_with_app_id_returns_source_ref_bindings(monkeypatch): from langflow.services.adapters.deployment.watsonx_orchestrate.core import tools as tools_core_module fake_clients = SimpleNamespace( tool=SimpleNamespace(), connections=SimpleNamespace(), ) async def mock_validate_connection(connections_client, *, app_id): # noqa: ARG001 return SimpleNamespace(connection_id="conn-123") async def mock_create_and_upload_wxo_flow_tools( *, clients, # noqa: ARG001 flow_payloads, # noqa: ARG001 connections, # noqa: ARG001 ): return ["tool-1", "tool-2"] monkeypatch.setattr( "langflow.services.adapters.deployment.watsonx_orchestrate.core.config.validate_connection", mock_validate_connection, ) monkeypatch.setattr( tools_core_module, "create_and_upload_wxo_flow_tools", mock_create_and_upload_wxo_flow_tools, ) result = await tools_core_module.process_raw_flows_with_app_id( clients=fake_clients, app_id="app-1", flows=[ BaseFlowArtifact[payloads_module.WatsonxFlowArtifactProviderData]( id=UUID("00000000-0000-0000-0000-000000000001"), name="snapshot-one", description="desc", data={"nodes": [], "edges": []}, tags=[], provider_data=payloads_module.WatsonxFlowArtifactProviderData( project_id="project-1", source_ref="fv-1", tool_display_name="Snapshot One", ), ), BaseFlowArtifact[payloads_module.WatsonxFlowArtifactProviderData]( id=UUID("00000000-0000-0000-0000-000000000002"), name="snapshot-two", description="desc", data={"nodes": [], "edges": []}, tags=[], provider_data=payloads_module.WatsonxFlowArtifactProviderData( project_id="project-1", source_ref="fv-2", tool_display_name="Snapshot Two", ), ), ], ) assert [ { "source_ref": binding.source_ref, "tool_id": binding.tool_id, } for binding in result ] == [ { "source_ref": "fv-1", "tool_id": "tool-1", }, { "source_ref": "fv-2", "tool_id": "tool-2", }, ] @pytest.mark.anyio async def test_process_raw_flows_with_app_id_accepts_typed_provider_data(monkeypatch): from langflow.services.adapters.deployment.watsonx_orchestrate.core import tools as tools_core_module fake_clients = SimpleNamespace( tool=SimpleNamespace(), connections=SimpleNamespace(), ) async def mock_validate_connection(connections_client, *, app_id): # noqa: ARG001 return SimpleNamespace(connection_id="conn-123") async def mock_create_and_upload_wxo_flow_tools( *, clients, # noqa: ARG001 flow_payloads, # noqa: ARG001 connections, # noqa: ARG001 ): return ["tool-1"] monkeypatch.setattr( "langflow.services.adapters.deployment.watsonx_orchestrate.core.config.validate_connection", mock_validate_connection, ) monkeypatch.setattr( tools_core_module, "create_and_upload_wxo_flow_tools", mock_create_and_upload_wxo_flow_tools, ) result = await tools_core_module.process_raw_flows_with_app_id( clients=fake_clients, app_id="app-1", flows=[ BaseFlowArtifact[payloads_module.WatsonxFlowArtifactProviderData]( id=UUID("00000000-0000-0000-0000-000000000001"), name="snapshot-one", description="desc", data={"nodes": [], "edges": []}, tags=[], provider_data=payloads_module.WatsonxFlowArtifactProviderData( project_id="project-1", source_ref="fv-typed-1", tool_display_name="Snapshot One", ), ), ], ) assert len(result) == 1 assert result[0].source_ref == "fv-typed-1" assert result[0].tool_id == "tool-1" @pytest.mark.anyio async def test_process_raw_flows_with_app_id_rejects_plain_dict_provider_data(monkeypatch): from langflow.services.adapters.deployment.watsonx_orchestrate.core import tools as tools_core_module fake_clients = SimpleNamespace( tool=SimpleNamespace(), connections=SimpleNamespace(), ) async def mock_validate_connection(connections_client, *, app_id): # noqa: ARG001 return SimpleNamespace(connection_id="conn-123") async def mock_create_and_upload_wxo_flow_tools( *, clients, # noqa: ARG001 flow_payloads, # noqa: ARG001 connections, # noqa: ARG001 ): return ["tool-1"] monkeypatch.setattr( "langflow.services.adapters.deployment.watsonx_orchestrate.core.config.validate_connection", mock_validate_connection, ) monkeypatch.setattr( tools_core_module, "create_and_upload_wxo_flow_tools", mock_create_and_upload_wxo_flow_tools, ) with pytest.raises( InvalidContentError, match=r"Flow payload provider_data must be a WatsonxFlowArtifactProviderData model instance\.", ): await tools_core_module.process_raw_flows_with_app_id( clients=fake_clients, app_id="app-1", flows=[ BaseFlowArtifact( id=UUID("00000000-0000-0000-0000-000000000001"), name="snapshot-one", description="desc", data={"nodes": [], "edges": []}, tags=[], provider_data={"project_id": "project-1", "source_ref": "fv-dict-1"}, ), ], ) def test_create_wxo_flow_tool_keeps_load_from_db_global_values_unprefixed(monkeypatch): captured_tool_definition = {} flow_payload = BaseFlowArtifact[payloads_module.WatsonxFlowArtifactProviderData]( id="00000000-0000-0000-0000-000000000001", name="flow", description="desc", data={ "nodes": [ { "data": { "node": { "template": { "api_key": { "load_from_db": True, "value": "OPENAI_API_KEY", }, "plain_value": { "load_from_db": False, "value": "DO_NOT_TOUCH", }, } } } } ], "edges": [], }, tags=[], provider_data=payloads_module.WatsonxFlowArtifactProviderData( project_id="project-123", source_ref="fv-flow-template-1", tool_display_name="Flow Template", ), ) fake_tool = SimpleNamespace( __tool_spec__=SimpleNamespace( model_dump=lambda **kwargs: {"name": "flow"}, # noqa: ARG005 ) ) def mock_create_langflow_tool(*, tool_definition, connections, show_details): # noqa: ARG001 assert show_details is False captured_tool_definition.update(tool_definition) return fake_tool monkeypatch.setattr(tools_module, "create_langflow_tool", mock_create_langflow_tool) monkeypatch.setattr( tools_module, "build_langflow_artifact_bytes", lambda **kwargs: b"artifact", # noqa: ARG005 ) from langflow.services.adapters.deployment.watsonx_orchestrate.core.tools import create_wxo_flow_tool create_wxo_flow_tool( flow_payload=flow_payload, connections={}, ) template = captured_tool_definition["data"]["nodes"][0]["data"]["node"]["template"] assert template["api_key"]["value"] == "OPENAI_API_KEY" assert template["plain_value"]["value"] == "DO_NOT_TOUCH" def test_create_wxo_flow_tool_excludes_provider_data_from_artifact(monkeypatch): """provider_data must not leak into the flow JSON zipped inside the artifact. WxO's tool runtime chokes on unexpected top-level keys like ``provider_data`` in the flow definition, causing executions to hang indefinitely. """ captured_flow_definition = {} flow_payload = BaseFlowArtifact[payloads_module.WatsonxFlowArtifactProviderData]( id="00000000-0000-0000-0000-000000000001", name="flow", description="desc", data={ "nodes": [ { "data": { "type": "ChatInput", "id": "ChatInput-1", "node": {"template": {"_type": "CustomComponent"}}, } }, { "data": { "type": "ChatOutput", "id": "ChatOutput-1", "node": {"template": {"_type": "CustomComponent"}}, } }, ], "edges": [], }, tags=[], provider_data=payloads_module.WatsonxFlowArtifactProviderData( project_id="project-123", source_ref="src-ref-1", tool_display_name="Flow", ), ) fake_tool = SimpleNamespace( __tool_spec__=SimpleNamespace( model_dump=lambda **kwargs: {"name": "flow"}, # noqa: ARG005 ) ) def mock_create_langflow_tool(*, tool_definition, connections, show_details): # noqa: ARG001 assert show_details is False captured_flow_definition.update(tool_definition) return fake_tool monkeypatch.setattr(tools_module, "create_langflow_tool", mock_create_langflow_tool) monkeypatch.setattr( tools_module, "build_langflow_artifact_bytes", lambda **kwargs: b"artifact", # noqa: ARG005 ) from langflow.services.adapters.deployment.watsonx_orchestrate.core.tools import create_wxo_flow_tool create_wxo_flow_tool( flow_payload=flow_payload, connections={}, ) assert "provider_data" not in captured_flow_definition assert "data" in captured_flow_definition assert "name" in captured_flow_definition assert "description" in captured_flow_definition def test_create_wxo_flow_tool_requires_provider_data_project_id(): flow_payload = BaseFlowArtifact( id="00000000-0000-0000-0000-000000000001", name="flow", description="desc", data={"nodes": [], "edges": []}, tags=[], ) from langflow.services.adapters.deployment.watsonx_orchestrate.core.tools import create_wxo_flow_tool with pytest.raises( InvalidContentError, match=r"Flow payload provider_data must be a WatsonxFlowArtifactProviderData model instance\.", ): create_wxo_flow_tool( flow_payload=flow_payload, connections={}, ) def test_create_wxo_flow_tool_normalizes_name_for_raw_payload(monkeypatch): flow_payload = BaseFlowArtifact[payloads_module.WatsonxFlowArtifactProviderData]( id="00000000-0000-0000-0000-000000000001", name="basicllmwxo", description="desc", data={"nodes": [], "edges": []}, tags=[], provider_data=payloads_module.WatsonxFlowArtifactProviderData( project_id="project-123", source_ref="fv-prefix-1", tool_name="basicllmwxo", tool_display_name="Basic LLM wxO", ), ) fake_tool = SimpleNamespace( __tool_spec__=SimpleNamespace( model_dump=lambda **kwargs: {"name": "basicllmwxo"}, # noqa: ARG005 ) ) monkeypatch.setattr( tools_module, "create_langflow_tool", lambda **kwargs: fake_tool, # noqa: ARG005 ) monkeypatch.setattr( tools_module, "build_langflow_artifact_bytes", lambda **kwargs: b"artifact", # noqa: ARG005 ) from langflow.services.adapters.deployment.watsonx_orchestrate.core.tools import create_wxo_flow_tool tool_payload, artifact_bytes = create_wxo_flow_tool( flow_payload=flow_payload, connections={}, ) assert tool_payload["name"] == "basicllmwxo" assert tool_payload["binding"]["langflow"]["project_id"] == "project-123" assert artifact_bytes == b"artifact" def test_create_wxo_flow_tool_uses_provider_data_technical_name(monkeypatch): flow_payload = BaseFlowArtifact[payloads_module.WatsonxFlowArtifactProviderData]( id="00000000-0000-0000-0000-000000000001", name="raw-correlation-key", description="desc", data={"nodes": [], "edges": []}, tags=[], provider_data=payloads_module.WatsonxFlowArtifactProviderData( project_id="project-123", source_ref="fv-prefix-1", tool_display_name="Basic LLM wxO", ), ) fake_tool = SimpleNamespace( __tool_spec__=SimpleNamespace( model_dump=lambda **kwargs: {"name": "raw-correlation-key"}, # noqa: ARG005 ) ) monkeypatch.setattr( tools_module, "create_langflow_tool", lambda **kwargs: fake_tool, # noqa: ARG005 ) monkeypatch.setattr( tools_module, "build_langflow_artifact_bytes", lambda **kwargs: b"artifact", # noqa: ARG005 ) tool_payload, artifact_bytes = tools_module.create_wxo_flow_tool( flow_payload=flow_payload, connections={}, ) _assert_langflow_agent_name(tool_payload["name"], display_name="Basic LLM wxO") assert tool_payload["display_name"] == "Basic LLM wxO" assert artifact_bytes == b"artifact" @pytest.mark.anyio async def test_rollback_create_result_cleans_up_agent_tools_and_apps(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace() captured: dict[str, object] = {} async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients async def mock_rollback_created_resources(*, clients, agent_id, tool_ids, app_ids=None): captured["clients"] = clients captured["agent_id"] = agent_id captured["tool_ids"] = list(tool_ids) captured["app_ids"] = list(app_ids or []) monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) monkeypatch.setattr(service_module, "rollback_created_resources", mock_rollback_created_resources) await service.rollback_create_result( user_id="user-1", deployment_id="dep-created", provider_result={ "display_name": "Agent Display Name", "app_ids": ["cfg"], "tools_with_refs": [ {"source_ref": "fv-1", "tool_id": "tool-1"}, {"source_ref": "fv-2", "tool_id": "tool-2"}, ], }, db=object(), ) assert captured == { "clients": fake_clients, "agent_id": "dep-created", "tool_ids": ["tool-1", "tool-2"], "app_ids": ["cfg"], } @pytest.mark.anyio async def test_create_execution_posts_runs_payload(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "tools": []}) fake_base = FakeBaseClient() fake_clients = _with_wxo_wrappers( SimpleNamespace( _base=fake_base, agent=fake_agent, tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.create_execution( user_id="user-1", db=object(), payload=ExecutionCreate( deployment_id="dep-1", provider_data={"input": "hello from test"}, ), ) assert result.deployment_id == "dep-1" assert result.execution_id == "run-1" assert result.provider_result["status"] == "accepted" assert result.provider_result["execution_id"] == "run-1" assert result.provider_result["thread_id"] == "thread-1" assert fake_base.post_calls path, payload = fake_base.post_calls[0] assert path == "/runs" assert payload["agent_id"] == "dep-1" assert payload["message"] == {"role": "user", "content": "hello from test"} @pytest.mark.anyio async def test_get_execution_returns_completed_output(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "tools": []}) fake_base = FakeBaseClient( get_payloads={ "/runs/run-1": { "id": "run-1", "status": "completed", "agent_id": "dep-1", "completed_at": "2026-03-08T18:23:25.277362Z", "result": {"data": "Final assistant response"}, } } ) fake_clients = _with_wxo_wrappers( SimpleNamespace( _base=fake_base, agent=fake_agent, tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.get_execution( user_id="user-1", db=object(), execution_id="run-1", ) assert result.execution_id == "run-1" assert result.provider_result["status"] == "completed" assert result.provider_result["agent_id"] == "dep-1" assert result.provider_result["execution_id"] == "run-1" assert result.provider_result["completed_at"] == "2026-03-08T18:23:25.277362Z" @pytest.mark.anyio async def test_get_execution_fetches_result_payload(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "tools": []}) fake_base = FakeBaseClient( get_payloads={ "/runs/run-1": { "id": "run-1", "status": "completed", "agent_id": "dep-1", "result": {"output": "some result"}, } } ) fake_clients = _with_wxo_wrappers( SimpleNamespace( _base=fake_base, agent=fake_agent, tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.get_execution( user_id="user-1", db=object(), execution_id="run-1", ) assert result.provider_result["status"] == "completed" assert result.provider_result["agent_id"] == "dep-1" assert result.provider_result["result"] == {"output": "some result"} @pytest.mark.anyio async def test_get_execution_requires_execution_id(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "tools": []}) fake_base = FakeBaseClient() fake_clients = _with_wxo_wrappers( SimpleNamespace( _base=fake_base, agent=fake_agent, tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(ValueError, match="execution_id"): await service.get_execution( user_id="user-1", db=object(), execution_id="", ) @pytest.mark.anyio async def test_get_execution_handles_client_api_exception(monkeypatch): from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_base = FakeBaseClient() def failing_get(path: str, params=None): # noqa: ARG001 resp = SimpleNamespace(status_code=500, text='{"detail": "internal error"}') raise ClientAPIException(response=resp) fake_base._get = failing_get fake_clients = _with_wxo_wrappers( SimpleNamespace( _base=fake_base, agent=FakeAgentClient({"id": "dep-1", "tools": []}), tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(DeploymentError, match="getting a deployment execution"): await service.get_execution( user_id="user-1", db=object(), execution_id="run-1", ) @pytest.mark.anyio async def test_list_configs_single_deployment_scope(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}) fake_tool = FakeToolClient( [ { "id": "tool-1", "name": "tool-one", "binding": { "langflow": { "connections": { "cfg-1": "conn-1", } } }, "created_at": "2026-03-08T18:23:25.277362Z", } ] ) connections_client = FakeConnectionsClient() connections_client._draft_entries_by_id = [ ListConfigsResponse.model_validate( { "connection_id": "conn-1", "app_id": "cfg-1", "security_scheme": "key_value_creds", "environment": "live", } ) ] fake_clients = SimpleNamespace( agent=fake_agent, tool=fake_tool, connections=connections_client, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.list_configs( user_id="user-1", db=object(), params=ConfigListParams(deployment_ids=["dep-1"]), ) assert result.configs == [] @pytest.mark.anyio async def test_list_configs_deployment_scope_filters_to_key_value_creds(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "tools": ["tool-1", "tool-2"]}) fake_tool = FakeToolClient( [ {"id": "tool-1", "name": "tool-one", "binding": {"langflow": {"connections": {"cfg-1": "conn-1"}}}}, {"id": "tool-2", "name": "tool-two", "binding": {"langflow": {"connections": {"cfg-2": "conn-2"}}}}, ] ) connections_client = FakeConnectionsClient() connections_client._draft_entries_by_id = [ ListConfigsResponse.model_validate( { "connection_id": "conn-1", "app_id": "cfg-1", "security_scheme": "key_value_creds", "environment": "draft", } ), ListConfigsResponse.model_validate({"connection_id": "conn-2", "app_id": "cfg-2", "security_scheme": "oauth2"}), ] fake_clients = SimpleNamespace( agent=fake_agent, tool=fake_tool, connections=connections_client, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.list_configs( user_id="user-1", db=object(), params=ConfigListParams(deployment_ids=["dep-1"]), ) assert [config.id for config in result.configs] == ["conn-1"] assert [config.name for config in result.configs] == ["cfg-1"] assert [config.provider_data for config in result.configs] == [{"type": "key_value_creds", "environment": "draft"}] @pytest.mark.anyio async def test_list_configs_deployment_scope_warns_on_stale_tool_ids(monkeypatch, caplog): """Stale tool IDs can be deleted between reads; keep resolved configs and warn.""" service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "tools": ["tool-1", "deleted-tool"]}) fake_tool = FakeToolClient( [ { "id": "tool-1", "name": "tool-one", "binding": {"langflow": {"connections": {"cfg-1": "conn-1"}}}, } ] ) connections_client = FakeConnectionsClient() connections_client._draft_entries_by_id = [ ListConfigsResponse.model_validate( { "connection_id": "conn-1", "app_id": "cfg-1", "security_scheme": "key_value_creds", "environment": "live", } ) ] fake_clients = SimpleNamespace( agent=fake_agent, tool=fake_tool, connections=connections_client, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) import logging with caplog.at_level(logging.WARNING): result = await service.list_configs( user_id="user-1", db=object(), params=ConfigListParams(deployment_ids=["dep-1"]), ) assert result.configs == [] assert "tool IDs not returned by provider" in caplog.text assert "deleted-tool" in caplog.text assert "dep-1" in caplog.text @pytest.mark.anyio async def test_list_configs_deployment_scope_fails_fast_when_type_enrichment_fails(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}) fake_tool = FakeToolClient( [ { "id": "tool-1", "name": "tool-one", "binding": {"langflow": {"connections": {"cfg-1": "conn-1"}}}, } ] ) connections_client = FakeConnectionsClient() fake_clients = SimpleNamespace( agent=fake_agent, tool=fake_tool, connections=connections_client, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients def get_drafts_by_ids_raises(conn_ids): # noqa: ARG001 msg = "provider timeout" raise RuntimeError(msg) monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) monkeypatch.setattr(connections_client, "get_drafts_by_ids", get_drafts_by_ids_raises) with pytest.raises(DeploymentError, match="listing deployment configs"): await service.list_configs( user_id="user-1", db=object(), params=ConfigListParams(deployment_ids=["dep-1"]), ) @pytest.mark.anyio async def test_list_configs_deployment_scope_accepts_schema_compatible_detailed_connection(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}) fake_tool = FakeToolClient( [ { "id": "tool-1", "name": "tool-one", "binding": {"langflow": {"connections": {"cfg-1": "conn-1"}}}, } ] ) connections_client = FakeConnectionsClient() connections_client._draft_entries_by_id = [ SimpleNamespace(connection_id="conn-1", app_id="cfg-1", security_scheme="key_value_creds", environment="draft") ] fake_clients = SimpleNamespace( agent=fake_agent, tool=fake_tool, connections=connections_client, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.list_configs( user_id="user-1", db=object(), params=ConfigListParams(deployment_ids=["dep-1"]), ) assert len(result.configs) == 1 assert result.configs[0].id == "conn-1" assert result.configs[0].name == "cfg-1" assert result.configs[0].provider_data == {"type": "key_value_creds", "environment": "draft"} @pytest.mark.anyio async def test_list_configs_deployment_scope_warns_when_referenced_connection_missing(monkeypatch, caplog): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}) fake_tool = FakeToolClient( [ { "id": "tool-1", "name": "tool-one", "binding": {"langflow": {"connections": {"cfg-1": "conn-1"}}}, } ] ) connections_client = FakeConnectionsClient() connections_client._draft_entries_by_id = [] fake_clients = SimpleNamespace( agent=fake_agent, tool=fake_tool, connections=connections_client, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) import logging with caplog.at_level(logging.WARNING): result = await service.list_configs( user_id="user-1", db=object(), params=ConfigListParams(deployment_ids=["dep-1"]), ) assert result.configs == [] assert "connection IDs not returned by provider" in caplog.text assert "conn-1" in caplog.text @pytest.mark.anyio async def test_list_configs_tenant_scope_raises_on_provider_list_failure(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) connections_client = FakeConnectionsClient() def list_raises(): msg = "provider list failed" raise RuntimeError(msg) monkeypatch.setattr(connections_client, "list", list_raises) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": []}), tool=FakeToolClient([]), connections=connections_client, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(DeploymentError, match="listing deployment configs"): await service.list_configs(user_id="user-1", db=object(), params=None) @pytest.mark.anyio async def test_list_configs_tenant_scope_handles_none_response(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) connections_client = FakeConnectionsClient() connections_client._list_entries = None fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": []}), tool=FakeToolClient([]), connections=connections_client, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.list_configs(user_id="user-1", db=object(), params=None) assert result.configs == [] assert result.provider_result == {} @pytest.mark.anyio async def test_list_configs_deployment_scope_raises_on_agent_fetch_failure(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}) fake_clients = SimpleNamespace( agent=fake_agent, tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) def get_draft_by_id_raises(deployment_id): # noqa: ARG001 msg = "provider agent fetch failed" raise RuntimeError(msg) monkeypatch.setattr(fake_agent, "get_draft_by_id", get_draft_by_id_raises) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(DeploymentError, match="listing deployment configs"): await service.list_configs( user_id="user-1", db=object(), params=ConfigListParams(deployment_ids=["dep-1"]), ) @pytest.mark.anyio async def test_list_configs_deployment_scope_raises_when_agent_not_found(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient(None), tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(DeploymentNotFoundError, match="Deployment 'dep-1' not found"): await service.list_configs( user_id="user-1", db=object(), params=ConfigListParams(deployment_ids=["dep-1"]), ) @pytest.mark.anyio async def test_list_configs_deployment_scope_raises_on_non_dict_agent_payload(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient(["dep-1"]), tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(InvalidContentError, match="unexpected deployment payload type"): await service.list_configs( user_id="user-1", db=object(), params=ConfigListParams(deployment_ids=["dep-1"]), ) @pytest.mark.anyio async def test_list_configs_deployment_scope_trusts_non_list_tools_payload(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) connections_client = FakeConnectionsClient() connections_client._draft_entries_by_id = [ ListConfigsResponse.model_validate( { "connection_id": "conn-1", "app_id": "cfg-1", "security_scheme": "key_value_creds", "environment": "draft", } ) ] fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": ("tool-1",)}), tool=FakeToolClient( [ { "id": "tool-1", "binding": {"langflow": {"connections": {"cfg-1": "conn-1"}}}, } ] ), connections=connections_client, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.list_configs( user_id="user-1", db=object(), params=ConfigListParams(deployment_ids=["dep-1"]), ) assert [config.id for config in result.configs] == ["conn-1"] assert [config.name for config in result.configs] == ["cfg-1"] @pytest.mark.anyio async def test_list_configs_deployment_scope_returns_early_when_no_tools(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": []}), tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.list_configs( user_id="user-1", db=object(), params=ConfigListParams(deployment_ids=["dep-1"]), ) assert result.configs == [] assert result.provider_result == {"deployment_id": "dep-1", "tool_ids": []} @pytest.mark.anyio async def test_list_configs_deployment_scope_handles_none_tools_payload(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": None}), tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.list_configs( user_id="user-1", db=object(), params=ConfigListParams(deployment_ids=["dep-1"]), ) assert result.configs == [] assert result.provider_result == {"deployment_id": "dep-1", "tool_ids": []} @pytest.mark.anyio async def test_list_configs_deployment_scope_raises_on_tool_fetch_failure(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_tool = FakeToolClient([]) def get_drafts_by_ids_raises(tool_ids): # noqa: ARG001 msg = "provider tool fetch failed" raise RuntimeError(msg) monkeypatch.setattr(fake_tool, "get_drafts_by_ids", get_drafts_by_ids_raises) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}), tool=fake_tool, connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(DeploymentError, match="listing deployment configs"): await service.list_configs( user_id="user-1", db=object(), params=ConfigListParams(deployment_ids=["dep-1"]), ) @pytest.mark.anyio async def test_list_configs_deployment_scope_uses_latest_binding_for_same_app(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) connections_client = FakeConnectionsClient() connections_client._draft_entries_by_id = [ ListConfigsResponse.model_validate( { "connection_id": "conn-2", "app_id": "cfg-1", "security_scheme": "key_value_creds", "environment": "draft", } ) ] fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": ["tool-1", "tool-2"]}), tool=FakeToolClient( [ {"id": "tool-1", "binding": {"langflow": {"connections": {"cfg-1": "conn-1"}}}}, {"id": "tool-2", "binding": {"langflow": {"connections": {"cfg-1": "conn-2"}}}}, ] ), connections=connections_client, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.list_configs( user_id="user-1", db=object(), params=ConfigListParams(deployment_ids=["dep-1"]), ) assert len(result.configs) == 1 assert result.configs[0].id == "conn-2" assert result.configs[0].name == "cfg-1" @pytest.mark.anyio async def test_list_configs_deployment_scope_skips_enrichment_when_no_connections(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_tool = FakeToolClient([{"id": "tool-1", "name": "tool-without-connections"}]) connections_client = FakeConnectionsClient() def fail_if_called(conn_ids): # noqa: ARG001 msg = "connection enrichment should not be called" raise AssertionError(msg) monkeypatch.setattr(connections_client, "get_drafts_by_ids", fail_if_called) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}), tool=fake_tool, connections=connections_client, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.list_configs( user_id="user-1", db=object(), params=ConfigListParams(deployment_ids=["dep-1"]), ) assert result.configs == [] @pytest.mark.anyio async def test_list_configs_deployment_scope_raises_on_malformed_detailed_connection(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_tool = FakeToolClient([{"id": "tool-1", "binding": {"langflow": {"connections": {"cfg-1": "conn-1"}}}}]) connections_client = FakeConnectionsClient() monkeypatch.setattr( connections_client, "get_drafts_by_ids", lambda conn_ids: [SimpleNamespace(security_scheme="key_value_creds")], # noqa: ARG005 ) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}), tool=fake_tool, connections=connections_client, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(AttributeError): await service.list_configs( user_id="user-1", db=object(), params=ConfigListParams(deployment_ids=["dep-1"]), ) @pytest.mark.anyio async def test_list_configs_scopes_return_same_normalized_item_shape(monkeypatch): """Tenant-scope and deployment-scope must return the same ConfigListItem shape.""" service = WatsonxOrchestrateDeploymentService(DummySettingsService()) connections_client = FakeConnectionsClient() connections_client._list_entries = [ ListConfigsResponse.model_validate( { "connection_id": "conn-1", "app_id": "cfg-1", "name": "Config One", "security_scheme": "key_value_creds", "environment": "draft", } ) ] connections_client._draft_entries_by_id = [ ListConfigsResponse.model_validate( { "connection_id": "conn-1", "app_id": "cfg-1", "security_scheme": "key_value_creds", "environment": "draft", } ) ] fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}), tool=FakeToolClient( [ { "id": "tool-1", "name": "Tool One", "display_name": "Tool One", "binding": {"langflow": {"connections": {"cfg-1": "conn-1"}}}, } ] ), connections=connections_client, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) tenant_result = await service.list_configs(user_id="user-1", db=object(), params=None) deployment_result = await service.list_configs( user_id="user-1", db=object(), params=ConfigListParams(deployment_ids=["dep-1"]), ) expected = {"id": "conn-1", "name": "cfg-1", "provider_data": {"type": "key_value_creds", "environment": "draft"}} assert len(tenant_result.configs) == 1 assert len(deployment_result.configs) == 1 assert tenant_result.configs[0].model_dump(exclude_none=True) == expected assert deployment_result.configs[0].model_dump(exclude_none=True) == expected @pytest.mark.anyio async def test_list_snapshots_single_deployment_scope(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": ["tool-1", "tool-2"]}), tool=FakeToolClient( [ {"id": "tool-1", "name": "Tool One", "display_name": "Tool One"}, {"id": "tool-2", "name": "Tool Two", "display_name": "Tool Two"}, ] ), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.list_snapshots( user_id="user-1", db=object(), params=SnapshotListParams(deployment_ids=["dep-1"]), ) assert [snapshot.id for snapshot in result.snapshots] == ["tool-1", "tool-2"] assert [snapshot.name for snapshot in result.snapshots] == ["Tool One", "Tool Two"] @pytest.mark.anyio async def test_list_snapshots_stale_tool_ids_returns_empty(monkeypatch): """When the agent references tool IDs that no longer exist, return no snapshots.""" service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": ["deleted-tool-1", "deleted-tool-2"]}), tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.list_snapshots( user_id="user-1", db=object(), params=SnapshotListParams(deployment_ids=["dep-1"]), ) assert result.snapshots == [] @pytest.mark.anyio async def test_list_snapshots_partial_resolution_logs_stale_ids(monkeypatch, caplog): """When some tool IDs resolve and others don't, return only resolved ones and log the stale IDs.""" service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": ["tool-1", "deleted-tool"]}), tool=FakeToolClient([{"id": "tool-1", "name": "Tool One", "display_name": "Tool One"}]), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) import logging with caplog.at_level(logging.WARNING): result = await service.list_snapshots( user_id="user-1", db=object(), params=SnapshotListParams(deployment_ids=["dep-1"]), ) assert [snapshot.id for snapshot in result.snapshots] == ["tool-1"] assert "deleted-tool" in caplog.text assert "dep-1" in caplog.text @pytest.mark.anyio async def test_list_snapshots_single_deployment_scope_extracts_connections(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}), tool=FakeToolClient( [ { "id": "tool-1", "name": "Tool One", "display_name": "Tool One", "binding": {"langflow": {"connections": {"cfg-1": "conn-1"}}}, } ] ), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.list_snapshots( user_id="user-1", db=object(), params=SnapshotListParams(deployment_ids=["dep-1"]), ) assert [snapshot.id for snapshot in result.snapshots] == ["tool-1"] assert result.snapshots[0].provider_data == { "name": "Tool One", "display_name": "Tool One", "connections": {"cfg-1": "conn-1"}, } @pytest.mark.anyio async def test_list_configs_without_deployment_id_lists_tenant_scope(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) connections_client = FakeConnectionsClient() connections_client._list_entries = [ ListConfigsResponse.model_validate( { "connection_id": "conn-1", "app_id": "cfg-1", "name": "Config One", "security_scheme": "key_value_creds", "environment": "draft", } ), ListConfigsResponse.model_validate( { "connection_id": "conn-2", "app_id": "cfg-2", "name": "Config Two", "security_scheme": "key_value_creds", "environment": "live", } ), ] fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": []}), tool=FakeToolClient([]), connections=connections_client, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.list_configs(user_id="user-1", db=object(), params=None) assert [config.id for config in result.configs] == ["conn-1"] assert [config.name for config in result.configs] == ["cfg-1"] assert result.provider_result == {} @pytest.mark.anyio async def test_list_configs_tenant_scope_handles_sdk_models(monkeypatch): """SDK ConnectionsClient.list() returns ListConfigsResponse objects.""" service = WatsonxOrchestrateDeploymentService(DummySettingsService()) connections_client = FakeConnectionsClient() connections_client._list_entries = [ ListConfigsResponse.model_validate( { "connection_id": "conn-pydantic-1", "app_id": "cfg-pydantic-1", "name": "Pydantic One", "security_scheme": "key_value_creds", "environment": "draft", } ), ListConfigsResponse.model_validate( { "connection_id": "conn-pydantic-2", "app_id": "cfg-pydantic-2", "name": "Pydantic Two", "security_scheme": "key_value_creds", "environment": "live", } ), ] fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": []}), tool=FakeToolClient([]), connections=connections_client, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.list_configs(user_id="user-1", db=object(), params=None) assert [config.id for config in result.configs] == ["conn-pydantic-1"] assert [config.name for config in result.configs] == ["cfg-pydantic-1"] assert result.provider_result == {} @pytest.mark.anyio async def test_list_configs_tenant_scope_filters_to_key_value_creds(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) connections_client = FakeConnectionsClient() connections_client._list_entries = [ ListConfigsResponse.model_validate( { "connection_id": "conn-oauth", "app_id": "cfg-oauth", "name": "Oauth Config", "security_scheme": "oauth2", } ), ListConfigsResponse.model_validate( { "connection_id": "conn-auth", "app_id": "cfg-auth", "name": "Auth Config", "security_scheme": "key_value_creds", "environment": "draft", } ), ] fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": []}), tool=FakeToolClient([]), connections=connections_client, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.list_configs(user_id="user-1", db=object(), params=None) assert len(result.configs) == 1 assert result.configs[0].id == "conn-auth" assert result.configs[0].name == "cfg-auth" assert result.configs[0].provider_data == {"type": "key_value_creds", "environment": "draft"} @pytest.mark.anyio async def test_list_configs_tenant_scope_fails_fast_on_dict_entries(monkeypatch): """Tenant-scope list_configs raises when wxO returns unexpected entry types.""" service = WatsonxOrchestrateDeploymentService(DummySettingsService()) connections_client = FakeConnectionsClient() connections_client._list_entries = [ {"connection_id": "dict-conn", "app_id": "dict-cfg", "name": "Dict Config"}, ListConfigsResponse.model_validate({"connection_id": "model-conn", "app_id": "model-cfg", "name": "Model"}), ] fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": []}), tool=FakeToolClient([]), connections=connections_client, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(InvalidContentError, match="unexpected connection entry type: dict"): await service.list_configs(user_id="user-1", db=object(), params=None) @pytest.mark.anyio async def test_list_configs_tenant_scope_preserves_duplicates(monkeypatch): """Tenant-scope list_configs keeps duplicate wxO entries as returned.""" service = WatsonxOrchestrateDeploymentService(DummySettingsService()) connections_client = FakeConnectionsClient() connections_client._list_entries = [ ListConfigsResponse.model_validate( { "connection_id": "conn-dup", "app_id": "cfg-dup", "name": "First", "security_scheme": "key_value_creds", "environment": "draft", } ), ListConfigsResponse.model_validate( { "connection_id": "conn-dup", "app_id": "cfg-dup", "name": "Second", "security_scheme": "key_value_creds", "environment": "draft", } ), ListConfigsResponse.model_validate( { "connection_id": "conn-unique", "app_id": "cfg-unique", "name": "Unique", "security_scheme": "key_value_creds", "environment": "draft", } ), ] fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": []}), tool=FakeToolClient([]), connections=connections_client, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.list_configs(user_id="user-1", db=object(), params=None) assert [config.id for config in result.configs] == ["conn-dup", "conn-dup", "conn-unique"] assert [config.name for config in result.configs] == ["cfg-dup", "cfg-dup", "cfg-unique"] @pytest.mark.anyio async def test_list_configs_tenant_scope_fails_fast_on_non_sdk_entries(monkeypatch): """Tenant-scope list_configs raises on the first non-SDK entry.""" service = WatsonxOrchestrateDeploymentService(DummySettingsService()) connections_client = FakeConnectionsClient() connections_client._list_entries = [ "just-a-string", 42, {"connection_id": "dict-conn", "app_id": "dict-cfg"}, ListConfigsResponse.model_validate({"connection_id": "valid-conn", "app_id": "valid-cfg", "name": "Valid"}), ListConfigsResponse.model_validate({"connection_id": "", "app_id": "missing-connection-id"}), ListConfigsResponse.model_validate({"connection_id": "missing-app-id", "app_id": ""}), ] fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": []}), tool=FakeToolClient([]), connections=connections_client, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(InvalidContentError, match="unexpected connection entry type: str"): await service.list_configs(user_id="user-1", db=object(), params=None) @pytest.mark.anyio async def test_list_snapshots_without_deployment_id_lists_tenant_scope(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_base = FakeBaseClient( get_payloads={ "/tools": [ { "id": "tool-1", "name": "Tool One", "display_name": "Tool One", "binding": {"langflow": {"connections": {"cfg-1": "conn-1"}}}, }, {"id": "tool-2", "name": "Tool Two", "display_name": "Tool Two"}, ] } ) fake_clients = _with_wxo_wrappers( SimpleNamespace( _base=fake_base, agent=FakeAgentClient({"id": "dep-1", "tools": []}), tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.list_snapshots(user_id="user-1", db=object(), params=None) assert [snapshot.id for snapshot in result.snapshots] == ["tool-1", "tool-2"] assert result.snapshots[0].provider_data == { "name": "Tool One", "display_name": "Tool One", "connections": {"cfg-1": "conn-1"}, } assert result.snapshots[1].provider_data == {"name": "Tool Two", "display_name": "Tool Two", "connections": {}} assert result.provider_result == {} @pytest.mark.anyio async def test_list_snapshots_snapshot_ids_returns_verified_provider_data(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": []}), tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients async def mock_verify_tools_by_ids(clients, snapshot_ids): # noqa: ARG001 return SnapshotListResult( snapshots=[ SnapshotItem( id="tool-1", name="tool_one", provider_data={ "name": "tool_one", "display_name": "Tool One", "connections": {"cfg-1": "conn-1"}, }, ) ] ) monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) monkeypatch.setattr(service_module, "verify_tools_by_ids", mock_verify_tools_by_ids) result = await service.list_snapshots( user_id="user-1", db=object(), params=SnapshotListParams(snapshot_ids=["tool-1"]), ) assert [snapshot.id for snapshot in result.snapshots] == ["tool-1"] assert result.snapshots[0].provider_data == { "name": "tool_one", "display_name": "Tool One", "connections": {"cfg-1": "conn-1"}, } @pytest.mark.anyio async def test_list_snapshots_snapshot_ids_trusts_verified_results_without_revalidation(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": []}), tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients async def mock_verify_tools_by_ids(clients, snapshot_ids): # noqa: ARG001 return SnapshotListResult( snapshots=[ SnapshotItem( id="tool-1", name="Tool One", provider_data={"unexpected": "value"}, ) ] ) monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) monkeypatch.setattr(service_module, "verify_tools_by_ids", mock_verify_tools_by_ids) result = await service.list_snapshots( user_id="user-1", db=object(), params=SnapshotListParams(snapshot_ids=["tool-1"]), ) assert result.snapshots[0].provider_data == {"unexpected": "value"} @pytest.mark.anyio async def test_verify_tools_by_ids_returns_tool_metadata_provider_data(): fake_clients = SimpleNamespace( tool=FakeToolClient( [ { "id": "tool-1", "name": "Tool One", "display_name": "Tool One", "binding": {"langflow": {"connections": {"cfg-1": "conn-1"}}}, "extra": "ignored", }, { "id": "tool-2", "name": "Tool Two", "display_name": "Tool Two", "binding": {"langflow": {"connections": {}}}, "extra": "ignored", }, ] ) ) result = await tools_module.verify_tools_by_ids(fake_clients, ["tool-1", "tool-2"]) assert [snapshot.id for snapshot in result.snapshots] == ["tool-1", "tool-2"] assert result.snapshots[0].provider_data == { "name": "Tool One", "display_name": "Tool One", "connections": {"cfg-1": "conn-1"}, } assert result.snapshots[1].provider_data == {"name": "Tool Two", "display_name": "Tool Two", "connections": {}} @pytest.mark.anyio async def test_verify_tools_by_ids_tolerates_malformed_connections_payload(): fake_clients = SimpleNamespace( tool=FakeToolClient( [ { "id": "tool-1", "name": "Tool One", "display_name": "Tool One", "binding": {"langflow": {"connections": ["not-a-dict"]}}, } ] ) ) result = await tools_module.verify_tools_by_ids(fake_clients, ["tool-1"]) assert len(result.snapshots) == 1 assert result.snapshots[0].id == "tool-1" assert result.snapshots[0].provider_data == {"name": "Tool One", "display_name": "Tool One", "connections": {}} @pytest.mark.anyio async def test_verify_tools_by_ids_preserves_provider_connection_values(): fake_clients = SimpleNamespace( tool=FakeToolClient( [ { "id": "tool-1", "name": "Tool One", "display_name": "Tool One", "binding": {"langflow": {"connections": {"cfg-1": " "}}}, } ] ) ) result = await tools_module.verify_tools_by_ids(fake_clients, ["tool-1"]) assert len(result.snapshots) == 1 assert result.snapshots[0].id == "tool-1" assert result.snapshots[0].provider_data == { "name": "Tool One", "display_name": "Tool One", "connections": {"cfg-1": " "}, } @pytest.mark.anyio async def test_verify_tools_by_ids_preserves_provider_connection_mapping(): fake_clients = SimpleNamespace( tool=FakeToolClient( [ { "id": "tool-1", "name": "Tool One", "display_name": "Tool One", "binding": { "langflow": { "connections": { "cfg-1": "conn-1", "cfg-2": " ", " ": "conn-3", "cfg-4": 123, } } }, } ] ) ) result = await tools_module.verify_tools_by_ids(fake_clients, ["tool-1"]) assert len(result.snapshots) == 1 assert result.snapshots[0].id == "tool-1" assert result.snapshots[0].provider_data == { "name": "Tool One", "display_name": "Tool One", "connections": { "cfg-1": "conn-1", "cfg-2": " ", " ": "conn-3", "cfg-4": 123, }, } # --------------------------------------------------------------------------- # Retry / backoff tests # --------------------------------------------------------------------------- @pytest.mark.anyio async def test_retry_with_backoff_succeeds_on_first_try(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.retry import retry_with_backoff call_count = 0 async def op(): nonlocal call_count call_count += 1 return "ok" result = await retry_with_backoff(op, max_attempts=3) assert result == "ok" assert call_count == 1 @pytest.mark.anyio async def test_retry_with_backoff_forwards_args_and_kwargs(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.retry import retry_with_backoff received: list[tuple[str, str]] = [] async def op(prefix: str, *, suffix: str) -> str: received.append((prefix, suffix)) return f"{prefix}-{suffix}" result = await retry_with_backoff(op, 3, "left", suffix="right") assert result == "left-right" assert received == [("left", "right")] @pytest.mark.anyio async def test_retry_create_with_to_thread_forwards_kwargs(): import asyncio from langflow.services.adapters.deployment.watsonx_orchestrate.core.retry import retry_create def sync_add(a: int, *, b: int) -> int: return a + b result = await retry_create(asyncio.to_thread, sync_add, 2, b=5) assert result == 7 @pytest.mark.anyio async def test_retry_with_backoff_retries_then_succeeds(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.retry import retry_with_backoff call_count = 0 async def op(): nonlocal call_count call_count += 1 if call_count < 3: msg = "transient" raise RuntimeError(msg) return "ok" result = await retry_with_backoff(op, max_attempts=3) assert result == "ok" assert call_count == 3 @pytest.mark.anyio async def test_retry_with_backoff_gives_up_after_max_attempts(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.retry import retry_with_backoff call_count = 0 async def op(): nonlocal call_count call_count += 1 msg = "always fails" raise RuntimeError(msg) with pytest.raises(RuntimeError, match="always fails"): await retry_with_backoff(op, max_attempts=3) assert call_count == 3 @pytest.mark.anyio async def test_retry_with_backoff_respects_should_retry_predicate(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.retry import retry_with_backoff call_count = 0 async def op(): nonlocal call_count call_count += 1 msg = "non-retryable" raise ValueError(msg) with pytest.raises(ValueError, match="non-retryable"): await retry_with_backoff( op, max_attempts=5, should_retry=lambda exc: not isinstance(exc, ValueError), ) assert call_count == 1 def test_is_retryable_create_exception_non_retryable_status_codes(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.retry import is_retryable_create_exception non_retryable = {400, 401, 403, 404, 409, 422} for code in non_retryable: exc = HTTPException(status_code=code) assert is_retryable_create_exception(exc) is False, f"status {code} should not be retryable" def test_is_retryable_create_exception_retryable_status_codes(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.retry import is_retryable_create_exception for code in (500, 502, 503, 429): exc = HTTPException(status_code=code) assert is_retryable_create_exception(exc) is True, f"status {code} should be retryable" def test_is_retryable_create_exception_domain_exceptions_not_retryable(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.retry import is_retryable_create_exception assert is_retryable_create_exception(ResourceConflictError()) is False assert is_retryable_create_exception(InvalidContentError()) is False assert is_retryable_create_exception(InvalidDeploymentOperationError()) is False def test_is_retryable_create_exception_generic_exception_is_retryable(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.retry import is_retryable_create_exception assert is_retryable_create_exception(RuntimeError("boom")) is True @pytest.mark.anyio async def test_rollback_created_resources_deletes_all(monkeypatch): from langflow.services.adapters.deployment.watsonx_orchestrate.core import retry as retry_module deleted = {"agents": [], "tools": [], "configs": []} async def fake_delete_agent(clients, *, agent_id): # noqa: ARG001 deleted["agents"].append(agent_id) async def fake_delete_tool(clients, *, tool_id): # noqa: ARG001 deleted["tools"].append(tool_id) async def fake_delete_config(clients, *, app_id): # noqa: ARG001 deleted["configs"].append(app_id) monkeypatch.setattr(retry_module, "delete_agent_if_exists", fake_delete_agent) monkeypatch.setattr(retry_module, "delete_tool_if_exists", fake_delete_tool) monkeypatch.setattr(retry_module, "delete_config_if_exists", fake_delete_config) fake_clients = SimpleNamespace() await retry_module.rollback_created_resources( clients=fake_clients, agent_id="agent-1", tool_ids=["tool-1", "tool-2"], app_ids=["app-1"], ) assert deleted["agents"] == ["agent-1"] assert deleted["tools"] == ["tool-2", "tool-1"] assert deleted["configs"] == ["app-1"] @pytest.mark.anyio async def test_rollback_continues_after_individual_failures(monkeypatch): from langflow.services.adapters.deployment.watsonx_orchestrate.core import retry as retry_module deleted = {"configs": []} async def fail_delete_agent(clients, *, agent_id): # noqa: ARG001 msg = "agent delete failed" raise RuntimeError(msg) async def fail_delete_tool(clients, *, tool_id): # noqa: ARG001 msg = "tool delete failed" raise RuntimeError(msg) async def fake_delete_config(clients, *, app_id): # noqa: ARG001 deleted["configs"].append(app_id) monkeypatch.setattr(retry_module, "delete_agent_if_exists", fail_delete_agent) monkeypatch.setattr(retry_module, "delete_tool_if_exists", fail_delete_tool) monkeypatch.setattr(retry_module, "delete_config_if_exists", fake_delete_config) monkeypatch.setattr(retry_module, "ROLLBACK_MAX_RETRIES", 1) fake_clients = SimpleNamespace() await retry_module.rollback_created_resources( clients=fake_clients, agent_id="agent-1", tool_ids=["tool-1"], app_ids=["app-1"], ) assert deleted["configs"] == ["app-1"] @pytest.mark.anyio async def test_rollback_update_resources_restores_then_deletes(monkeypatch): from langflow.services.adapters.deployment.watsonx_orchestrate.core import retry as retry_module restored: list[tuple[str, dict]] = [] deleted = {"tools": [], "configs": []} fake_clients = SimpleNamespace( tool=SimpleNamespace(update=lambda tool_id, payload: restored.append((tool_id, payload))), ) async def fake_delete_tool(clients, *, tool_id): # noqa: ARG001 deleted["tools"].append(tool_id) async def fake_delete_config(clients, *, app_id): # noqa: ARG001 deleted["configs"].append(app_id) monkeypatch.setattr(retry_module, "delete_tool_if_exists", fake_delete_tool) monkeypatch.setattr(retry_module, "delete_config_if_exists", fake_delete_config) await retry_module.rollback_update_resources( clients=fake_clients, created_tool_ids=["tool-new-1", "tool-new-2"], created_app_id="cfg-new", original_tools={ "tool-old-1": {"name": "t1"}, "tool-old-2": {"name": "t2"}, }, ) assert [tool_id for tool_id, _payload in restored] == ["tool-old-2", "tool-old-1"] assert deleted["tools"] == ["tool-new-2", "tool-new-1"] assert deleted["configs"] == ["cfg-new"] # --------------------------------------------------------------------------- # Service method tests: get, delete, get_status, update happy path # --------------------------------------------------------------------------- @pytest.mark.anyio async def test_get_deployment_returns_agent(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient( { "id": "dep-1", "name": "my_agent", "display_name": "My Agent", "description": "desc", "tools": [], "environments": [], }, ), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.get(user_id="user-1", deployment_id="dep-1", db=object()) assert result.id == "dep-1" assert result.name == "my_agent" @pytest.mark.anyio async def test_get_deployment_includes_tool_ids_and_environment_in_provider_data(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient( { "id": "dep-1", "name": "my_agent", "display_name": "My Agent", "description": "desc", "tools": ["tool-1", "tool-2"], "environments": [{"name": "draft"}], "llm": "meta-llm", }, ), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.get(user_id="user-1", deployment_id="dep-1", db=object()) assert result.provider_data == { "display_name": "My Agent", "description": "desc", "tool_ids": ["tool-1", "tool-2"], "environments": ["draft"], "llm": "meta-llm", } @pytest.mark.anyio @pytest.mark.parametrize("missing_field", ["display_name", "description"]) async def test_get_deployment_requires_provider_display_metadata(monkeypatch, missing_field): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) agent = { "id": "dep-1", "name": "my_agent", "display_name": "My Agent", "description": "desc", "tools": [], "environments": [], } agent.pop(missing_field) fake_clients = SimpleNamespace(agent=FakeAgentClient(agent, fill_missing_metadata=False)) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(KeyError, match=missing_field): await service.get(user_id="user-1", deployment_id="dep-1", db=object()) @pytest.mark.anyio async def test_get_deployment_defaults_tool_ids_to_empty_list(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient( { "id": "dep-1", "name": "my_agent", "display_name": "My Agent", "description": "desc", "tools": [], "environments": [], }, ), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.get(user_id="user-1", deployment_id="dep-1", db=object()) assert result.provider_data == { "display_name": "My Agent", "description": "desc", "tool_ids": [], "llm": TEST_WXO_LLM, "environments": [], } @pytest.mark.anyio async def test_get_deployment_not_found_raises(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient(None), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(DeploymentNotFoundError, match="not found"): await service.get(user_id="user-1", deployment_id="dep-1", db=object()) @pytest.mark.anyio async def test_get_deployment_handles_client_api_exception(monkeypatch): from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException service = WatsonxOrchestrateDeploymentService(DummySettingsService()) class FailingAgentClient(FakeAgentClient): def get_draft_by_id(self, deployment_id: str): # noqa: ARG002 resp = SimpleNamespace(status_code=500, text='{"detail": "internal error"}') raise ClientAPIException(response=resp) fake_clients = SimpleNamespace( agent=FailingAgentClient(None), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(DeploymentError, match="getting a deployment"): await service.get(user_id="user-1", deployment_id="dep-1", db=object()) @pytest.mark.anyio async def test_delete_deployment_calls_agent_delete(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "tools": []}) fake_clients = SimpleNamespace( agent=fake_agent, tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.delete(user_id="user-1", deployment_id="dep-1", db=object()) assert result.id == "dep-1" assert fake_agent.delete_calls == ["dep-1"] @pytest.mark.anyio async def test_delete_deployment_not_found_raises(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) class FailingAgentClient(FakeAgentClient): def delete(self, deployment_id: str): # noqa: ARG002 resp = SimpleNamespace(status_code=status.HTTP_404_NOT_FOUND, text="not found") from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException raise ClientAPIException(response=resp) fake_clients = SimpleNamespace( agent=FailingAgentClient({"id": "dep-1", "tools": []}), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(DeploymentNotFoundError, match="not found"): await service.delete(user_id="user-1", deployment_id="dep-1", db=object()) @pytest.mark.anyio async def test_delete_only_deletes_agent_not_tools_or_configs(monkeypatch): """Delete only removes the agent — tools and connections are left untouched.""" service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient( {"id": "dep-1", "tools": ["tool-1", "tool-2"]}, ) fake_tool = FakeToolClient( [ { "id": "tool-1", "binding": {"langflow": {"connections": {"app-1": {}}}}, }, { "id": "tool-2", "binding": {"langflow": {"connections": {"app-2": {}}}}, }, ] ) fake_conn = FakeConnectionsClient() fake_clients = FakeWXOClients( agent=fake_agent, tool=fake_tool, connections=fake_conn, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.delete(user_id="user-1", deployment_id="dep-1", db=object()) assert result.id == "dep-1" assert fake_agent.delete_calls == ["dep-1"] assert fake_tool.delete_calls == [] assert fake_conn.delete_calls == [] @pytest.mark.anyio async def test_get_status_not_configured(): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) with pytest.raises(DeploymentNotConfiguredError, match="Deployment status is not configured"): await service.get_status(user_id="user-1", deployment_id="dep-1", db=object()) @pytest.mark.anyio async def test_update_deployment_display_name_and_description_renames_technical_agent(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient({"id": "dep-1", "name": "langflow_old_name_87654321", "tools": ["tool-1"]}) fake_clients = SimpleNamespace( agent=fake_agent, tool=FakeToolClient([{"id": "tool-1"}]), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) monkeypatch.setattr(payloads_module, "uuid4", lambda: SimpleNamespace(hex="abcdef1234567890")) update_data = DeploymentUpdate( spec=BaseDeploymentDataUpdate(description="new desc"), provider_data={"display_name": "new name"}, ) result = await service.update( user_id="user-1", deployment_id="dep-1", payload=update_data, db=object(), ) assert result.id == "dep-1" assert len(fake_agent.update_calls) == 1 agent_id, payload = fake_agent.update_calls[0] assert agent_id == "dep-1" assert payload["display_name"] == "new name" assert payload["name"] == "langflow_new_name_abcdef12" assert payload["description"] == "new desc" assert result.provider_result is not None assert result.provider_result.name == payload["name"] assert result.provider_result.display_name == "new name" assert result.provider_result.description == "new desc" @pytest.mark.anyio async def test_update_deployment_technical_name_still_uses_explicit_spec_name(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient( { "id": "dep-1", "name": "provider_agent_name", "display_name": "old label", "description": "provider desc", "tools": [], } ) fake_clients = SimpleNamespace( agent=fake_agent, tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.update( user_id="user-1", deployment_id="dep-1", payload=DeploymentUpdate( spec=BaseDeploymentDataUpdate(name="new_technical_name"), provider_data={"display_name": "!!!"}, ), db=object(), ) _, payload = fake_agent.update_calls[0] assert payload == {"display_name": "!!!", "name": "new_technical_name"} assert result.provider_result is not None assert result.provider_result.name == "new_technical_name" assert result.provider_result.display_name == "!!!" assert result.provider_result.description == "provider desc" @pytest.mark.anyio async def test_update_deployment_rejects_unnormalized_explicit_technical_name(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "name": "provider_agent_name", "tools": []}), tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(InvalidContentError, match="Agent name must only contain letters, numbers, and underscores"): await service.update( user_id="user-1", deployment_id="dep-1", payload=DeploymentUpdate(spec=BaseDeploymentDataUpdate(name="new technical name")), db=object(), ) def test_build_update_payload_rejects_explicit_null_technical_name(): spec = BaseDeploymentDataUpdate.model_construct(name=None, _fields_set={"name"}) with pytest.raises(InvalidContentError, match="Agent name cannot be set to null"): update_core_module.build_update_payload_from_spec(spec) @pytest.mark.parametrize("description", ["", " "]) def test_build_update_payload_rejects_empty_description(description): spec = BaseDeploymentDataUpdate(description=description) with pytest.raises(InvalidContentError, match="Agent description cannot be empty"): update_core_module.build_update_payload_from_spec(spec) @pytest.mark.parametrize( ("core_update", "field_label"), [ (WatsonxDeploymentUpdatePayload(llm=None), "Agent llm"), (WatsonxDeploymentUpdatePayload(display_name=None), "Agent display name"), ], ) def test_build_update_payload_rejects_explicit_null_core_update_fields(core_update, field_label): with pytest.raises(InvalidContentError, match=f"{field_label} cannot be set to null"): update_core_module.build_update_payload_from_spec(None, core_update=core_update) @pytest.mark.parametrize( ("core_update", "field_label"), [ (WatsonxDeploymentUpdatePayload.model_construct(llm="", _fields_set={"llm"}), "Agent llm"), ( WatsonxDeploymentUpdatePayload.model_construct(display_name=" ", _fields_set={"display_name"}), "Agent display name", ), ], ) def test_build_update_payload_rejects_empty_core_update_fields(core_update, field_label): with pytest.raises(InvalidContentError, match=f"{field_label} cannot be empty"): update_core_module.build_update_payload_from_spec(None, core_update=core_update) def test_build_update_payload_uses_resource_fallback_for_symbol_only_display_name(monkeypatch): monkeypatch.setattr(payloads_module, "uuid4", lambda: SimpleNamespace(hex="abcdef1234567890")) core_update = WatsonxDeploymentUpdatePayload(display_name="!!!") payload = update_core_module.build_update_payload_from_spec(None, core_update=core_update) assert payload == {"display_name": "!!!", "name": "langflow_agent_abcdef12"} @pytest.mark.anyio async def test_update_deployment_not_found_raises(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient(None), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) update_data = DeploymentUpdate(spec=BaseDeploymentDataUpdate(name="x")) with pytest.raises(DeploymentNotFoundError, match="not found"): await service.update( user_id="user-1", deployment_id="dep-1", payload=update_data, db=object(), ) @pytest.mark.anyio async def test_list_deployments_without_params(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient( {"id": "dep-1", "tools": []}, listed_agents=[ { "id": "dep-1", "name": "agent-1", "display_name": "Agent One", "description": "desc", "tools": [], "environments": [{"name": "draft"}], }, ], ) fake_clients = _with_wxo_wrappers( SimpleNamespace( _base=fake_agent, agent=fake_agent, ) ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.list(user_id="user-1", db=object(), params=None) assert len(result.deployments) == 1 assert result.deployments[0].id == "dep-1" @pytest.mark.anyio @pytest.mark.parametrize("missing_field", ["display_name", "description"]) async def test_list_deployments_requires_provider_display_metadata(monkeypatch, missing_field): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) agent = { "id": "dep-1", "name": "agent-1", "display_name": "Agent One", "description": "desc", "tools": [], "environments": [{"name": "draft"}], } agent.pop(missing_field) fake_agent = FakeAgentClient({"id": "dep-1", "tools": []}, listed_agents=[agent], fill_missing_metadata=False) fake_clients = _with_wxo_wrappers( SimpleNamespace( _base=fake_agent, agent=fake_agent, ) ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(DeploymentError, match="listing deployments"): await service.list(user_id="user-1", db=object(), params=None) @pytest.mark.anyio async def test_list_deployments_ignores_removed_names_filter(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient( {"id": "dep-1", "tools": []}, listed_agents=[ { "id": "dep-1", "name": "agent-1", "display_name": "Agent One", "description": "desc", "tools": [], "environments": [{"name": "draft"}], }, ], ) fake_clients = _with_wxo_wrappers( SimpleNamespace( _base=fake_agent, agent=fake_agent, ) ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) params = DeploymentListParams() result = await service.list(user_id="user-1", db=object(), params=params) assert len(result.deployments) == 1 assert result.deployments[0].id == "dep-1" assert fake_agent.last_list_params is None @pytest.mark.anyio async def test_list_conflict_does_not_force_agent_resource_hint(monkeypatch): from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException service = WatsonxOrchestrateDeploymentService(DummySettingsService()) class FailingListAgentClient(FakeAgentClient): def _get(self, path: str, params: dict | None = None): # noqa: ARG002 if path == "/agents": raise ClientAPIException( response=SimpleNamespace(status_code=409, text='{"detail":"resource already exists"}') ) return {} fake_agent = FailingListAgentClient({"id": "dep-1", "tools": []}) fake_clients = _with_wxo_wrappers( SimpleNamespace( _base=fake_agent, agent=fake_agent, ) ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(ResourceConflictError) as exc_info: await service.list(user_id="user-1", db=object(), params=None) assert exc_info.value.resource is None assert exc_info.value.resource_name is None @pytest.mark.anyio async def test_list_types_returns_supported_types(): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) result = await service.list_types(user_id="user-1", db=object()) assert DeploymentType.AGENT in result.deployment_types assert len(result.deployment_types) == 1 @pytest.mark.anyio async def test_list_llms_returns_normalized_model_names(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient( {"id": "dep-1", "tools": []}, get_payloads={ "/models": [ {"model_name": "granite-3.1-8b"}, {"model_name": "granite-3.3-8b"}, {"model_name": "granite-3.1-8b"}, ] }, ) fake_clients = _with_wxo_wrappers( SimpleNamespace( _base=fake_agent, agent=fake_agent, ) ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.list_llms(user_id="user-1", db=object()) assert result.provider_result == { "models": [ {"model_name": "groq/openai/gpt-oss-120b"}, {"model_name": "bedrock/openai.gpt-oss-120b-1:0"}, {"model_name": "granite-3.1-8b"}, {"model_name": "granite-3.3-8b"}, {"model_name": "granite-3.1-8b"}, ] } @pytest.mark.anyio async def test_list_llms_invalid_payload_raises_deployment_error(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient( {"id": "dep-1", "tools": []}, get_payloads={"/models": [{"id": "missing-model-name"}]}, ) fake_clients = _with_wxo_wrappers( SimpleNamespace( _base=fake_agent, agent=fake_agent, ) ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(DeploymentError, match="listing deployment LLMs"): await service.list_llms(user_id="user-1", db=object()) @pytest.mark.anyio async def test_update_spec_only_description_sends_update(monkeypatch): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_agent = FakeAgentClient( {"id": "dep-1", "name": "existing_agent", "display_name": "Existing Agent", "tools": []} ) fake_clients = SimpleNamespace( agent=fake_agent, ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) result = await service.update( user_id="user-1", deployment_id="dep-1", payload=DeploymentUpdate(spec=BaseDeploymentDataUpdate(description="only desc")), db=object(), ) assert result.id == "dep-1" assert len(fake_agent.update_calls) == 1 _, payload = fake_agent.update_calls[0] assert payload == {"description": "only desc"} assert "name" not in payload assert result.provider_result is not None assert result.provider_result.display_name == "Existing Agent" # --------------------------------------------------------------------------- # Client authentication tests # --------------------------------------------------------------------------- def test_get_authenticator_ibm_cloud(): from langflow.services.adapters.deployment.watsonx_orchestrate.client import get_authenticator auth = get_authenticator("https://api.region-foobar.cloud.ibm.com", "test-key") from ibm_cloud_sdk_core.authenticators import IAMAuthenticator assert isinstance(auth, IAMAuthenticator) def test_get_authenticator_mcsp(): from langflow.services.adapters.deployment.watsonx_orchestrate.client import get_authenticator auth = get_authenticator("https://api.wxo.ibm.com", "test-key") from ibm_cloud_sdk_core.authenticators import MCSPAuthenticator assert isinstance(auth, MCSPAuthenticator) def test_get_authenticator_unknown_url(): from langflow.services.adapters.deployment.watsonx_orchestrate.client import get_authenticator from lfx.services.adapters.deployment.exceptions import AuthSchemeError with pytest.raises(AuthSchemeError, match="Could not determine"): get_authenticator("https://example.com", "test-key") @pytest.mark.parametrize( "bypass_url", [ # Path smuggling: IBM domain appears in the path, not the hostname. "https://evil.example.com/.cloud.ibm.com", "https://evil.example.com/.ibm.com", # Query-string smuggling: IBM domain in query parameters. "https://evil.example.com/?host=.cloud.ibm.com", # Userinfo smuggling: IBM domain in userinfo before @. "https://cloud.ibm.com@evil.example.com/", # Fragment smuggling. "https://evil.example.com/#.cloud.ibm.com", ], ) def test_get_authenticator_rejects_substring_bypass(bypass_url): """Regression for CodeQL py/incomplete-url-substring-sanitization: scheme selection must be hostname-based.""" from langflow.services.adapters.deployment.watsonx_orchestrate.client import get_authenticator from lfx.services.adapters.deployment.exceptions import AuthSchemeError with pytest.raises(AuthSchemeError, match="Could not determine"): get_authenticator(bypass_url, "test-key") def test_get_authenticator_sets_http_timeout_on_iam(): from langflow.services.adapters.deployment.watsonx_orchestrate.client import get_authenticator auth = get_authenticator("https://api.region-foobar.cloud.ibm.com", "test-key") assert auth.token_manager.http_config == {"timeout": (10, 30)} def test_get_authenticator_sets_http_timeout_on_mcsp(): from langflow.services.adapters.deployment.watsonx_orchestrate.client import get_authenticator auth = get_authenticator("https://api.wxo.ibm.com", "test-key") assert auth.token_manager.http_config == {"timeout": (10, 30)} def test_get_authenticator_uses_default_iam_urls_when_unset(monkeypatch): try: with monkeypatch.context() as context: context.delenv("IBM_IAM_MCSP_DEV_URL_OVERRIDE", raising=False) context.delenv("IBM_IAM_DEV_URL_OVERRIDE", raising=False) reloaded_client_module = _reload_wxo_auth_modules() iam_auth = reloaded_client_module.get_authenticator("https://api.region-foobar.cloud.ibm.com", "test-key") mcsp_auth = reloaded_client_module.get_authenticator("https://api.wxo.ibm.com", "test-key") assert iam_auth.token_manager.url == "https://iam.cloud.ibm.com" assert mcsp_auth.token_manager.url == "https://iam.platform.saas.ibm.com" finally: _reload_wxo_auth_modules() @pytest.mark.parametrize("env_value", ["", " "]) def test_get_authenticator_empty_or_whitespace_env_var_falls_through_to_default(monkeypatch, env_value): try: with monkeypatch.context() as context: context.setenv("IBM_IAM_MCSP_DEV_URL_OVERRIDE", env_value) context.setenv("IBM_IAM_DEV_URL_OVERRIDE", env_value) reloaded_client_module = _reload_wxo_auth_modules() iam_auth = reloaded_client_module.get_authenticator("https://api.region-foobar.cloud.ibm.com", "test-key") mcsp_auth = reloaded_client_module.get_authenticator("https://api.wxo.ibm.com", "test-key") assert iam_auth.token_manager.url == "https://iam.cloud.ibm.com" assert mcsp_auth.token_manager.url == "https://iam.platform.saas.ibm.com" finally: _reload_wxo_auth_modules() def test_get_authenticator_uses_override_iam_urls(monkeypatch): custom_mcsp_url = " https://iam.platform.saas.ibm.com/custom-mcsp " custom_iam_url = " https://iam.cloud.ibm.com/custom-iam " try: with monkeypatch.context() as context: context.setenv("IBM_IAM_MCSP_DEV_URL_OVERRIDE", custom_mcsp_url) context.setenv("IBM_IAM_DEV_URL_OVERRIDE", custom_iam_url) reloaded_client_module = _reload_wxo_auth_modules() iam_auth = reloaded_client_module.get_authenticator("https://api.region-foobar.cloud.ibm.com", "test-key") mcsp_auth = reloaded_client_module.get_authenticator("https://api.wxo.ibm.com", "test-key") assert iam_auth.token_manager.url == custom_iam_url.strip() assert mcsp_auth.token_manager.url == custom_mcsp_url.strip() finally: _reload_wxo_auth_modules() @pytest.mark.anyio async def test_get_provider_clients_uses_request_scoped_context_memoization(monkeypatch): resolve_calls = 0 async def mock_resolve_wxo_client_credentials(*, user_id, db, provider_id): # noqa: ARG001 nonlocal resolve_calls resolve_calls += 1 return WxOCredentials( instance_url="https://api.region-foobar.cloud.ibm.com/", authenticator=object(), ) monkeypatch.setattr(client_module, "resolve_wxo_client_credentials", mock_resolve_wxo_client_credentials) client_module.clear_provider_clients_request_context() provider_context = deployment_context_module.DeploymentAdapterContext(provider_id=UUID(int=1)) with deployment_context_module.DeploymentProviderIDContext.scope(provider_context): first = await client_module.get_provider_clients(user_id="user-1", db=object()) second = await client_module.get_provider_clients(user_id="user-1", db=object()) assert first is second assert resolve_calls == 1 @pytest.mark.anyio async def test_get_provider_clients_rejects_mixed_provider_contexts(monkeypatch): resolve_calls = 0 async def mock_resolve_wxo_client_credentials(*, user_id, db, provider_id): # noqa: ARG001 nonlocal resolve_calls resolve_calls += 1 return WxOCredentials( instance_url="https://api.region-foobar.cloud.ibm.com/", authenticator=object(), ) monkeypatch.setattr(client_module, "resolve_wxo_client_credentials", mock_resolve_wxo_client_credentials) client_module.clear_provider_clients_request_context() with deployment_context_module.DeploymentProviderIDContext.scope( deployment_context_module.DeploymentAdapterContext(provider_id=UUID(int=1)) ): await client_module.get_provider_clients(user_id="user-1", db=object()) with ( deployment_context_module.DeploymentProviderIDContext.scope( deployment_context_module.DeploymentAdapterContext(provider_id=UUID(int=2)) ), pytest.raises(CredentialResolutionError, match="different deployment provider context"), ): await client_module.get_provider_clients(user_id="user-1", db=object()) assert resolve_calls == 1 @pytest.mark.anyio async def test_deployment_provider_scope_reuses_clients_within_scope(monkeypatch): resolve_calls = 0 async def mock_resolve_wxo_client_credentials(*, user_id, db, provider_id): # noqa: ARG001 nonlocal resolve_calls resolve_calls += 1 return WxOCredentials( instance_url="https://api.region-foobar.cloud.ibm.com/", authenticator=object(), ) monkeypatch.setattr(client_module, "resolve_wxo_client_credentials", mock_resolve_wxo_client_credentials) client_module.clear_provider_clients_request_context() with deployment_context_module.deployment_provider_scope(UUID(int=1)): first = await client_module.get_provider_clients(user_id="user-1", db=object()) second = await client_module.get_provider_clients(user_id="user-1", db=object()) assert first is second assert resolve_calls == 1 @pytest.mark.anyio async def test_deployment_provider_scope_resets_wxo_context_between_provider_scopes(monkeypatch): resolve_calls = 0 async def mock_resolve_wxo_client_credentials(*, user_id, db, provider_id): # noqa: ARG001 nonlocal resolve_calls resolve_calls += 1 return WxOCredentials( instance_url="https://api.region-foobar.cloud.ibm.com/", authenticator=object(), ) monkeypatch.setattr(client_module, "resolve_wxo_client_credentials", mock_resolve_wxo_client_credentials) # ``deployment_provider_scope`` only composes ``wxo_scope`` (which is what # actually resets the wxo provider-clients context between scopes) when the # wxo deployments feature flag is enabled. monkeypatch.setattr(deployment_context_module.FEATURE_FLAGS, "wxo_deployments", True) client_module.clear_provider_clients_request_context() with deployment_context_module.deployment_provider_scope(UUID(int=1)): await client_module.get_provider_clients(user_id="user-1", db=object()) with deployment_context_module.deployment_provider_scope(UUID(int=2)): await client_module.get_provider_clients(user_id="user-1", db=object()) assert resolve_calls == 2 @pytest.mark.anyio async def test_deployment_provider_scope_rejects_mixed_users_within_same_scope(monkeypatch): resolve_calls = 0 async def mock_resolve_wxo_client_credentials(*, user_id, db, provider_id): # noqa: ARG001 nonlocal resolve_calls resolve_calls += 1 return WxOCredentials( instance_url="https://api.region-foobar.cloud.ibm.com/", authenticator=object(), ) monkeypatch.setattr(client_module, "resolve_wxo_client_credentials", mock_resolve_wxo_client_credentials) client_module.clear_provider_clients_request_context() with deployment_context_module.deployment_provider_scope(UUID(int=1)): await client_module.get_provider_clients(user_id="user-1", db=object()) with pytest.raises(CredentialResolutionError, match="different deployment provider context"): await client_module.get_provider_clients(user_id="user-2", db=object()) assert resolve_calls == 1 @pytest.mark.anyio async def test_resolve_wxo_client_credentials_reads_provider_url_from_account(monkeypatch): from langflow.services.database.models.deployment_provider_account.model import DeploymentProviderAccount provider_account = DeploymentProviderAccount( id=UUID("00000000-0000-0000-0000-000000000099"), user_id=UUID("00000000-0000-0000-0000-000000000100"), name="prod", provider_tenant_id="tenant-1", provider_key="watsonx-orchestrate", provider_url="https://api.us-south.wxo.cloud.ibm.com/instances/tenant-1", api_key="encrypted-api-key", # pragma: allowlist secret ) async def mock_get_provider_account_by_id(*args, **kwargs): # noqa: ARG001 return provider_account monkeypatch.setattr(client_module, "get_provider_account_by_id", mock_get_provider_account_by_id) monkeypatch.setattr( client_module.auth_utils, "decrypt_api_key", lambda _encrypted_api_key: "decrypted-api-key", # pragma: allowlist secret ) credentials = await client_module.resolve_wxo_client_credentials( user_id="user-1", db=object(), provider_id=UUID("00000000-0000-0000-0000-000000000001"), ) assert credentials.instance_url == provider_account.provider_url assert credentials.authenticator is not None def test_wxo_client_initializes_subclients_eagerly(monkeypatch): init_counts = {"tool": 0, "connections": 0, "agent": 0, "base": 0} class FakeToolClient: def __init__(self, base_url: str, authenticator): # noqa: ARG002 init_counts["tool"] += 1 self.base_url = base_url class FakeConnectionsClient: def __init__(self, base_url: str, authenticator): # noqa: ARG002 init_counts["connections"] += 1 self.base_url = base_url class FakeAgentClient: def __init__(self, base_url: str, authenticator): # noqa: ARG002 init_counts["agent"] += 1 self.base_url = base_url class FakeBaseWXOClient: def __init__(self, base_url: str, authenticator): # noqa: ARG002 init_counts["base"] += 1 self.base_url = base_url monkeypatch.setattr(types_module, "ToolClient", FakeToolClient) monkeypatch.setattr(types_module, "ConnectionsClient", FakeConnectionsClient) monkeypatch.setattr(types_module, "AgentClient", FakeAgentClient) monkeypatch.setattr(types_module, "BaseWXOClient", FakeBaseWXOClient) wxo_client = types_module.WxOClient( instance_url="https://api.region-foobar.cloud.ibm.com", authenticator=object(), ) # All sub-clients should be created eagerly at construction assert init_counts == {"tool": 1, "connections": 1, "agent": 1, "base": 1} # Subsequent access does not re-create _ = wxo_client.agent assert init_counts["agent"] == 1 # --------------------------------------------------------------------------- # Utility function tests # --------------------------------------------------------------------------- def test_normalize_wxo_name(): from langflow.services.adapters.deployment.watsonx_orchestrate.payloads import normalize_wxo_name assert normalize_wxo_name("Hello World!") == "Hello_World" assert normalize_wxo_name("test-name-123") == "test_name_123" assert normalize_wxo_name(" spaces ") == "__spaces__" assert normalize_wxo_name("") == "" def test_validate_wxo_name_valid(): from langflow.services.adapters.deployment.watsonx_orchestrate.payloads import validate_wxo_name assert validate_wxo_name("my_deployment", field_label="Tool name") == "my_deployment" assert validate_wxo_name("My Deployment!", field_label="Tool name") == "My_Deployment" def test_validate_wxo_name_empty(): from langflow.services.adapters.deployment.watsonx_orchestrate.payloads import validate_wxo_name with pytest.raises(InvalidContentError, match="Tool name must include at least one alphanumeric character"): validate_wxo_name("!!!", field_label="Tool name") def test_validate_wxo_name_starts_with_digit(): from langflow.services.adapters.deployment.watsonx_orchestrate.payloads import validate_wxo_name with pytest.raises(InvalidContentError, match="Tool name must start with a letter"): validate_wxo_name("123abc", field_label="Tool name") @pytest.mark.anyio async def test_create_and_upload_wxo_flow_tools_with_bindings_journals_created_ids_on_upload_failure(monkeypatch): created_calls: list[dict] = [] def mock_create_tool(payload: dict): created_calls.append(payload) return {"id": f"tool-{len(created_calls)}"} def mock_upload_tool_artifact(tool_id: str, files: dict): # noqa: ARG001 if tool_id == "tool-1": raise InvalidContentError(message="artifact upload failed") return {"status": "ok"} fake_clients = SimpleNamespace( tool=SimpleNamespace(create=mock_create_tool), upload_tool_artifact=mock_upload_tool_artifact, ) def _mock_create_wxo_flow_tool(*, flow_payload, connections): # noqa: ARG001 return ( {"name": flow_payload.name, "description": flow_payload.description}, b"artifact", ) monkeypatch.setattr(tools_module, "create_wxo_flow_tool", _mock_create_wxo_flow_tool) bindings = [ tools_module.FlowToolBindingSpec( flow_payload=BaseFlowArtifact( id=UUID("00000000-0000-0000-0000-000000000031"), name="snapshot-one", description="desc", data={"nodes": [], "edges": []}, tags=[], provider_data={"project_id": "project-1", "source_ref": "fv-binding-1"}, ), connections={"cfg-1": "conn-1"}, ), tools_module.FlowToolBindingSpec( flow_payload=BaseFlowArtifact( id=UUID("00000000-0000-0000-0000-000000000032"), name="snapshot-two", description="desc", data={"nodes": [], "edges": []}, tags=[], provider_data={"project_id": "project-1", "source_ref": "fv-binding-2"}, ), connections={"cfg-1": "conn-1"}, ), ] with pytest.raises(tools_module.ToolUploadBatchError) as exc: await tools_module.create_and_upload_wxo_flow_tools_with_bindings( clients=fake_clients, tool_bindings=bindings, ) assert set(exc.value.created_tool_ids) == {"tool-1", "tool-2"} assert len(created_calls) == 2 @pytest.mark.anyio async def test_upload_wxo_flow_tool_maps_tool_conflict_with_structured_resource(): from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException def mock_create_tool(payload: dict): # noqa: ARG001 raise ClientAPIException( response=SimpleNamespace( status_code=409, text='{"detail":"Tool with name \'Simple_Agent\' already exists for this tenant."}', ) ) fake_clients = SimpleNamespace( tool=SimpleNamespace(create=mock_create_tool), upload_tool_artifact=lambda tool_id, files: {"id": tool_id}, # noqa: ARG005 ) with pytest.raises(ResourceConflictError) as exc_info: await tools_module.upload_wxo_flow_tool( clients=fake_clients, tool_payload={"name": "Simple_Agent"}, artifact_bytes=b"artifact", ) assert exc_info.value.resource == "tool" assert exc_info.value.resource_name == "Simple_Agent" @pytest.mark.anyio async def test_create_agent_deployment_maps_agent_conflict_with_structured_resource(): from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException def mock_create_agent(payload: dict): # noqa: ARG001 raise ClientAPIException( response=SimpleNamespace( status_code=409, text='{"detail":"Agent with name \'my_agent\' already exists for this tenant."}', ) ) fake_clients = SimpleNamespace(agent=SimpleNamespace(create=mock_create_agent)) with pytest.raises(ResourceConflictError) as exc_info: await create_core_module.create_agent_deployment( clients=fake_clients, payload={ "name": "my_agent", "display_name": "My Agent", "description": "desc", "tools": ["tool-1"], "style": "default", "llm": TEST_WXO_LLM, }, ) assert exc_info.value.resource == "agent" assert exc_info.value.resource_name == "my_agent" def test_extract_error_detail_json_string(): from langflow.services.adapters.deployment.watsonx_orchestrate.utils import extract_error_detail assert extract_error_detail('{"detail": "something went wrong"}') == "something went wrong" def test_extract_error_detail_json_list(): from langflow.services.adapters.deployment.watsonx_orchestrate.utils import extract_error_detail assert extract_error_detail('{"detail": [{"msg": "field required"}]}') == "field required" def test_extract_error_detail_json_dict(): from langflow.services.adapters.deployment.watsonx_orchestrate.utils import extract_error_detail result = extract_error_detail('{"detail": {"msg": "invalid"}}') assert result == "invalid" def test_extract_error_detail_non_json(): from langflow.services.adapters.deployment.watsonx_orchestrate.utils import extract_error_detail assert extract_error_detail("plain text error") == "plain text error" def test_extract_error_detail_with_null_detail_falls_back_to_body(): from langflow.services.adapters.deployment.watsonx_orchestrate.utils import extract_error_detail assert extract_error_detail('{"detail": null}') == '{"detail": null}' def test_extract_error_detail_uses_message_field_when_detail_missing(): from langflow.services.adapters.deployment.watsonx_orchestrate.utils import extract_error_detail payload = '{"statusCode":409,"message":"The connection ID already exists.","details":"duplicate"}' assert extract_error_detail(payload) == "The connection ID already exists." def test_dedupe_list(): from langflow.services.adapters.deployment.watsonx_orchestrate.utils import dedupe_list assert dedupe_list(["a", "b", "a", "c", "b"]) == ["a", "b", "c"] assert dedupe_list([]) == [] def test_raise_as_deployment_error_wraps_service_error_by_default(): from langflow.services.adapters.deployment.watsonx_orchestrate.constants import ErrorPrefix from langflow.services.adapters.deployment.watsonx_orchestrate.utils import raise_as_deployment_error original = InvalidContentError(message="invalid payload") with pytest.raises(DeploymentError, match="Please check server logs for details"): raise_as_deployment_error( original, error_prefix=ErrorPrefix.LIST, log_msg="unexpected list failure", ) def test_raise_as_deployment_error_reraises_allowed_service_error(): from langflow.services.adapters.deployment.watsonx_orchestrate.constants import ErrorPrefix from langflow.services.adapters.deployment.watsonx_orchestrate.utils import raise_as_deployment_error original = InvalidContentError(message="invalid payload") with pytest.raises(InvalidContentError, match="invalid payload"): raise_as_deployment_error( original, error_prefix=ErrorPrefix.LIST, log_msg="unexpected list failure", pass_through=(InvalidContentError,), ) def test_raise_as_deployment_error_client_api_falls_back_to_raw_body(): from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException from langflow.services.adapters.deployment.watsonx_orchestrate.constants import ErrorPrefix from langflow.services.adapters.deployment.watsonx_orchestrate.utils import raise_as_deployment_error resp = SimpleNamespace(status_code=500, text='{"error":"boom"}') exc = ClientAPIException(response=resp) with pytest.raises(DeploymentError, match="error details: boom"): raise_as_deployment_error( exc, error_prefix=ErrorPrefix.LIST_CONFIGS, log_msg="unexpected config list failure", ) def test_raise_as_deployment_error_http_exception_uses_detail(): from langflow.services.adapters.deployment.watsonx_orchestrate.constants import ErrorPrefix from langflow.services.adapters.deployment.watsonx_orchestrate.utils import raise_as_deployment_error exc = HTTPException(status_code=400, detail="bad request") with pytest.raises(DeploymentError, match="error details: bad request"): raise_as_deployment_error( exc, error_prefix=ErrorPrefix.HEALTH, log_msg="unexpected health check failure", ) def test_raise_as_deployment_error_maps_not_found(): from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException from langflow.services.adapters.deployment.watsonx_orchestrate.constants import ErrorPrefix from langflow.services.adapters.deployment.watsonx_orchestrate.utils import raise_as_deployment_error resp = SimpleNamespace(status_code=500, text='{"detail":"Agent \'abc\' not found"}') exc = ClientAPIException(response=resp) with pytest.raises(ResourceNotFoundError, match="not found"): raise_as_deployment_error( exc, error_prefix=ErrorPrefix.UPDATE, log_msg="unexpected update failure", ) def test_raise_as_deployment_error_maps_conflict(): from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException from langflow.services.adapters.deployment.watsonx_orchestrate.constants import ErrorPrefix from langflow.services.adapters.deployment.watsonx_orchestrate.utils import raise_as_deployment_error resp = SimpleNamespace(status_code=500, text='{"detail":"resource already exists"}') exc = ClientAPIException(response=resp) with pytest.raises(ResourceConflictError, match="already exists"): raise_as_deployment_error( exc, error_prefix=ErrorPrefix.UPDATE, log_msg="unexpected update conflict", resource="agent", resource_name="my-agent", ) def test_raise_as_deployment_error_maps_unprocessable_content(): from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException from langflow.services.adapters.deployment.watsonx_orchestrate.constants import ErrorPrefix from langflow.services.adapters.deployment.watsonx_orchestrate.utils import raise_as_deployment_error resp = SimpleNamespace(status_code=422, text='{"detail":"unprocessable"}') exc = ClientAPIException(response=resp) with pytest.raises(InvalidContentError, match="unprocessable"): raise_as_deployment_error( exc, error_prefix=ErrorPrefix.UPDATE, log_msg="unexpected update validation failure", ) def test_raise_as_deployment_error_maps_forbidden_to_authorization_error(): from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException from langflow.services.adapters.deployment.watsonx_orchestrate.constants import ErrorPrefix from langflow.services.adapters.deployment.watsonx_orchestrate.utils import raise_as_deployment_error resp = SimpleNamespace(status_code=403, text='{"detail":"forbidden"}') exc = ClientAPIException(response=resp) with pytest.raises(AuthorizationError, match="forbidden"): raise_as_deployment_error( exc, error_prefix=ErrorPrefix.UPDATE, log_msg="unexpected update authorization failure", ) def test_build_agent_payload_from_values_structure(): from langflow.services.adapters.deployment.watsonx_orchestrate.utils import build_agent_payload_from_values payload = build_agent_payload_from_values( agent_name="agent_name", agent_display_name="Agent Name", description="test description", tool_ids=["tool-1", "tool-2"], llm=TEST_WXO_LLM, ) assert payload["name"] == "agent_name" assert payload["display_name"] == "Agent Name" assert payload["description"] == "test description" assert payload["tools"] == ["tool-1", "tool-2"] assert payload["llm"] == TEST_WXO_LLM def test_build_agent_payload_from_values_uses_display_name_for_default_description(): from langflow.services.adapters.deployment.watsonx_orchestrate.utils import build_agent_payload_from_values payload = build_agent_payload_from_values( agent_name="agent_name", agent_display_name="Agent Name", description=" ", tool_ids=["tool-1"], llm=TEST_WXO_LLM, ) assert payload["description"] == "Langflow deployment Agent Name" # --------------------------------------------------------------------------- # Config helper unit tests (core/config.py standalone functions) # --------------------------------------------------------------------------- def test_normalize_optional_text_strips_and_returns_none_for_empty(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.config import normalize_optional_text assert normalize_optional_text(None) is None assert normalize_optional_text("") is None assert normalize_optional_text(" ") is None assert normalize_optional_text("hello") == "hello" assert normalize_optional_text(" spaced ") == "spaced" def test_normalize_optional_text_rejects_non_str(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.config import normalize_optional_text with pytest.raises(TypeError, match=r"expected str \| None"): normalize_optional_text(("value",)) with pytest.raises(TypeError, match=r"expected str \| None"): normalize_optional_text(42) def test_normalize_optional_text_handles_str_enum(): from enum import Enum from langflow.services.adapters.deployment.watsonx_orchestrate.core.config import normalize_optional_text class FakeEnum(str, Enum): KEY_VALUE = "key_value_creds" assert normalize_optional_text(FakeEnum.KEY_VALUE) == "key_value_creds" def test_build_config_list_item_valid(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.config import build_config_list_item from langflow.services.adapters.deployment.watsonx_orchestrate.payloads import WatsonxConfigItemProviderData from lfx.services.adapters.payload import PayloadSlot slot = PayloadSlot(WatsonxConfigItemProviderData) item = build_config_list_item( config_item_data_slot=slot, connection_id="conn-1", app_id="app-1", config_type="key_value_creds", environment="draft", ) assert item.id == "conn-1" assert item.name == "app-1" assert isinstance(item.provider_data, dict) assert item.provider_data["type"] == "key_value_creds" assert item.provider_data["environment"] == "draft" def test_build_config_list_item_missing_environment_for_key_value_creds(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.config import build_config_list_item from langflow.services.adapters.deployment.watsonx_orchestrate.payloads import WatsonxConfigItemProviderData from lfx.services.adapters.payload import PayloadSlot slot = PayloadSlot(WatsonxConfigItemProviderData) with pytest.raises(InvalidContentError, match="key_value_creds connection without a required environment"): build_config_list_item( config_item_data_slot=slot, connection_id="conn-1", app_id="app-1", config_type="key_value_creds", environment=None, ) def test_build_config_list_item_invalid_payload(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.config import build_config_list_item from langflow.services.adapters.deployment.watsonx_orchestrate.payloads import WatsonxConfigItemProviderData from lfx.services.adapters.payload import PayloadSlot slot = PayloadSlot(WatsonxConfigItemProviderData) with pytest.raises(InvalidContentError, match="invalid config item provider_data payload"): build_config_list_item( config_item_data_slot=slot, connection_id="conn-1", app_id="app-1", config_type="unknown_type", environment=None, ) def test_warn_if_expected_ids_missing_logs_warning(caplog): import logging from langflow.services.adapters.deployment.watsonx_orchestrate.core.config import warn_if_expected_ids_missing with caplog.at_level(logging.WARNING): warn_if_expected_ids_missing( deployment_id="dep-1", resource_name="tool", expected_ids=["t1", "t2", "t3"], resolved_ids={"t1"}, ) assert "t2" in caplog.text assert "t3" in caplog.text def test_warn_if_expected_ids_missing_no_warning_when_all_resolved(caplog): import logging from langflow.services.adapters.deployment.watsonx_orchestrate.core.config import warn_if_expected_ids_missing with caplog.at_level(logging.WARNING): warn_if_expected_ids_missing( deployment_id="dep-1", resource_name="tool", expected_ids=["t1", "t2"], resolved_ids={"t1", "t2"}, ) assert "list_configs" not in caplog.text # --------------------------------------------------------------------------- # Execution helper tests # --------------------------------------------------------------------------- def test_resolve_execution_message_string(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import resolve_execution_message result = resolve_execution_message("hello") assert result == {"role": "user", "content": "hello"} def test_resolve_execution_message_dict_with_role_content(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import resolve_execution_message msg = {"role": "assistant", "content": "hi"} assert resolve_execution_message(msg) == msg def test_resolve_execution_message_dict_with_nested_message(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import resolve_execution_message msg = {"message": {"role": "user", "content": "nested"}} assert resolve_execution_message(msg) == {"role": "user", "content": "nested"} def test_resolve_execution_message_empty_string_raises(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import resolve_execution_message with pytest.raises(ValueError, match="must not be empty"): resolve_execution_message(" ") def test_resolve_execution_message_none_raises(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import resolve_execution_message with pytest.raises(ValueError, match="requires input content"): resolve_execution_message(None) def test_create_agent_run_result_empty_raises(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import create_agent_run_result with pytest.raises(DeploymentError, match="empty response"): create_agent_run_result(None) with pytest.raises(DeploymentError, match="empty response"): create_agent_run_result({}) def test_create_agent_run_result_with_run_id(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import create_agent_run_result result = create_agent_run_result({"status": "running", "run_id": "r-1"}) assert result == {"status": "running", "execution_id": "r-1"} def test_create_agent_run_result_extracts_thread_id(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import create_agent_run_result result = create_agent_run_result({"status": "running", "run_id": "r-1", "thread_id": "t-1"}) assert result["thread_id"] == "t-1" def test_create_agent_run_result_omits_thread_id_when_absent(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import create_agent_run_result result = create_agent_run_result({"status": "running", "run_id": "r-1"}) assert "thread_id" not in result # --------------------------------------------------------------------------- # Status helper tests # --------------------------------------------------------------------------- def test_get_agent_environments_dedupes_preserving_order(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.status import get_agent_environments agent = { "environments": [ {"name": "draft"}, {"name": "draft"}, {"name": "live"}, {"name": "future-env"}, ] } assert get_agent_environments(agent) == ["draft", "live", "future-env"] def test_get_agent_environments_returns_empty_list_when_provider_returns_empty(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.status import get_agent_environments assert get_agent_environments({"environments": []}) == [] def test_get_agent_environments_raises_when_environments_key_missing(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.status import get_agent_environments with pytest.raises(KeyError): get_agent_environments({}) def test_get_agent_environments_raises_when_env_entry_missing_name(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.status import get_agent_environments with pytest.raises(KeyError): get_agent_environments({"environments": [{"not_name": "draft"}]}) # --------------------------------------------------------------------------- # Artifact build tests # --------------------------------------------------------------------------- def test_build_langflow_artifact_bytes_structure(): from langflow.services.adapters.deployment.watsonx_orchestrate.core.tools import ( build_langflow_artifact_bytes, ) flow_definition = {"nodes": [{"id": "n1"}], "edges": []} tool = SimpleNamespace( __tool_spec__=SimpleNamespace(name="test_tool"), requirements=["lfx>=0.3.0"], ) artifact_bytes = build_langflow_artifact_bytes( tool=tool, flow_definition=flow_definition, ) assert isinstance(artifact_bytes, bytes) with zipfile.ZipFile(io.BytesIO(artifact_bytes), "r") as zf: names = zf.namelist() assert "test_tool.json" in names assert "requirements.txt" in names assert "bundle-format" in names def _read_artifact_requirements(artifact_bytes: bytes) -> list[str]: """Return the non-empty requirements.txt lines from a wxO artifact zip.""" with zipfile.ZipFile(io.BytesIO(artifact_bytes), "r") as zf: content = zf.read("requirements.txt").decode("utf-8") return [line.strip() for line in content.splitlines() if line.strip()] def _provider_model_flow(provider: str, field: str = "model") -> dict: """Minimal flow whose single unified-model node selects *provider*. ``field`` is the model-selection template field (``model`` for Language Model / Embedding, ``agent_llm`` for the Agent component). """ return { "id": "flow-1", "name": "Provider Flow", "data": { "nodes": [ { "id": "lm-1", "type": "genericNode", "data": { "type": "LanguageModelComponent", "node": { "template": { "code": { "type": "code", "value": "from lfx.base.models.model import LCModelComponent", }, field: {"value": [{"provider": provider, "name": "x"}]}, } }, }, } ], "edges": [], }, } def test_build_langflow_artifact_requirements_pin_lfx_first_line(): """requirements.txt for an empty flow pins lfx as its first (and only) line. Locks the bundle-split contract that the wxO runner installs a pinned lfx plus only the third-party packages dependency detection emits. """ from langflow.services.adapters.deployment.watsonx_orchestrate.core.tools import ( build_langflow_artifact_bytes, ) from packaging.requirements import Requirement tool = SimpleNamespace(__tool_spec__=SimpleNamespace(name="empty_tool")) artifact_bytes = build_langflow_artifact_bytes(tool=tool, flow_definition={"data": {"nodes": [], "edges": []}}) reqs = _read_artifact_requirements(artifact_bytes) assert reqs, "requirements.txt should never be empty (lfx must always be pinned)" assert reqs[0].startswith("lfx=="), f"first requirement should pin lfx, got {reqs[0]!r}" # Every line must be a valid, installable PEP 508 specifier. for line in reqs: Requirement(line) def test_build_langflow_artifact_requirements_for_openai_provider_flow(): """A bundle-resident provider flow ships the langchain SDK, never a bundle dist. The runner builds the model from the langchain class directly, so it needs ``langchain-openai`` — and must NOT be told to install ``lfx-openai`` / ``lfx-bundles`` (those are not on the runner and are not required). """ from langflow.services.adapters.deployment.watsonx_orchestrate.core.tools import ( build_langflow_artifact_bytes, ) tool = SimpleNamespace(__tool_spec__=SimpleNamespace(name="openai_tool")) artifact_bytes = build_langflow_artifact_bytes(tool=tool, flow_definition=_provider_model_flow("OpenAI")) reqs = _read_artifact_requirements(artifact_bytes) names = {line.split("==")[0] for line in reqs} assert "langchain-openai" in names assert "lfx-openai" not in names assert "lfx-bundles" not in names assert any(line.startswith("lfx==") for line in reqs) def test_build_langflow_artifact_requirements_for_openrouter_provider_flow(): """OpenRouter is unified-selectable and runs on langchain-openai at runtime. Regression guard for the gap where OpenRouter resolved to no packages and produced an artifact that ImportErrors on the wxO runner. """ from langflow.services.adapters.deployment.watsonx_orchestrate.core.tools import ( build_langflow_artifact_bytes, ) tool = SimpleNamespace(__tool_spec__=SimpleNamespace(name="openrouter_tool")) artifact_bytes = build_langflow_artifact_bytes(tool=tool, flow_definition=_provider_model_flow("OpenRouter")) names = {line.split("==")[0] for line in _read_artifact_requirements(artifact_bytes)} assert "langchain-openai" in names # --------------------------------------------------------------------------- # WxOCredentials repr masking # --------------------------------------------------------------------------- def test_wxo_credentials_repr_does_not_expose_authenticator_data(): class _FakeAuthenticator: pass creds = WxOCredentials(instance_url="https://example.com", authenticator=_FakeAuthenticator()) repr_str = repr(creds) assert repr_str == "WxOCredentials(instance_url='https://example.com', authenticator=_FakeAuthenticator)" # --------------------------------------------------------------------------- # NotImplementedError stubs # --------------------------------------------------------------------------- @pytest.mark.anyio async def test_redeploy_raises_operation_not_supported(): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) with pytest.raises(OperationNotSupportedError, match="Redeployment is not supported"): await service.redeploy(user_id="user-1", deployment_id="dep-1", db=object()) @pytest.mark.anyio async def test_duplicate_raises_operation_not_supported(): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) with pytest.raises(OperationNotSupportedError, match="duplication is not supported"): await service.duplicate(user_id="user-1", deployment_id="dep-1", db=object()) @pytest.mark.anyio async def test_teardown_succeeds(): service = WatsonxOrchestrateDeploymentService(DummySettingsService()) await service.teardown() # --------------------------------------------------------------------------- # Tests for review fixes # --------------------------------------------------------------------------- @pytest.mark.anyio async def test_get_agent_run_empty_response_raises(monkeypatch): """get_agent_run raises DeploymentError when provider returns empty payload.""" from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import get_agent_run async def fake_to_thread(fn, *args, **kwargs): # noqa: ARG001 return None import asyncio monkeypatch.setattr(asyncio, "to_thread", fake_to_thread) fake_client = SimpleNamespace(get_run=lambda _run_id: None) with pytest.raises(DeploymentError, match="empty response"): await get_agent_run(fake_client, run_id="run-123") def test_retry_rollback_uses_retryable_filter(): """retry_rollback should use is_retryable_create_exception to skip non-retryable errors. Validates that the filter correctly identifies non-retryable HTTP status codes (via HTTPException, which is checked by is_retryable_create_exception). """ from langflow.services.adapters.deployment.watsonx_orchestrate.core.retry import is_retryable_create_exception # Non-retryable status codes should not be retried for code in [400, 401, 403, 404, 409, 422]: assert not is_retryable_create_exception(HTTPException(status_code=code)), ( f"HTTPException with status {code} should NOT be retryable" ) # Retryable status codes should be retried for code in [500, 502, 503, 504]: assert is_retryable_create_exception(HTTPException(status_code=code)), ( f"HTTPException with status {code} should be retryable" ) # Domain exceptions that are non-retryable from lfx.services.adapters.deployment.exceptions import InvalidContentError, ResourceConflictError assert not is_retryable_create_exception(ResourceConflictError()) assert not is_retryable_create_exception(InvalidContentError()) # Generic exceptions are retryable (e.g. transient network errors) assert is_retryable_create_exception(RuntimeError("transient")) @pytest.mark.anyio async def test_credential_resolution_catches_arbitrary_exceptions(monkeypatch): """resolve_wxo_client_credentials wraps unexpected exceptions as CredentialResolutionError.""" from langflow.services.adapters.deployment.watsonx_orchestrate.client import resolve_wxo_client_credentials from lfx.services.adapters.deployment.exceptions import CredentialResolutionError class FakeSQLAlchemyError(Exception): pass async def mock_get_provider(*args, **kwargs): # noqa: ARG001 error_message = "connection refused" raise FakeSQLAlchemyError(error_message) monkeypatch.setattr( "langflow.services.adapters.deployment.watsonx_orchestrate.client.get_provider_account_by_id", mock_get_provider, ) with pytest.raises(CredentialResolutionError, match="unexpected error"): await resolve_wxo_client_credentials( user_id="user-1", db=object(), provider_id=UUID("00000000-0000-0000-0000-000000000001"), ) def test_wxo_client_eagerly_constructs_sub_clients(): """WxOClient eagerly builds tool/connections/agent from instance_url and authenticator.""" types_module = importlib.import_module("langflow.services.adapters.deployment.watsonx_orchestrate.types") wxo_client_cls = types_module.WxOClient from ibm_cloud_sdk_core.authenticators import NoAuthAuthenticator client = wxo_client_cls(instance_url="https://test.example.com", authenticator=NoAuthAuthenticator()) # Sub-clients should be eagerly created at construction assert client.tool is not None assert client.connections is not None assert client.agent is not None assert client.base is not None def test_wxo_client_is_frozen(): """WxOClient is frozen and rejects post-construction mutation.""" types_module = importlib.import_module("langflow.services.adapters.deployment.watsonx_orchestrate.types") wxo_client_cls = types_module.WxOClient from ibm_cloud_sdk_core.authenticators import NoAuthAuthenticator client = wxo_client_cls(instance_url="https://test.example.com", authenticator=NoAuthAuthenticator()) with pytest.raises(AttributeError): client.instance_url = "https://evil.example.com" def test_wxo_client_strips_trailing_slash(): """WxOClient normalizes instance_url by stripping trailing slashes.""" types_module = importlib.import_module("langflow.services.adapters.deployment.watsonx_orchestrate.types") wxo_client_cls = types_module.WxOClient from ibm_cloud_sdk_core.authenticators import NoAuthAuthenticator client = wxo_client_cls(instance_url="https://test.example.com/", authenticator=NoAuthAuthenticator()) assert client.instance_url == "https://test.example.com" def test_wxo_client_rejects_empty_url(): """WxOClient rejects empty instance_url at construction.""" types_module = importlib.import_module("langflow.services.adapters.deployment.watsonx_orchestrate.types") wxo_client_cls = types_module.WxOClient from ibm_cloud_sdk_core.authenticators import NoAuthAuthenticator with pytest.raises(ValueError, match="non-empty"): wxo_client_cls(instance_url="", authenticator=NoAuthAuthenticator()) def test_wxo_credentials_is_frozen(): """WxOCredentials is frozen and rejects post-construction mutation.""" from ibm_cloud_sdk_core.authenticators import NoAuthAuthenticator creds = WxOCredentials(instance_url="https://test.example.com", authenticator=NoAuthAuthenticator()) with pytest.raises(AttributeError): creds.instance_url = "https://evil.example.com" def test_wxo_credentials_rejects_empty_url(): """WxOCredentials rejects empty instance_url at construction.""" from ibm_cloud_sdk_core.authenticators import NoAuthAuthenticator with pytest.raises(ValueError, match="non-empty"): WxOCredentials(instance_url="", authenticator=NoAuthAuthenticator()) def test_raise_for_status_separates_status_codes_from_string_heuristics(): """Status code dispatch and string heuristic dispatch are now separate cascading checks. Previously, status_code == 404 and 'not found' in detail were combined with `or`, meaning ANY status code with 'not found' text would raise DeploymentNotFoundError. Now, the 404 status code check is standalone, and 'not found' string heuristic only fires as a fallback for unmapped status codes. """ from lfx.services.adapters.deployment.exceptions import ResourceNotFoundError, raise_as_deployment_error # status_code=404 raises ResourceNotFoundError regardless of detail text with pytest.raises(ResourceNotFoundError): raise_as_deployment_error(status_code=404, detail="anything", message_prefix="test") # status_code=409 raises ResourceConflictError regardless of detail text with pytest.raises(ResourceConflictError): raise_as_deployment_error(status_code=409, detail="anything", message_prefix="test") # String heuristics still work as fallback for unmapped/None status codes with pytest.raises(ResourceNotFoundError): raise_as_deployment_error(status_code=None, detail="agent not found", message_prefix="test") with pytest.raises(ResourceConflictError): raise_as_deployment_error(status_code=None, detail="resource already exists", message_prefix="test") # --------------------------------------------------------------------------- # Test Coverage Gap #1: create — 409 conflict via ClientAPIException # --------------------------------------------------------------------------- @pytest.mark.anyio async def test_create_maps_409_conflict_to_deployment_conflict_error(): """Create raises ResourceConflictError when the provider agent create returns a 409.""" from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException service = WatsonxOrchestrateDeploymentService(DummySettingsService()) clients = FakeWXOClients( agent=FakeAgentClient( {"id": "dep-1", "tools": []}, create_exception=ClientAPIException( response=SimpleNamespace(status_code=409, text='{"detail":"already exists"}') ), ), tool=FakeToolClient([{"id": "tool-existing-1", "binding": {"langflow": {}}}]), connections=FakeConnectionsClient(existing_app_id="app-existing-1"), ) _attach_provider_clients(service, clients) with pytest.raises(ResourceConflictError, match="already exist"): await service.create( user_id="user-1", db=object(), payload=DeploymentCreate( spec=BaseDeploymentData( name="my_deployment", description="desc", type=DeploymentType.AGENT, ), provider_data=_create_provider_spec(), ), ) @pytest.mark.anyio async def test_create_tool_conflict_detail_is_not_misclassified_as_agent(monkeypatch): """Service-layer catch does not carry resource hints — resource stays None. This proves the conflict is not misclassified as 'agent'. The inner-layer test ``test_upload_wxo_flow_tool_maps_tool_conflict_with_structured_resource`` covers the path where ``resource='tool'`` is correctly attached. """ from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException service = WatsonxOrchestrateDeploymentService(DummySettingsService()) _attach_provider_clients(service, FakeWXOClients()) async def mock_apply_provider_create_plan_with_rollback(**kwargs): # noqa: ARG001 raise ClientAPIException( response=SimpleNamespace( status_code=409, text='{"detail":"Tool with name \'Simple_Agent\' already exists for this tenant."}', ) ) monkeypatch.setattr( service_module, "apply_provider_create_plan_with_rollback", mock_apply_provider_create_plan_with_rollback, ) with pytest.raises(ResourceConflictError) as exc_info: await service.create( user_id="user-1", db=object(), payload=DeploymentCreate( spec=BaseDeploymentData( name="ahhahahaqwerg", description="desc", type=DeploymentType.AGENT, ), provider_data=_create_provider_spec(), ), ) assert exc_info.value.resource is None assert exc_info.value.resource_name is None @pytest.mark.anyio async def test_create_maps_422_to_invalid_content_error(): """Create raises InvalidContentError when the provider agent create returns a 422.""" from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException service = WatsonxOrchestrateDeploymentService(DummySettingsService()) clients = FakeWXOClients( agent=FakeAgentClient( {"id": "dep-1", "tools": []}, create_exception=ClientAPIException( response=SimpleNamespace(status_code=422, text='{"detail":"validation error"}') ), ), tool=FakeToolClient([{"id": "tool-existing-1", "binding": {"langflow": {}}}]), connections=FakeConnectionsClient(existing_app_id="app-existing-1"), ) _attach_provider_clients(service, clients) with pytest.raises(InvalidContentError, match="validation error"): await service.create( user_id="user-1", db=object(), payload=DeploymentCreate( spec=BaseDeploymentData( name="my_deployment", description="desc", type=DeploymentType.AGENT, ), provider_data=_create_provider_spec(), ), ) # --------------------------------------------------------------------------- # Test Coverage Gap #2: create — unsupported deployment type rejection # --------------------------------------------------------------------------- @pytest.mark.anyio async def test_create_rejects_unsupported_deployment_type(monkeypatch): """Create raises DeploymentSupportError for non-AGENT deployment types.""" service = WatsonxOrchestrateDeploymentService(DummySettingsService()) # Simulate a deployment type that is not in SUPPORTED_ADAPTER_DEPLOYMENT_TYPES # by creating a spec with AGENT type but monkeypatching the check. spec = BaseDeploymentData( name="my_deployment", description="desc", type=DeploymentType.AGENT, ) # Override the type to a fake unsupported value after construction monkeypatch.setattr(spec, "type", SimpleNamespace(value="unsupported_type")) with pytest.raises(DeploymentSupportError, match="not supported"): await service.create( user_id="user-1", db=object(), payload=DeploymentCreate(spec=spec), ) # --------------------------------------------------------------------------- # Test Coverage Gap #3: update — empty provider_data + no spec → InvalidContentError # --------------------------------------------------------------------------- @pytest.mark.anyio async def test_update_rejects_empty_provider_data_with_no_spec_changes(monkeypatch): """Update raises InvalidContentError when provider_data is None and spec produces no update fields.""" service = WatsonxOrchestrateDeploymentService(DummySettingsService()) fake_clients = SimpleNamespace( agent=FakeAgentClient({"id": "dep-1", "tools": ["tool-1"]}), tool=FakeToolClient([{"id": "tool-1", "binding": {}}]), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) # Construct a payload that bypasses Pydantic validation to reach the guard payload = DeploymentUpdate.model_construct( spec=None, snapshot=None, config=None, provider_data=None, ) with pytest.raises(InvalidContentError, match="No data was provided for updating the deployment"): await service.update( user_id="user-1", deployment_id="dep-1", payload=payload, db=object(), ) # --------------------------------------------------------------------------- # Test Coverage Gap #4: extract_langflow_artifact_from_zip — all error paths # --------------------------------------------------------------------------- def test_extract_langflow_artifact_from_zip_success(): """extract_langflow_artifact_from_zip returns parsed JSON from a valid zip.""" import json from langflow.services.adapters.deployment.watsonx_orchestrate.core.tools import ( extract_langflow_artifact_from_zip, ) flow_data = {"name": "test_flow", "nodes": []} buf = io.BytesIO() with zipfile.ZipFile(buf, "w") as zf: zf.writestr("flow.json", json.dumps(flow_data)) result = extract_langflow_artifact_from_zip(buf.getvalue(), snapshot_id="snap-1") assert result == flow_data def test_extract_langflow_artifact_from_zip_no_json(): """extract_langflow_artifact_from_zip raises InvalidContentError when no JSON in zip.""" from langflow.services.adapters.deployment.watsonx_orchestrate.core.tools import ( extract_langflow_artifact_from_zip, ) buf = io.BytesIO() with zipfile.ZipFile(buf, "w") as zf: zf.writestr("readme.txt", "hello") with pytest.raises(InvalidContentError, match="does not include a flow JSON"): extract_langflow_artifact_from_zip(buf.getvalue(), snapshot_id="snap-1") def test_extract_langflow_artifact_from_zip_bad_zip(): """extract_langflow_artifact_from_zip raises InvalidContentError for invalid zip data.""" from langflow.services.adapters.deployment.watsonx_orchestrate.core.tools import ( extract_langflow_artifact_from_zip, ) with pytest.raises(InvalidContentError, match="not a valid zip"): extract_langflow_artifact_from_zip(b"not a zip file", snapshot_id="snap-1") def test_extract_langflow_artifact_from_zip_invalid_utf8(): """extract_langflow_artifact_from_zip raises InvalidContentError for non-UTF-8 content.""" from langflow.services.adapters.deployment.watsonx_orchestrate.core.tools import ( extract_langflow_artifact_from_zip, ) buf = io.BytesIO() with zipfile.ZipFile(buf, "w") as zf: zf.writestr("flow.json", b"\xff\xfe invalid utf-8") with pytest.raises(InvalidContentError, match="not valid UTF-8"): extract_langflow_artifact_from_zip(buf.getvalue(), snapshot_id="snap-1") def test_extract_langflow_artifact_from_zip_invalid_json(): """extract_langflow_artifact_from_zip raises InvalidContentError for malformed JSON.""" from langflow.services.adapters.deployment.watsonx_orchestrate.core.tools import ( extract_langflow_artifact_from_zip, ) buf = io.BytesIO() with zipfile.ZipFile(buf, "w") as zf: zf.writestr("flow.json", "not valid json {{{") with pytest.raises(InvalidContentError, match="invalid JSON"): extract_langflow_artifact_from_zip(buf.getvalue(), snapshot_id="snap-1") # --------------------------------------------------------------------------- # Test Coverage Gap #5: validate_connection — all negative paths # --------------------------------------------------------------------------- @pytest.mark.anyio async def test_validate_connection_missing_connection(): """validate_connection raises InvalidContentError when connection not found.""" from langflow.services.adapters.deployment.watsonx_orchestrate.core.config import validate_connection connections_client = FakeConnectionsClient() # no existing connections with pytest.raises(InvalidContentError, match="not found"): await validate_connection(connections_client, app_id="missing_app") @pytest.mark.anyio async def test_validate_connection_missing_config(monkeypatch): """validate_connection raises InvalidContentError when config not found.""" from langflow.services.adapters.deployment.watsonx_orchestrate.core.config import validate_connection connections_client = FakeConnectionsClient(existing_app_id="my_app") def get_config_none(app_id, env): # noqa: ARG001 return None monkeypatch.setattr(connections_client, "get_config", get_config_none) with pytest.raises(InvalidContentError, match="missing draft config"): await validate_connection(connections_client, app_id="my_app") @pytest.mark.anyio async def test_validate_connection_wrong_security_scheme(monkeypatch): """validate_connection raises InvalidContentError for non-key-value security scheme.""" from langflow.services.adapters.deployment.watsonx_orchestrate.core.config import validate_connection connections_client = FakeConnectionsClient(existing_app_id="my_app") def get_config_wrong_scheme(app_id, env): # noqa: ARG001 return SimpleNamespace(security_scheme="oauth2") monkeypatch.setattr(connections_client, "get_config", get_config_wrong_scheme) with pytest.raises(InvalidContentError, match="key-value credentials"): await validate_connection(connections_client, app_id="my_app") @pytest.mark.anyio async def test_validate_connection_missing_credentials(monkeypatch): """validate_connection raises InvalidContentError when credentials are missing.""" from ibm_watsonx_orchestrate_core.types.connections import ConnectionSecurityScheme from langflow.services.adapters.deployment.watsonx_orchestrate.core.config import validate_connection connections_client = FakeConnectionsClient(existing_app_id="my_app") def get_config_ok(app_id, env): # noqa: ARG001 return SimpleNamespace(security_scheme=ConnectionSecurityScheme.KEY_VALUE) def get_credentials_none(app_id, env, *, use_app_credentials): # noqa: ARG001 return None monkeypatch.setattr(connections_client, "get_config", get_config_ok) monkeypatch.setattr(connections_client, "get_credentials", get_credentials_none) with pytest.raises(InvalidContentError, match="missing draft runtime credentials"): await validate_connection(connections_client, app_id="my_app") # --------------------------------------------------------------------------- # Additional coverage: create_agent_run_result raises on missing run_id # --------------------------------------------------------------------------- def test_create_agent_run_result_raises_on_missing_run_id(): """create_agent_run_result raises DeploymentError when response has no execution identifier.""" from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import create_agent_run_result with pytest.raises(DeploymentError, match="did not return an execution identifier"): create_agent_run_result({"status": "accepted"}) def test_create_agent_run_result_extracts_run_id(): """create_agent_run_result translates WXO run_id to execution_id.""" from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import create_agent_run_result result = create_agent_run_result({"status": "accepted", "run_id": "run-123"}) assert result["execution_id"] == "run-123" assert result["status"] == "accepted" def test_create_agent_run_result_falls_back_to_id_field(): """create_agent_run_result uses 'id' field when 'run_id' is absent.""" from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import create_agent_run_result result = create_agent_run_result({"status": "running", "id": "id-456"}) assert result["execution_id"] == "id-456" # --------------------------------------------------------------------------- # Additional coverage: require_single_deployment_id — multiple IDs # --------------------------------------------------------------------------- def test_require_single_deployment_id_rejects_multiple_ids(): """require_single_deployment_id raises InvalidContentError for multiple IDs.""" from langflow.services.adapters.deployment.watsonx_orchestrate.utils import require_single_deployment_id params = ConfigListParams(deployment_ids=["id-1", "id-2"]) with pytest.raises(InvalidContentError, match="exactly one deployment_id"): require_single_deployment_id(params, resource_label="config") # --------------------------------------------------------------------------- # Additional coverage: exception chain preserved (from exc, not from None) # --------------------------------------------------------------------------- @pytest.mark.anyio async def test_create_preserves_exception_chain_on_unexpected_error(): """Create preserves exception chain with 'from exc' instead of 'from None'.""" service = WatsonxOrchestrateDeploymentService(DummySettingsService()) original_error = RuntimeError("unexpected db error") clients = FakeWXOClients( agent=FakeAgentClient({"id": "dep-1", "tools": []}, create_exception=original_error), tool=FakeToolClient([{"id": "tool-existing-1", "binding": {"langflow": {}}}]), connections=FakeConnectionsClient(existing_app_id="app-existing-1"), ) _attach_provider_clients(service, clients) with pytest.raises(DeploymentError, match="Please check server logs for details") as exc_info: await service.create( user_id="user-1", db=object(), payload=DeploymentCreate( spec=BaseDeploymentData( name="my_deployment", description="desc", type=DeploymentType.AGENT, ), provider_data=_create_provider_spec(), ), ) inner = exc_info.value.__cause__ assert isinstance(inner, DeploymentError) assert inner.__cause__ is original_error @pytest.mark.anyio async def test_delete_preserves_exception_chain_on_unexpected_error(monkeypatch): """Delete preserves exception chain with 'from exc' instead of 'from None'.""" service = WatsonxOrchestrateDeploymentService(DummySettingsService()) original_error = RuntimeError("unexpected error") fake_agent = FakeAgentClient({"id": "dep-1", "tools": []}) fake_agent.delete = lambda _deployment_id: (_ for _ in ()).throw(original_error) fake_clients = SimpleNamespace( agent=fake_agent, tool=FakeToolClient([]), connections=FakeConnectionsClient(), ) async def mock_get_provider_clients(*, user_id, db): # noqa: ARG001 return fake_clients monkeypatch.setattr(service, "_get_provider_clients", mock_get_provider_clients) with pytest.raises(DeploymentError) as exc_info: await service.delete( user_id="user-1", deployment_id="dep-1", db=object(), ) assert exc_info.value.__cause__ is original_error # --------------------------------------------------------------------------- # Additional coverage: _ensure_dict logs warning on non-dict replacement # --------------------------------------------------------------------------- def test_ensure_dict_logs_warning_on_non_dict(): """_ensure_dict logs a warning when replacing a non-dict value.""" from unittest.mock import patch from langflow.services.adapters.deployment.watsonx_orchestrate.core.tools import _ensure_dict parent = {"binding": "not a dict"} with patch("langflow.services.adapters.deployment.watsonx_orchestrate.core.tools.logger") as mock_logger: result = _ensure_dict(parent, "binding") assert result == {} assert parent["binding"] == {} mock_logger.warning.assert_called_once() call_args = mock_logger.warning.call_args assert "Expected dict" in call_args[0][0] assert call_args[0][2] == "str" # --------------------------------------------------------------------------- # get_agent_run — happy path: id → execution_id + passthrough fields # --------------------------------------------------------------------------- @pytest.mark.anyio async def test_get_agent_run_translates_run_id_to_execution_id(monkeypatch): """get_agent_run maps WXO id to execution_id and passes through other fields.""" import asyncio as _asyncio from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import get_agent_run wxo_payload = { "id": "r-42", "status": "completed", "agent_id": "agent-1", "started_at": "2026-01-01T00:00:00Z", "completed_at": "2026-01-01T00:01:00Z", "result": {"output": "hello"}, } async def fake_to_thread(fn, *args, **kwargs): # noqa: ARG001 return wxo_payload monkeypatch.setattr(_asyncio, "to_thread", fake_to_thread) fake_client = SimpleNamespace(get_run=lambda _run_id: wxo_payload) result = await get_agent_run(fake_client, run_id="r-42") assert result["execution_id"] == "r-42" assert "run_id" not in result assert result["status"] == "completed" assert result["agent_id"] == "agent-1" assert result["started_at"] == "2026-01-01T00:00:00Z" assert result["completed_at"] == "2026-01-01T00:01:00Z" assert result["result"] == {"output": "hello"} @pytest.mark.anyio async def test_get_agent_run_passes_through_error_fields(monkeypatch): """get_agent_run forwards failed_at, cancelled_at, and last_error.""" import asyncio as _asyncio from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import get_agent_run wxo_payload = { "id": "r-fail", "status": "failed", "failed_at": "2026-01-01T00:02:00Z", "last_error": "timeout exceeded", } async def fake_to_thread(fn, *args, **kwargs): # noqa: ARG001 return wxo_payload monkeypatch.setattr(_asyncio, "to_thread", fake_to_thread) fake_client = SimpleNamespace(get_run=lambda _run_id: wxo_payload) result = await get_agent_run(fake_client, run_id="r-fail") assert result["execution_id"] == "r-fail" assert result["status"] == "failed" assert result["failed_at"] == "2026-01-01T00:02:00Z" assert result["last_error"] == "timeout exceeded" # --------------------------------------------------------------------------- # get_agent_run — WXO payload omits id # --------------------------------------------------------------------------- @pytest.mark.anyio async def test_get_agent_run_falls_back_to_param_run_id(monkeypatch): """get_agent_run uses the run_id parameter when WXO payload omits id.""" import asyncio as _asyncio from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import get_agent_run wxo_payload = {"status": "in_progress", "agent_id": "agent-1"} async def fake_to_thread(fn, *args, **kwargs): # noqa: ARG001 return wxo_payload monkeypatch.setattr(_asyncio, "to_thread", fake_to_thread) fake_client = SimpleNamespace(get_run=lambda _run_id: wxo_payload) result = await get_agent_run(fake_client, run_id="r-99") assert result["execution_id"] == "r-99" assert result["status"] == "in_progress" assert result["agent_id"] == "agent-1" # --------------------------------------------------------------------------- # build_orchestrate_run_payload — simplified MVP payload # --------------------------------------------------------------------------- def test_build_orchestrate_run_payload_uses_message_directly(): """build_orchestrate_run_payload passes message from provider_data when present. The run target is always the deployment's own (owner-pinned) resource key: a caller-supplied provider_data["agent_id"] must be ignored so it cannot redirect the run to an arbitrary agent in the owner's WxO tenant. """ from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import build_orchestrate_run_payload message = {"role": "user", "content": "direct message"} result = build_orchestrate_run_payload( provider_data={"message": message, "agent_id": "a-1"}, deployment_id="dep-fallback", ) assert result["message"] is message assert result["agent_id"] == "dep-fallback" assert len(result) == 2 def test_build_orchestrate_run_payload_falls_back_to_deployment_id(): """build_orchestrate_run_payload uses deployment_id when agent_id is absent.""" from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import build_orchestrate_run_payload result = build_orchestrate_run_payload( provider_data={"input": "hello"}, deployment_id="dep-fallback", ) assert result["agent_id"] == "dep-fallback" assert result["message"] == {"role": "user", "content": "hello"} assert len(result) == 2 def test_build_orchestrate_run_payload_excludes_extra_fields(): """build_orchestrate_run_payload does not forward extra fields besides thread_id.""" from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import build_orchestrate_run_payload result = build_orchestrate_run_payload( provider_data={ "input": "hi", "thread_id": "t-1", "llm_params": {"model": "gpt-4"}, "guardrails": True, "stream": True, }, deployment_id="dep-1", ) assert result["thread_id"] == "t-1" assert "llm_params" not in result assert "guardrails" not in result assert "stream" not in result assert len(result) == 3 def test_build_orchestrate_run_payload_omits_thread_id_when_absent(): """build_orchestrate_run_payload does not include thread_id when not provided.""" from langflow.services.adapters.deployment.watsonx_orchestrate.core.execution import build_orchestrate_run_payload result = build_orchestrate_run_payload( provider_data={"input": "hi"}, deployment_id="dep-1", ) assert "thread_id" not in result assert len(result) == 2 # --------------------------------------------------------------------------- # WatsonxAgentExecutionResultData — adapter schema explicit fields # --------------------------------------------------------------------------- def test_adapter_execution_schema_parses_all_explicit_fields(): """WatsonxAgentExecutionResultData parses all execution response fields.""" from langflow.services.adapters.deployment.watsonx_orchestrate.payloads import WatsonxAgentExecutionResultData data = { "execution_id": "e-1", "agent_id": "a-1", "status": "completed", "result": {"output": "answer"}, "started_at": "2026-01-01T00:00:00Z", "completed_at": "2026-01-01T00:01:00Z", "failed_at": None, "cancelled_at": None, "last_error": None, } parsed = WatsonxAgentExecutionResultData.model_validate(data) assert parsed.execution_id == "e-1" assert parsed.agent_id == "a-1" assert parsed.status == "completed" assert parsed.result == {"output": "answer"} assert parsed.started_at == "2026-01-01T00:00:00Z" assert parsed.completed_at == "2026-01-01T00:01:00Z" def test_adapter_execution_schema_has_no_run_id_field(): """WatsonxAgentExecutionResultData does not expose run_id as a named field.""" from langflow.services.adapters.deployment.watsonx_orchestrate.payloads import WatsonxAgentExecutionResultData assert "run_id" not in WatsonxAgentExecutionResultData.model_fields # --------------------------------------------------------------------------- # WatsonxApiAgentExecution{Create,Status}ResultData — API schema explicit fields # --------------------------------------------------------------------------- def test_api_execution_create_schema_parses_all_explicit_fields(): """WatsonxApiAgentExecutionCreateResultData parses all execution response fields.""" from langflow.api.v1.mappers.deployments.watsonx_orchestrate.payloads import ( WatsonxApiAgentExecutionCreateResultData, ) data = { "id": "e-1", "agent_id": "a-1", "status": "accepted", "result": None, "started_at": "2026-01-01T00:00:00Z", } parsed = WatsonxApiAgentExecutionCreateResultData.model_validate(data) assert parsed.id == "e-1" assert parsed.agent_id == "a-1" assert parsed.status == "accepted" assert parsed.started_at == "2026-01-01T00:00:00Z" def test_api_execution_status_schema_parses_all_explicit_fields(): """WatsonxApiAgentExecutionStatusResultData parses all execution response fields.""" from langflow.api.v1.mappers.deployments.watsonx_orchestrate.payloads import ( WatsonxApiAgentExecutionStatusResultData, ) data = { "id": "e-1", "agent_id": "a-1", "status": "failed", "result": None, "started_at": "2026-01-01T00:00:00Z", "failed_at": "2026-01-01T00:00:05Z", "last_error": "something broke", } parsed = WatsonxApiAgentExecutionStatusResultData.model_validate(data) assert parsed.id == "e-1" assert parsed.agent_id == "a-1" assert parsed.status == "failed" assert parsed.failed_at == "2026-01-01T00:00:05Z" assert parsed.last_error == "something broke" def test_api_execution_schemas_have_no_run_id_field(): """Neither create nor status schema exposes run_id as a named field.""" from langflow.api.v1.mappers.deployments.watsonx_orchestrate.payloads import ( WatsonxApiAgentExecutionCreateResultData, WatsonxApiAgentExecutionStatusResultData, ) assert "run_id" not in WatsonxApiAgentExecutionCreateResultData.model_fields assert "run_id" not in WatsonxApiAgentExecutionStatusResultData.model_fields def test_api_execution_schemas_omit_langflow_owned_fields(): """deployment_id (Langflow DB UUID) belongs on the top-level response, not in provider_data.""" from langflow.api.v1.mappers.deployments.watsonx_orchestrate.payloads import ( WatsonxApiAgentExecutionCreateResultData, WatsonxApiAgentExecutionStatusResultData, ) for schema in (WatsonxApiAgentExecutionCreateResultData, WatsonxApiAgentExecutionStatusResultData): assert "deployment_id" not in schema.model_fields assert "id" in schema.model_fields assert not hasattr(schema, "resolved_deployment_id") def test_api_execution_schema_validates_provider_owned_id_fields_without_rewriting(): """Both create and status schemas reject blank provider IDs without stripping valid values.""" from langflow.api.v1.mappers.deployments.watsonx_orchestrate.payloads import ( WatsonxApiAgentExecutionCreateResultData, WatsonxApiAgentExecutionStatusResultData, ) for schema in (WatsonxApiAgentExecutionCreateResultData, WatsonxApiAgentExecutionStatusResultData): parsed = schema.model_validate( { "id": " e-1 ", "agent_id": " a-1 ", "thread_id": " t-1 ", } ) assert parsed.id == " e-1 " assert parsed.agent_id == " a-1 " assert parsed.thread_id == " t-1 " with pytest.raises(ValidationError): schema.model_validate( { "id": " ", "agent_id": "", "thread_id": " ", } ) # --------------------------------------------------------------------------- # Mapper shapers: shape_execution_create_result / shape_execution_status_result # --------------------------------------------------------------------------- def test_shape_execution_create_result_maps_all_fields(): """shape_execution_create_result maps adapter fields to API response.""" from langflow.api.v1.mappers.deployments.watsonx_orchestrate.mapper import WatsonxOrchestrateDeploymentMapper mapper = WatsonxOrchestrateDeploymentMapper() deployment_id = UUID("00000000-0000-0000-0000-000000000001") adapter_result = ExecutionCreateResult( execution_id="e-1", deployment_id="agent-1", provider_result={ "execution_id": "e-1", "agent_id": "agent-1", "status": "accepted", "started_at": "2026-01-01T00:00:00Z", }, ) response = mapper.shape_execution_create_result(adapter_result, deployment_id=deployment_id) assert response.deployment_id == deployment_id assert response.provider_data["id"] == "e-1" assert response.provider_data["status"] == "accepted" assert response.provider_data["started_at"] == "2026-01-01T00:00:00Z" assert response.provider_data["agent_id"] == "agent-1" assert "deployment_id" not in response.provider_data assert "execution_id" not in response.provider_data def test_shape_execution_status_result_maps_all_fields(): """shape_execution_status_result maps adapter fields to API response.""" from langflow.api.v1.mappers.deployments.watsonx_orchestrate.mapper import WatsonxOrchestrateDeploymentMapper mapper = WatsonxOrchestrateDeploymentMapper() deployment_id = UUID("00000000-0000-0000-0000-000000000002") adapter_result = ExecutionStatusResult( execution_id="e-2", deployment_id="agent-2", provider_result={ "execution_id": "e-2", "agent_id": "agent-2", "status": "completed", "result": {"output": "done"}, "completed_at": "2026-01-01T00:01:00Z", }, ) response = mapper.shape_execution_status_result(adapter_result, deployment_id=deployment_id) assert response.deployment_id == deployment_id assert response.provider_data["id"] == "e-2" assert response.provider_data["status"] == "completed" assert response.provider_data["result"] == {"output": "done"} assert response.provider_data["completed_at"] == "2026-01-01T00:01:00Z" assert "deployment_id" not in response.provider_data assert "execution_id" not in response.provider_data def test_shape_execution_status_result_none_execution_id(): """When adapter has no execution_id, provider_data includes it as None.""" from langflow.api.v1.mappers.deployments.watsonx_orchestrate.mapper import WatsonxOrchestrateDeploymentMapper mapper = WatsonxOrchestrateDeploymentMapper() deployment_id = UUID("00000000-0000-0000-0000-000000000003") adapter_result = ExecutionStatusResult( execution_id=None, deployment_id="agent-3", provider_result={ "agent_id": "agent-3", "status": "in_progress", }, ) response = mapper.shape_execution_status_result( adapter_result, deployment_id=deployment_id, ) assert response.provider_data["id"] is None assert response.provider_data["status"] == "in_progress" # --------------------------------------------------------------------------- # verify_credentials # --------------------------------------------------------------------------- @pytest.mark.anyio async def test_verify_credentials_success(monkeypatch): """verify_credentials returns VerifyCredentialsResult on valid credentials.""" from lfx.services.adapters.deployment.schema import VerifyCredentials, VerifyCredentialsResult class FakeTokenManager: def get_token(self): return "fake-token" class FakeAuthenticator: token_manager = FakeTokenManager() monkeypatch.setattr( service_module, "get_authenticator", lambda **_kwargs: FakeAuthenticator(), ) monkeypatch.setattr(service_module, "fetch_models_adapter", lambda *_args, **_kwargs: {}) svc = WatsonxOrchestrateDeploymentService(settings_service=DummySettingsService()) payload = VerifyCredentials( base_url="https://api.us-south.wxo.cloud.ibm.com", provider_data={"api_key": "valid-key"}, # pragma: allowlist secret ) result = await svc.verify_credentials(user_id="u1", payload=payload) assert isinstance(result, VerifyCredentialsResult) @pytest.mark.anyio async def test_verify_credentials_invalid_key_raises(monkeypatch): """verify_credentials raises AuthenticationError when provider rejects credentials.""" from ibm_cloud_sdk_core import ApiException from lfx.services.adapters.deployment.exceptions import AuthenticationError from lfx.services.adapters.deployment.schema import VerifyCredentials class FakeTokenManager: def get_token(self): raise ApiException(401, message="invalid api key") class FailingAuthenticator: token_manager = FakeTokenManager() monkeypatch.setattr( service_module, "get_authenticator", lambda **_kwargs: FailingAuthenticator(), ) svc = WatsonxOrchestrateDeploymentService(settings_service=DummySettingsService()) payload = VerifyCredentials( base_url="https://api.us-south.wxo.cloud.ibm.com", provider_data={"api_key": "bad-key"}, # pragma: allowlist secret ) with pytest.raises(AuthenticationError, match="Credential verification"): await svc.verify_credentials(user_id="u1", payload=payload) @pytest.mark.anyio async def test_verify_credentials_instance_probe_forbidden(monkeypatch): """403 from wxO models probe maps to AuthorizationError (wrong instance for key).""" from ibm_watsonx_orchestrate_clients.tools.tool_client import ClientAPIException from lfx.services.adapters.deployment.exceptions import AuthorizationError from lfx.services.adapters.deployment.schema import VerifyCredentials from requests import Response class FakeTokenManager: def get_token(self): return "fake-token" class FakeAuthenticator: token_manager = FakeTokenManager() monkeypatch.setattr( service_module, "get_authenticator", lambda **_kwargs: FakeAuthenticator(), ) def _fail_fetch_models(*_args, **_kwargs): response = Response() response.status_code = 403 raise ClientAPIException(response=response) monkeypatch.setattr(service_module, "fetch_models_adapter", _fail_fetch_models) svc = WatsonxOrchestrateDeploymentService(settings_service=DummySettingsService()) payload = VerifyCredentials( base_url="https://api.us-south.wxo.cloud.ibm.com", provider_data={"api_key": "valid-key"}, # pragma: allowlist secret ) with pytest.raises(AuthorizationError, match="Credential verification"): await svc.verify_credentials(user_id="u1", payload=payload) @pytest.mark.anyio async def test_verify_credentials_malformed_key_from_authenticator_constructor_raises(): """verify_credentials raises InvalidContentError when authenticator creation fails validation.""" from lfx.services.adapters.deployment.exceptions import InvalidContentError from lfx.services.adapters.deployment.schema import VerifyCredentials svc = WatsonxOrchestrateDeploymentService(settings_service=DummySettingsService()) payload = VerifyCredentials( base_url="https://api.us-south.wxo.cloud.ibm.com", provider_data={"api_key": "{bad-key}"}, # pragma: allowlist secret ) with pytest.raises(InvalidContentError, match="malformed"): await svc.verify_credentials(user_id="u1", payload=payload) @pytest.mark.anyio async def test_verify_credentials_missing_provider_data_raises(): """verify_credentials raises when provider_data is missing.""" from lfx.services.adapters.deployment.schema import VerifyCredentials from lfx.services.adapters.payload import AdapterPayloadMissingError svc = WatsonxOrchestrateDeploymentService(settings_service=DummySettingsService()) payload = VerifyCredentials( base_url="https://api.us-south.wxo.cloud.ibm.com", provider_data=None, ) with pytest.raises(AdapterPayloadMissingError): await svc.verify_credentials(user_id="u1", payload=payload) @pytest.mark.anyio async def test_verify_credentials_bad_auth_scheme_raises(): """verify_credentials raises AuthSchemeError for unrecognised URLs.""" from lfx.services.adapters.deployment.exceptions import AuthSchemeError from lfx.services.adapters.deployment.schema import VerifyCredentials svc = WatsonxOrchestrateDeploymentService(settings_service=DummySettingsService()) payload = VerifyCredentials( base_url="https://unknown-provider.example.com", provider_data={"api_key": "some-key"}, # pragma: allowlist secret ) with pytest.raises(AuthSchemeError): await svc.verify_credentials(user_id="u1", payload=payload) @pytest.mark.anyio async def test_verify_credentials_authenticator_construction_unexpected_error(monkeypatch): """verify_credentials raises DeploymentError when get_authenticator() throws unexpectedly.""" from lfx.services.adapters.deployment.exceptions import DeploymentError from lfx.services.adapters.deployment.schema import VerifyCredentials def _exploding_authenticator(**_kwargs): msg = "something unexpected" raise RuntimeError(msg) monkeypatch.setattr( service_module, "get_authenticator", _exploding_authenticator, ) svc = WatsonxOrchestrateDeploymentService(settings_service=DummySettingsService()) payload = VerifyCredentials( base_url="https://api.us-south.wxo.cloud.ibm.com", provider_data={"api_key": "some-key"}, # pragma: allowlist secret ) with pytest.raises(DeploymentError, match="failed unexpectedly"): await svc.verify_credentials(user_id="u1", payload=payload) @pytest.mark.anyio async def test_verify_credentials_forbidden_key_raises(monkeypatch): """verify_credentials raises AuthorizationError when provider returns 403.""" from ibm_cloud_sdk_core import ApiException from lfx.services.adapters.deployment.exceptions import AuthorizationError from lfx.services.adapters.deployment.schema import VerifyCredentials class FakeTokenManager: def get_token(self): raise ApiException(403, message="forbidden") class ForbiddenAuthenticator: token_manager = FakeTokenManager() monkeypatch.setattr( service_module, "get_authenticator", lambda **_kwargs: ForbiddenAuthenticator(), ) svc = WatsonxOrchestrateDeploymentService(settings_service=DummySettingsService()) payload = VerifyCredentials( base_url="https://api.us-south.wxo.cloud.ibm.com", provider_data={"api_key": "some-key"}, # pragma: allowlist secret ) with pytest.raises(AuthorizationError, match="Credential verification"): await svc.verify_credentials(user_id="u1", payload=payload) @pytest.mark.anyio async def test_verify_credentials_provider_unreachable(monkeypatch): """verify_credentials raises DeploymentError on network failure.""" from lfx.services.adapters.deployment.exceptions import DeploymentError from lfx.services.adapters.deployment.schema import VerifyCredentials class FakeTokenManager: def get_token(self): msg = "connection refused" raise ConnectionError(msg) class UnreachableAuthenticator: token_manager = FakeTokenManager() monkeypatch.setattr( service_module, "get_authenticator", lambda **_kwargs: UnreachableAuthenticator(), ) svc = WatsonxOrchestrateDeploymentService(settings_service=DummySettingsService()) payload = VerifyCredentials( base_url="https://api.us-south.wxo.cloud.ibm.com", provider_data={"api_key": "some-key"}, # pragma: allowlist secret ) with pytest.raises(DeploymentError, match="failed unexpectedly"): await svc.verify_credentials(user_id="u1", payload=payload) # --------------------------------------------------------------------------- # Ownership checks: binding.langflow verification # --------------------------------------------------------------------------- def _make_langflow_tool(tool_id: str, *, connections: dict[str, str] | None = None) -> dict[str, Any]: """Build a tool dict that looks Langflow-managed (has binding.langflow).""" return { "id": tool_id, "name": f"tool_{tool_id}", "binding": { "langflow": { "project_id": "proj-1", "connections": connections or {}, } }, } def _make_external_tool(tool_id: str) -> dict[str, Any]: """Build a tool dict that is NOT Langflow-managed (no binding.langflow).""" return { "id": tool_id, "name": f"external_{tool_id}", "binding": {"some_other_platform": {}}, } def _make_unbound_tool(tool_id: str) -> dict[str, Any]: """Build a tool dict with no binding at all.""" return {"id": tool_id, "name": f"bare_{tool_id}"} @pytest.mark.anyio async def test_update_connection_deltas_rejects_non_langflow_tool(): """_update_existing_tools must refuse to modify tools without binding.langflow.""" _update_deltas = update_core_module._update_existing_tools external_tool = _make_external_tool("ext-1") clients = FakeWXOClients(tool=FakeToolClient([external_tool])) ops = ToolConnectionOps(bind=OrderedUniqueStrs.from_values(["app-1"])) with pytest.raises(InvalidContentError, match="does not have a Langflow binding"): await _update_deltas( clients=clients, existing_tool_deltas={"ext-1": ops}, tool_renames={}, resolved_connections={"app-1": "conn-1"}, operation_to_provider_app_id={"app-1": "app-1"}, original_tools={}, tool_by_id={"ext-1": external_tool}, ) @pytest.mark.anyio async def test_update_connection_deltas_accepts_langflow_tool(): """_update_existing_tools succeeds for tools with binding.langflow.""" _update_deltas = update_core_module._update_existing_tools lf_tool = _make_langflow_tool("lf-1") clients = FakeWXOClients(tool=FakeToolClient([lf_tool])) ops = ToolConnectionOps(bind=OrderedUniqueStrs.from_values(["app-1"])) original_tools: dict[str, dict] = {} await _update_deltas( clients=clients, existing_tool_deltas={"lf-1": ops}, tool_renames={}, resolved_connections={"app-1": "conn-1"}, operation_to_provider_app_id={"app-1": "app-1"}, original_tools=original_tools, tool_by_id={"lf-1": lf_tool}, ) assert "lf-1" in original_tools assert clients.tool.update_calls @pytest.mark.anyio async def test_bind_existing_tools_for_create_rejects_non_langflow_tool(): """_bind_existing_tools_for_create must refuse to modify tools without binding.langflow.""" _bind_existing = create_core_module._bind_existing_tools_for_create external_tool = _make_external_tool("ext-1") clients = FakeWXOClients(tool=FakeToolClient([external_tool])) with pytest.raises(InvalidContentError, match="does not have a Langflow binding"): await _bind_existing( clients=clients, existing_tool_bindings={"ext-1": ["app-1"]}, operation_to_provider_app_id={"app-1": "app-1"}, resolved_connections={"app-1": "conn-1"}, original_tools={}, ) @pytest.mark.anyio async def test_bind_existing_tools_for_create_accepts_langflow_tool(): """_bind_existing_tools_for_create succeeds for tools with binding.langflow.""" _bind_existing = create_core_module._bind_existing_tools_for_create lf_tool = _make_langflow_tool("lf-1") clients = FakeWXOClients(tool=FakeToolClient([lf_tool])) original_tools: dict[str, dict] = {} await _bind_existing( clients=clients, existing_tool_bindings={"lf-1": ["app-1"]}, operation_to_provider_app_id={"app-1": "app-1"}, resolved_connections={"app-1": "conn-1"}, original_tools=original_tools, ) assert "lf-1" in original_tools assert clients.tool.update_calls @pytest.mark.anyio async def test_update_existing_tool_connection_bindings_rejects_non_langflow_tool(): """update_existing_tool_connection_bindings must refuse to modify tools without binding.langflow.""" _update_bindings = tools_module.update_existing_tool_connection_bindings external_tool = _make_external_tool("ext-1") clients = FakeWXOClients(tool=FakeToolClient([external_tool])) with pytest.raises(InvalidContentError, match="does not have a Langflow binding"): await _update_bindings( clients=clients, existing_target_tool_ids=["ext-1"], resolved_connections={"app-1": "conn-1"}, original_tools={}, ) @pytest.mark.anyio async def test_update_existing_tool_connection_bindings_rejects_unbound_tool(): """update_existing_tool_connection_bindings must refuse tools with no binding at all.""" _update_bindings = tools_module.update_existing_tool_connection_bindings bare_tool = _make_unbound_tool("bare-1") clients = FakeWXOClients(tool=FakeToolClient([bare_tool])) with pytest.raises(InvalidContentError, match="does not have a Langflow binding"): await _update_bindings( clients=clients, existing_target_tool_ids=["bare-1"], resolved_connections={"app-1": "conn-1"}, original_tools={}, ) # --------------------------------------------------------------------------- # Tool rename safety # --------------------------------------------------------------------------- @pytest.mark.anyio async def test_apply_tool_renames_succeeds_for_langflow_tool(): """_update_existing_tools renames a Langflow-owned tool on the agent.""" _apply_renames = update_core_module._update_existing_tools lf_tool = _make_langflow_tool("lf-1") clients = FakeWXOClients(tool=FakeToolClient([lf_tool])) original_tools: dict[str, dict] = {} await _apply_renames( clients=clients, existing_tool_deltas={}, tool_renames={"lf-1": "New Name"}, resolved_connections={}, operation_to_provider_app_id={}, original_tools=original_tools, tool_by_id={"lf-1": lf_tool}, ) assert clients.tool.update_calls tool_id, payload = clients.tool.update_calls[0] assert tool_id == "lf-1" _assert_langflow_agent_name(payload["name"], display_name="New Name") assert payload["display_name"] == "New Name" assert "lf-1" in original_tools @pytest.mark.anyio async def test_apply_tool_renames_uses_loaded_tool_without_fetching(): """Rename should reuse the update-apply tool map instead of fetching the tool again.""" _apply_renames = update_core_module._update_existing_tools lf_tool = _make_langflow_tool("lf-1") fake_tool = FakeToolClient([]) clients = FakeWXOClients(tool=fake_tool) await _apply_renames( clients=clients, existing_tool_deltas={}, tool_renames={"lf-1": "New Name"}, resolved_connections={}, operation_to_provider_app_id={}, original_tools={}, tool_by_id={"lf-1": lf_tool}, ) assert fake_tool.get_drafts_by_ids_calls == [] assert fake_tool.update_calls @pytest.mark.anyio async def test_apply_tool_renames_rejects_non_langflow_tool(): """_update_existing_tools must refuse to rename tools without binding.langflow.""" _apply_renames = update_core_module._update_existing_tools external_tool = _make_external_tool("ext-1") clients = FakeWXOClients(tool=FakeToolClient([external_tool])) with pytest.raises(InvalidContentError, match="does not have a Langflow binding"): await _apply_renames( clients=clients, existing_tool_deltas={}, tool_renames={"ext-1": "stolen_name"}, resolved_connections={}, operation_to_provider_app_id={}, original_tools={}, tool_by_id={"ext-1": external_tool}, ) assert not clients.tool.update_calls @pytest.mark.anyio async def test_apply_tool_renames_rejects_missing_tool(): """_update_existing_tools must fail if tool doesn't exist on provider.""" _apply_renames = update_core_module._update_existing_tools clients = FakeWXOClients(tool=FakeToolClient([])) with pytest.raises(InvalidContentError, match="not found in provider"): await _apply_renames( clients=clients, existing_tool_deltas={}, tool_renames={"ghost-1": "New Name"}, resolved_connections={}, operation_to_provider_app_id={}, original_tools={}, tool_by_id={}, ) @pytest.mark.anyio async def test_apply_tool_renames_captures_original_for_rollback(): """_update_existing_tools must capture original payload before renaming for rollback.""" _apply_renames = update_core_module._update_existing_tools lf_tool = _make_langflow_tool("lf-1") lf_tool["name"] = "original_name" lf_tool["display_name"] = "original_name" clients = FakeWXOClients(tool=FakeToolClient([lf_tool])) original_tools: dict[str, dict] = {} await _apply_renames( clients=clients, existing_tool_deltas={}, tool_renames={"lf-1": "New Name"}, resolved_connections={}, operation_to_provider_app_id={}, original_tools=original_tools, tool_by_id={"lf-1": lf_tool}, ) assert original_tools["lf-1"]["name"] == "original_name" @pytest.mark.anyio async def test_apply_tool_renames_preserves_latest_connections_when_original_already_captured(): """Rename should keep connection updates already applied earlier in the transaction.""" _apply_renames = update_core_module._update_existing_tools lf_tool = _make_langflow_tool("lf-1", connections={"app-1": "conn-1", "app-2": "conn-2"}) lf_tool["name"] = "current_name" lf_tool["display_name"] = "current_name" clients = FakeWXOClients(tool=FakeToolClient([lf_tool])) # Simulate pre-captured rollback payload from a prior connection-delta step. original_tools: dict[str, dict] = { "lf-1": { "id": "lf-1", "name": "pre_delta_name", "display_name": "pre_delta_name", "binding": {"langflow": {"project_id": "proj-1", "connections": {"app-1": "conn-1"}}}, } } await _apply_renames( clients=clients, existing_tool_deltas={}, tool_renames={"lf-1": "New Name"}, resolved_connections={}, operation_to_provider_app_id={}, original_tools=original_tools, tool_by_id={"lf-1": lf_tool}, ) _, payload = clients.tool.update_calls[0] _assert_langflow_agent_name(payload["name"], display_name="New Name") assert payload["display_name"] == "New Name" assert payload["binding"]["langflow"]["connections"] == {"app-1": "conn-1", "app-2": "conn-2"} # Pre-captured rollback state must remain unchanged. assert original_tools["lf-1"]["name"] == "pre_delta_name" assert original_tools["lf-1"]["binding"]["langflow"]["connections"] == {"app-1": "conn-1"} @pytest.mark.anyio async def test_apply_tool_renames_preserves_latest_connections_for_add_and_remove_delta(): """Rename should preserve the latest provider connections after mixed add/remove updates.""" _apply_renames = update_core_module._update_existing_tools # Simulate post-delta provider state (one app removed, one app added). lf_tool = _make_langflow_tool("lf-1", connections={"cfg-keep": "conn-keep", "cfg-add": "conn-add"}) lf_tool["name"] = "current_name" lf_tool["display_name"] = "current_name" clients = FakeWXOClients(tool=FakeToolClient([lf_tool])) # Simulate pre-delta rollback snapshot captured earlier. original_tools: dict[str, dict] = { "lf-1": { "id": "lf-1", "name": "pre_delta_name", "display_name": "pre_delta_name", "binding": { "langflow": { "project_id": "proj-1", "connections": {"cfg-keep": "conn-keep", "cfg-remove": "conn-remove"}, } }, } } await _apply_renames( clients=clients, existing_tool_deltas={}, tool_renames={"lf-1": "Renamed Tool"}, resolved_connections={}, operation_to_provider_app_id={}, original_tools=original_tools, tool_by_id={"lf-1": lf_tool}, ) _, payload = clients.tool.update_calls[0] _assert_langflow_agent_name(payload["name"], display_name="Renamed Tool") assert payload["display_name"] == "Renamed Tool" assert payload["binding"]["langflow"]["connections"] == {"cfg-keep": "conn-keep", "cfg-add": "conn-add"} # Rollback snapshot remains pre-delta. assert original_tools["lf-1"]["binding"]["langflow"]["connections"] == { "cfg-keep": "conn-keep", "cfg-remove": "conn-remove", } # --------------------------------------------------------------------------- # Tool technical-name generation # --------------------------------------------------------------------------- def test_flow_artifact_provider_data_generates_provider_technical_name(): """WatsonxFlowArtifactProviderData derives a provider-valid technical name from a display label.""" provider_data = payloads_module.WatsonxFlowArtifactProviderData( project_id="project-1", source_ref="fv-1", tool_display_name="My Flow", ) _assert_langflow_agent_name(provider_data.tool_name, display_name="My Flow") def test_flow_artifact_provider_data_uses_resource_fallback_for_symbol_only_label(): """Symbol-only display labels still get a provider-valid generated name.""" provider_data = payloads_module.WatsonxFlowArtifactProviderData( project_id="project-1", source_ref="fv-1", tool_display_name="!@#$%", ) assert provider_data.tool_name.startswith("langflow_tool_") assert ( payloads_module.validate_wxo_name(provider_data.tool_name, field_label="Tool name") == provider_data.tool_name ) def test_flow_artifact_provider_data_accepts_leading_digit_display_label(): """Display labels are not validated as provider technical names.""" provider_data = payloads_module.WatsonxFlowArtifactProviderData( project_id="project-1", source_ref="fv-1", tool_display_name="123flow", ) assert provider_data.tool_name.startswith("langflow_123flow_") assert ( payloads_module.validate_wxo_name(provider_data.tool_name, field_label="Tool name") == provider_data.tool_name ) def test_flow_artifact_provider_data_validates_explicit_provider_technical_name(): provider_data = payloads_module.WatsonxFlowArtifactProviderData( project_id="project-1", source_ref="fv-1", tool_name="explicit_tool_name", tool_display_name="My Flow", ) assert provider_data.tool_name == "explicit_tool_name" def test_flow_artifact_provider_data_requires_display_name(): with pytest.raises(ValidationError, match="tool_display_name"): payloads_module.WatsonxFlowArtifactProviderData( project_id="project-1", source_ref="fv-1", tool_name="explicit_tool_name", ) def test_flow_artifact_provider_data_rejects_invalid_explicit_provider_technical_name(): with pytest.raises(ValidationError, match="Tool name must start with a letter"): payloads_module.WatsonxFlowArtifactProviderData( project_id="project-1", source_ref="fv-1", tool_name="123_tool_name", tool_display_name="My Flow", ) # --------------------------------------------------------------------------- # Rename operation in plan builder # --------------------------------------------------------------------------- def test_build_update_plan_includes_rename(): """build_provider_update_plan collects rename_tool operations into tool_renames.""" build_plan = update_core_module.build_provider_update_plan agent = {"id": "agent-1", "tools": ["tool-1"]} payload = WatsonxDeploymentUpdatePayload( llm=TEST_WXO_LLM, operations=[ { "op": "rename_tool", "tool": {"source_ref": "fv-1", "tool_id": "tool-1"}, "tool_display_name": "Better Name", }, ], ) plan = build_plan(agent=agent, provider_update=payload) assert plan.tool_renames == {"tool-1": "Better Name"} def test_build_update_plan_without_renames_has_empty_dict(): """build_provider_update_plan returns empty tool_renames when no renames present.""" build_plan = update_core_module.build_provider_update_plan agent = {"id": "agent-1", "tools": ["tool-1"]} payload = WatsonxDeploymentUpdatePayload( llm=TEST_WXO_LLM, operations=[ { "op": "remove_tool", "tool": {"source_ref": "fv-1", "tool_id": "tool-1"}, }, ], ) plan = build_plan(agent=agent, provider_update=payload) assert plan.tool_renames == {} # --------------------------------------------------------------------------- # WXO_LFX_REQUIREMENT_OVERRIDE # --------------------------------------------------------------------------- def test_resolve_lfx_requirement_uses_override(monkeypatch): """_resolve_lfx_requirement returns the env var value when set.""" _resolve = tools_module._resolve_lfx_requirement monkeypatch.setenv("WXO_LFX_REQUIREMENT_OVERRIDE", "lfx-nightly==0.4.0.dev32") assert _resolve() == "lfx-nightly==0.4.0.dev32" def test_resolve_lfx_requirement_ignores_blank_override(monkeypatch): """_resolve_lfx_requirement ignores empty/whitespace-only override.""" _resolve = tools_module._resolve_lfx_requirement monkeypatch.setenv("WXO_LFX_REQUIREMENT_OVERRIDE", " ") # Should not return blank — should fall through to installed version or raise result = _resolve() assert result.strip() assert result != " " def test_rename_tool_provider_payload_parses(): """WatsonxRenameToolOperation parses correctly at provider level.""" op = WatsonxRenameToolOperation( op="rename_tool", tool={"source_ref": "fv-1", "tool_id": "tool-1"}, tool_display_name="Better Name", ) assert op.tool_display_name == "Better Name" assert op.tool.tool_id == "tool-1" # --------------------------------------------------------------------------- # DeploymentFlowVersionListItem carries provider tool names under provider_data # --------------------------------------------------------------------------- def test_flow_version_list_item_includes_tool_names_in_provider_data(): """DeploymentFlowVersionListItem serializes provider tool metadata under provider_data.""" from langflow.api.v1.schemas.deployments import DeploymentFlowVersionListItem item = DeploymentFlowVersionListItem( id="00000000-0000-0000-0000-000000000001", flow_id="00000000-0000-0000-0000-000000000002", flow_name="My Flow", version_number=1, provider_data={"tool_name": "my_custom_tool", "tool_display_name": "My Custom Tool"}, ) data = item.model_dump() assert data["provider_data"]["tool_name"] == "my_custom_tool" assert data["provider_data"]["tool_display_name"] == "My Custom Tool" def test_flow_version_list_item_provider_data_defaults_to_none(): """DeploymentFlowVersionListItem defaults provider_data to None.""" from langflow.api.v1.schemas.deployments import DeploymentFlowVersionListItem item = DeploymentFlowVersionListItem( id="00000000-0000-0000-0000-000000000001", flow_id="00000000-0000-0000-0000-000000000002", version_number=1, ) assert item.provider_data is None