xinyin025/LangBot
1
0
Fork 0
mirror of https://github.com/langbot-app/LangBot.git synced 2026-08-09 04:40:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into dev/4.11.x

Browse Source 
# Conflicts:
#	pyproject.toml
#	src/langbot/pkg/pipeline/controller.py
#	uv.lock
#	web/src/app/home/bots/components/bot-form/BotForm.tsx
#	web/src/app/home/components/home-sidebar/HomeSidebar.tsx
#	web/src/app/home/components/home-sidebar/SidebarDataContext.tsx
#	web/src/app/home/plugin-pages/page.tsx
#	web/src/app/infra/entities/api/index.ts
This commit is contained in:
huanghuoguoguo
2026-08-04 12:16:27 +08:00
parent 551593b173 e263a5d1d7
commit 679b8346cb
75 changed files with 3809 additions and 713 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/integration/api/test_smoke.py
+24
View File
@@ -9,6 +9,8 @@ Run: uv run pytest tests/integration/api/test_smoke.py -q
from __future__ import annotations
from types import SimpleNamespace
import pytest
from unittest.mock import MagicMock, AsyncMock, Mock
@@ -304,12 +306,34 @@ class TestUserInitEndpoint:
data = await response.get_json()
assert data['data'] == {
'initialized': True,
'authenticated_invitation_acceptance_enabled': False,
'password_login_enabled': True,
'space_login_enabled': False,
}
fake_api_app.user_service.get_login_capabilities.assert_awaited_once_with()
fake_api_app.user_service.get_first_user.assert_not_awaited()
@pytest.mark.asyncio
async def test_account_info_enables_authenticated_invitation_acceptance_in_cloud(
self, quart_test_client, fake_api_app
):
fake_api_app.deployment = SimpleNamespace(mode='cloud')
fake_api_app.user_service.is_initialized.return_value = True
fake_api_app.user_service.get_login_capabilities = AsyncMock(
return_value={'password_login_enabled': True, 'space_login_enabled': True}
)
response = await quart_test_client.get('/api/v1/user/account-info')
assert response.status_code == 200
data = await response.get_json()
assert data['data'] == {
'initialized': True,
'authenticated_invitation_acceptance_enabled': True,
'password_login_enabled': False,
'space_login_enabled': True,
}
@pytest.mark.asyncio
async def test_recovery_key_resets_any_existing_account(self, quart_test_client, fake_api_app, monkeypatch):
fake_api_app.user_service.is_initialized.return_value = True
tests/integration/api/test_support_admin_launch.py
-1
View File
@@ -333,7 +333,6 @@ async def test_support_admin_request_context_has_actor_owner_and_no_membership(s
assert Permission.RESOURCE_MANAGE.value in permissions
assert not permissions.intersection(
{
Permission.OWNER_TRANSFER.value,
Permission.MEMBER_VIEW.value,
Permission.MEMBER_INVITE.value,
Permission.MEMBER_UPDATE_ROLE.value,
tests/integration/api/test_user_space_oauth.py
+27 -1
View File
@@ -272,9 +272,10 @@ async def test_space_credits_are_resolved_from_workspace_owner(space_oauth_api):
'/api/v1/user/space-credits',
headers={'Authorization': 'Bearer account-token', 'X-Workspace-Id': WORKSPACE_UUID},
)
payload = await response.get_json()
assert response.status_code == 200
assert (await response.get_json())['data'] == {
assert payload['data'] == {
'credits': 25000,
'owner_space_bound': True,
'is_workspace_owner': True,
@@ -282,6 +283,31 @@ async def test_space_credits_are_resolved_from_workspace_owner(space_oauth_api):
application.space_service.get_credits.assert_awaited_once_with('owner@example.com')
@pytest.mark.asyncio
async def test_cloud_workspace_owner_is_always_space_bound_after_login(space_oauth_api):
application, client = space_oauth_api
application.deployment.mode = 'cloud'
application.user_service.get_workspace_owner = AsyncMock(return_value=None)
application.space_service.get_credits = AsyncMock()
application.cloud_model_catalog_service = SimpleNamespace(
get_workspace_credits=lambda workspace_uuid: 25000 if workspace_uuid == WORKSPACE_UUID else None
)
response = await client.get(
'/api/v1/user/space-credits',
headers={'Authorization': 'Bearer account-token', 'X-Workspace-Id': WORKSPACE_UUID},
)
payload = await response.get_json()
assert response.status_code == 200
assert payload['data'] == {
'credits': 25000,
'owner_space_bound': True,
'is_workspace_owner': True,
}
application.space_service.get_credits.assert_not_awaited()
@pytest.mark.asyncio
async def test_bind_callback_uses_opaque_state_and_never_treats_it_as_jwt(space_oauth_api):
application, client = space_oauth_api
tests/integration/api/test_workspaces.py
+9
View File
@@ -188,6 +188,7 @@ async def test_owner_invites_second_account_and_secret_is_not_persisted(workspac
workspace_uuid = current['workspace']['uuid']
assert current['membership']['role'] == 'owner'
assert 'member.invite' in current['permissions']
assert 'owner.transfer' not in current['permissions']
invite_response = await client.post(
f'/api/v1/workspaces/{workspace_uuid}/invitations',
@@ -263,6 +264,14 @@ async def test_owner_invites_second_account_and_secret_is_not_persisted(workspac
assert member_current['membership']['role'] == 'viewer'
assert 'member.invite' not in member_current['permissions']
transfer_response = await client.patch(
f'/api/v1/workspaces/{workspace_uuid}/members/{member_current["membership"]["account_uuid"]}',
headers=_auth(owner_token, workspace_uuid),
json={'role': 'owner'},
)
assert transfer_response.status_code == 403
assert (await transfer_response.get_json())['code'] == 'permission_denied'
forbidden_invite = await client.post(
f'/api/v1/workspaces/{workspace_uuid}/invitations',
headers=_auth(member_token, workspace_uuid),
tests/integration/persistence/test_migrations.py
+24
View File
@@ -95,6 +95,30 @@ 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()
assert _get_script_head() == '0020_merge_agent_cloud_heads'
@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()
assert _get_script_head() == '0020_merge_agent_cloud_heads'
@pytest.mark.asyncio
async def test_upgrade_from_baseline_to_head(self, sqlite_engine):
"""
tests/integration/persistence/test_single_owner_migration.py
+100
View File
@@ -0,0 +1,100 @@
from __future__ import annotations
import pytest
import sqlalchemy as sa
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from langbot.pkg.entity.persistence.base import Base
from langbot.pkg.entity.persistence.user import User
from langbot.pkg.entity.persistence.workspace import Workspace, WorkspaceMembership
from langbot.pkg.persistence.alembic_runner import run_alembic_stamp, run_alembic_upgrade
@pytest.mark.asyncio
async def test_single_owner_migration_demotes_historical_extra_owner_and_installs_unique_index(tmp_path):
engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "single-owner.db"}')
try:
async with engine.begin() as connection:
await connection.run_sync(Base.metadata.create_all)
await connection.execute(sa.text('DROP INDEX uq_workspace_memberships_one_active_owner'))
session_factory = async_sessionmaker(engine, expire_on_commit=False)
workspace_uuid = '00000000-0000-4000-8000-000000000001'
creator_uuid = '00000000-0000-4000-8000-000000000010'
promoted_uuid = '00000000-0000-4000-8000-000000000020'
async with session_factory() as session:
session.add_all(
[
User(
uuid=creator_uuid,
user='creator@example.test',
normalized_email='creator@example.test',
password='hash',
account_type='local',
),
User(
uuid=promoted_uuid,
user='promoted@example.test',
normalized_email='promoted@example.test',
password='hash',
account_type='local',
),
Workspace(
uuid=workspace_uuid,
instance_uuid='instance-test',
name='Workspace',
slug='workspace',
type='team',
status='active',
source='local',
created_by_account_uuid=creator_uuid,
),
WorkspaceMembership(
uuid='00000000-0000-4000-8000-000000000100',
workspace_uuid=workspace_uuid,
account_uuid=creator_uuid,
role='owner',
status='active',
),
WorkspaceMembership(
uuid='00000000-0000-4000-8000-000000000200',
workspace_uuid=workspace_uuid,
account_uuid=promoted_uuid,
role='owner',
status='active',
),
]
)
await session.commit()
await run_alembic_stamp(engine, '0018_merge_launch_replay')
await run_alembic_upgrade(engine, 'head')
async with engine.connect() as connection:
roles = dict(
(
await connection.execute(
sa.text(
'SELECT account_uuid, role FROM workspace_memberships '
'WHERE workspace_uuid = :workspace_uuid ORDER BY account_uuid'
),
{'workspace_uuid': workspace_uuid},
)
).all()
)
assert roles == {creator_uuid: 'owner', promoted_uuid: 'admin'}
indexes = await connection.run_sync(
lambda sync_connection: {
index['name'] for index in sa.inspect(sync_connection).get_indexes('workspace_memberships')
}
)
assert 'uq_workspace_memberships_one_active_owner' in indexes
with pytest.raises(sa.exc.IntegrityError):
async with engine.begin() as connection:
await connection.execute(
sa.text("UPDATE workspace_memberships SET role = 'owner' WHERE account_uuid = :account_uuid"),
{'account_uuid': promoted_uuid},
)
finally:
await engine.dispose()
tests/integration/persistence/test_workspace_migration.py
+1 -1
View File
@@ -180,7 +180,7 @@ async def test_workspace_upgrade_repairs_ownerless_existing_local_workspace(lega
assert membership['status'] == 'active'
async def test_published_owner_repair_revision_upgrades_to_merged_head(legacy_engine):
async def test_development_owner_repair_revision_upgrades_to_merged_head(legacy_engine):
await run_alembic_upgrade(legacy_engine, '0017_local_owner_repair')
assert await get_alembic_current(legacy_engine) == '0017_local_owner_repair'
tests/unit_tests/api/http/test_authz.py
+1 -2
View File
@@ -27,10 +27,9 @@ def test_owner_has_every_fixed_permission():
assert ctx.workspace.permissions == frozenset(permission.value for permission in authz.Permission)
def test_admin_cannot_transfer_owner_delete_workspace_or_link_billing():
def test_admin_cannot_delete_workspace_or_link_billing():
ctx = _context(authz.WorkspaceRole.ADMIN)
assert not authz.has_permission(ctx, authz.Permission.OWNER_TRANSFER)
assert not authz.has_permission(ctx, authz.Permission.WORKSPACE_DELETE)
assert not authz.has_permission(ctx, authz.Permission.BILLING_LINK_MANAGE)
assert authz.has_permission(ctx, authz.Permission.MEMBER_INVITE)
tests/unit_tests/api/service/test_cloud_managed_model_protection.py
+112
View File
@@ -0,0 +1,112 @@
"""Cloud Runtime write protection for the managed LangBot Models catalog."""
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
from langbot.pkg.api.http.service import model as model_service_module
from langbot.pkg.api.http.service.model import (
EmbeddingModelsService,
LLMModelsService,
RerankModelsService,
_assert_cloud_managed_provider_mutable,
)
from langbot.pkg.cloud.model_catalog import LANGBOT_MODELS_PROVIDER_REQUESTER
WORKSPACE = 'workspace-a'
PROVIDER = 'managed-provider'
MODEL = 'managed-model'
@pytest.mark.asyncio
async def test_managed_provider_guard_is_cloud_only(monkeypatch) -> None:
async def managed_provider(_ap, _context, provider_uuid):
assert provider_uuid == PROVIDER
return {'uuid': PROVIDER, 'requester': LANGBOT_MODELS_PROVIDER_REQUESTER}
monkeypatch.setattr(model_service_module, '_require_workspace_provider', managed_provider)
application = SimpleNamespace(persistence_mgr=SimpleNamespace(mode=SimpleNamespace(value='cloud_runtime')))
with pytest.raises(ValueError, match='managed by Cloud'):
await _assert_cloud_managed_provider_mutable(
application,
WORKSPACE,
PROVIDER,
)
application.persistence_mgr.mode.value = 'normal'
await _assert_cloud_managed_provider_mutable(
application,
WORKSPACE,
PROVIDER,
)
@pytest.mark.parametrize(
('service_type', 'create_method', 'model_data'),
[
(LLMModelsService, 'create_llm_model', {'provider_uuid': PROVIDER, 'name': 'chat', 'abilities': []}),
(EmbeddingModelsService, 'create_embedding_model', {'provider_uuid': PROVIDER, 'name': 'embedding'}),
(RerankModelsService, 'create_rerank_model', {'provider_uuid': PROVIDER, 'name': 'rerank'}),
],
)
@pytest.mark.asyncio
async def test_all_model_types_reject_creation_under_managed_provider(
monkeypatch,
service_type,
create_method: str,
model_data: dict,
) -> None:
guard = AsyncMock(side_effect=ValueError('LangBot Models is managed by Cloud and cannot be modified'))
monkeypatch.setattr(model_service_module, '_assert_cloud_managed_provider_mutable', guard)
application = SimpleNamespace(
persistence_mgr=SimpleNamespace(),
provider_service=SimpleNamespace(
get_provider=AsyncMock(return_value={'uuid': PROVIDER, 'requester': LANGBOT_MODELS_PROVIDER_REQUESTER})
),
model_mgr=None,
)
service = service_type(application)
with pytest.raises(ValueError, match='managed by Cloud'):
await getattr(service, create_method)(WORKSPACE, model_data)
guard.assert_awaited_once()
@pytest.mark.parametrize(
('service_type', 'get_method', 'write_method', 'payload'),
[
(LLMModelsService, 'get_llm_model', 'update_llm_model', {'name': 'changed'}),
(LLMModelsService, 'get_llm_model', 'delete_llm_model', None),
(EmbeddingModelsService, 'get_embedding_model', 'update_embedding_model', {'name': 'changed'}),
(EmbeddingModelsService, 'get_embedding_model', 'delete_embedding_model', None),
(RerankModelsService, 'get_rerank_model', 'update_rerank_model', {'name': 'changed'}),
(RerankModelsService, 'get_rerank_model', 'delete_rerank_model', None),
],
)
@pytest.mark.asyncio
async def test_all_model_types_reject_update_and_delete_for_managed_provider(
monkeypatch,
service_type,
get_method: str,
write_method: str,
payload: dict | None,
) -> None:
guard = AsyncMock(side_effect=ValueError('LangBot Models is managed by Cloud and cannot be modified'))
monkeypatch.setattr(model_service_module, '_assert_cloud_managed_provider_mutable', guard)
application = SimpleNamespace(persistence_mgr=SimpleNamespace(mode=SimpleNamespace(value='cloud_runtime')))
service = service_type(application)
monkeypatch.setattr(
service,
get_method,
AsyncMock(return_value={'uuid': MODEL, 'provider_uuid': PROVIDER, 'extra_args': {}}),
)
args = (WORKSPACE, MODEL) if payload is None else (WORKSPACE, MODEL, payload)
with pytest.raises(ValueError, match='managed by Cloud'):
await getattr(service, write_method)(*args)
guard.assert_awaited_once()
tests/unit_tests/api/service/test_provider_service.py
+54
View File
@@ -25,6 +25,7 @@ from langbot.pkg.workspace.errors import WorkspaceNotFoundError
pytestmark = pytest.mark.asyncio
WORKSPACE_UUID = 'workspace-a'
SYSTEM_REQUESTER = 'space-chat-completions'
def _create_mock_provider(
@@ -1005,3 +1006,56 @@ class TestProviderSecretRoundtrip:
)
ap.persistence_mgr.execute_async.assert_not_awaited()
class TestCloudManagedProviderProtection:
@staticmethod
def _service() -> ModelProviderService:
ap = SimpleNamespace(
persistence_mgr=SimpleNamespace(
mode=SimpleNamespace(value='cloud_runtime'),
execute_async=AsyncMock(),
),
model_mgr=SimpleNamespace(),
)
return ModelProviderService(ap)
async def test_cloud_rejects_user_created_system_requester(self):
service = self._service()
with pytest.raises(ValueError, match='reserved'):
await service.create_provider(
WORKSPACE_UUID,
{
'name': 'Fake LangBot Models',
'requester': SYSTEM_REQUESTER,
'base_url': 'https://example.invalid/v1',
'api_keys': ['fake'],
},
)
with pytest.raises(ValueError, match='reserved'):
await service.find_or_create_provider(
WORKSPACE_UUID,
SYSTEM_REQUESTER,
'https://api.langbot.cloud/v1',
['fake'],
)
service.ap.persistence_mgr.execute_async.assert_not_awaited()
async def test_cloud_rejects_update_and_delete_of_managed_provider(self):
service = self._service()
service.get_provider = AsyncMock(
return_value={'uuid': 'system-provider', 'requester': SYSTEM_REQUESTER}
)
with pytest.raises(ValueError, match='managed by Cloud'):
await service.update_provider(WORKSPACE_UUID, 'system-provider', {'name': 'Renamed'})
with pytest.raises(ValueError, match='managed by Cloud'):
await service.delete_provider(WORKSPACE_UUID, 'system-provider')
service.ap.persistence_mgr.execute_async.assert_not_awaited()
async def test_oss_does_not_reserve_space_requester(self):
ap = SimpleNamespace(persistence_mgr=SimpleNamespace(mode=SimpleNamespace(value='oss_compat')))
service = ModelProviderService(ap)
assert service._system_requester_is_reserved(SYSTEM_REQUESTER) is False
tests/unit_tests/cloud/test_bootstrap.py
+19 -1
View File
@@ -66,6 +66,10 @@ class _Provider:
def __init__(self):
self.manifest_provider = _Manifest()
async def fetch_model_catalog(self, instance_uuid: str):
del instance_uuid
raise AssertionError('not used by bootstrap contract tests')
def bootstrap(self, *, instance_uuid: str, instance_config: dict):
del instance_config
return VerifiedCloudDeployment(
@@ -79,6 +83,7 @@ class _Provider:
entitlement_provider=_Entitlements(),
directory_provider=_Directory(),
manifest_provider=self.manifest_provider,
model_catalog_provider=self,
verification_key_id='root-2026',
)
@@ -228,10 +233,23 @@ async def test_cloud_pgvector_contract_is_fail_closed(pgvector_config, message):
)
async def test_cloud_runtime_allows_explicitly_disabled_box():
config = _cloud_config()
config['box']['enabled'] = False
deployment = await resolve_deployment(
instance_uuid='instance-a',
instance_config=config,
entry_points=lambda: _EntryPoints([_EntryPoint(_Provider())]),
now=1_000,
)
assert isinstance(deployment, VerifiedCloudDeployment)
@pytest.mark.parametrize(
('mutate', 'message'),
[
(lambda config: config['box'].update(enabled=False), 'box.enabled=true'),
(lambda config: config['box'].update(backend='docker'), 'box.backend=nsjail'),
(lambda config: config['box']['runtime'].update(endpoint=''), 'box.runtime.endpoint'),
(
tests/unit_tests/cloud/test_model_catalog.py
+378
View File
@@ -0,0 +1,378 @@
from __future__ import annotations
import asyncio
import logging
from datetime import UTC, datetime
from types import SimpleNamespace
import pytest
import sqlalchemy
from sqlalchemy.ext.asyncio import create_async_engine
from langbot.pkg.cloud.model_catalog import (
CloudModelCatalogSnapshot,
CloudModelCatalogSyncService,
system_model_uuid,
system_provider_uuid,
)
from langbot.pkg.entity.persistence.base import Base
from langbot.pkg.entity.persistence.model import EmbeddingModel, LLMModel, ModelProvider
from langbot.pkg.entity.persistence.workspace import Workspace
from langbot.pkg.persistence.mgr import PersistenceManager, PersistenceMode
pytestmark = pytest.mark.asyncio
INSTANCE_UUID = 'instance-model-catalog'
WORKSPACE_A = '00000000-0000-4000-8000-000000000001'
WORKSPACE_B = '00000000-0000-4000-8000-000000000002'
OWNER_A = '10000000-0000-4000-8000-000000000001'
OWNER_B = '10000000-0000-4000-8000-000000000002'
class _CatalogProvider:
def __init__(self, snapshot: CloudModelCatalogSnapshot) -> None:
self.snapshot = snapshot
async def fetch_model_catalog(self, instance_uuid: str) -> CloudModelCatalogSnapshot:
assert instance_uuid == INSTANCE_UUID
return self.snapshot
def _snapshot(
*,
key_a: str | None = 'owner-a-key',
model_id: str = 'gpt-test',
include_embedding: bool = True,
) -> CloudModelCatalogSnapshot:
models = [
{
'uuid': 'upstream-chat',
'model_id': model_id,
'category': 'chat',
'llm_abilities': ['chat', 'vision'],
'is_featured': True,
'featured_order': 7,
}
]
if include_embedding:
models.append(
{
'uuid': 'upstream-embedding',
'model_id': 'embedding-test',
'category': 'embedding',
}
)
return CloudModelCatalogSnapshot.model_validate(
{
'instance_uuid': INSTANCE_UUID,
'generated_at': datetime.now(UTC),
'base_url': 'https://api.langbot.cloud/v1/',
'models': models,
'workspaces': [
{
'workspace_uuid': WORKSPACE_A,
'owner_account_uuid': OWNER_A,
'api_key': key_a,
'credits': 25000,
},
{
'workspace_uuid': WORKSPACE_B,
'owner_account_uuid': OWNER_B,
'api_key': 'owner-b-key',
'credits': 5000,
},
],
}
)
async def test_catalog_snapshot_treats_null_model_abilities_as_empty() -> None:
payload = _snapshot().model_dump(mode='json')
payload['models'][0]['llm_abilities'] = None
snapshot = CloudModelCatalogSnapshot.model_validate(payload)
assert snapshot.models[0].llm_abilities == ()
async def test_catalog_reconciles_every_workspace_idempotently_and_tracks_owner_and_downlisting(tmp_path) -> None:
engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "model-catalog.db"}')
manager = PersistenceManager(object(), mode=PersistenceMode.CLOUD_RUNTIME)
manager.db = SimpleNamespace(get_engine=lambda: engine)
bindings = [
SimpleNamespace(instance_uuid=INSTANCE_UUID, workspace_uuid=WORKSPACE_A, placement_generation=1),
SimpleNamespace(instance_uuid=INSTANCE_UUID, workspace_uuid=WORKSPACE_B, placement_generation=1),
]
workspace_service = SimpleNamespace(list_active_execution_bindings=lambda: _async_value(bindings))
reload_counter = _AsyncCounter()
runtime_reload = SimpleNamespace(load_models_from_db=reload_counter)
app = SimpleNamespace(
persistence_mgr=manager,
workspace_service=workspace_service,
model_mgr=runtime_reload,
logger=logging.getLogger(__name__),
)
provider = _CatalogProvider(_snapshot())
service = CloudModelCatalogSyncService(app, provider, INSTANCE_UUID)
try:
async with engine.begin() as connection:
await connection.run_sync(Base.metadata.create_all)
await connection.execute(
sqlalchemy.insert(Workspace),
[
{
'uuid': WORKSPACE_A,
'instance_uuid': INSTANCE_UUID,
'name': 'A',
'slug': 'a',
'source': 'cloud_projection',
},
{
'uuid': WORKSPACE_B,
'instance_uuid': INSTANCE_UUID,
'name': 'B',
'slug': 'b',
'source': 'cloud_projection',
},
],
)
await connection.execute(
sqlalchemy.insert(ModelProvider).values(
uuid='custom-provider',
workspace_uuid=WORKSPACE_A,
name='Custom',
requester='openai-chat-completions',
base_url='https://custom.example/v1',
api_keys=['custom-key'],
)
)
await connection.execute(
sqlalchemy.insert(LLMModel).values(
uuid='custom-model',
workspace_uuid=WORKSPACE_A,
name='custom-model',
provider_uuid='custom-provider',
abilities=['chat'],
extra_args={},
prefered_ranking=0,
)
)
first = await service.sync_once()
assert first == {'workspaces': 2, 'created': 6, 'updated': 0, 'deleted': 0}
assert reload_counter.calls == 1
assert service.get_workspace_credits(WORKSPACE_A) == 25000
assert service.get_workspace_credits(WORKSPACE_B) == 5000
async with engine.connect() as connection:
providers = (
await connection.execute(
sqlalchemy.select(
ModelProvider.uuid,
ModelProvider.workspace_uuid,
ModelProvider.api_keys,
).where(ModelProvider.requester == 'space-chat-completions')
)
).all()
assert {item.workspace_uuid for item in providers} == {WORKSPACE_A, WORKSPACE_B}
assert {item.uuid for item in providers} == {
system_provider_uuid(WORKSPACE_A),
system_provider_uuid(WORKSPACE_B),
}
assert {item.workspace_uuid: item.api_keys for item in providers} == {
WORKSPACE_A: ['owner-a-key'],
WORKSPACE_B: ['owner-b-key'],
}
assert await connection.scalar(sqlalchemy.select(sqlalchemy.func.count()).select_from(LLMModel)) == 3
assert await connection.scalar(sqlalchemy.select(sqlalchemy.func.count()).select_from(EmbeddingModel)) == 2
second = await service.sync_once()
assert second == {'workspaces': 2, 'created': 0, 'updated': 0, 'deleted': 0}
assert reload_counter.calls == 1
provider.snapshot = _snapshot(
key_a='new-owner-key',
model_id='gpt-renamed',
include_embedding=False,
)
third = await service.sync_once()
assert third == {'workspaces': 2, 'created': 0, 'updated': 3, 'deleted': 2}
assert reload_counter.calls == 2
async with engine.connect() as connection:
provider_a_keys = await connection.scalar(
sqlalchemy.select(ModelProvider.api_keys).where(ModelProvider.uuid == system_provider_uuid(WORKSPACE_A))
)
assert provider_a_keys == ['new-owner-key']
system_model_names = (
(
await connection.execute(
sqlalchemy.select(LLMModel.name).where(
LLMModel.provider_uuid.in_(
[system_provider_uuid(WORKSPACE_A), system_provider_uuid(WORKSPACE_B)]
)
)
)
)
.scalars()
.all()
)
assert set(system_model_names) == {'gpt-renamed'}
assert await connection.scalar(sqlalchemy.select(sqlalchemy.func.count()).select_from(EmbeddingModel)) == 0
assert (
await connection.scalar(
sqlalchemy.select(sqlalchemy.func.count())
.select_from(ModelProvider)
.where(ModelProvider.uuid == 'custom-provider')
)
== 1
)
assert (
await connection.scalar(
sqlalchemy.select(sqlalchemy.func.count())
.select_from(LLMModel)
.where(LLMModel.uuid == 'custom-model')
)
== 1
)
provider.snapshot = _snapshot(key_a=None, model_id='gpt-renamed', include_embedding=False)
fourth = await service.sync_once()
assert fourth == {'workspaces': 2, 'created': 0, 'updated': 1, 'deleted': 0}
assert reload_counter.calls == 3
async with engine.connect() as connection:
provider_a_keys = await connection.scalar(
sqlalchemy.select(ModelProvider.api_keys).where(ModelProvider.uuid == system_provider_uuid(WORKSPACE_A))
)
assert provider_a_keys == []
finally:
await engine.dispose()
def test_workspace_scoped_ids_are_stable_and_secrets_are_redacted() -> None:
assert system_provider_uuid(WORKSPACE_A) == system_provider_uuid(WORKSPACE_A)
assert system_provider_uuid(WORKSPACE_A) != system_provider_uuid(WORKSPACE_B)
assert system_model_uuid(WORKSPACE_A, 'chat', 'upstream') != system_model_uuid(WORKSPACE_B, 'chat', 'upstream')
snapshot = _snapshot()
assert 'owner-a-key' not in repr(snapshot)
async def test_snapshot_must_cover_every_active_workspace() -> None:
snapshot = _snapshot().model_copy(update={'workspaces': _snapshot().workspaces[:1]})
app = SimpleNamespace(
workspace_service=SimpleNamespace(
list_active_execution_bindings=lambda: _async_value(
[SimpleNamespace(workspace_uuid=WORKSPACE_A), SimpleNamespace(workspace_uuid=WORKSPACE_B)]
)
),
logger=logging.getLogger(__name__),
)
service = CloudModelCatalogSyncService(app, _CatalogProvider(snapshot), INSTANCE_UUID)
with pytest.raises(ValueError, match='missing billing projections for 1 active Workspaces'):
await service.sync_once()
async def _async_value(value):
return value
class _AsyncCounter:
def __init__(self) -> None:
self.calls = 0
async def __call__(self) -> None:
self.calls += 1
async def test_partial_workspace_failure_reloads_already_committed_changes() -> None:
bindings = [
SimpleNamespace(workspace_uuid=WORKSPACE_A),
SimpleNamespace(workspace_uuid=WORKSPACE_B),
]
reload_counter = _AsyncCounter()
app = SimpleNamespace(
workspace_service=SimpleNamespace(list_active_execution_bindings=lambda: _async_value(bindings)),
model_mgr=SimpleNamespace(load_models_from_db=reload_counter),
logger=logging.getLogger(__name__),
)
service = CloudModelCatalogSyncService(app, _CatalogProvider(_snapshot()), INSTANCE_UUID)
calls = 0
async def sync_workspace(*_args):
nonlocal calls
calls += 1
if calls == 1:
return {'created': 1, 'updated': 0, 'deleted': 0}
raise RuntimeError('second Workspace failed')
service._sync_workspace = sync_workspace # type: ignore[method-assign]
with pytest.raises(RuntimeError, match='second Workspace failed'):
await service.sync_once()
assert service.get_workspace_credits(WORKSPACE_A) == 25000
assert service.get_workspace_credits(WORKSPACE_B) is None
assert reload_counter.calls == 1
async def test_failed_runtime_reload_is_retried_after_noop_sync() -> None:
bindings = [SimpleNamespace(workspace_uuid=WORKSPACE_A)]
class _FlakyReload:
def __init__(self) -> None:
self.calls = 0
async def __call__(self) -> None:
self.calls += 1
if self.calls == 1:
raise RuntimeError('reload failed')
runtime_reload = _FlakyReload()
app = SimpleNamespace(
workspace_service=SimpleNamespace(list_active_execution_bindings=lambda: _async_value(bindings)),
model_mgr=SimpleNamespace(load_models_from_db=runtime_reload),
logger=logging.getLogger(__name__),
)
service = CloudModelCatalogSyncService(app, _CatalogProvider(_snapshot()), INSTANCE_UUID)
calls = 0
async def sync_workspace(*_args):
nonlocal calls
calls += 1
if calls == 1:
return {'created': 1, 'updated': 0, 'deleted': 0}
return {'created': 0, 'updated': 0, 'deleted': 0}
service._sync_workspace = sync_workspace # type: ignore[method-assign]
with pytest.raises(RuntimeError, match='reload failed'):
await service.sync_once()
summary = await service.sync_once()
assert summary == {'workspaces': 1, 'created': 0, 'updated': 0, 'deleted': 0}
assert runtime_reload.calls == 2
async def test_background_sync_log_redacts_exception_message(caplog) -> None:
secret = 'owner-secret-api-key'
attempted = asyncio.Event()
class _FailingProvider:
async def fetch_model_catalog(self, instance_uuid: str) -> CloudModelCatalogSnapshot:
del instance_uuid
attempted.set()
raise RuntimeError(f'database parameters include {secret}')
app = SimpleNamespace(logger=logging.getLogger(__name__))
service = CloudModelCatalogSyncService(app, _FailingProvider(), INSTANCE_UUID)
service.sync_interval_seconds = 0.001
task = asyncio.create_task(service.run())
try:
await asyncio.wait_for(attempted.wait(), timeout=1)
await asyncio.sleep(0.01)
finally:
task.cancel()
with pytest.raises(asyncio.CancelledError):
await task
assert secret not in caplog.text
assert 'Cloud model catalog synchronization failed (RuntimeError)' in caplog.text
tests/unit_tests/persistence/test_tenant_uow.py
+1
View File
@@ -961,6 +961,7 @@ async def test_scoped_session_rejects_raw_or_unapproved_sql(
sa.select(sa.func.coalesce(sa.func.sum(sa.literal(1)), sa.literal(0))),
sa.select(
sa.func.now(),
sa.func.date_trunc('hour', sa.column('timestamp')),
sa.func.length(sa.literal('value')),
sa.func.nullif(sa.literal('value'), sa.literal('')),
),
tests/unit_tests/pipeline/test_controller_tenancy.py
+29
View File
@@ -11,6 +11,7 @@ from sqlalchemy.ext.asyncio import create_async_engine
from langbot.pkg.persistence.mgr import PersistenceManager, PersistenceMode
from langbot.pkg.persistence.tenant_uow import PersistenceScopeKind
from langbot.pkg.pipeline.controller import Controller
from langbot.pkg.pipeline.pool import QueryPool
from langbot.pkg.workspace.errors import WorkspaceGenerationMismatchError
@@ -143,3 +144,31 @@ async def test_controller_revalidates_generation_before_running_pipeline(
runtime_pipeline.run.assert_awaited_once_with(sample_query)
query_pool.remove_query.assert_awaited_once_with(sample_query)
session._semaphore.release.assert_called_once_with()
@pytest.mark.asyncio
async def test_controller_schedules_query_without_removing_it_twice(mock_app, sample_query):
query_pool = QueryPool()
query_pool.queries.append(sample_query)
mock_app.query_pool = query_pool
mock_app.sess_mgr.get_session = AsyncMock(return_value=SimpleNamespace(_semaphore=asyncio.Semaphore(1)))
scheduler_errors: list[str] = []
def stop_on_scheduler_error(message):
scheduler_errors.append(str(message))
raise asyncio.CancelledError
def stop_after_scheduling(process_coro, **_kwargs):
process_coro.close()
raise asyncio.CancelledError
mock_app.logger.error.side_effect = stop_on_scheduler_error
mock_app.task_mgr.create_task.side_effect = stop_after_scheduling
controller = Controller(mock_app)
with pytest.raises(asyncio.CancelledError):
await controller.consumer()
assert scheduler_errors == []
assert query_pool.queries == []
tests/unit_tests/platform/test_websocket_session_isolation.py
+43
View File
@@ -1,6 +1,7 @@
"""Regression tests for isolated embed-widget conversations."""
import asyncio
import contextvars
from pathlib import Path
from unittest.mock import AsyncMock, Mock
@@ -207,6 +208,48 @@ async def test_embed_event_uses_stable_session_launcher(monkeypatch):
assert received[0].sender.id == f'websocket_pipeline-1:{session_id}'
@pytest.mark.asyncio
async def test_pipeline_override_survives_detached_listener_task(monkeypatch):
manager = WebSocketConnectionManager()
connection = await manager.add_connection(
websocket=Mock(),
scope=SCOPE_A,
pipeline_uuid='pipeline-1',
session_type='person',
)
monkeypatch.setattr(websocket_adapter_module, 'ws_connection_manager', manager)
class DetachedTaskManager:
def __init__(self):
self.tasks = []
def create_task(self, coro, **_kwargs):
task = asyncio.create_task(coro, context=contextvars.Context())
self.tasks.append(task)
return Mock(task=task)
task_manager = DetachedTaskManager()
adapter = WebSocketAdapter.model_construct(
ap=Mock(task_mgr=task_manager),
logger=_adapter_logger(),
)
adapter.websocket_person_session = WebSocketSession(id='person')
adapter.websocket_group_session = WebSocketSession(id='group')
pipeline_overrides = []
async def listener(_event, callback_adapter):
pipeline_overrides.append(callback_adapter.get_pipeline_uuid_override())
adapter.listeners = {platform_events.FriendMessage: listener}
await adapter.handle_websocket_message(
connection,
{'message': [{'type': 'Plain', 'text': 'hello'}], 'stream': False},
)
await asyncio.gather(*task_manager.tasks)
assert pipeline_overrides == ['pipeline-1']
@pytest.mark.asyncio
async def test_embed_group_event_uses_stable_session_launcher(monkeypatch):
manager = WebSocketConnectionManager()
tests/unit_tests/workspace/test_workspace_collaboration.py
+15 -14
View File
@@ -203,20 +203,21 @@ async def test_last_owner_cannot_be_demoted(collaboration_context):
second_membership,
)
promoted = await service.update_member_role(
workspace.uuid,
second.uuid,
'owner',
owner_membership,
)
assert promoted.role == 'owner'
demoted = await service.update_member_role(
workspace.uuid,
owner_membership.account_uuid,
'admin',
owner_membership,
)
assert demoted.role == 'admin'
with pytest.raises(MembershipPermissionError, match='cannot be transferred'):
await service.update_member_role(
workspace.uuid,
second.uuid,
'owner',
owner_membership,
)
with pytest.raises(LastOwnerError):
await service.update_member_role(
workspace.uuid,
owner_membership.account_uuid,
'admin',
owner_membership,
)
async def test_workspace_selector_requires_membership(collaboration_context):