wehub-resource-sync
534bb94eea
Test Migrations / Migrations (SQLite) (push) Has been cancelled
Build Dev Image / build-dev-image (push) Has been cancelled
Check i18n Keys / Check i18n Key Consistency (push) Has been cancelled
Lint / Ruff Lint & Format (push) Has been cancelled
Lint / Frontend Lint (push) Has been cancelled
Test Migrations / Migrations (PostgreSQL) (push) Has been cancelled
33 lines
1.0 KiB
Python
33 lines
1.0 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from langbot.pkg.plugin.connector import PluginRuntimeConnector, PluginRuntimeNotConnectedError
|
|
|
|
|
|
def make_connector() -> PluginRuntimeConnector:
|
|
app = SimpleNamespace(instance_config=SimpleNamespace(data={'plugin': {'enable': True}}))
|
|
return PluginRuntimeConnector(app, AsyncMock())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ping_plugin_runtime_raises_specific_error_when_not_connected():
|
|
connector = make_connector()
|
|
|
|
with pytest.raises(PluginRuntimeNotConnectedError, match='Plugin runtime is not connected'):
|
|
await connector.ping_plugin_runtime()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ping_plugin_runtime_delegates_to_connected_handler():
|
|
connector = make_connector()
|
|
connector.handler = SimpleNamespace(ping=AsyncMock(return_value='pong'))
|
|
|
|
result = await connector.ping_plugin_runtime()
|
|
|
|
assert result == 'pong'
|
|
connector.handler.ping.assert_awaited_once()
|