test(agent): verify persistence and denied action side effects

tests/unit_tests/plugin/conftest.py

@@ -0,0 +1,38 @@
+"""Database fixture for plugin persistence behavior tests."""
+
+import pytest
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+
+from langbot.pkg.entity.persistence.base import Base
+from langbot.pkg.entity.persistence.bstorage import BinaryStorage
+from langbot.pkg.entity.persistence.plugin import PluginSetting
+from langbot.pkg.entity.persistence.user import User
+from langbot.pkg.entity.persistence.workspace import Workspace
+
+
+@pytest.fixture
+async def plugin_db():
+    engine = create_async_engine('sqlite+aiosqlite:///:memory:')
+    try:
+        async with engine.begin() as connection:
+            await connection.execute(text('PRAGMA foreign_keys=ON'))
+            await connection.run_sync(
+                Base.metadata.create_all,
+                tables=[User.__table__, Workspace.__table__, PluginSetting.__table__, BinaryStorage.__table__],
+            )
+        async with AsyncSession(engine, expire_on_commit=False) as session:
+            session.add_all(
+                Workspace(
+                    uuid=workspace,
+                    instance_uuid='instance-a',
+                    name=workspace,
+                    slug=workspace,
+                    source='cloud_projection',
+                )
+                for workspace in ('workspace-a', 'workspace-b')
+            )
+            await session.flush()
+            yield session
+    finally:
+        await engine.dispose()

tests/unit_tests/plugin/test_connector_pure.py

@@ -123,6 +123,7 @@ class TestExtractDepsMetadata:
 
         # Should find requirements.txt in subdirectory
         assert task_context.metadata['deps_total'] == 2
+        assert task_context.metadata['deps_list'] == ['pytest', 'black']
 
     def test_archive_preview_rejects_extreme_compression_ratio(self):
         from langbot.pkg.plugin.connector import inspect_plugin_archive_metadata

tests/unit_tests/plugin/test_connector_reconcile.py

@@ -583,11 +583,15 @@ async def test_local_install_cleans_untracked_legacy_plugin_before_runtime_apply
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize(('remaining_references', 'statement_count'), [(1, 1), (0, 2)])
+@pytest.mark.parametrize('remaining_references', [1, 0])
 async def test_artifact_cleanup_is_reference_counted_within_workspace(
     remaining_references: int,
-    statement_count: int,
+    plugin_db,
 ):
