""" Comprehensive Unit Tests for Chat Multi-Channel Architecture & Service. Verifies Channel Policy, Rate Limiting, Two-Tier Moderation, HMAC Item Linking, Asynchronous Cluster Router Fan-out, and Redis 7 Sharded Pub/Sub. """ import asyncio import time import unittest from typing import List from server.chat.chat_cluster_router import RedisShardedPubSubBridge from server.chat.chat_types import ( ChatChannelType, ChatMessageDTO, ItemAffixDTO, SendChatRequestDTO, ) from server.chat.chat_service import ChatService class TestChatService(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: self.secret_key = b"FREEEXILE_SECRET_SIGNING_KEY_2026" self.chat_service = ChatService(item_secret_key=self.secret_key, cluster_shards=16) async def test_world_chat_level_requirement(self) -> None: """Level < 20 cannot send World chat.""" req_underlevel = SendChatRequestDTO( sender_id=1001, sender_name="Newbie", sender_level=15, channel=ChatChannelType.WORLD, content="Hello world!" ) res = await self.chat_service.handle_send_chat(req_underlevel) self.assertFalse(res.success) self.assertIn("Cần đạt cấp 20", res.error_message) req_valid = SendChatRequestDTO( sender_id=1002, sender_name="Expert", sender_level=25, channel=ChatChannelType.WORLD, content="Hello world from veteran!" ) res_valid = await self.chat_service.handle_send_chat(req_valid) self.assertTrue(res_valid.success) async def test_recruit_chat_level_requirement(self) -> None: """Level < 10 cannot send Recruit chat.""" req_underlevel = SendChatRequestDTO( sender_id=1003, sender_name="Novice", sender_level=8, channel=ChatChannelType.RECRUIT, content="Looking for party!" ) res = await self.chat_service.handle_send_chat(req_underlevel) self.assertFalse(res.success) self.assertIn("Cần đạt cấp 10", res.error_message) req_valid = SendChatRequestDTO( sender_id=1004, sender_name="Leader", sender_level=12, channel=ChatChannelType.RECRUIT, content="Tuyển thành viên đi phụ bản!" ) res_valid = await self.chat_service.handle_send_chat(req_valid) self.assertTrue(res_valid.success) async def test_world_chat_cooldown_rate_limit(self) -> None: """World chat enforces 15-second cooldown token bucket.""" req = SendChatRequestDTO( sender_id=2001, sender_name="Warrior", sender_level=30, channel=ChatChannelType.WORLD, content="Message 1" ) res1 = await self.chat_service.handle_send_chat(req) self.assertTrue(res1.success) # Immediate repeat attempt res2 = await self.chat_service.handle_send_chat(req) self.assertFalse(res2.success) self.assertIn("Gửi tin quá nhanh", res2.error_message) self.assertGreater(res2.cooldown_remaining_ms, 0) async def test_two_tier_moderation_profanity_masking(self) -> None: """Profane words with dots/leetspeak are masked with asterisks.""" req = SendChatRequestDTO( sender_id=3001, sender_name="PlayerA", sender_level=20, channel=ChatChannelType.ZONE, zone_id="zone_wasteland", content="thang d.m may c_a_c" ) res = await self.chat_service.handle_send_chat(req) self.assertTrue(res.success) self.assertIn("***", res.filtered_content) self.assertNotIn("d.m", res.filtered_content) async def test_two_tier_moderation_rmt_auto_mute_and_persistence(self) -> None: """RMT keywords and phone numbers trigger auto-mute and block subsequent messages.""" req_rmt = SendChatRequestDTO( sender_id=4001, sender_name="Scammer", sender_level=50, channel=ChatChannelType.ZONE, zone_id="zone_city", content="Ban vang gia re chuyen khoan atm qua zalo 0987654321" ) res = await self.chat_service.handle_send_chat(req_rmt) self.assertFalse(res.success) self.assertIn("khóa chat", res.error_message) self.assertGreaterEqual(res.risk_score, 60) # Subsequent harmless message from same sender must be blocked by persistent mute req_clean = SendChatRequestDTO( sender_id=4001, sender_name="Scammer", sender_level=50, channel=ChatChannelType.ZONE, zone_id="zone_city", content="Xin loi toi khong spam nua" ) res_blocked = await self.chat_service.handle_send_chat(req_clean) self.assertFalse(res_blocked.success) self.assertIn("tạm khóa chat", res_blocked.error_message) # Admin un-mutes the player self.chat_service.unmute_player(4001) res_allowed = await self.chat_service.handle_send_chat(req_clean) self.assertTrue(res_allowed.success) async def test_hmac_item_link_snapshot_and_spoof_prevention(self) -> None: """Item snapshots are signed by HMAC-SHA256 and reject spoofed stats.""" snap = self.chat_service.item_link_service.create_item_snapshot( item_uuid="sword_001", item_name_key="Thanh Phong Huyen Kiem", rarity=4, element=1, quality=20, item_level=90, crafter_name="Kiem Thanh", affixes=(ItemAffixDTO(affix_id="aff_phys", stat_name="Sat Thuong", stat_value=1200.0),) ) req = SendChatRequestDTO( sender_id=5001, sender_name="LuckyCraftsman", sender_level=60, channel=ChatChannelType.ZONE, zone_id="zone_sanctuary", content="Xem kiem ta vua duc!", link_item_uuids=("sword_001",) ) res = await self.chat_service.handle_send_chat(req) self.assertTrue(res.success) query_res = self.chat_service.query_item_snapshot("sword_001", snap.hmac_signature) self.assertTrue(query_res.is_valid) self.assertIsNotNone(query_res.item_snapshot) self.assertEqual(query_res.item_snapshot.crafter_name, "Kiem Thanh") self.assertEqual(query_res.item_snapshot.item_level, 90) fake_query = self.chat_service.query_item_snapshot("sword_001", "fake_signature_abc123") self.assertFalse(fake_query.is_valid) self.assertIn("mismatch", fake_query.error_message) async def test_item_link_snapshot_expiration(self) -> None: """Item snapshots expire after TTL and are purged from cache.""" snap = self.chat_service.item_link_service.create_item_snapshot( item_uuid="expired_sword", item_name_key="Kiem Cu", rarity=1, element=2, quality=0, item_level=10, crafter_name="Tho Ren", ttl_seconds=-1 # Already expired ) query_res = self.chat_service.query_item_snapshot("expired_sword", snap.hmac_signature) self.assertFalse(query_res.is_valid) self.assertIn("expired", query_res.error_message.lower()) async def test_whisper_policies(self) -> None: """Whisper rejects self-messaging, invalid targets, and blocked players.""" # Self whisper req_self = SendChatRequestDTO( sender_id=6001, sender_name="Solo", sender_level=20, channel=ChatChannelType.WHISPER, target_id=6001, content="Note to self" ) res_self = await self.chat_service.handle_send_chat(req_self) self.assertFalse(res_self.success) self.assertIn("chính mình", res_self.error_message) # Invalid target <= 0 req_invalid = SendChatRequestDTO( sender_id=6001, sender_name="Solo", sender_level=20, channel=ChatChannelType.WHISPER, target_id=0, content="Lost message" ) res_invalid = await self.chat_service.handle_send_chat(req_invalid) self.assertFalse(res_invalid.success) self.assertIn("không hợp lệ", res_invalid.error_message) # Blacklist check self.chat_service.set_blacklist_entry(player_id=6002, blocked_player_id=6001, is_blocked=True) req_blocked = SendChatRequestDTO( sender_id=6001, sender_name="Stalker", sender_level=20, channel=ChatChannelType.WHISPER, target_id=6002, content="Can we talk?" ) res_blocked = await self.chat_service.handle_send_chat(req_blocked) self.assertFalse(res_blocked.success) self.assertIn("đã chặn", res_blocked.error_message) async def test_duplicate_spam_check(self) -> None: """Duplicate messages on World or Zone within 5s are rejected.""" req1 = SendChatRequestDTO( sender_id=7001, sender_name="PlayerA", sender_level=30, channel=ChatChannelType.ZONE, zone_id="zone_market", content="Can mua Kim Cuong So Thach!" ) res1 = await self.chat_service.handle_send_chat(req1) self.assertTrue(res1.success) # Fast forward cooldown past 3s, but still within 5s duplicate spam window (3.5s elapsed) self.chat_service.channel_manager._last_send_time[(7001, ChatChannelType.ZONE)] = time.time() - 3.5 # Same sender sends identical content on Zone within 5s res2 = await self.chat_service.handle_send_chat(req1) self.assertFalse(res2.success) self.assertIn("Phát hiện spam", res2.error_message) async def test_ring_buffer_history_capped_at_100(self) -> None: """Channel history is capped at exactly 100 messages.""" channel_key = "zone:zone_arena" for i in range(105): msg = ChatMessageDTO( message_id=f"msg_{i}", sender_id=8000 + i, sender_name=f"Duelist_{i}", sender_title="", sender_guild_tag="", channel=ChatChannelType.ZONE, target_id=0, raw_content=f"Fight round {i}", filtered_content=f"Fight round {i}", timestamp_ms=int(time.time() * 1000) ) self.chat_service.channel_manager.add_to_history(channel_key, msg) history = self.chat_service.get_channel_history(channel_key) self.assertEqual(len(history), 100) self.assertEqual(history[0].message_id, "msg_5") self.assertEqual(history[-1].message_id, "msg_104") async def test_cluster_router_sync_and_async_fanout(self) -> None: """Router supports both synchronous sinks and async coroutine subscribers.""" sync_received: List[ChatMessageDTO] = [] async_received: List[ChatMessageDTO] = [] def sync_sink(msg: ChatMessageDTO) -> None: sync_received.append(msg) async def async_sink(msg: ChatMessageDTO) -> None: async_received.append(msg) ch_key = "zone:battlefield" self.chat_service.subscribe_client(ch_key, 9001, sync_sink) self.chat_service.subscribe_client(ch_key, 9002, async_sink) req = SendChatRequestDTO( sender_id=9003, sender_name="General", sender_level=50, channel=ChatChannelType.ZONE, zone_id="battlefield", content="Tien len!" ) res = await self.chat_service.handle_send_chat(req) self.assertTrue(res.success) await asyncio.sleep(0.01) self.assertEqual(len(sync_received), 1) self.assertEqual(len(async_received), 1) async def test_redis_sharded_pubsub_bridge_integration(self) -> None: """Tests Redis 7 Sharded Pub/Sub bridge formatting and broadcast hook.""" bridge = RedisShardedPubSubBridge() self.assertEqual(bridge.to_sharded_channel("world"), "{world}") self.assertEqual(bridge.to_sharded_channel("zone:barrow"), "{zone:barrow}") self.assertEqual(bridge.to_sharded_channel("guild:101"), "{guild:101}") self.chat_service.cluster_router.attach_redis_bridge(bridge) req = SendChatRequestDTO( sender_id=9501, sender_name="Broadcaster", sender_level=30, channel=ChatChannelType.ZONE, zone_id="ancient_barrow", content="Thong bao toan khu vuc!" ) res = await self.chat_service.handle_send_chat(req) self.assertTrue(res.success) self.assertGreaterEqual(len(bridge.published_messages), 1) self.assertEqual(bridge.published_messages[0][0], "{zone:ancient_barrow}") if __name__ == "__main__": unittest.main()