mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-11-01 21:35:25 +08:00
293 lines
9.9 KiB
Python
293 lines
9.9 KiB
Python
"""Blocking interaction and callback authorization tests."""
|
|
|
|
import asyncio
|
|
import time
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
|
|
from langbot_plugin.api.entities.builtin.runner.interaction import (
|
|
InteractionDeliveryCapabilities,
|
|
InteractionHandle,
|
|
InteractionRequest,
|
|
)
|
|
from langbot.pkg.agent.runner.interaction_manager import InteractionManager
|
|
from langbot.pkg.agent.runner.interaction_store import (
|
|
InteractionAlreadySubmittedError,
|
|
InteractionExpiredError,
|
|
InteractionNotFoundError,
|
|
InteractionScopeError,
|
|
InteractionStore,
|
|
)
|
|
from langbot.pkg.entity.persistence.base import Base
|
|
|
|
|
|
def _session():
|
|
return {
|
|
'run_id': 'run-1',
|
|
'runner_id': 'plugin:test/Runner/default',
|
|
'runtime_context': {'deadline_at': time.time() + 300},
|
|
'authorization': {
|
|
'binding_id': 'binding-1',
|
|
'processor_type': 'agent',
|
|
'processor_id': 'agent-1',
|
|
'bot_id': 'bot-1',
|
|
'workspace_id': 'workspace-1',
|
|
'thread_id': None,
|
|
'interactions_enabled': True,
|
|
'interactions_authorized': True,
|
|
'platform_context': {
|
|
'actor': {'actor_id': 'user-1'},
|
|
'delivery': {'reply_target': {'target_type': 'group', 'target_id': 'chat-1'}},
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def _request(**changes):
|
|
return {
|
|
'interaction_id': 'approval',
|
|
'title': 'Approve?',
|
|
'actions': [{'id': 'yes', 'label': 'Yes'}, {'id': 'no', 'label': 'No'}],
|
|
**changes,
|
|
}
|
|
|
|
|
|
class Adapter:
|
|
def __init__(self, native=True):
|
|
self.native = native
|
|
self.sent = asyncio.Event()
|
|
self.token = None
|
|
self.updates = []
|
|
self.messages = []
|
|
|
|
def get_supported_apis(self):
|
|
return ['send_interaction', 'update_interaction', 'send_message'] if self.native else ['send_message']
|
|
|
|
def get_interaction_capabilities(self):
|
|
return InteractionDeliveryCapabilities(
|
|
field_types=['text', 'select'] if self.native else [],
|
|
action_styles=['default', 'primary'] if self.native else [],
|
|
max_fields=8 if self.native else None,
|
|
max_actions=6 if self.native else None,
|
|
supports_updates=self.native,
|
|
)
|
|
|
|
async def send_interaction(self, target_type, target_id, request, callback_token, reply_to=None):
|
|
self.token = callback_token
|
|
self.sent.set()
|
|
return InteractionHandle(message_id='card-1')
|
|
|
|
async def update_interaction(self, handle, state, submission=None):
|
|
self.updates.append(state)
|
|
return handle
|
|
|
|
async def send_message(self, target_type, target_id, message):
|
|
self.messages.append(str(message))
|
|
self.sent.set()
|
|
|
|
|
|
@pytest.fixture
|
|
async def store(tmp_path):
|
|
engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "interactions.db"}')
|
|
async with engine.begin() as connection:
|
|
await connection.run_sync(Base.metadata.create_all)
|
|
yield InteractionStore(engine)
|
|
await engine.dispose()
|
|
|
|
|
|
async def _start(manager, adapter, *, request=None):
|
|
task = asyncio.create_task(
|
|
manager.request_interaction(
|
|
session=_session(),
|
|
adapter=adapter,
|
|
request_payload=request or _request(),
|
|
timeout=2,
|
|
on_unsupported='error',
|
|
responders='actor',
|
|
)
|
|
)
|
|
await asyncio.wait_for(adapter.sent.wait(), 1)
|
|
return task
|
|
|
|
|
|
def _callback(token, **overrides):
|
|
return {
|
|
**dict(
|
|
callback_token=token,
|
|
submission={'action_ref': 0},
|
|
bot_id='bot-1',
|
|
conversation_id='group_chat-1',
|
|
actor_id='user-1',
|
|
),
|
|
**overrides,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_callback_is_single_use_and_stays_in_same_run(store):
|
|
adapter = Adapter()
|
|
manager = InteractionManager(SimpleNamespace(), store=store)
|
|
task = await _start(manager, adapter)
|
|
record = await manager.consume_callback(**_callback(adapter.token, submission={'action_ref': 0, 'action_id': 'no'}))
|
|
assert record['run_id'] == 'run-1'
|
|
assert manager.wake_waiter(record, 'user-1')
|
|
assert (await task)['submission']['action_id'] == 'yes'
|
|
assert adapter.updates == ['submitted']
|
|
with pytest.raises(InteractionAlreadySubmittedError):
|
|
await manager.consume_callback(**_callback(adapter.token))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_forged_token_actor_and_conversation_do_not_consume(store):
|
|
adapter = Adapter()
|
|
manager = InteractionManager(SimpleNamespace(), store=store)
|
|
task = await _start(manager, adapter)
|
|
with pytest.raises(InteractionNotFoundError):
|
|
await manager.consume_callback(**_callback('forged'))
|
|
for override in ({'actor_id': 'attacker'}, {'conversation_id': 'group-other'}, {'bot_id': 'bot-other'}):
|
|
with pytest.raises(InteractionScopeError):
|
|
await manager.consume_callback(**_callback(adapter.token, **override))
|
|
assert (await store.get_request('run-1', 'approval'))['status'] == 'pending'
|
|
record = await manager.consume_callback(**_callback(adapter.token))
|
|
assert manager.wake_waiter(record, 'user-1')
|
|
await task
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_responder_list_rejects_other_actor_before_consuming(store):
|
|
adapter = Adapter()
|
|
manager = InteractionManager(SimpleNamespace(), store=store)
|
|
task = asyncio.create_task(
|
|
manager.request_interaction(
|
|
session=_session(),
|
|
adapter=adapter,
|
|
request_payload=_request(),
|
|
timeout=2,
|
|
on_unsupported='error',
|
|
responders=['user-1'],
|
|
)
|
|
)
|
|
await asyncio.wait_for(adapter.sent.wait(), 1)
|
|
with pytest.raises(ValueError, match='allowed responder'):
|
|
await manager.consume_callback(**_callback(adapter.token, actor_id='attacker'))
|
|
assert (await store.get_request('run-1', 'approval'))['status'] == 'pending'
|
|
record = await manager.consume_callback(**_callback(adapter.token))
|
|
assert manager.wake_waiter(record, 'user-1')
|
|
await task
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_restart_cancels_pending_and_old_callback_expires(store):
|
|
adapter = Adapter()
|
|
manager = InteractionManager(SimpleNamespace(), store=store)
|
|
task = await _start(manager, adapter)
|
|
assert await store.cancel_orphaned_pending(set()) == 1
|
|
assert (await store.get_request('run-1', 'approval'))['status'] == 'cancelled'
|
|
with pytest.raises(InteractionExpiredError):
|
|
await manager.consume_callback(**_callback(adapter.token))
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_timeout_updates_card(store):
|
|
adapter = Adapter()
|
|
manager = InteractionManager(SimpleNamespace(), store=store)
|
|
result = await manager.request_interaction(
|
|
session=_session(),
|
|
adapter=adapter,
|
|
request_payload=_request(),
|
|
timeout=0.01,
|
|
on_unsupported='error',
|
|
responders='actor',
|
|
)
|
|
assert result['status'] == 'timeout'
|
|
assert adapter.updates == ['expired']
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize('policy', ['degrade', 'error'])
|
|
async def test_multi_field_form_never_splits_into_text_questions(store, policy):
|
|
adapter = Adapter(native=False)
|
|
manager = InteractionManager(SimpleNamespace(), store=store)
|
|
result = await manager.request_interaction(
|
|
session=_session(),
|
|
adapter=adapter,
|
|
request_payload=_request(
|
|
actions=[],
|
|
fields=[
|
|
{'id': 'first', 'label': 'First', 'type': 'text'},
|
|
{'id': 'second', 'label': 'Second', 'type': 'text'},
|
|
],
|
|
),
|
|
timeout=1,
|
|
on_unsupported=policy,
|
|
responders='actor',
|
|
)
|
|
assert result['unsupported']
|
|
assert adapter.messages == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_visual_degradation_keeps_native_form(store):
|
|
adapter = Adapter()
|
|
manager = InteractionManager(SimpleNamespace(), store=store)
|
|
task = await _start(
|
|
manager,
|
|
adapter,
|
|
request=_request(
|
|
actions=[],
|
|
fields=[
|
|
{'id': 'name', 'label': 'Name', 'type': 'text', 'placeholder': 'Your name', 'default': 'Alex'},
|
|
],
|
|
),
|
|
)
|
|
record = await store.get_request('run-1', 'approval')
|
|
assert record['request']['fields'][0]['placeholder'] is None
|
|
assert record['request']['fields'][0]['default'] is None
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
|
|
|
|
@pytest.mark.parametrize('policy', ['error', 'degrade'])
|
|
def test_structural_policy_and_visual_degradation_are_independent(policy):
|
|
capabilities = InteractionDeliveryCapabilities(
|
|
field_types=['text'],
|
|
action_styles=['default'],
|
|
max_fields=8,
|
|
max_actions=2,
|
|
text_fallback=True,
|
|
supports_markdown=False,
|
|
)
|
|
structural = InteractionRequest.model_validate(
|
|
_request(
|
|
actions=[],
|
|
fields=[{'id': 'count', 'label': 'Count', 'type': 'number'}],
|
|
)
|
|
)
|
|
_, _, use_text, issues = InteractionManager._normalize_request(structural, capabilities, policy)
|
|
if policy == 'error':
|
|
assert use_text is False
|
|
assert issues == [{'path': 'fields[0].type', 'reason': 'unsupported_field_type'}]
|
|
else:
|
|
assert use_text is True
|
|
assert issues == []
|
|
|
|
visual = InteractionRequest.model_validate(
|
|
_request(
|
|
description='**Approve**',
|
|
actions=[{'id': 'yes', 'label': 'Yes', 'style': 'primary'}],
|
|
)
|
|
)
|
|
normalized, degraded, use_text, issues = InteractionManager._normalize_request(visual, capabilities, policy)
|
|
assert not use_text and not issues
|
|
assert normalized.actions[0].style == 'default'
|
|
assert {(item['path'], item['reason']) for item in degraded} == {
|
|
('actions[0].style', 'unsupported_style'),
|
|
('description', 'unsupported_markdown'),
|
|
}
|