Spaces:
Running
Running
File size: 1,888 Bytes
372531f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 |
import pytest
from unittest.mock import AsyncMock
from fastapi import WebSocket
from src.logs_handler import CustomLogsHandler
import os
import json
@pytest.mark.asyncio
async def test_custom_logs_handler():
# Mock websocket
mock_websocket = AsyncMock()
mock_websocket.send_json = AsyncMock()
# Test initialization
handler = CustomLogsHandler(mock_websocket, "test_query")
# Verify log file creation
assert os.path.exists(handler.log_file)
# Test sending log data
test_data = {
"type": "logs",
"message": "Test log message"
}
await handler.send_json(test_data)
# Verify websocket was called with correct data
mock_websocket.send_json.assert_called_once_with(test_data)
# Verify log file contents
with open(handler.log_file, 'r') as f:
log_data = json.load(f)
assert len(log_data['events']) == 1
assert log_data['events'][0]['data'] == test_data
@pytest.mark.asyncio
async def test_content_update():
"""Test handling of non-log type data that updates content"""
mock_websocket = AsyncMock()
mock_websocket.send_json = AsyncMock()
handler = CustomLogsHandler(mock_websocket, "test_query")
# Test content update
content_data = {
"query": "test query",
"sources": ["source1", "source2"],
"report": "test report"
}
await handler.send_json(content_data)
mock_websocket.send_json.assert_called_once_with(content_data)
# Verify log file contents
with open(handler.log_file, 'r') as f:
log_data = json.load(f)
assert log_data['content']['query'] == "test query"
assert log_data['content']['sources'] == ["source1", "source2"]
assert log_data['content']['report'] == "test report" |