"""Plugin foundation contracts and Workspace isolation without a transport.""" from contextlib import asynccontextmanager from io import BytesIO from types import SimpleNamespace from unittest.mock import AsyncMock, Mock from zipfile import ZipFile import pytest from langbot_plugin.entities.io.errors import ActionCallError from pydantic import TypeAdapter, ValidationError from langbot.pkg.api.application import plugin_operations as ops from langbot.pkg.api.application.errors import error_from_exception from langbot.pkg.api.context import ExecutionContext from langbot.pkg.workspace.errors import WorkspaceGenerationMismatchError, WorkspaceNotFoundError CONTEXT = SimpleNamespace(instance_uuid='instance-a', workspace_uuid='workspace-a', placement_generation=3) def app(): execution = ExecutionContext(instance_uuid='instance-a', workspace_uuid='workspace-a', placement_generation=3) return SimpleNamespace( plugin_connector=SimpleNamespace(require_workspace_context=AsyncMock(return_value=execution)), instance_config=SimpleNamespace(data={'system': {'limitation': {}}}), ) @pytest.mark.asyncio async def test_plugin_reads_mask_secrets_and_config_write_restores_persisted_values(): ap = app() raw = {'plugin_config': {'apiKey': 'runtime-secret'}, 'debug': {'plugin_debug_key': 'debug-secret'}} ap.plugin_connector.get_plugin_info = AsyncMock(return_value=raw) ap.plugin_connector.set_plugin_config = AsyncMock() ap.persistence_mgr = SimpleNamespace( execute_async=AsyncMock( return_value=SimpleNamespace( scalar_one_or_none=lambda: { 'apiKey': 'stored-secret', 'nested': {'token': 'nested-secret'}, 'enabled': False, } ) ) ) key = ops.PluginInput(author='a', plugin_name='b') assert (await ops.get_plugin(ap, CONTEXT, key))['plugin']['plugin_config']['apiKey'] == '***' assert raw['plugin_config']['apiKey'] == 'runtime-secret' assert (await ops.get_config(ap, CONTEXT, key))['config']['nested']['token'] == '***' result = await ops.update_config( ap, CONTEXT, ops.ConfigInput( author='a', plugin_name='b', config={'apiKey': '***', 'nested': {'token': '***'}, 'enabled': True} ), ) assert result == {'changes': [{'field': 'enabled', 'before': False, 'after': True}]} assert ap.plugin_connector.set_plugin_config.await_args.args == ( 'a', 'b', {'apiKey': 'stored-secret', 'nested': {'token': 'nested-secret'}, 'enabled': True}, ) changed = await ops.update_config( ap, CONTEXT, ops.ConfigInput( author='a', plugin_name='b', config={ 'apiKey': 'replacement-secret', 'nested': {'token': 'changed-secret', 'mode': 'new'}, 'enabled': False, }, ), ) assert {item['field'] for item in changed['changes']} == {'nested'} assert all( value not in str(changed) for value in ('stored-secret', 'nested-secret', 'replacement-secret', 'changed-secret') ) assert 'new' in str(changed) with pytest.raises(ValueError, match='no existing value'): await ops.update_config(ap, CONTEXT, ops.ConfigInput(author='a', plugin_name='b', config={'password': '***'})) @pytest.mark.asyncio async def test_queued_plugin_task_revalidates_generation_before_runtime_io(): ap = app() ap.plugin_connector.delete_plugin = AsyncMock() queued = [] scopes = [] @asynccontextmanager async def tenant_scope(workspace): scopes.append(workspace) yield def create_task(coro, **metadata): queued.append((coro, metadata)) return SimpleNamespace(id=17) ap.persistence_mgr = SimpleNamespace(tenant_scope=tenant_scope) ap.task_mgr = SimpleNamespace(create_user_task=create_task) assert await ops.delete(ap, CONTEXT, ops.DeleteInput(author='a', plugin_name='b')) == {'task_id': 17} assert TypeAdapter(ops.TaskResult).validate_python({'task_id': 17}, strict=True) == {'task_id': 17} assert queued[0][1]['workspace_uuid'] == 'workspace-a' assert queued[0][1]['placement_generation'] == 3 ap.plugin_connector.require_workspace_context.side_effect = WorkspaceGenerationMismatchError('stale placement') with pytest.raises(WorkspaceGenerationMismatchError): await queued[0][0] assert scopes == ['workspace-a'] ap.plugin_connector.delete_plugin.assert_not_awaited() @pytest.mark.asyncio async def test_install_quota_counts_plugins_and_mcp_before_queueing(): ap = app() ap.instance_config.data['system']['limitation']['max_extensions'] = 2 ap.plugin_connector.list_plugins = AsyncMock(return_value=[{}]) ap.mcp_service = SimpleNamespace(get_mcp_servers=AsyncMock(return_value=[{}])) ap.task_mgr = SimpleNamespace(create_user_task=Mock()) with pytest.raises(ValueError, match='Maximum number'): await ops.install_marketplace(ap, CONTEXT, ops.MarketplaceInput(plugin_author='a', plugin_name='b')) ap.task_mgr.create_user_task.assert_not_called() @pytest.mark.asyncio async def test_asset_path_rejects_escape_before_runtime_read_and_returns_exact_bytes(): ap = app() ap.plugin_connector.get_plugin_assets = AsyncMock(return_value={'asset_base64': 'AP8K', 'mime_type': 'image/png'}) for path in ('../secret', '/etc/passwd', '..\\secret'): with pytest.raises(WorkspaceNotFoundError): await ops.asset(ap, CONTEXT, ops.AssetInput(author='a', plugin_name='b', filepath=path)) ap.plugin_connector.get_plugin_assets.assert_not_awaited() result = await ops.asset(ap, CONTEXT, ops.AssetInput(author='a', plugin_name='b', filepath='image.png')) assert result == {'content': b'\x00\xff\n', 'mime_type': 'image/png'} assert ap.plugin_connector.get_plugin_assets.await_args.args == ('a', 'b', 'assets/image.png') @pytest.mark.asyncio @pytest.mark.parametrize('operation', ['icon', 'asset']) @pytest.mark.parametrize( 'error', [ActionCallError('Plugin asset unavailable'), ValueError('Plugin a/b is not installed in this Workspace')] ) async def test_missing_plugin_media_maps_to_not_found(operation, error): ap = app() ap.plugin_connector.get_plugin_icon = AsyncMock(side_effect=error) ap.plugin_connector.get_plugin_assets = AsyncMock(side_effect=error) payload = ops.AssetInput(author='a', plugin_name='b', filepath='image.png') with pytest.raises(WorkspaceNotFoundError) as failure: await getattr(ops, operation)(ap, CONTEXT, payload) result = error_from_exception(failure.value) assert (result.code, result.status_code) == ('resource_not_found', 404) @pytest.mark.asyncio @pytest.mark.parametrize('result', [None, {}, {'plugin_icon_base64': ''}]) async def test_empty_plugin_icon_is_not_found(result): ap = app() ap.plugin_connector.get_plugin_icon = AsyncMock(return_value=result) with pytest.raises(WorkspaceNotFoundError): await ops.icon(ap, CONTEXT, ops.PluginInput(author='a', plugin_name='b')) @pytest.mark.asyncio async def test_plugin_icon_defaults_to_svg_and_preserves_other_validation_errors(): ap = app() ap.plugin_connector.get_plugin_icon = AsyncMock(return_value={'plugin_icon_base64': 'PHN2Zy8+'}) payload = ops.PluginInput(author='a', plugin_name='b') assert await ops.icon(ap, CONTEXT, payload) == {'content': b'', 'mime_type': 'image/svg+xml'} ap.plugin_connector.get_plugin_icon.side_effect = ValueError('Invalid plugin identity') with pytest.raises(ValueError, match='Invalid plugin identity'): await ops.icon(ap, CONTEXT, payload) @pytest.mark.asyncio async def test_config_files_are_workspace_scoped_and_size_checked(): ap = app() ap.storage_mgr = SimpleNamespace( save_scoped=AsyncMock(return_value='scoped-key'), is_scoped_object_key=Mock(return_value=False), delete_scoped_object_key=AsyncMock(), ) assert await ops.upload_config_file(ap, CONTEXT, ops.FileInput(content=b'payload', filename='data.txt')) == { 'file_key': 'scoped-key' } args = ap.storage_mgr.save_scoped.await_args assert args.args == (CONTEXT,) assert args.kwargs['owner'] == 'workspace-a' assert args.kwargs['owner_type'] == 'plugin_config' assert args.kwargs['value'] == b'payload' with pytest.raises(ValueError, match='10MB'): await ops.upload_config_file(ap, CONTEXT, ops.FileInput(content=b'x' * (10 * 1024 * 1024 + 1))) with pytest.raises(ValueError, match='Invalid file key'): await ops.delete_config_file(ap, CONTEXT, ops.FileKeyInput(file_key='foreign/raw')) ap.storage_mgr.delete_scoped_object_key.assert_not_awaited() @pytest.mark.asyncio async def test_archive_preview_uses_archive_admission_and_component_metadata(): ap = app() archive = BytesIO() with ZipFile(archive, 'w') as output: output.writestr( 'manifest.yaml', 'metadata:\n author: example\n name: sample\nspec:\n components:\n tools:\n fromDirs: [components/tools]\n', ) output.writestr('components/tools/tool.yaml', 'name: test') output.writestr('requirements.txt', 'httpx\n') result = await ops.preview_local(ap, CONTEXT, ops.FileInput(content=archive.getvalue(), filename='sample.lbpkg')) assert result['metadata'] == {'author': 'example', 'name': 'sample'} assert result['component_counts'] == {'tools': 1} assert result['requirements'] == ['httpx'] with pytest.raises(ValueError): await ops.preview_local(ap, CONTEXT, ops.FileInput(content=b'invalid')) def test_install_inputs_do_not_coerce_force_flags_or_github_path_segments(): with pytest.raises(ValidationError): ops.LocalInstallInput(content=b'zip', administrator_force='true') with pytest.raises(ValidationError): ops.GithubAssetsInput(owner='a/b', repo='repo', release_id=1) with pytest.raises(ValidationError): ops.DeleteInput(author='a', plugin_name='b', delete_data='false')