""" Unit tests for FreeExile Endgame Infinite World Map & Map Content System. Verifies: 1. Endgame Phase unlock conditions upon completing basic quests/campaign. 2. Infinite Procedural Node Graph generation with deterministic seeds. 3. Node progression: depth scaling, tier allocation (T1 to T16+ Über), monster levels. 4. Procedural Map Content generation: random count (1-5) and distinct savage encounters per map. 5. Node lifecycle: ACCESSIBLE -> IN_PROGRESS -> CLEARED -> Frontier Expansion. 6. Server-Authoritative traversal rules and integration with 6-Portal Hideout Map Device. """ import unittest import time from typing import List, Dict, Set from server.world.quest_engine import QuestEngine, create_default_quest_engine, QuestStatus from server.world.hideout_engine import HideoutEngine, HideoutTheme from server.world.endgame_types import ( EndGameStatus, MapBiomeType, ContentType, MapNodeStatus, MapContent, MapNode, AtlasWorldState, ) from server.world.endgame_content_catalog import ( ALL_CONTENT_TYPES, CONTENT_DEFINITIONS, get_content_definition, calculate_content_count_for_tier, ) from server.world.endgame_atlas_engine import ( EndGameAtlasEngine, CAMPAIGN_COMPLETION_MIN_LEVEL, CAMPAIGN_COMPLETION_QUEST_ID, ) class TestEndGameAtlasSystem(unittest.TestCase): def setUp(self): self.quest_engine = create_default_quest_engine() self.hideout_engine = HideoutEngine() self.atlas_engine = EndGameAtlasEngine( quest_engine=self.quest_engine, hideout_engine=self.hideout_engine ) self.player_id = "exile_pioneer_01" def test_endgame_locked_initially_for_new_player(self): """New players with no completed quests must have Endgame status LOCKED.""" self.quest_engine.initialize_player(self.player_id, player_level=1) can_unlock = self.atlas_engine.check_endgame_eligibility(self.player_id) self.assertFalse(can_unlock, "New player level 1 without campaign completion cannot unlock Endgame") status = self.atlas_engine.get_endgame_status(self.player_id) self.assertEqual(status, EndGameStatus.LOCKED) def test_endgame_unlocks_when_campaign_completed(self): """Completing the final basic campaign quest unlocks Endgame phase.""" self.quest_engine.initialize_player(self.player_id, player_level=CAMPAIGN_COMPLETION_MIN_LEVEL) # Complete prerequisite campaign quests for q_id in [ "quest_norm_01_thuc_tinh_tang_kiem", "quest_norm_02_khai_thong_dan_dien", "quest_norm_03_luyen_khi_hoa_thach", "quest_norm_04_diet_u_hon_kiem_no", ]: self.quest_engine.accept_quest(self.player_id, q_id, player_level=CAMPAIGN_COMPLETION_MIN_LEVEL) q = self.quest_engine.get_player_quest(self.player_id, q_id) if q: for obj in q.objectives.values(): q.objectives[obj.objective_id] = obj.with_increment(obj.target_count) q.status = QuestStatus.COMPLETED # Accept and complete the milestone campaign quest 05 self.quest_engine.accept_quest(self.player_id, CAMPAIGN_COMPLETION_QUEST_ID, player_level=CAMPAIGN_COMPLETION_MIN_LEVEL) self.quest_engine.record_action( self.player_id, "action_vault_instant_buyout", count=1, ) # Check quest is completed q = self.quest_engine.get_player_quest(self.player_id, CAMPAIGN_COMPLETION_QUEST_ID) self.assertIsNotNone(q) self.assertEqual(q.status, QuestStatus.COMPLETED) # Now player should be eligible for Endgame is_eligible = self.atlas_engine.check_endgame_eligibility(self.player_id) self.assertTrue(is_eligible, "Player who completed campaign quest must be eligible for Endgame") # Perform unlock atlas_state = self.atlas_engine.unlock_endgame(self.player_id, seed=1337) self.assertTrue(atlas_state.is_endgame_unlocked) self.assertEqual(self.atlas_engine.get_endgame_status(self.player_id), EndGameStatus.UNLOCKED) self.assertGreater(len(atlas_state.nodes), 0, "Atlas must generate initial node cluster") def test_infinite_atlas_initial_node_cluster_structure(self): """Initial Atlas cluster must contain origin node at Depth 0 and branching children at Depth 1-2.""" self._unlock_player_endgame(seed=98765) state = self.atlas_engine.get_atlas_state(self.player_id) self.assertIsNotNone(state) # Origin node must be Depth 0 and accessible origin = state.nodes.get(state.current_origin_node_id) self.assertIsNotNone(origin, "Origin node must exist") self.assertEqual(origin.depth, 0) self.assertEqual(origin.tier, 1) self.assertIn(origin.status, (MapNodeStatus.ACCESSIBLE, MapNodeStatus.CLEARED)) # Origin must connect to at least 2 forward branching nodes at depth 1 self.assertGreaterEqual(len(origin.connected_node_ids), 2) for child_id in origin.connected_node_ids: child = state.nodes.get(child_id) self.assertIsNotNone(child, f"Child node {child_id} must exist in state") self.assertEqual(child.depth, 1) self.assertGreaterEqual(child.tier, 1) def test_node_depth_scaling_and_monster_levels(self): """Node tier and monster level must scale proportionally with depth from origin.""" self._unlock_player_endgame(seed=42) state = self.atlas_engine.get_atlas_state(self.player_id) # Verify depth to tier scaling across initial cluster for node in state.nodes.values(): expected_monster_lvl = 67 + node.tier self.assertEqual( node.monster_level, expected_monster_lvl, f"Node {node.node_id} tier {node.tier} must have monster level {expected_monster_lvl}" ) # Higher depth nodes should have greater or equal tiers self.assertGreaterEqual(node.tier, max(1, min(16, node.depth))) def test_random_map_content_density_and_tier_scaling(self): """Every map node must generate a random count of contents (1 to 5) scaling with tier.""" self._unlock_player_endgame(seed=555) state = self.atlas_engine.get_atlas_state(self.player_id) for node in state.nodes.values(): contents = node.contents min_c, max_c = calculate_content_count_for_tier(node.tier) self.assertGreaterEqual( len(contents), min_c, f"Node {node.node_id} (Tier {node.tier}) must have at least {min_c} contents" ) self.assertLessEqual( len(contents), max_c, f"Node {node.node_id} (Tier {node.tier}) must not exceed {max_c} contents" ) # Ensure all content types are valid savage encounters for content in contents: self.assertIn(content.content_type, ALL_CONTENT_TYPES) self.assertTrue(len(content.name) > 0) self.assertTrue(len(content.description) > 0) self.assertGreaterEqual(content.difficulty_rating, 1) self.assertLessEqual(content.difficulty_rating, 5) def test_clearing_node_unlocks_adjacent_and_grants_atlas_points(self): """Clearing a node transitions it to CLEARED, awards 1 Atlas Passive Point, and unlocks children.""" self._unlock_player_endgame(seed=777) state = self.atlas_engine.get_atlas_state(self.player_id) origin_id = state.current_origin_node_id initial_points = state.atlas_passive_points # Complete origin node result = self.atlas_engine.complete_map_node(self.player_id, origin_id) self.assertTrue(result["success"]) self.assertEqual(result["atlas_points_awarded"], 1) updated_state = self.atlas_engine.get_atlas_state(self.player_id) self.assertEqual(updated_state.atlas_passive_points, initial_points + 1) self.assertEqual(updated_state.total_maps_cleared, 1) # Origin must now be CLEARED origin_node = updated_state.nodes[origin_id] self.assertEqual(origin_node.status, MapNodeStatus.CLEARED) # Connected child nodes must now be ACCESSIBLE for child_id in origin_node.connected_node_ids: child = updated_state.nodes[child_id] self.assertEqual(child.status, MapNodeStatus.ACCESSIBLE) def test_infinite_frontier_procedural_expansion(self): """Clearing child nodes expands the infinite frontier, generating deeper depths deterministically.""" self._unlock_player_endgame(seed=2026) state = self.atlas_engine.get_atlas_state(self.player_id) initial_node_count = len(state.nodes) # Clear origin self.atlas_engine.complete_map_node(self.player_id, state.current_origin_node_id) state = self.atlas_engine.get_atlas_state(self.player_id) origin_node = state.nodes[state.current_origin_node_id] # Pick first accessible child at depth 1 and complete it depth_1_node_id = origin_node.connected_node_ids[0] res = self.atlas_engine.complete_map_node(self.player_id, depth_1_node_id) self.assertTrue(res["success"]) # Frontier must have expanded: new nodes generated at deeper layers new_state = self.atlas_engine.get_atlas_state(self.player_id) self.assertGreater( len(new_state.nodes), initial_node_count, "Frontier expansion must add newly discovered/accessible nodes at deeper depths" ) self.assertGreaterEqual(new_state.unlocked_depth, 2) def test_node_activation_opens_six_portals_in_hideout_device(self): """Activating an accessible node configures the Hideout Map Device with exactly 6 Portals.""" self._unlock_player_endgame(seed=101) state = self.atlas_engine.get_atlas_state(self.player_id) origin_id = state.current_origin_node_id # Activate origin node activation = self.atlas_engine.activate_map_node(self.player_id, origin_id) self.assertTrue(activation["success"]) self.assertEqual(activation["portals_opened"], 6) self.assertIn("instance_id", activation) # Verify hideout device state hideout = self.hideout_engine.get_or_create_hideout(self.player_id) self.assertEqual(hideout.map_device.portals_remaining, 6) self.assertIsNotNone(hideout.map_device.active_map) self.assertEqual(hideout.map_device.active_map.tier, 1) def test_server_authoritative_traversal_rejects_unreachable_node(self): """Attempting to activate an unreachable/locked node must be strictly rejected.""" self._unlock_player_endgame(seed=999) # Attempt to activate a fictitious or distant locked node fake_node_id = "node_d99_b99_fake" res = self.atlas_engine.activate_map_node(self.player_id, fake_node_id) self.assertFalse(res["success"]) self.assertEqual(res["error"], "NODE_NOT_FOUND") # Helper method def _unlock_player_endgame(self, seed: int = 12345) -> None: self.quest_engine.initialize_player(self.player_id, player_level=CAMPAIGN_COMPLETION_MIN_LEVEL) for q_id in [ "quest_norm_01_thuc_tinh_tang_kiem", "quest_norm_02_khai_thong_dan_dien", "quest_norm_03_luyen_khi_hoa_thach", "quest_norm_04_diet_u_hon_kiem_no", ]: self.quest_engine.accept_quest(self.player_id, q_id, player_level=CAMPAIGN_COMPLETION_MIN_LEVEL) q = self.quest_engine.get_player_quest(self.player_id, q_id) if q: for obj in q.objectives.values(): q.objectives[obj.objective_id] = obj.with_increment(obj.target_count) q.status = QuestStatus.COMPLETED self.quest_engine.accept_quest(self.player_id, CAMPAIGN_COMPLETION_QUEST_ID, player_level=CAMPAIGN_COMPLETION_MIN_LEVEL) self.quest_engine.record_action( self.player_id, "action_vault_instant_buyout", count=1, ) self.atlas_engine.unlock_endgame(self.player_id, seed=seed) if __name__ == "__main__": unittest.main()