mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-11-01 05:15:22 +08:00
RockChinQandmultica-agent
8c5ee20283
LANG-231: Fix SDK 0.7.19 input compatibility regressions by accepting optional cursor in ExecutionRuntimesInput and float | None stale_after_seconds in ExecutionRuntimeReconcileInput. Update test_all_69_operations_route_through_execution_api to drive real SDK 0.7.19 payloads through handler and assert raw execution dispatch. Add autouse fixture in plugin tests to isolate transfer root environment leaks. Co-authored-by: multica-agent <[email protected]>
51 lines
1.8 KiB
Python
51 lines
1.8 KiB
Python
"""Database fixture for plugin persistence behavior tests."""
|
|
|
|
import os
|
|
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
|
|
from langbot_plugin.runtime.security import PLUGIN_FILE_STORAGE_DIR_ENV
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clean_file_transfer_env():
|
|
original = os.environ.get(PLUGIN_FILE_STORAGE_DIR_ENV)
|
|
yield
|
|
if original is None:
|
|
os.environ.pop(PLUGIN_FILE_STORAGE_DIR_ENV, None)
|
|
else:
|
|
os.environ[PLUGIN_FILE_STORAGE_DIR_ENV] = original
|
|
|
|
|
|
@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()
|