diff --git a/skills/skills/langbot-mcp-ops/SKILL.md b/skills/skills/langbot-mcp-ops/SKILL.md index a9d723de3..330d330b6 100644 --- a/skills/skills/langbot-mcp-ops/SKILL.md +++ b/skills/skills/langbot-mcp-ops/SKILL.md @@ -60,6 +60,12 @@ To inspect key identity and permissions, call `GET /api/v1/system/context` with ## Tool surface +`create_bot` requires `bot_data.name` and `bot_data.adapter` as non-empty strings, +and `bot_data.adapter_config` as an object with the adapter's settings. +For `update_pipeline`, fetch the current pipeline before editing: `config` is a +complete replacement containing `trigger`, `safety`, `ai`, and `output` objects, +not a partial patch. Invalid configurations are rejected before persistence. + Slack quick setup: create a disabled `slack-omni` bot draft first, then use `start_slack_setup` with the user's App Configuration access/refresh tokens. Choose `socket_mode=true` with a browser-reachable OAuth `redirect_url`, or diff --git a/src/langbot/pkg/api/application/operation_docs.py b/src/langbot/pkg/api/application/operation_docs.py index b2f882070..a1b08f575 100644 --- a/src/langbot/pkg/api/application/operation_docs.py +++ b/src/langbot/pkg/api/application/operation_docs.py @@ -245,7 +245,7 @@ OPERATION_DOCS: dict[str, OperationDoc] = { domain='bot', label={'zh_Hans': '创建机器人', 'en_US': 'Create Bot'}, summary='Create a new chat bot connected to an IM adapter.', - description='Create and configure a new chat bot in the workspace. Provide bot_data containing name, adapter settings, and optional pipeline binding. Side effect: Instantiates a new bot entity.', + description='Create a chat bot. bot_data requires non-empty name and adapter strings, and an adapter_config object with adapter-specific settings. Optional event_bindings routes events to a pipeline or agent. Side effect: Instantiates a new bot entity.', risk='write', exposure='all', ), @@ -997,7 +997,7 @@ OPERATION_DOCS: dict[str, OperationDoc] = { domain='pipeline', label={'zh_Hans': '更新流水线', 'en_US': 'Update Pipeline'}, summary='Update topology, nodes, or settings of a pipeline.', - description='Modify pipeline nodes, model assignments, tool attachments, or name. Prerequisites: Call pipeline.list to obtain pipeline_uuid. Side effect: Updates pipeline definition.', + description='Modify pipeline settings or name. Call pipeline.get first: config, when supplied, replaces the complete configuration and must contain trigger, safety, ai, and output objects. Partial config is rejected without writing. Runtime initialization is checked before persistence. Side effect: Updates pipeline definition.', risk='write', exposure='all', ), diff --git a/src/langbot/pkg/api/application/service_operations.py b/src/langbot/pkg/api/application/service_operations.py index 9c04a1430..cfeb30b2d 100644 --- a/src/langbot/pkg/api/application/service_operations.py +++ b/src/langbot/pkg/api/application/service_operations.py @@ -834,6 +834,15 @@ def _contract(operation_id, method): annotation = annotation | None default = ... if default is inspect.Parameter.empty else default constraints = {} + if operation_id == 'bot.create' and name == 'bot_data': + constraints['json_schema_extra'] = { + 'required': ['name', 'adapter', 'adapter_config'], + 'properties': { + 'name': {'type': 'string', 'minLength': 1}, + 'adapter': {'type': 'string', 'minLength': 1}, + 'adapter_config': {'type': 'object'}, + }, + } if name in {'limit', 'max_count', 'max_entries'}: constraints = {'ge': 1, 'le': 100_000 if '.export.' in operation_id else 1000} elif name in {'offset', 'from_index', 'after_sequence', 'before_id'}: diff --git a/src/langbot/pkg/api/http/service/bot.py b/src/langbot/pkg/api/http/service/bot.py index 506fb4de1..7aac59081 100644 --- a/src/langbot/pkg/api/http/service/bot.py +++ b/src/langbot/pkg/api/http/service/bot.py @@ -759,6 +759,13 @@ class BotService: async def create_bot(self, context: TenantContext, bot_data: dict) -> str: """Create bot""" workspace_uuid = require_workspace_uuid(context) + if not isinstance(bot_data, dict): + raise ValueError('bot_data must be an object') + for field in ('name', 'adapter'): + if not isinstance(bot_data.get(field), str) or not bot_data[field].strip(): + raise ValueError(f'bot_data.{field} is required and must be a non-empty string') + if not isinstance(bot_data.get('adapter_config'), dict): + raise ValueError('bot_data.adapter_config is required and must be an object') # Check limitation from .workspace_limits import workspace_limits @@ -769,7 +776,6 @@ class BotService: if len(existing_bots) >= max_bots: raise ValueError(f'Maximum number of bots ({max_bots}) reached') - # TODO: 检查配置信息格式 bot_data = await self._prepare_bot_data(context, bot_data, include_uuid=True) await require_workspace_platform(self.ap, workspace_uuid, bot_data['adapter']) bot_data['uuid'] = str(uuid.uuid4()) diff --git a/src/langbot/pkg/api/http/service/pipeline.py b/src/langbot/pkg/api/http/service/pipeline.py index 650adad60..cc0afcada 100644 --- a/src/langbot/pkg/api/http/service/pipeline.py +++ b/src/langbot/pkg/api/http/service/pipeline.py @@ -4,6 +4,7 @@ import uuid import json import sqlalchemy import typing +import copy from ....core import app from ....agent.runner.config_resolver import RunnerConfigResolver @@ -229,10 +230,16 @@ class PipelineService: for protected_field in ('uuid', 'workspace_uuid', 'for_version', 'stages', 'is_default'): pipeline_data.pop(protected_field, None) + current_pipeline = await self.get_pipeline(context, pipeline_uuid, include_secret=True) + if current_pipeline is None: + raise WorkspaceNotFoundError('Pipeline not found') if 'config' in pipeline_data: - current_pipeline = await self.get_pipeline(context, pipeline_uuid, include_secret=True) - if current_pipeline is None: - raise WorkspaceNotFoundError('Pipeline not found') + config = pipeline_data['config'] + if not isinstance(config, dict): + raise ValueError('config must be an object containing trigger, safety, ai, and output') + for section in ('trigger', 'safety', 'ai', 'output'): + if not isinstance(config.get(section), dict): + raise ValueError(f'config.{section} must be an object; config replaces the complete configuration') current_config = current_pipeline.get('config', {}) old_ai = current_config.get('ai', {}) if isinstance(current_config, dict) else {} old_runner = old_ai.get('runner') if isinstance(old_ai, dict) else None @@ -268,6 +275,13 @@ class PipelineService: new_prefs['workspace_operations'] = old_ops self._validate_extension_preferences(pipeline_data['extensions_preferences']) + candidate_data = copy.deepcopy(current_pipeline) + candidate_data.update(copy.deepcopy(pipeline_data)) + try: + candidate = await self.ap.pipeline_mgr.prepare_pipeline(context, candidate_data) + except (KeyError, TypeError, ValueError) as exc: + raise ValueError('Invalid pipeline configuration: runtime initialization failed') from exc + result = await self.ap.persistence_mgr.execute_async( scope_statement( sqlalchemy.update(persistence_pipeline.LegacyPipeline) @@ -280,12 +294,7 @@ class PipelineService: if getattr(result, 'rowcount', None) == 0: raise WorkspaceNotFoundError('Pipeline not found') - pipeline = await self.get_pipeline(context, pipeline_uuid, include_secret=True) - if pipeline is None: - raise WorkspaceNotFoundError('Pipeline not found') - - await self.ap.pipeline_mgr.remove_pipeline(context, pipeline_uuid) - await self.ap.pipeline_mgr.load_pipeline(context, pipeline) + self.ap.pipeline_mgr.publish_pipeline(candidate) # update all conversation that use this pipeline for session in self.ap.sess_mgr.session_list: diff --git a/src/langbot/pkg/api/mcp/server.py b/src/langbot/pkg/api/mcp/server.py index 7d680495d..979727100 100644 --- a/src/langbot/pkg/api/mcp/server.py +++ b/src/langbot/pkg/api/mcp/server.py @@ -357,7 +357,8 @@ class LangBotMCPServer: @mcp.tool( description=( 'Create a bot. Cloud Workspace platform restrictions also apply to this tool. `bot_data` is a JSON object matching the LangBot ' - 'POST /api/v1/platform/bots body (e.g. name, adapter, config). ' + 'POST /api/v1/platform/bots body: name and adapter are required non-empty strings; ' + 'adapter_config is a required object containing adapter-specific settings. ' 'Use event_bindings for exclusive Agent/Pipeline routes; plugin_processors is an array ' 'of {processor_uuid, enabled} subscriptions that automatically match Runner-declared events. ' 'Subscriptions run independently of each other and of event_bindings. Returns the new bot UUID.' @@ -449,7 +450,9 @@ class LangBotMCPServer: } ) - @mcp.tool(description='Update a pipeline by UUID. `pipeline_data` matches the PUT body.') + @mcp.tool( + description='Update a pipeline by UUID. `pipeline_data` matches the PUT body. Fetch the pipeline first: config replaces the complete configuration and requires trigger, safety, ai, and output objects. Invalid config is rejected before writing.' + ) async def update_pipeline(pipeline_uuid: str, pipeline_data: dict) -> str: context = _authorized(Permission.RESOURCE_MANAGE) await ap.pipeline_service.update_pipeline(context, pipeline_uuid, pipeline_data) diff --git a/tests/integration/api/test_pipeline_bot_write_validation.py b/tests/integration/api/test_pipeline_bot_write_validation.py new file mode 100644 index 000000000..382653257 --- /dev/null +++ b/tests/integration/api/test_pipeline_bot_write_validation.py @@ -0,0 +1,312 @@ +import copy +import asyncio +import json +import logging +import socket +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +import quart +import sqlalchemy as sa +import aiohttp +from hypercorn.asyncio import serve +from hypercorn.config import Config +from sqlalchemy.ext.asyncio import create_async_engine + +from langbot.pkg.api.application import ApplicationAPI, ApplicationAPIException +from langbot.pkg.api.http.controller.groups.pipelines.pipelines import PipelinesRouterGroup +from langbot.pkg.api.http.controller.groups.platform.bots import BotsRouterGroup +from langbot.pkg.api.http.service.bot import BotService +from langbot.pkg.api.http.service.pipeline import PipelineService +from langbot.pkg.entity.persistence.bot import Bot +from langbot.pkg.entity.persistence.base import Base +from langbot.pkg.entity.persistence.metadata import WorkspaceMetadata +from langbot.pkg.entity.persistence.pipeline import LegacyPipeline +from langbot.pkg.entity.persistence.workspace import Workspace, WorkspaceExecutionState +from langbot.pkg.persistence.mgr import PersistenceManager +from langbot.pkg.pipeline.cntfilter.cntfilter import ContentFilterStage +from langbot.pkg.pipeline.pipelinemgr import PipelineManager +from langbot.pkg.utils.paths import get_resource_path +from langbot.pkg.workspace.service import WorkspaceService +from tests.integration.api.test_application_foundation import context + + +@pytest.fixture +async def write_foundation(tmp_path): + engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "writes.db"}') + try: + async with engine.begin() as connection: + await connection.run_sync( + lambda sync: Base.metadata.create_all( + sync, + tables=[ + model.__table__ + for model in (Workspace, WorkspaceExecutionState, WorkspaceMetadata, LegacyPipeline, Bot) + ], + ) + ) + await connection.execute( + sa.insert(Workspace).values( + uuid='workspace-a', + instance_uuid='foundation-instance', + name='Test', + slug='test', + source='cloud_projection', + ) + ) + await connection.execute( + sa.insert(WorkspaceExecutionState).values( + workspace_uuid='workspace-a', + instance_uuid='foundation-instance', + active_generation=1, + source='cloud', + ) + ) + ap = SimpleNamespace(instance_config=SimpleNamespace(data={}), logger=logging.getLogger('write-validation')) + ap.persistence_mgr = PersistenceManager(ap) + ap.persistence_mgr.db = SimpleNamespace(get_engine=lambda: engine) + ap.workspace_service = WorkspaceService(ap, instance_uuid='foundation-instance') + yield ApplicationAPI(ap), ap, engine + finally: + await engine.dispose() + + +@pytest.fixture +async def writes(write_foundation): + api, ap, engine = write_foundation + ap.pipeline_service = PipelineService(ap) + ap.bot_service = BotService(ap) + ap.pipeline_mgr = PipelineManager(ap) + ap.pipeline_mgr.stage_dict = {'ContentFilterStage': ContentFilterStage} + ap.sess_mgr = SimpleNamespace(session_list=[]) + with open(get_resource_path('templates/default-pipeline-config.json')) as stream: + config = json.load(stream) + async with engine.begin() as connection: + await connection.execute( + sa.insert(LegacyPipeline).values( + uuid='pipeline-a', + workspace_uuid='workspace-a', + name='Original', + description='', + for_version='test', + stages=['ContentFilterStage'], + config=config, + extensions_preferences={}, + is_default=True, + ) + ) + async with ap.persistence_mgr.tenant_scope('workspace-a'): + pipeline = await ap.pipeline_service.get_pipeline(context(), 'pipeline-a', include_secret=True) + await ap.pipeline_mgr.load_pipeline(context(), pipeline) + ap.user_service = SimpleNamespace( + get_authenticated_account=AsyncMock(return_value=SimpleNamespace(uuid='account-test', user='test@example.com')) + ) + ap.apikey_service = SimpleNamespace(authenticate_api_key=AsyncMock(return_value=None)) + ap.workspace_collaboration_service = SimpleNamespace( + resolve_account_workspace=AsyncMock( + return_value=SimpleNamespace( + workspace=SimpleNamespace(uuid='workspace-a'), + membership=SimpleNamespace(uuid='membership-test', role='owner', projection_revision=1), + execution=SimpleNamespace(instance_uuid='foundation-instance', placement_generation=1), + ) + ) + ) + app = quart.Quart(__name__) + await PipelinesRouterGroup(ap, app).initialize() + await BotsRouterGroup(ap, app).initialize() + return api, ap, engine, app.test_client(), config + + +async def stored_pipeline(engine): + async with engine.connect() as connection: + return dict((await connection.execute(sa.select(LegacyPipeline))).mappings().one()) + + +@pytest.mark.asyncio +@pytest.mark.parametrize('transport', ['http', 'application', 'service']) +@pytest.mark.parametrize('invalid', ['partial', 'runtime', 'null']) +async def test_invalid_pipeline_preserves_database_and_runtime(writes, transport, invalid): + api, ap, engine, client, config = writes + original = await stored_pipeline(engine) + runtime = ap.pipeline_mgr.pipelines[0] + if invalid == 'partial': + config = {'ai': config['ai']} + elif invalid == 'runtime': + config['safety'] = {} + else: + config = None + payload = {'name': 'Must not persist', 'config': config} + if transport == 'http': + response = await client.put( + '/api/v1/pipelines/pipeline-a', json=payload, headers={'Authorization': 'Bearer test-token'} + ) + assert response.status_code == 400, await response.get_json() + print('pipeline.update', invalid, response.status_code, await response.get_json()) + elif transport == 'application': + with pytest.raises(ApplicationAPIException) as failure: + await api.execute('pipeline.update', context(), {'pipeline_uuid': 'pipeline-a', 'pipeline_data': payload}) + assert failure.value.error.status_code == 400 + else: + async with ap.persistence_mgr.tenant_scope('workspace-a'): + with pytest.raises(ValueError): + await ap.pipeline_service.update_pipeline(context(), 'pipeline-a', payload) + assert await stored_pipeline(engine) == original + assert ap.pipeline_mgr.pipelines == [runtime] + + +@pytest.mark.asyncio +async def test_valid_pipeline_replaces_runtime_after_write(writes): + _, ap, engine, client, config = writes + previous = ap.pipeline_mgr.pipelines[0] + response = await client.put( + '/api/v1/pipelines/pipeline-a', + json={'name': 'Updated', 'config': config}, + headers={'Authorization': 'Bearer test-token'}, + ) + assert response.status_code == 200, await response.get_json() + assert (await stored_pipeline(engine))['name'] == 'Updated' + assert ap.pipeline_mgr.pipelines[0] is not previous + assert ap.pipeline_mgr.pipelines[0].pipeline_entity.name == 'Updated' + + +@pytest.mark.asyncio +async def test_failed_pipeline_write_does_not_publish_candidate(writes, monkeypatch): + _, ap, engine, _, config = writes + original = await stored_pipeline(engine) + runtime = ap.pipeline_mgr.pipelines[0] + execute = ap.persistence_mgr.execute_async + + async def fail_update(statement, *args, **kwargs): + if isinstance(statement, sa.sql.dml.Update): + raise RuntimeError('Database unavailable') + return await execute(statement, *args, **kwargs) + + monkeypatch.setattr(ap.persistence_mgr, 'execute_async', fail_update) + async with ap.persistence_mgr.tenant_scope('workspace-a'): + with pytest.raises(RuntimeError, match='Database unavailable'): + await ap.pipeline_service.update_pipeline(context(), 'pipeline-a', {'config': config, 'name': 'Unsaved'}) + assert await stored_pipeline(engine) == original + assert ap.pipeline_mgr.pipelines == [runtime] + + +@pytest.mark.asyncio +@pytest.mark.parametrize('transport', ['http', 'application', 'service']) +@pytest.mark.parametrize( + ('field', 'value'), + [ + ('adapter_config', 'missing'), + ('adapter_config', None), + ('adapter_config', []), + ('adapter', 'missing'), + ('name', 'missing'), + ('adapter', ''), + ], +) +async def test_invalid_bot_has_no_database_side_effects(writes, transport, field, value): + api, ap, engine, client, _ = writes + pipeline = await stored_pipeline(engine) + payload = {'name': 'Bot', 'adapter': 'aiocqhttp', 'adapter_config': {}} + if value == 'missing': + payload.pop(field) + else: + payload[field] = value + original = copy.deepcopy(payload) + if transport == 'http': + response = await client.post( + '/api/v1/platform/bots', json=payload, headers={'Authorization': 'Bearer test-token'} + ) + assert response.status_code == 400, await response.get_json() + assert field in (await response.get_json())['msg'] + print('bot.create', field, response.status_code, await response.get_json()) + elif transport == 'application': + with pytest.raises(ApplicationAPIException) as failure: + await api.execute('bot.create', context(), {'bot_data': payload}) + assert failure.value.error.status_code == 400 + assert field in failure.value.error.message + else: + async with ap.persistence_mgr.tenant_scope('workspace-a'): + with pytest.raises(ValueError, match=field): + await ap.bot_service.create_bot(context(), payload) + async with engine.connect() as connection: + assert (await connection.execute(sa.select(sa.func.count()).select_from(Bot))).scalar_one() == 0 + assert await stored_pipeline(engine) == pipeline + assert payload == original + + +@pytest.mark.asyncio +async def test_bot_create_schema_lists_required_fields(writes): + api, _, _, _, _ = writes + operation = next(item for item in api.operation_catalog() if item['id'] == 'bot.create') + schema = operation['input_schema']['properties']['bot_data'] + assert schema['required'] == ['name', 'adapter', 'adapter_config'] + assert schema['properties']['adapter_config']['type'] == 'object' + + +@pytest.mark.asyncio +async def test_valid_bot_binds_existing_pipeline(writes): + _, _, engine, client, _ = writes + response = await client.post( + '/api/v1/platform/bots', + json={ + 'name': 'Bound bot', + 'adapter': 'aiocqhttp', + 'adapter_config': {}, + 'enable': False, + }, + headers={'Authorization': 'Bearer test-token'}, + ) + assert response.status_code == 200, await response.get_json() + async with engine.connect() as connection: + bot = (await connection.execute(sa.select(Bot))).mappings().one() + assert bot['adapter_config'] == {} + assert all(binding['target_uuid'] == 'pipeline-a' for binding in bot['event_bindings']) + print('bot.create valid', response.status_code, await response.get_json(), 'bindings:', bot['event_bindings']) + + +@pytest.mark.asyncio +async def test_live_http_rejects_invalid_writes_without_dirty_rows(writes): + _, ap, engine, client, _ = writes + original = await stored_pipeline(engine) + runtime = ap.pipeline_mgr.pipelines[0] + listener = socket.socket() + listener.bind(('127.0.0.1', 0)) + listener.listen() + port = listener.getsockname()[1] + config = Config() + config.bind = [f'fd://{listener.detach()}'] + shutdown = asyncio.Event() + server = asyncio.create_task(serve(client.app, config, shutdown_trigger=shutdown.wait)) + try: + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=15)) as session: + for method, path, payload in ( + ('PUT', '/api/v1/pipelines/pipeline-a', {'name': 'Unsaved', 'config': {'ai': {}}}), + ('POST', '/api/v1/platform/bots', {'name': 'Invalid bot', 'adapter': 'aiocqhttp'}), + ): + async with session.request( + method, + f'http://127.0.0.1:{port}{path}', + json=payload, + headers={'Authorization': 'Bearer test-token'}, + ) as response: + body = await response.json() + print( + json.dumps( + { + 'method': method, + 'path': path, + 'request': payload, + 'status': response.status, + 'response': body, + } + ) + ) + assert response.status == 400 + assert await stored_pipeline(engine) == original + async with engine.connect() as connection: + assert (await connection.execute(sa.select(sa.func.count()).select_from(Bot))).scalar_one() == 0 + assert ap.pipeline_mgr.pipelines == [runtime] + print('Independent SQLite readback: pipeline unchanged; bots=0; original runtime retained') + finally: + shutdown.set() + await server diff --git a/tests/unit_tests/api/service/test_bot_service.py b/tests/unit_tests/api/service/test_bot_service.py index 187635475..b9b24caf9 100644 --- a/tests/unit_tests/api/service/test_bot_service.py +++ b/tests/unit_tests/api/service/test_bot_service.py @@ -369,7 +369,7 @@ class TestBotServiceCreateBot: # Execute & Verify with pytest.raises(ValueError, match='Maximum number of bots'): - await service.create_bot(WORKSPACE_UUID, {'name': 'New Bot'}) + await service.create_bot(WORKSPACE_UUID, {'name': 'New Bot', 'adapter': 'aiocqhttp', 'adapter_config': {}}) assert all(not call.args[0].is_insert for call in ap.persistence_mgr.execute_async.await_args_list) ap.platform_mgr.load_bot.assert_not_awaited() diff --git a/tests/unit_tests/api/service/test_pipeline_service.py b/tests/unit_tests/api/service/test_pipeline_service.py index 6f324c9b8..eb0fc1439 100644 --- a/tests/unit_tests/api/service/test_pipeline_service.py +++ b/tests/unit_tests/api/service/test_pipeline_service.py @@ -411,6 +411,8 @@ class TestPipelineServiceUpdatePipeline: ap.pipeline_mgr = SimpleNamespace() ap.pipeline_mgr.remove_pipeline = AsyncMock() ap.pipeline_mgr.load_pipeline = AsyncMock() + ap.pipeline_mgr.prepare_pipeline = AsyncMock() + ap.pipeline_mgr.publish_pipeline = Mock() ap.sess_mgr = SimpleNamespace() ap.sess_mgr.session_list = [] ap.bot_service = None # No bot_service when not updating name @@ -444,6 +446,8 @@ class TestPipelineServiceUpdatePipeline: ap.pipeline_mgr = SimpleNamespace() ap.pipeline_mgr.remove_pipeline = AsyncMock() ap.pipeline_mgr.load_pipeline = AsyncMock() + ap.pipeline_mgr.prepare_pipeline = AsyncMock() + ap.pipeline_mgr.publish_pipeline = Mock() ap.sess_mgr = SimpleNamespace() ap.sess_mgr.session_list = [] ap.bot_service = SimpleNamespace() @@ -497,6 +501,8 @@ class TestPipelineServiceUpdatePipeline: ap.pipeline_mgr = SimpleNamespace() ap.pipeline_mgr.remove_pipeline = AsyncMock() ap.pipeline_mgr.load_pipeline = AsyncMock() + ap.pipeline_mgr.prepare_pipeline = AsyncMock() + ap.pipeline_mgr.publish_pipeline = Mock() ap.sess_mgr = SimpleNamespace() # Mock session with conversation using this pipeline @@ -965,7 +971,7 @@ class TestPipelineSecretRoundtrip: write_result = Mock(rowcount=1) ap = SimpleNamespace( persistence_mgr=SimpleNamespace(execute_async=AsyncMock(return_value=write_result)), - pipeline_mgr=SimpleNamespace(remove_pipeline=AsyncMock(), load_pipeline=AsyncMock()), + pipeline_mgr=SimpleNamespace(prepare_pipeline=AsyncMock(), publish_pipeline=Mock()), sess_mgr=SimpleNamespace(session_list=[]), ) service = PipelineService(ap) @@ -976,13 +982,16 @@ class TestPipelineSecretRoundtrip: 'pipeline-secret', { 'config': { + 'trigger': {}, + 'safety': {}, + 'output': {}, 'ai': { 'n8n': { 'webhook-url': '***', 'headers': {'Authorization': '***'}, 'timeout': 60, } - } + }, } }, ) @@ -990,6 +999,9 @@ class TestPipelineSecretRoundtrip: statement = ap.persistence_mgr.execute_async.await_args.args[0] stored_config = next(value.value for column, value in statement._values.items() if column.key == 'config') assert stored_config == { + 'trigger': {}, + 'safety': {}, + 'output': {}, 'ai': { 'n8n': { 'webhook-url': 'https://hook.invalid/bearer-secret', diff --git a/tests/unit_tests/pipeline/test_pipeline_service.py b/tests/unit_tests/pipeline/test_pipeline_service.py index 20e08187e..95c91510f 100644 --- a/tests/unit_tests/pipeline/test_pipeline_service.py +++ b/tests/unit_tests/pipeline/test_pipeline_service.py @@ -11,14 +11,14 @@ WORKSPACE_UUID = 'workspace-a' @pytest.mark.asyncio async def test_update_pipeline_filters_protected_fields_without_mutating_input(mock_app): service = PipelineService(mock_app) - loaded_pipeline = Mock() + loaded_pipeline = {'uuid': 'pipeline-uuid'} service.get_pipeline = AsyncMock(return_value=loaded_pipeline) bot = Mock(uuid='bot-uuid') bot_result = Mock(all=Mock(return_value=[bot])) mock_app.persistence_mgr.execute_async = AsyncMock(side_effect=[None, bot_result]) mock_app.bot_service = Mock(update_bot=AsyncMock()) - mock_app.pipeline_mgr = Mock(remove_pipeline=AsyncMock(), load_pipeline=AsyncMock()) + mock_app.pipeline_mgr = Mock(prepare_pipeline=AsyncMock(), publish_pipeline=Mock()) mock_app.sess_mgr.session_list = [] pipeline_data = { @@ -39,5 +39,7 @@ async def test_update_pipeline_filters_protected_fields_without_mutating_input(m assert updated_fields == {'name'} mock_app.bot_service.update_bot.assert_not_awaited() - mock_app.pipeline_mgr.remove_pipeline.assert_awaited_once_with('workspace-a', 'pipeline-uuid') - mock_app.pipeline_mgr.load_pipeline.assert_awaited_once_with('workspace-a', loaded_pipeline) + mock_app.pipeline_mgr.prepare_pipeline.assert_awaited_once_with( + 'workspace-a', {'uuid': 'pipeline-uuid', 'name': 'Updated pipeline'} + ) + mock_app.pipeline_mgr.publish_pipeline.assert_called_once_with(mock_app.pipeline_mgr.prepare_pipeline.return_value)