48c3ce76f0
- Introduced WS_Message_V3 standardized Pydantic model and WS_Manager_V3. - Implemented /v3/ws/ endpoint with granular Redis routing to solve "noisy neighbor" scaling issues. - Added presence tracking using Redis Sets for group coordination. - Comprehensive test suite added (unit and integration) covering models, manager, and routing logic. - Documentation: Created V3 Frontend WebSocket Guide and Project design spec. - Updated main Frontend API guide and tests README with new standards.
75 lines
2.6 KiB
Python
75 lines
2.6 KiB
Python
import sys
|
|
import os
|
|
import asyncio
|
|
import unittest
|
|
from unittest.mock import MagicMock, AsyncMock, patch
|
|
from datetime import datetime, timezone
|
|
|
|
# Add project root to path
|
|
sys.path.append(os.getcwd())
|
|
|
|
# Mock app.config BEFORE imports
|
|
mock_config = MagicMock()
|
|
mock_config.settings = MagicMock()
|
|
mock_config.settings.REDIS = {'server': 'localhost', 'port': 6379}
|
|
sys.modules["app.config"] = mock_config
|
|
|
|
from app.lib_websockets_v3 import WS_Message_V3, WS_Manager_V3
|
|
|
|
class TestWSV3Library(unittest.TestCase):
|
|
|
|
def test_message_model_validation(self):
|
|
print("\n--- Testing WS_Message_V3 Validation ---")
|
|
data = {
|
|
"msg_type": "cmd",
|
|
"target": "group",
|
|
"from_id": "client_abc",
|
|
"group_id": "group_123",
|
|
"cmd": "RELOAD",
|
|
"payload": {"force": True}
|
|
}
|
|
msg = WS_Message_V3(**data)
|
|
self.assertEqual(msg.version, "3")
|
|
self.assertEqual(msg.cmd, "RELOAD")
|
|
self.assertTrue(isinstance(msg.sent_at.isoformat(), str))
|
|
print("✅ Model validation passed.")
|
|
|
|
def test_channel_name_generation(self):
|
|
print("\n--- Testing Channel Name Generation ---")
|
|
manager = WS_Manager_V3()
|
|
channels = manager.get_channel_names("client_abc", "group_123")
|
|
|
|
self.assertIn("ws:client:client_abc", channels)
|
|
self.assertIn("ws:group:group_123", channels)
|
|
self.assertIn("ws:broadcast", channels)
|
|
print("✅ Channel name generation passed.")
|
|
|
|
@patch('redis.asyncio.Redis.from_url')
|
|
def test_publish_routing(self, mock_redis_factory):
|
|
print("\n--- Testing Publish Routing ---")
|
|
mock_redis = AsyncMock()
|
|
mock_redis_factory.return_value = mock_redis
|
|
|
|
manager = WS_Manager_V3()
|
|
|
|
async def run_test():
|
|
# 1. Test Group Routing
|
|
msg_group = WS_Message_V3(
|
|
msg_type="msg", target="group", from_id="sender", group_id="target_group"
|
|
)
|
|
await manager.publish_message(msg_group)
|
|
mock_redis.publish.assert_called_with("ws:group:target_group", unittest.mock.ANY)
|
|
|
|
# 2. Test Direct Routing
|
|
msg_direct = WS_Message_V3(
|
|
msg_type="msg", target="direct", from_id="sender", to_id="target_client"
|
|
)
|
|
await manager.publish_message(msg_direct)
|
|
mock_redis.publish.assert_called_with("ws:client:target_client", unittest.mock.ANY)
|
|
|
|
asyncio.run(run_test())
|
|
print("✅ Publish routing logic passed.")
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|