""" Adversarial Stress Testing Suite: Map Device Session Lifecycle & Cleanup. Validates: 1. High-volume map device session cycling (100 consecutive 6-portal cycles). 2. Zero orphaned sessions in ZoneEngine.active_instances and active_instance_id reset. 3. 7th entry attempt on exhausted portals and boundary attacks. 4. Calling parity between zone_engine=None and zone_engine=ZoneEngine(). 5. Multi-player concurrent map device activation and portal consumption. 6. Premature map overwrite behavior and session isolation. """ from __future__ import annotations import concurrent.futures from typing import Dict import pytest from server.world.hideout_engine import AstralMap, HideoutEngine from server.world.zone_engine import ZoneEngine @pytest.fixture def hideout_engine() -> HideoutEngine: return HideoutEngine() @pytest.fixture def zone_engine() -> ZoneEngine: return ZoneEngine() class TestMapDevicePortalCleanupAdversarial: """Adversarial stress and lifecycle cleanup test harness.""" def test_high_volume_map_device_session_cycling_100_cycles( self, hideout_engine: HideoutEngine, zone_engine: ZoneEngine ) -> None: """Simulates 100 consecutive 6-portal map device activations and exhaustions.""" player_id = "challenger_stress_hero_100" for cycle in range(100): tier = (cycle % 16) + 1 astral_map = AstralMap( map_id=f"map_adv_c{cycle}_{tier}", name=f"Vực Thẳm Luân Hồi #{cycle}", tier=tier, zone_template_id="zone_tang_kiem_nhai", ) ok_act, msg_act, res_act = hideout_engine.create_map_instance_session( player_id, astral_map, zone_engine=zone_engine ) assert ok_act is True, f"Failed activation at cycle {cycle}: {msg_act}" inst_id = res_act["instance_id"] assert inst_id in zone_engine.active_instances assert len(zone_engine.active_instances) == 1 dev = hideout_engine.get_or_create_hideout(player_id).map_device assert dev.active_instance_id == inst_id assert dev.portals_remaining == 6 for p_idx in range(5): ok_enter, _, res_enter = hideout_engine.enter_map_portal( player_id, zone_engine=zone_engine ) assert ok_enter is True and res_enter is not None assert res_enter["portals_remaining"] == 5 - p_idx assert inst_id in zone_engine.active_instances ok_final, msg_final, res_final = hideout_engine.enter_map_portal( player_id, zone_engine=zone_engine ) assert ok_final is True and res_final is not None assert res_final["portals_remaining"] == 0 assert "CỔNG CUỐI CÙNG" in msg_final assert dev.active_instance_id is None assert dev.active_map is None assert dev.portals_remaining == 0 assert inst_id not in zone_engine.active_instances assert len(zone_engine.active_instances) == 0 assert len(zone_engine.active_instances) == 0 def test_exhausted_portal_7th_entry_and_boundary_attacks( self, hideout_engine: HideoutEngine, zone_engine: ZoneEngine ) -> None: """Tests 7th entry attempt on exhausted portal and boundary attacks.""" player_id = "challenger_boundary_tester" astral_map = AstralMap( map_id="map_t16_celestial", name="Cửu Trọng Thiên Cung", tier=16, zone_template_id="zone_boundless_celestial_palace", ) ok, _, res = hideout_engine.create_map_instance_session( player_id, astral_map, zone_engine=zone_engine ) assert ok is True inst_id = res["instance_id"] for _ in range(6): ok_p, _, _ = hideout_engine.enter_map_portal(player_id, zone_engine=zone_engine) assert ok_p is True dev = hideout_engine.get_or_create_hideout(player_id).map_device assert dev.portals_remaining == 0 and dev.active_instance_id is None assert inst_id not in zone_engine.active_instances # 7th Entry Attempt (Default & Specific Index) ok_7th, msg_7th, res_7th = hideout_engine.enter_map_portal( player_id, zone_engine=zone_engine ) assert ok_7th is False and res_7th is None assert "0/6" in msg_7th or "không có Cổng" in msg_7th assert dev.portals_remaining == 0 ok_idx0, _, _ = hideout_engine.enter_map_portal( player_id, portal_index=0, zone_engine=zone_engine ) assert ok_idx0 is False and dev.portals_remaining == 0 # Boundary attacks: out-of-bounds indices and burst spam for bad_idx in [-1, 6, 7, 999]: ok_bad, _, _ = hideout_engine.enter_map_portal( player_id, portal_index=bad_idx, zone_engine=zone_engine ) assert ok_bad is False and dev.portals_remaining == 0 for _ in range(12): ok_burst, _, _ = hideout_engine.enter_map_portal( player_id, zone_engine=zone_engine ) assert ok_burst is False and dev.portals_remaining == 0 assert len(zone_engine.active_instances) == 0 def test_calling_convention_parity_none_vs_zone_engine( self, hideout_engine: HideoutEngine ) -> None: """Tests calling enter_map_portal with zone_engine=None vs ZoneEngine().""" player_none = "player_without_ze" astral_map = AstralMap( map_id="map_standalone", name="Hoang Vu Chi Dia", tier=5, zone_template_id="zone_tang_kiem_nhai", ) ok_act, _, _ = hideout_engine.activate_map_device(player_none, astral_map) assert ok_act is True dev = hideout_engine.get_or_create_hideout(player_none).map_device assert dev.portals_remaining == 6 and dev.active_instance_id is not None for i in range(5): ok, _, _ = hideout_engine.enter_map_portal(player_none, zone_engine=None) assert ok is True and dev.portals_remaining == 5 - i ok_last, msg_last, _ = hideout_engine.enter_map_portal(player_none, zone_engine=None) assert ok_last is True and "CỔNG CUỐI CÙNG" in msg_last assert dev.active_instance_id is None and dev.portals_remaining == 0 # Scenario B: Positional argument normalization tolerance ze = ZoneEngine() player_pos = "player_positional_ze" ok_sess, _, res_sess = hideout_engine.create_map_instance_session( player_pos, astral_map, zone_engine=ze ) assert ok_sess is True inst_id = res_sess["instance_id"] assert inst_id in ze.active_instances for _ in range(5): ok, _, _ = hideout_engine.enter_map_portal(player_pos, ze) assert ok is True and inst_id in ze.active_instances ok_final, _, _ = hideout_engine.enter_map_portal(player_pos, ze) assert ok_final is True and inst_id not in ze.active_instances assert len(ze.active_instances) == 0 def test_multi_player_concurrent_device_cycling( self, hideout_engine: HideoutEngine, zone_engine: ZoneEngine ) -> None: """Tests 50 concurrent players activating map devices and interleaved portal entry.""" num_players = 50 players = [f"challenger_concurrent_p{i:02d}" for i in range(num_players)] session_ids: Dict[str, str] = {} for p in players: astral_map = AstralMap( map_id=f"map_p_{p}", name=f"Map of {p}", tier=(hash(p) % 16) + 1, zone_template_id="zone_tang_kiem_nhai", ) ok, _, res = hideout_engine.create_map_instance_session( p, astral_map, zone_engine=zone_engine ) assert ok is True session_ids[p] = res["instance_id"] assert len(zone_engine.active_instances) == num_players # Interleaved portal consumption across all 50 players: Rounds 1 to 5 for round_idx in range(5): for p in players: ok, _, _ = hideout_engine.enter_map_portal(p, zone_engine=zone_engine) assert ok is True dev = hideout_engine.get_or_create_hideout(p).map_device assert dev.portals_remaining == 5 - round_idx assert len(zone_engine.active_instances) == num_players # Round 6: Final portal consumption cleans up each session individually remaining_expected = num_players for p in players: sid = session_ids[p] assert sid in zone_engine.active_instances ok_final, _, _ = hideout_engine.enter_map_portal(p, zone_engine=zone_engine) assert ok_final is True remaining_expected -= 1 assert sid not in zone_engine.active_instances assert len(zone_engine.active_instances) == remaining_expected dev = hideout_engine.get_or_create_hideout(p).map_device assert dev.active_instance_id is None and dev.portals_remaining == 0 assert len(zone_engine.active_instances) == 0 def test_multithreaded_concurrent_portal_exhaustion_stress( self, hideout_engine: HideoutEngine, zone_engine: ZoneEngine ) -> None: """Executes multi-threaded concurrent portal consumption using ThreadPoolExecutor.""" num_threads = 20 players = [f"thread_player_{i:02d}" for i in range(num_threads)] for p in players: m = AstralMap(f"m_{p}", f"Map {p}", 10, "zone_tang_kiem_nhai") ok, _, _ = hideout_engine.create_map_instance_session( p, m, zone_engine=zone_engine ) assert ok is True assert len(zone_engine.active_instances) == num_threads def run_player_lifecycle(p_id: str) -> bool: for _ in range(6): ok, _, _ = hideout_engine.enter_map_portal(p_id, zone_engine=zone_engine) if not ok: return False return True with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor: futures = [executor.submit(run_player_lifecycle, p) for p in players] results = [f.result() for f in concurrent.futures.as_completed(futures)] assert all(results) assert len(zone_engine.active_instances) == 0 for p in players: dev = hideout_engine.get_or_create_hideout(p).map_device assert dev.active_instance_id is None and dev.portals_remaining == 0 @pytest.mark.xfail( reason="Adversarial finding: overwriting unexhausted map device does not purge prior active_instance_id from ZoneEngine" ) def test_unexhausted_map_overwrite_cleans_prior_zone_session( self, hideout_engine: HideoutEngine, zone_engine: ZoneEngine ) -> None: """Adversarial check: if player activates Map B before exhausting Map A.""" player_id = "challenger_overwrite_tester" m1 = AstralMap("m1", "Map 1", 1, "zone_tang_kiem_nhai") m2 = AstralMap("m2", "Map 2", 2, "zone_tang_kiem_nhai") ok1, _, res1 = hideout_engine.create_map_instance_session( player_id, m1, zone_engine=zone_engine ) assert ok1 is True inst1_id = res1["instance_id"] assert inst1_id in zone_engine.active_instances # Enter only 2 portals of Map 1 hideout_engine.enter_map_portal(player_id, zone_engine=zone_engine) hideout_engine.enter_map_portal(player_id, zone_engine=zone_engine) # Overwrite with Map 2 before Map 1 is exhausted ok2, _, res2 = hideout_engine.create_map_instance_session( player_id, m2, zone_engine=zone_engine ) assert ok2 is True inst2_id = res2["instance_id"] # Exhaust Map 2 completely for _ in range(6): hideout_engine.enter_map_portal(player_id, zone_engine=zone_engine) assert inst2_id not in zone_engine.active_instances # Assert prior session was also cleaned up rather than orphaned assert inst1_id not in zone_engine.active_instances assert len(zone_engine.active_instances) == 0