xinyin025/LangBot
1
0
Fork 0
mirror of https://github.com/langbot-app/LangBot.git synced 2026-10-22 16:35:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

test(persistence): verify migration data and startup task bounds

Browse Source
This commit is contained in:
huanghuoguoguo
2026-10-17 16:08:08 +08:00
parent c05aac538f
commit 0f522e5286
3 changed files with 108 additions and 229 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/integration/persistence/test_migration_branch_convergence.py
+73 -2
View File
@@ -108,6 +108,18 @@ async def _seed_common(engine):
base_url='https://fixture.invalid',
api_keys=['fixture-key'],
)
await _insert(
conn,
'llm_models',
uuid='model-1',
workspace_uuid=WORKSPACE,
name='Fixture model',
provider_uuid='provider-1',
abilities=['vision'],
extra_args={'sentinel': 'preserve-model'},
prefered_ranking=7,
reasoning_config={'level': 'provider_default'},
)
await _insert(
conn,
'legacy_pipelines',
@@ -261,6 +273,31 @@ async def _assert_merged_schema_and_data(engine, source):
assert await conn.scalar(sa.text('SELECT password FROM users')) == 'hash'
assert await conn.scalar(sa.text('SELECT COUNT(*) FROM workspace_memberships')) == 1
assert await conn.scalar(sa.text('SELECT name FROM model_providers')) == 'Fixture provider'
models = await conn.run_sync(lambda sync: sa.Table('llm_models', sa.MetaData(), autoload_with=sync))
model = (
await conn.execute(
sa.select(
models.c.name,
models.c.abilities,
models.c.extra_args,
models.c.prefered_ranking,
models.c.reasoning_config,
)
)
).one()
assert tuple(model) == (
'Fixture model',
['vision'],
{'sentinel': 'preserve-model'},
7,
{'level': 'provider_default'},
)
reasoning_column = await conn.run_sync(
lambda sync: next(
column for column in sa.inspect(sync).get_columns('llm_models') if column['name'] == 'reasoning_config'
)
)
assert reasoning_column['nullable'] is False
assert await conn.scalar(sa.text('SELECT description FROM legacy_pipelines')) == 'preserve pipeline'
assert (
await conn.execute(sa.text('SELECT message_content FROM monitoring_messages ORDER BY id'))
@@ -306,8 +343,18 @@ async def _assert_merged_schema_and_data(engine, source):
@pytest.mark.asyncio
@pytest.mark.parametrize('source', ['baseline', 'master', 'beta'])
async def test_historical_schema_and_populated_branch_upgrade(convergence_engine, source):
@pytest.mark.parametrize(
('source', 'checkpoint'),
[
('baseline', None),
('baseline', '0016_space_launch_replay'),
('baseline', '0018_merge_workspace_heads'),
('baseline', '0018_llm_reasoning_config'),
('master', None),
('beta', None),
],
)
async def test_historical_schema_and_populated_branch_upgrade(convergence_engine, source, checkpoint):
engine = convergence_engine
fixture = json.loads((FIXTURES / f'{source}_schema.json').read_text())
async with engine.begin() as conn:
@@ -338,6 +385,9 @@ async def test_historical_schema_and_populated_branch_upgrade(convergence_engine
await run_alembic_upgrade(engine, '0010_scope_resources')
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
if checkpoint is not None:
await run_alembic_upgrade(engine, checkpoint)
assert await get_alembic_current(engine) == checkpoint
await run_alembic_upgrade(engine, 'head')
await _assert_merged_schema_and_data(engine, source)
await run_alembic_upgrade(engine, 'head')
@@ -351,8 +401,29 @@ async def test_empty_database_startup_schema_then_real_migrations(convergence_en
assert await conn.run_sync(lambda sync: sa.inspect(sync).get_table_names()) == []
# This is the documented fresh-install contract: create_all precedes Alembic.
await conn.run_sync(Base.metadata.create_all)
await conn.execute(
Base.metadata.tables['metadata'].insert().values(key='instance_uuid', value='migration-convergence-fresh')
)
await conn.execute(
Base.metadata.tables['users']
.insert()
.values(
uuid=ACCOUNT, user='[email protected]', normalized_email='[email protected]', password='fresh-password'
)
)
await run_alembic_upgrade(engine, 'head')
assert await get_alembic_current(engine) == get_alembic_head()
async with engine.connect() as conn:
tables = await conn.run_sync(lambda sync: set(sa.inspect(sync).get_table_names()))
assert {
'users',
'bots',
'agents',
'runner_state',
'directory_projection_inbox',
'passkey_credentials',
} <= tables
assert await conn.scalar(sa.text('SELECT password FROM users')) == 'fresh-password'
def _legacy_shared_certification(**overrides):
tests/integration/persistence/test_migrations.py
+8 -170
View File
@@ -122,39 +122,6 @@ class TestSQLiteMigrationBaseline:
class TestSQLiteMigrationUpgrade:
"""Tests for upgrade to head workflow."""
@pytest.mark.asyncio
async def test_upgrade_from_published_space_launch_head_to_merged_head(self, sqlite_engine):
"""A database released at the production-only 0016 head must remain upgradable."""
async with sqlite_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
await run_alembic_stamp(sqlite_engine, '0016_space_launch_replay')
await run_alembic_upgrade(sqlite_engine, 'head')
assert await get_alembic_current(sqlite_engine) == _get_script_head()
@pytest.mark.asyncio
async def test_upgrade_from_development_workspace_head_to_merged_head(self, sqlite_engine):
"""A database at the 4.11 development head must absorb later master migrations."""
async with sqlite_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
await run_alembic_stamp(sqlite_engine, '0018_merge_workspace_heads')
await run_alembic_upgrade(sqlite_engine, 'head')
assert await get_alembic_current(sqlite_engine) == _get_script_head()
@pytest.mark.asyncio
async def test_upgrade_from_reasoning_config_head_to_merged_head(self, sqlite_engine):
"""A database that already ran the feature migration must remain upgradable."""
async with sqlite_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
await run_alembic_stamp(sqlite_engine, '0018_llm_reasoning_config')
await run_alembic_upgrade(sqlite_engine, 'head')
assert await get_alembic_current(sqlite_engine) == _get_script_head()
@pytest.mark.asyncio
async def test_upgrade_removes_agent_enabled_column(self, sqlite_engine):
async with sqlite_engine.begin() as conn:
@@ -171,62 +138,6 @@ class TestSQLiteMigrationUpgrade:
assert 'enabled' not in columns
@pytest.mark.asyncio
async def test_upgrade_from_baseline_to_head(self, sqlite_engine):
"""
Upgrade from baseline to head applies all migrations.
Workflow:
1. Create tables
2. Stamp baseline
3. Upgrade to head
4. Verify current revision is head
"""
# Create tables
async with sqlite_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
# Stamp baseline
await run_alembic_stamp(sqlite_engine, '0001_baseline')
# Upgrade to head
await run_alembic_upgrade(sqlite_engine, 'head')
# Verify revision
rev = await get_alembic_current(sqlite_engine)
assert rev is not None, 'Expected a revision after upgrade'
# Head should be the latest migration. Resolve the actual head from the
# Alembic script directory instead of hardcoding a revision number, so
# adding a new migration doesn't require editing this assertion.
assert rev == _get_script_head(), f'Expected head {_get_script_head()}, got {rev}'
@pytest.mark.asyncio
async def test_upgrade_idempotent(self, sqlite_engine):
"""
Running upgrade to head multiple times is idempotent.
Workflow:
1. Upgrade to head
2. Get revision
3. Upgrade to head again
4. Verify same revision
"""
# Create tables
async with sqlite_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
# Stamp and upgrade
await run_alembic_stamp(sqlite_engine, '0001_baseline')
await run_alembic_upgrade(sqlite_engine, 'head')
rev1 = await get_alembic_current(sqlite_engine)
# Upgrade again - should be idempotent
await run_alembic_upgrade(sqlite_engine, 'head')
rev2 = await get_alembic_current(sqlite_engine)
assert rev2 == rev1, f'Expected {rev1}, got {rev2}'
@pytest.mark.asyncio
async def test_upgrade_from_0012_adds_knowledge_base_embedding_dimension(self, sqlite_engine):
"""The PostgreSQL pgvector revision also evolves the OSS ORM schema."""
@@ -340,87 +251,14 @@ class TestSQLiteMigrationFreshDatabase:
assert await get_alembic_current(sqlite_engine) == '0022_codex_credentials'
@pytest.mark.asyncio
async def test_fresh_db_upgrade_from_scratch(self, tmp_path):
"""
Fresh database (no tables) can be upgraded directly to head.
Workflow:
1. Create fresh engine with new DB file
2. Create tables
3. Upgrade to head
4. Verify revision
"""
# Use different DB file for fresh test
fresh_db_file = tmp_path / 'test_migrations_fresh.db'
fresh_url = f'sqlite+aiosqlite:///{fresh_db_file}'
fresh_engine = create_async_engine(fresh_url)
# Create tables on fresh DB
async with fresh_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
# Upgrade to head directly (no baseline stamp)
await run_alembic_upgrade(fresh_engine, 'head')
# Verify revision
rev = await get_alembic_current(fresh_engine)
assert rev is not None, 'Expected a revision on fresh DB'
await fresh_engine.dispose()
@pytest.mark.asyncio
async def test_fresh_db_without_create_all_behavior(self, tmp_path):
"""
Fresh database without create_all - test actual behavior.
This tests what happens when migrations run on truly empty DB.
The behavior is determined by Alembic and migration scripts.
EXPECTED: Either:
1. Migration succeeds (if scripts handle empty DB)
2. Migration fails with specific error (if scripts require tables)
IMPORTANT: This test verifies the ACTUAL behavior, not accepting
any arbitrary failure with try-except pass.
"""
fresh_db_file = tmp_path / 'test_empty_migrations.db'
fresh_url = f'sqlite+aiosqlite:///{fresh_db_file}'
fresh_engine = create_async_engine(fresh_url)
# Capture the actual behavior
actual_result = None
actual_error = None
try:
await run_alembic_upgrade(fresh_engine, 'head')
rev = await get_alembic_current(fresh_engine)
actual_result = rev
except Exception as e:
actual_error = e
await fresh_engine.dispose()
# Verify specific behavior - one of two outcomes is expected
if actual_result is not None:
# Migration succeeded - verify revision exists
assert actual_result is not None, 'Revision should exist after successful migration'
else:
# Migration failed - verify the error type is known
# Alembic typically raises specific errors for missing tables
assert actual_error is not None, 'Error should be captured if migration failed'
# Log the error type for documentation (don't silently pass)
error_type = type(actual_error).__name__
# Acceptable error types for empty DB scenarios
acceptable_errors = [
'OperationalError', # SQLite table not found
'ProgrammingError', # SQLAlchemy errors
'CommandError', # Alembic command errors
]
assert error_type in acceptable_errors, (
f'Unexpected error type: {error_type}. '
f'This may indicate a regression in migration behavior. '
f'Error: {actual_error}'
)
async def test_fresh_db_without_create_all_behavior(self, sqlite_engine):
"""Bare Alembic reaches head but relies on ORM bootstrap for core tables."""
await run_alembic_upgrade(sqlite_engine, 'head')
assert await get_alembic_current(sqlite_engine) == _get_script_head()
async with sqlite_engine.connect() as conn:
tables = await conn.run_sync(lambda sync: set(sqlalchemy.inspect(sync).get_table_names()))
assert {'agents', 'runner_state', 'directory_projection_states', 'directory_projection_inbox'} <= tables
assert not {'users', 'bots', 'workspaces'} & tables
class TestSQLiteMigrationGetCurrent:
tests/integration/persistence/test_migrations_postgres.py
+27 -57
View File
@@ -424,62 +424,6 @@ class TestPostgreSQLMigrationBaseline:
class TestPostgreSQLMigrationUpgrade:
"""Tests for upgrade to head workflow on PostgreSQL."""
@pytest.mark.asyncio
async def test_postgres_upgrade_from_baseline_to_head(self, postgres_engine, clean_tables, clean_alembic_version):
"""
Upgrade from baseline to head applies all migrations.
Workflow:
1. Create tables
2. Stamp baseline
3. Upgrade to head
4. Verify current revision is head
"""
# Create tables
async with postgres_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
# Stamp baseline
await run_alembic_stamp(postgres_engine, '0001_baseline')
# Upgrade to head
await run_alembic_upgrade(postgres_engine, 'head')
# Verify revision
rev = await get_alembic_current(postgres_engine)
assert rev is not None, 'Expected a revision after upgrade'
# Head should be the latest migration. Resolve the actual head from the
# Alembic script directory instead of hardcoding a revision number, so
# adding a new migration doesn't require editing this assertion.
assert rev == _get_script_head(), f'Expected head {_get_script_head()}, got {rev}'
@pytest.mark.asyncio
async def test_postgres_upgrade_idempotent(self, postgres_engine, clean_tables, clean_alembic_version):
"""
Running upgrade to head multiple times is idempotent.
Workflow:
1. Upgrade to head
2. Get revision
3. Upgrade to head again
4. Verify same revision
"""
# Create tables
async with postgres_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
# Stamp and upgrade
await run_alembic_stamp(postgres_engine, '0001_baseline')
await run_alembic_upgrade(postgres_engine, 'head')
rev1 = await get_alembic_current(postgres_engine)
# Upgrade again - should be idempotent
await run_alembic_upgrade(postgres_engine, 'head')
rev2 = await get_alembic_current(postgres_engine)
assert rev2 == rev1, f'Expected {rev1}, got {rev2}'
@pytest.mark.asyncio
@pytest.mark.parametrize('settings_type', ['TEXT', 'JSON'])
async def test_legacy_knowledge_restore_statement_accepts_historical_and_fresh_settings_types(
@@ -1070,7 +1014,27 @@ class TestPostgreSQLTenantRuntime:
await rag_manager.load_knowledge_bases_from_db()
mcp_loader = MCPLoader(runtime_application)
mcp_loader.host_mcp_server = AsyncMock()
active_hosts = 0
peak_active_hosts = 0
peak_host_tasks = 0
track_hosted_task = mcp_loader.track_hosted_task
def observe_host_task(task, context):
nonlocal peak_host_tasks
track_hosted_task(task, context)
peak_host_tasks = max(peak_host_tasks, len(mcp_loader._hosted_mcp_tasks))
async def host_probe(_context, _config):
nonlocal active_hosts, peak_active_hosts
active_hosts += 1
peak_active_hosts = max(peak_active_hosts, active_hosts)
try:
await asyncio.sleep(0)
finally:
active_hosts -= 1
mcp_loader.track_hosted_task = observe_host_task
mcp_loader.host_mcp_server = AsyncMock(side_effect=host_probe)
await mcp_loader.load_mcp_servers_from_db()
dispatch_tasks = tuple(mcp_loader._host_dispatch_tasks)
if dispatch_tasks:
@@ -1111,6 +1075,12 @@ class TestPostgreSQLTenantRuntime:
assert len(pipeline_manager.pipelines) == workspace_count
assert len(rag_manager.knowledge_bases) == workspace_count
assert mcp_loader.host_mcp_server.await_count == workspace_count
assert 0 < peak_active_hosts <= min(workspace_count, 8)
assert 0 < peak_host_tasks <= min(workspace_count, 8)
assert active_hosts == 0
logging.getLogger('postgres-capacity-runtime-test').info(
'MCP startup peaks: active_hosts=%d hosted_tasks=%d', peak_active_hosts, peak_host_tasks
)
assert not mcp_loader._host_dispatch_tasks
assert not mcp_loader._hosted_mcp_tasks
assert len(plugin_handler.reconciled) == workspace_count