+    import sqlalchemy
+    from langbot.pkg.entity.persistence.plugin import PluginSetting
+    from langbot.pkg.entity.persistence.bstorage import BinaryStorage
+
     app = SimpleNamespace(
         instance_config=SimpleNamespace(data={'plugin': {'enable': True}}),
         deployment=SimpleNamespace(mode='cloud'),
@@ -598,19 +602,38 @@ async def test_artifact_cleanup_is_reference_counted_within_workspace(
         workspace_uuid='workspace-a',
         placement_generation=1,
     )
-    statements = []
-
-    async def execute(statement):
-        statements.append(statement)
-        return SimpleNamespace(scalar_one=lambda: remaining_references)
+    digest = 'a' * 64
+    settings = [('workspace-b', digest), ('workspace-a', 'b' * 64)]
+    if remaining_references:
+        settings.append(('workspace-a', digest))
+    plugin_db.add_all(
+        PluginSetting(workspace_uuid=workspace, plugin_author='author', plugin_name=artifact, artifact_digest=artifact)
+        for workspace, artifact in settings
+    )
+    plugin_db.add_all(
+        BinaryStorage(
+            workspace_uuid=workspace, unique_key=key, key=key, owner_type=owner_type, owner=owner, value=b'pkg'
+        )
+        for workspace, key, owner_type, owner in [
+            ('workspace-a', 'target', 'plugin_artifact', digest),
+            ('workspace-b', 'other-workspace', 'plugin_artifact', digest),
+            ('workspace-a', 'other-artifact', 'plugin_artifact', 'b' * 64),
+            ('workspace-a', 'other-owner-type', 'plugin', digest),
+        ]
+    )
+    await plugin_db.flush()
 
     await connector._delete_artifact_if_unreferenced(
         execution_context,
-        'a' * 64,
-        execute=execute,
+        digest,
+        execute=plugin_db.execute,
     )
 
-    assert len(statements) == statement_count
+    remaining = (await plugin_db.execute(sqlalchemy.select(BinaryStorage.unique_key))).scalars().all()
+    expected = {'other-workspace', 'other-artifact', 'other-owner-type'}
+    if remaining_references:
+        expected.add('target')
+    assert set(remaining) == expected
 
 
 @pytest.mark.asyncio

tests/unit_tests/plugin/test_extract_deps.py

@@ -113,6 +113,7 @@ numpy"""
 
         # Should find nested requirements.txt (ends with 'requirements.txt')
         assert task_context.metadata.get('deps_total') == 2
+        assert task_context.metadata['deps_list'] == ['requests', 'flask']
 
     def test_no_requirements_in_zip(self):
         """Test handling ZIP without requirements.txt."""

tests/unit_tests/plugin/test_handler_actions.py

@@ -236,10 +236,16 @@ class TestInitializePluginSettings:
         app.persistence_mgr.execute_async.assert_awaited_once()
 
     @pytest.mark.asyncio
-    async def test_existing_desired_setting_remains_core_owned(self, app):
+    async def test_existing_desired_setting_remains_core_owned(self, app, plugin_db):
         """Runtime initialization validates identity without rewriting Core state."""
+        import sqlalchemy
+        from langbot.pkg.entity.persistence.plugin import PluginSetting
+
         runtime_handler = make_handler(app)
-        existing_setting = SimpleNamespace(
+        existing_setting = PluginSetting(
+            workspace_uuid='workspace-a',
+            plugin_author='test-author',
+            plugin_name='test-plugin',
             enabled=False,
             priority=5,
             config={'key': 'value'},
@@ -247,7 +253,29 @@ class TestInitializePluginSettings:
             runtime_revision=1,
             artifact_digest='a' * 64,
         )
-        app.persistence_mgr.execute_async.return_value = make_result(existing_setting)
+        other_setting = PluginSetting(
+            workspace_uuid='workspace-b',
+            plugin_author='test-author',
+            plugin_name='test-plugin',
+            enabled=True,
+        )
+        plugin_db.add_all([other_setting, existing_setting])
+        await plugin_db.flush()
+
+        async def execute(statement):
+            result = await plugin_db.execute(statement)
+            return result.scalars() if statement.is_select or statement.is_update else result
+
+        app.persistence_mgr.execute_async.side_effect = execute
+        before = (
+            (
+                await plugin_db.execute(
+                    sqlalchemy.select(*PluginSetting.__table__.c).order_by(PluginSetting.workspace_uuid)
+                )
+            )
+            .mappings()
+            .all()
+        )
 
         response = await runtime_handler.actions[RuntimeToLangBotAction.INITIALIZE_PLUGIN_SETTINGS.value](
             {
@@ -260,6 +288,17 @@ class TestInitializePluginSettings:
 
         assert response.code == 0
         app.persistence_mgr.execute_async.assert_awaited_once()
+        plugin_db.expire_all()
+        after = (
+            (
+                await plugin_db.execute(
+                    sqlalchemy.select(*PluginSetting.__table__.c).order_by(PluginSetting.workspace_uuid)
+                )
+            )
+            .mappings()
+            .all()
+        )
+        assert after == before
 
 
 class TestSetBinaryStorage:

tests/unit_tests/plugin/test_handler_helpers.py

@@ -107,41 +107,69 @@ class TestMakeRagErrorResponse:
 class TestCleanupPluginData:
     """Tests for cleanup_plugin_data method."""
 
-    @pytest.mark.asyncio
-    async def test_deletes_plugin_settings(self):
-        """Test that plugin settings are deleted."""
-        handler_module = get_handler_module()
+    @pytest.fixture
+    async def cleanup(self, plugin_db):
+        from langbot.pkg.entity.persistence.plugin import PluginSetting
+        from langbot.pkg.entity.persistence.bstorage import BinaryStorage
+        from langbot_plugin.entities.io.context import InstallationBinding
 
-        mock_app = Mock()
-        mock_app.persistence_mgr = AsyncMock()
-        mock_app.persistence_mgr.execute_async = AsyncMock()
-
-        # Mock the handler without connection (we only need ap)
-        handler_instance = Mock(spec=handler_module.RuntimeConnectionHandler)
-        handler_instance.ap = mock_app
-
-        # Call cleanup_plugin_data
-        await handler_module.RuntimeConnectionHandler.cleanup_plugin_data(
-            handler_instance, 'test-author', 'test-plugin'
+        plugin_db.add_all(
+            PluginSetting(workspace_uuid=workspace, plugin_author=author, plugin_name=name)
+            for workspace, author, name in [
+                ('workspace-a', 'test-author', 'test-plugin'),
+                ('workspace-b', 'test-author', 'test-plugin'),
+                ('workspace-a', 'other-author', 'test-plugin'),
+                ('workspace-a', 'test-author', 'other-plugin'),
+            ]
         )
-
-        # Verify plugin settings delete was called
-        calls = mock_app.persistence_mgr.execute_async.call_args_list
-        assert len(calls) >= 1
+        plugin_db.add_all(
+            BinaryStorage(
+                workspace_uuid=workspace, unique_key=key, key=key, owner_type=owner_type, owner=owner, value=b'data'
+            )
+            for workspace, key, owner_type, owner in [
+                ('workspace-a', 'target', 'plugin', 'test-author/test-plugin'),
+                ('workspace-b', 'other-workspace', 'plugin', 'test-author/test-plugin'),
+                ('workspace-a', 'other-owner', 'plugin', 'other-author/test-plugin'),
+                ('workspace-a', 'other-type', 'skill', 'test-author/test-plugin'),
+            ]
+        )
+        await plugin_db.flush()
+        app = Mock()
+        app.persistence_mgr.execute_async = plugin_db.execute
+        handler = get_handler_module().RuntimeConnectionHandler(Mock(), AsyncMock(), app)
+        binding = InstallationBinding(
+            instance_uuid='instance-a',
+            workspace_uuid='workspace-a',
+            placement_generation=1,
+            installation_uuid='00000000-0000-4000-8000-000000000001',
+            runtime_revision=1,
+            artifact_digest='a' * 64,
+        )
+        with handler.installation_scope(binding):
+            await handler.cleanup_plugin_data('test-author', 'test-plugin')
+        return plugin_db
 
     @pytest.mark.asyncio
-    async def test_deletes_binary_storage(self):
+    async def test_deletes_plugin_settings(self, cleanup):
+        import sqlalchemy
+        from langbot.pkg.entity.persistence.plugin import PluginSetting
+
+        rows = (
+            await cleanup.execute(
+                sqlalchemy.select(PluginSetting.workspace_uuid, PluginSetting.plugin_author, PluginSetting.plugin_name)
+            )
+        ).all()
+        assert set(rows) == {
+            ('workspace-b', 'test-author', 'test-plugin'),
+            ('workspace-a', 'other-author', 'test-plugin'),
+            ('workspace-a', 'test-author', 'other-plugin'),
+        }
+
+    @pytest.mark.asyncio
+    async def test_deletes_binary_storage(self, cleanup):
         """Test that binary storage is deleted."""
-        handler_module = get_handler_module()
+        import sqlalchemy
+        from langbot.pkg.entity.persistence.bstorage import BinaryStorage
 
-        mock_app = Mock()
-        mock_app.persistence_mgr = AsyncMock()
-        mock_app.persistence_mgr.execute_async = AsyncMock()
-
-        handler_instance = Mock(spec=handler_module.RuntimeConnectionHandler)
-        handler_instance.ap = mock_app
-
-        await handler_module.RuntimeConnectionHandler.cleanup_plugin_data(handler_instance, 'author', 'plugin-name')
-
-        # Should have at least 2 calls: one for settings, one for binary storage
-        assert mock_app.persistence_mgr.execute_async.call_count >= 2
+        keys = (await cleanup.execute(sqlalchemy.select(BinaryStorage.unique_key))).scalars().all()
+        assert set(keys) == {'other-workspace', 'other-owner', 'other-type'}