mirror of
https://github.com/open-webui/open-webui.git
synced 2026-10-06 02:48:04 +00:00
119 lines
3.4 KiB
Python
119 lines
3.4 KiB
Python
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from open_webui import events
|
|
|
|
|
|
class DummyRequest:
|
|
def __init__(self):
|
|
self.app = SimpleNamespace(state=SimpleNamespace(instance_id='instance-1', WEBUI_NAME='Test WebUI'))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_event_derives_resource_operation_and_sanitizes():
|
|
request = DummyRequest()
|
|
|
|
event = events.build_event(
|
|
request,
|
|
events.EVENTS.KNOWLEDGE_FILE_ADDED,
|
|
actor={
|
|
'id': 'user-1',
|
|
'name': 'Ada',
|
|
'email': 'ada@example.com',
|
|
'role': 'admin',
|
|
'api_key': 'secret',
|
|
},
|
|
subject_id='file-1',
|
|
data={
|
|
'safe': 'value',
|
|
'token': 'hidden',
|
|
'nested': {'refresh_token': 'hidden', 'name': 'visible'},
|
|
'content': 'x' * (events.MAX_STRING_LENGTH + 10),
|
|
},
|
|
)
|
|
|
|
payload = event.model_dump()
|
|
|
|
assert payload['schema'] == events.EVENT_VERSION
|
|
assert payload['event'] == 'knowledge.file.added'
|
|
assert payload['resource'] == 'knowledge.file'
|
|
assert payload['operation'] == 'added'
|
|
assert payload['instance_id'] == 'instance-1'
|
|
assert payload['actor'] == {
|
|
'id': 'user-1',
|
|
'name': 'Ada',
|
|
'email': 'ada@example.com',
|
|
'role': 'admin',
|
|
'type': 'user',
|
|
}
|
|
assert 'token' not in payload['data']
|
|
assert 'refresh_token' not in payload['data']['nested']
|
|
assert payload['data']['nested']['name'] == 'visible'
|
|
assert payload['data']['content'].endswith('...')
|
|
|
|
|
|
def test_build_event_accepts_events_enum():
|
|
request = DummyRequest()
|
|
|
|
event = events.build_event(
|
|
request,
|
|
events.EVENTS.MESSAGE_CREATED,
|
|
actor={'id': 'user-1'},
|
|
subject_id='message-1',
|
|
)
|
|
|
|
payload = event.model_dump()
|
|
assert payload['event'] == 'message.created'
|
|
assert payload['resource'] == 'message'
|
|
assert payload['operation'] == 'created'
|
|
assert events.EVENTS.MESSAGE_CREATED.value in events.EVENT_CATALOG
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_webhook_sink_sends_canonical_json(monkeypatch):
|
|
request = DummyRequest()
|
|
sent = {}
|
|
|
|
async def fake_config_get(key, default=None):
|
|
assert key == 'webhook_url'
|
|
return 'https://example.com/events'
|
|
|
|
async def fake_post_webhook(name, url, message, event_data):
|
|
sent.update({'name': name, 'url': url, 'message': message, 'event_data': event_data})
|
|
return True
|
|
|
|
monkeypatch.setattr(events.Config, 'get', fake_config_get)
|
|
monkeypatch.setattr(events, 'post_webhook', fake_post_webhook)
|
|
|
|
event = events.build_event(
|
|
request,
|
|
events.EVENTS.USER_CREATED,
|
|
actor={'id': 'admin-1', 'name': 'Admin', 'role': 'admin'},
|
|
subject_id='user-1',
|
|
)
|
|
|
|
await events.WebhookEventSink().handle_event(request.app, event)
|
|
|
|
assert sent['name'] == 'Test WebUI'
|
|
assert sent['url'] == 'https://example.com/events'
|
|
assert sent['event_data'] == event.model_dump()
|
|
assert sent['event_data']['event'] == 'user.created'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_publish_event_swallows_sink_failure(monkeypatch):
|
|
request = DummyRequest()
|
|
|
|
class FailingSink:
|
|
async def handle_event(self, app, event):
|
|
raise RuntimeError('boom')
|
|
|
|
monkeypatch.setattr(events, 'EVENT_SINKS', [FailingSink()])
|
|
|
|
await events.publish_event(
|
|
request,
|
|
events.EVENTS.USER_CREATED,
|
|
actor={'id': 'admin-1'},
|
|
subject_id='user-1',
|
|
)
|