""" Comprehensive Opaque-Box E2E Test Suite for FreeExile Distributed Multi-Channel Chat System (1,000,000 CCU). Enforces 4-Tier Testing Methodology covering Requirements R1 to R5: - Tier 1: Feature Coverage (8 Channels, Level Gating, Token Bucket Cooldowns, Spam, Ring Buffer, Trie, Sentinel RMT, HMAC) - Tier 2: Boundary & Corner Cases (Negative IDs, Empty/Max Strings, Zero-Width, Homoglyphs, Cooldown Limits, Expired Snapshots, Tampered Rarity) - Tier 3: Cross-Feature State Interleaving & Pairwise Combinations (Blacklist + Item Link, Guild RMT Auto-Mute, Profanity + Item Link, 64-Shard Broadcast) - Tier 4: Real-World End-to-End Application Scenarios (World Boss Burst, RMT Bot Flood, Guild Inspection Storm, High-Throughput Fan-Out) """ from __future__ import annotations import asyncio import time import unittest from typing import List from server.chat.chat_service import ChatService from server.chat.chat_types import ( ChatChannelType, ChatMessageDTO, ItemAffixDTO, SendChatRequestDTO, ) class TestTier1FeatureCoverage(unittest.IsolatedAsyncioTestCase): """Tier 1: Functional Feature Coverage (R1-R5 core capabilities).""" 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=64) async def test_01_all_eight_channels_delivery(self) -> None: """Verifies successful delivery and history recording across all 8 chat channels.""" channels_spec = [ (ChatChannelType.WORLD, {"sender_level": 25, "content": "World greeting"}), (ChatChannelType.ZONE, {"sender_level": 20, "zone_id": "zone_sword_grave", "content": "Zone alert"}), (ChatChannelType.GUILD, {"sender_level": 20, "guild_id": "guild_sect_1", "content": "Guild salute"}), (ChatChannelType.PARTY, {"sender_level": 20, "party_id": "party_raid_1", "content": "Party ready"}), (ChatChannelType.WHISPER, {"sender_level": 20, "target_id": 9002, "content": "Whisper private"}), (ChatChannelType.SYSTEM, {"sender_level": 1, "content": "System divine bulletin"}), (ChatChannelType.RECRUIT, {"sender_level": 15, "content": "Recruit LFG tier 16"}), (ChatChannelType.FEEDBACK, {"sender_level": 1, "content": "Feedback ticket bug report"}), ] for ch_type, extra_args in channels_spec: req = SendChatRequestDTO( sender_id=9001, sender_name="ExileHero", channel=ch_type, **extra_args, ) res = await self.chat_service.handle_send_chat(req) self.assertTrue(res.success, f"Channel {ch_type.name} failed: {res.error_message}") self.assertTrue(bool(res.message_id)) async def test_02_level_gating_world_and_recruit(self) -> None: """Enforces minimum level requirements (World: 20, Recruit: 10).""" under_world = SendChatRequestDTO( sender_id=101, sender_name="Noob", sender_level=19, channel=ChatChannelType.WORLD, content="Underlevel world" ) res_w_fail = await self.chat_service.handle_send_chat(under_world) self.assertFalse(res_w_fail.success) self.assertIn("cấp 20", res_w_fail.error_message) valid_world = SendChatRequestDTO( sender_id=102, sender_name="Veteran", sender_level=20, channel=ChatChannelType.WORLD, content="Valid world" ) self.assertTrue((await self.chat_service.handle_send_chat(valid_world)).success) under_recruit = SendChatRequestDTO( sender_id=103, sender_name="Youngster", sender_level=9, channel=ChatChannelType.RECRUIT, content="Underlevel recruit" ) res_r_fail = await self.chat_service.handle_send_chat(under_recruit) self.assertFalse(res_r_fail.success) self.assertIn("cấp 10", res_r_fail.error_message) valid_recruit = SendChatRequestDTO( sender_id=104, sender_name="Fighter", sender_level=10, channel=ChatChannelType.RECRUIT, content="Valid recruit" ) self.assertTrue((await self.chat_service.handle_send_chat(valid_recruit)).success) async def test_03_token_bucket_cooldown_enforcement(self) -> None: """Rejects rapid repeat requests exceeding channel cooldown token buckets.""" req = SendChatRequestDTO( sender_id=201, sender_name="Speedy", sender_level=30, channel=ChatChannelType.WORLD, content="Fast message 1" ) self.assertTrue((await self.chat_service.handle_send_chat(req)).success) repeat_res = await self.chat_service.handle_send_chat(req) self.assertFalse(repeat_res.success) self.assertIn("Gửi tin quá nhanh", repeat_res.error_message) self.assertGreater(repeat_res.cooldown_remaining_ms, 0) async def test_04_duplicate_spam_detection_world_zone(self) -> None: """Blocks identical message content sent within 5 seconds on World and Zone.""" req = SendChatRequestDTO( sender_id=301, sender_name="Spammer", sender_level=30, channel=ChatChannelType.ZONE, zone_id="z_wasteland", content="Identical content" ) self.assertTrue((await self.chat_service.handle_send_chat(req)).success) # Advance clock beyond 3.0s cooldown to isolate the 5.0s duplicate spam check self.chat_service.channel_manager._last_send_time[(301, ChatChannelType.ZONE)] = time.time() - 3.2 dup_res = await self.chat_service.handle_send_chat(req) self.assertFalse(dup_res.success) self.assertIn("spam", dup_res.error_message.lower()) async def test_05_channel_history_100_ring_buffer(self) -> None: """Maintains exactly 100 recent messages in memory per channel ring buffer.""" for i in range(105): req = SendChatRequestDTO( sender_id=400 + i, sender_name=f"Sys_{i}", sender_level=1, channel=ChatChannelType.SYSTEM, content=f"Broadcast notice #{i}" ) await self.chat_service.handle_send_chat(req) history = self.chat_service.get_channel_history("system") self.assertEqual(len(history), 100) self.assertEqual(history[0].raw_content, "Broadcast notice #5") self.assertEqual(history[-1].raw_content, "Broadcast notice #104") async def test_06_trie_profanity_censorship_and_risk(self) -> None: """Tier 1 Trie normalizes and censors profane roots with *** and sets risk score.""" req = SendChatRequestDTO( sender_id=501, sender_name="Rude", sender_level=20, channel=ChatChannelType.ZONE, zone_id="z_arena", 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) self.assertEqual(res.risk_score, 25) async def test_07_sentinel_rmt_detection_and_auto_mute(self) -> None: """Tier 2 Sentinel intercepts black market RMT advertising and triggers auto-mute.""" req = SendChatRequestDTO( sender_id=601, sender_name="RMTBot", sender_level=50, channel=ChatChannelType.ZONE, zone_id="z_market", content="Ban vang gia re chuyen khoan atm qua zalo 0987654321" ) res = await self.chat_service.handle_send_chat(req) self.assertFalse(res.success) self.assertIn("khóa chat", res.error_message) self.assertGreaterEqual(res.risk_score, 60) async def test_08_hmac_item_link_lifecycle(self) -> None: """Creates HMAC-SHA256 authenticated item snapshot, links in chat, and inspects tooltip.""" snap = self.chat_service.item_link_service.create_item_snapshot( item_uuid="item_primal_sword_01", item_name_key="Thanh Phong Huyền Kiếm", rarity=5, element=1, quality=20, item_level=100, crafter_name="Độc Cô Cầu Bại", affixes=(ItemAffixDTO(affix_id="aff_phys", stat_name="Sát Thương Vật Lý", stat_value=2500.0),) ) req = SendChatRequestDTO( sender_id=701, sender_name="Crafter", sender_level=50, channel=ChatChannelType.WORLD, content="Khoe bảo kiếm mới luyện!", link_item_uuids=("item_primal_sword_01",) ) res = await self.chat_service.handle_send_chat(req) self.assertTrue(res.success) q_res = self.chat_service.query_item_snapshot("item_primal_sword_01", snap.hmac_signature) self.assertTrue(q_res.is_valid) self.assertIsNotNone(q_res.item_snapshot) self.assertEqual(q_res.item_snapshot.crafter_name, "Độc Cô Cầu Bại") self.assertEqual(q_res.item_snapshot.rarity, 5) class TestTier2BoundaryAndCornerCases(unittest.IsolatedAsyncioTestCase): """Tier 2: Boundary, Extreme and Corner Case Testing.""" async def asyncSetUp(self) -> None: self.chat_service = ChatService() async def test_09_whisper_invalid_targets(self) -> None: """Validates error rejection for non-positive target IDs and self-whispering.""" for bad_id in (0, -1, -99): req = SendChatRequestDTO( sender_id=1001, sender_name="User", sender_level=20, channel=ChatChannelType.WHISPER, target_id=bad_id, content="Hi" ) res = await self.chat_service.handle_send_chat(req) self.assertFalse(res.success) self.assertIn("không hợp lệ", res.error_message) self_req = SendChatRequestDTO( sender_id=1001, sender_name="User", sender_level=20, channel=ChatChannelType.WHISPER, target_id=1001, content="Echo" ) self_res = await self.chat_service.handle_send_chat(self_req) self.assertFalse(self_res.success) self.assertIn("chính mình", self_res.error_message) async def test_10_missing_scope_identifiers(self) -> None: """Rejects Guild or Party requests when scope identifier is blank.""" guild_req = SendChatRequestDTO( sender_id=1002, sender_name="Loner", sender_level=20, channel=ChatChannelType.GUILD, guild_id="", content="Guild test" ) self.assertIn("Bang Hội", (await self.chat_service.handle_send_chat(guild_req)).error_message) party_req = SendChatRequestDTO( sender_id=1002, sender_name="Loner", sender_level=20, channel=ChatChannelType.PARTY, party_id="", content="Party test" ) self.assertIn("Tổ Đội", (await self.chat_service.handle_send_chat(party_req)).error_message) async def test_11_empty_and_max_string_boundaries(self) -> None: """Handles empty strings and extreme 1000-char content payloads gracefully.""" empty_req = SendChatRequestDTO( sender_id=1003, sender_name="Silent", sender_level=20, channel=ChatChannelType.ZONE, zone_id="z_cave", content="" ) self.assertTrue((await self.chat_service.handle_send_chat(empty_req)).success) large_content = "X" * 1000 large_req = SendChatRequestDTO( sender_id=1004, sender_name="Loud", sender_level=20, channel=ChatChannelType.ZONE, zone_id="z_cave", content=large_content ) res_large = await self.chat_service.handle_send_chat(large_req) self.assertTrue(res_large.success) self.assertEqual(len(res_large.filtered_content), 1000) async def test_12_unicode_zero_width_evasion(self) -> None: """Detects zero-width evasion tricks (\\u200B, \\uFEFF) hiding profanities.""" req = SendChatRequestDTO( sender_id=1005, sender_name="Tricky", sender_level=20, channel=ChatChannelType.ZONE, zone_id="z_hall", content="thang d\u200Bm may" ) res = await self.chat_service.handle_send_chat(req) self.assertTrue(res.success) self.assertGreaterEqual(res.risk_score, 25) async def test_13_homoglyph_and_leetspeak_evasion(self) -> None: """Folds homoglyphs and leetspeak substitutions (@, 0, !, 3, $) into standard root checks.""" req = SendChatRequestDTO( sender_id=1006, sender_name="Leet", sender_level=20, channel=ChatChannelType.ZONE, zone_id="z_hall", content="d0 l0n d!t bu0i" ) res = await self.chat_service.handle_send_chat(req) self.assertTrue(res.success) self.assertGreaterEqual(res.risk_score, 25) async def test_14_rapid_repeat_at_cooldown_boundary(self) -> None: """Enforces remaining cooldown time precision when repeated right before expiration.""" req = SendChatRequestDTO( sender_id=1007, sender_name="Recruiter", sender_level=20, channel=ChatChannelType.RECRUIT, content="Need healer" ) self.assertTrue((await self.chat_service.handle_send_chat(req)).success) # Advance clock to 9.9 seconds elapsed (0.1s before 10.0s cooldown expiry) self.chat_service.channel_manager._last_send_time[(1007, ChatChannelType.RECRUIT)] = time.time() - 9.9 repeat_res = await self.chat_service.handle_send_chat(req) self.assertFalse(repeat_res.success) self.assertGreater(repeat_res.cooldown_remaining_ms, 0) self.assertLessEqual(repeat_res.cooldown_remaining_ms, 200) async def test_15_expired_item_snapshot_ttl(self) -> None: """Rejects tooltip query when item snapshot TTL has elapsed (24h retention policy).""" snap = self.chat_service.item_link_service.create_item_snapshot( item_uuid="item_old_relic", item_name_key="Cổ Đao", rarity=3, element=2, quality=0, item_level=50, crafter_name="Tiền Bối", ttl_seconds=-1 ) q_res = self.chat_service.query_item_snapshot("item_old_relic", snap.hmac_signature) self.assertFalse(q_res.is_valid) self.assertIn("expired", q_res.error_message.lower()) async def test_16_corrupted_signature_and_tampered_rarity(self) -> None: """Blocks fraudulent item tooltip inspections with corrupted signature or tampered rarity.""" snap = self.chat_service.item_link_service.create_item_snapshot( item_uuid="item_legit_sword", item_name_key="Kiếm Thường", rarity=1, element=1, quality=0, item_level=10, crafter_name="Thợ Rèn" ) corrupted_sig = snap.hmac_signature[:-4] + "ffff" q_corrupt = self.chat_service.query_item_snapshot("item_legit_sword", corrupted_sig) self.assertFalse(q_corrupt.is_valid) self.assertIn("mismatch", q_corrupt.error_message.lower()) # Attempt to forge rarity from 1 (Phàm) to 5 (Thái Cổ) is_tampered_valid = self.chat_service.item_link_service.verify_item_signature( item_uuid="item_legit_sword", item_name_key="Kiếm Thường", rarity=5, claimed_signature=snap.hmac_signature ) self.assertFalse(is_tampered_valid) class TestTier3CrossFeatureCombinations(unittest.IsolatedAsyncioTestCase): """Tier 3: Cross-Feature State Interleaving & Pairwise Combinations.""" async def asyncSetUp(self) -> None: self.chat_service = ChatService(cluster_shards=64) async def test_17_whisper_blacklist_with_item_link(self) -> None: """Player blacklisted by recipient cannot deliver whisper even with valid item attached.""" self.chat_service.set_blacklist_entry(player_id=2002, blocked_player_id=2001, is_blocked=True) snap = self.chat_service.item_link_service.create_item_snapshot( item_uuid="item_gift_01", item_name_key="Bảo Hạp", rarity=4, element=3, quality=10, item_level=80, crafter_name="Người Tặng" ) req = SendChatRequestDTO( sender_id=2001, sender_name="Stalker", sender_level=30, channel=ChatChannelType.WHISPER, target_id=2002, content="Tặng ngươi bảo hạp này!", link_item_uuids=("item_gift_01",) ) res = await self.chat_service.handle_send_chat(req) self.assertFalse(res.success) self.assertIn("đã chặn", res.error_message) self.assertEqual(len(self.chat_service.get_channel_history("whisper:2001:2002")), 0) async def test_18_guild_chat_rmt_auto_mute(self) -> None: """RMT advertisement in Guild chat triggers auto-mute quarantine without broadcasting.""" req = SendChatRequestDTO( sender_id=3001, sender_name="Infiltrator", sender_level=45, channel=ChatChannelType.GUILD, guild_id="guild_vanguard", content="Ban acc lien he zalo 0912345678 chuyen khoan atm lay gia re" ) res = await self.chat_service.handle_send_chat(req) self.assertFalse(res.success) self.assertIn("khóa chat", res.error_message) self.assertGreaterEqual(res.risk_score, 60) self.assertEqual(len(self.chat_service.get_channel_history("guild:guild_vanguard")), 0) async def test_19_world_chat_profanity_with_valid_item_link(self) -> None: """Profane text is masked with *** while attached item link remains valid and queryable.""" snap = self.chat_service.item_link_service.create_item_snapshot( item_uuid="item_blade_xyz", item_name_key="Huyết Ma Đao", rarity=4, element=4, quality=15, item_level=88, crafter_name="Ma Đao Vương" ) req = SendChatRequestDTO( sender_id=4001, sender_name="Berserker", sender_level=35, channel=ChatChannelType.WORLD, content="Cây đao này d.m chém đau vcl", link_item_uuids=("item_blade_xyz",) ) 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) history = self.chat_service.get_channel_history("world") self.assertEqual(len(history), 1) self.assertTrue(history[0].is_censored) self.assertEqual(len(history[0].linked_items), 1) q_res = self.chat_service.query_item_snapshot("item_blade_xyz", snap.hmac_signature) self.assertTrue(q_res.is_valid) async def test_20_multi_channel_sharded_concurrent_broadcast(self) -> None: """Dispatches concurrent broadcasts across 64 shards without cross-talk or drops.""" received: dict[str, List[str]] = {"world": [], "zone:z1": [], "guild:g1": [], "party:p1": []} for ch_key in received.keys(): self.chat_service.subscribe_client(ch_key, 999, lambda msg, k=ch_key: received[k].append(msg.raw_content)) reqs = [ SendChatRequestDTO(sender_id=5001, sender_name="P1", sender_level=25, channel=ChatChannelType.WORLD, content="Msg World"), SendChatRequestDTO(sender_id=5002, sender_name="P2", sender_level=25, channel=ChatChannelType.ZONE, zone_id="z1", content="Msg Zone"), SendChatRequestDTO(sender_id=5003, sender_name="P3", sender_level=25, channel=ChatChannelType.GUILD, guild_id="g1", content="Msg Guild"), SendChatRequestDTO(sender_id=5004, sender_name="P4", sender_level=25, channel=ChatChannelType.PARTY, party_id="p1", content="Msg Party"), ] results = await asyncio.gather(*(self.chat_service.handle_send_chat(r) for r in reqs)) self.assertTrue(all(r.success for r in results)) for k in received.keys(): self.assertEqual(len(received[k]), 1, f"Channel {k} did not receive expected message") class TestTier4RealWorldScenarios(unittest.IsolatedAsyncioTestCase): """Tier 4: Real-World End-to-End Application Scenarios.""" async def asyncSetUp(self) -> None: self.chat_service = ChatService(cluster_shards=64) async def test_21_world_boss_announcement_burst(self) -> None: """System channel alerts spawn while 20 players discuss and link gear in parallel.""" sys_delivered: List[str] = [] self.chat_service.subscribe_client("system", 1, lambda msg: sys_delivered.append(msg.raw_content)) sys_req = SendChatRequestDTO( sender_id=1, sender_name="Server", sender_level=1, channel=ChatChannelType.SYSTEM, content="THIÊN HẠ ĐẠI NẠN: Thượng Cổ Thần Ma xuất thế!" ) sys_res = await self.chat_service.handle_send_chat(sys_req) self.assertTrue(sys_res.success) self.assertEqual(len(sys_delivered), 1) player_tasks = [] for i in range(20): p_req = SendChatRequestDTO( sender_id=6000 + i, sender_name=f"Player_{i}", sender_level=30, channel=ChatChannelType.ZONE, zone_id="zone_boss_lair", content=f"Tập hợp cổng #{i}" ) player_tasks.append(self.chat_service.handle_send_chat(p_req)) player_results = await asyncio.gather(*player_tasks) self.assertTrue(all(r.success for r in player_results)) self.assertEqual(len(self.chat_service.get_channel_history("zone:zone_boss_lair")), 20) async def test_22_rmt_bot_flood_mitigation(self) -> None: """Multi-layer defense halts an RMT bot on Recruit, World, and Zone channels sequentially.""" # Vector 1: Bot Level 1 tries Recruit -> blocked by level gate r1 = await self.chat_service.handle_send_chat(SendChatRequestDTO( sender_id=7001, sender_name="Bot1", sender_level=1, channel=ChatChannelType.RECRUIT, content="Shop account" )) self.assertFalse(r1.success) self.assertIn("cấp 10", r1.error_message) # Vector 2: Bot Level 1 tries World -> blocked by level gate r2 = await self.chat_service.handle_send_chat(SendChatRequestDTO( sender_id=7001, sender_name="Bot1", sender_level=1, channel=ChatChannelType.WORLD, content="Shop account" )) self.assertFalse(r2.success) self.assertIn("cấp 20", r2.error_message) # Vector 3: Bot moves to Zone without level gate, attempts phone RMT -> auto-muted by Sentinel r3 = await self.chat_service.handle_send_chat(SendChatRequestDTO( sender_id=7001, sender_name="Bot1", sender_level=1, channel=ChatChannelType.ZONE, zone_id="zone_novice", content="Mua ban vang gia re lh zalo 0933112233 chuyen khoan atm" )) self.assertFalse(r3.success) self.assertIn("khóa chat", r3.error_message) self.assertGreaterEqual(r3.risk_score, 60) async def test_23_guild_raid_item_inspection_storm(self) -> None: """50 guild members concurrently inspect a linked +15 weapon snapshot with < 2ms latency.""" snap = self.chat_service.item_link_service.create_item_snapshot( item_uuid="weapon_thai_co_15", item_name_key="Hỗn Độn Thần Kiếm +15", rarity=5, element=1, quality=20, item_level=100, crafter_name="Bang Chủ", affixes=(ItemAffixDTO(affix_id="aff_god", stat_name="Võ Đạo Cực Hạn", stat_value=9999.0),) ) t_start = time.perf_counter() queries = [self.chat_service.query_item_snapshot("weapon_thai_co_15", snap.hmac_signature) for _ in range(50)] elapsed_ms = (time.perf_counter() - t_start) * 1000.0 self.assertTrue(all(q.is_valid for q in queries)) self.assertTrue(all(q.item_snapshot.crafter_name == "Bang Chủ" for q in queries)) self.assertLess(elapsed_ms / 50.0, 2.0, f"Average query took {elapsed_ms / 50.0:.3f}ms (exceeded 2.0ms SLA)") async def test_24_high_throughput_fanout_burst(self) -> None: """Dispatches 100 rapid messages across 64 shards to 20 subscribers (2,000 deliveries).""" delivered_count = 0 def on_msg(_: ChatMessageDTO) -> None: nonlocal delivered_count delivered_count += 1 for i in range(20): self.chat_service.subscribe_client("zone:burst_raid", 8000 + i, on_msg) for m in range(100): req = SendChatRequestDTO( sender_id=9000 + m, sender_name=f"Striker_{m}", sender_level=40, channel=ChatChannelType.ZONE, zone_id="burst_raid", content=f"Strike sequence #{m}" ) res = await self.chat_service.handle_send_chat(req) self.assertTrue(res.success) self.assertEqual(delivered_count, 2000) stats = self.chat_service.cluster_router.get_latency_stats() self.assertLess(stats["p99"], 15.0, f"Fan-out p99 {stats['p99']}ms exceeded 15.0ms SLA") if __name__ == "__main__": unittest.main()