bruno
2f2a5e9b4a
Introduce an admin portal (React + Nginx), WebSocket routing, and API versioning middleware with `/api/v1/` prefix deprecation. Add master API key authentication, new Prometheus metrics for AI token consumption and active WebSockets, and extend S3 config with a public endpoint URL. Update test paths and fixtures to align with the new routing structure.
54 lines
2.0 KiB
Python
54 lines
2.0 KiB
Python
import sys
|
|
import os
|
|
import pytest
|
|
from unittest.mock import AsyncMock, patch, MagicMock
|
|
|
|
# Add sdk to path so we can import it locally if not installed
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'sdk')))
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
|
|
|
|
from integration.shaarli.hub_plugin import ShaarliHubPlugin
|
|
from imago_client.exceptions import ImagoError
|
|
from imago_client.models import AiData
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shaarli_plugin_process_bookmark_success():
|
|
"""Test processing a bookmark which correctly uses the HubClient."""
|
|
plugin = ShaarliHubPlugin(api_url="http://test/api/v1", api_key="sk-test")
|
|
|
|
# Mock the underlying HubClient.ai.draft_task
|
|
mock_ai = AsyncMock()
|
|
mock_ai.draft_task.return_value = {
|
|
"title": "Sample Title",
|
|
"description": "Sample description",
|
|
"suggested_tags": ["tech", "ai"]
|
|
}
|
|
|
|
# Context manager mock
|
|
plugin.client.ai = mock_ai
|
|
plugin.client.__aenter__ = AsyncMock(return_value=plugin.client)
|
|
plugin.client.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
result = await plugin.process_bookmark("https://example.com", "Test", "Test desc")
|
|
|
|
assert "suggested_tags" in result
|
|
assert result["suggested_tags"] == ["tech", "ai"]
|
|
mock_ai.draft_task.assert_awaited_once_with(url="https://example.com", prompt="Test desc")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shaarli_plugin_process_bookmark_error():
|
|
"""Test handling of ImagoError during processing."""
|
|
plugin = ShaarliHubPlugin(api_url="http://test/api/v1", api_key="sk-test")
|
|
|
|
mock_ai = AsyncMock()
|
|
mock_ai.draft_task.side_effect = ImagoError("Quota exceeded")
|
|
|
|
plugin.client.ai = mock_ai
|
|
plugin.client.__aenter__ = AsyncMock(return_value=plugin.client)
|
|
plugin.client.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
result = await plugin.process_bookmark("https://example.com", "Test", "Test desc")
|
|
|
|
assert "error" in result
|
|
assert result["error"] == "Quota exceeded"
|