""" TeleportService and TownPortalEngine reference engines for E2E testing. """ from __future__ import annotations import time from typing import Dict, Optional, Tuple from tests.e2e_social_party.test_harness_models import InterruptReason, TargetType class TeleportService: """Safe Teleportation with 3.5s cast, move/damage interrupt, and 5-gate validation.""" def __init__(self, party_service, social_service, guild_service) -> None: self.party_service = party_service self.social_service = social_service self.guild_service = guild_service self.active_channels: Dict[str, Dict[str, any]] = {} self.zone_states: Dict[str, Dict[str, any]] = {} def set_zone_state(self, zone_id: str, is_boss_active: bool = False, is_secret_active: bool = False, player_count: int = 1, max_capacity: int = 6, is_active: bool = True) -> None: self.zone_states[zone_id] = { "is_boss_active": is_boss_active, "is_secret_active": is_secret_active, "player_count": player_count, "max_capacity": max_capacity, "is_active": is_active } def validate_destination_5_gates(self, player_id: str, target_id: str, target_type: TargetType, target_zone_id: str, target_alive: bool = True) -> Tuple[bool, str]: # Gate 1: Target Liveness if not target_alive: return False, "GATE_1_TARGET_DEAD: Mục tiêu đã tử vong" # Gate 2: Relationship Auth if target_type == TargetType.PARTY_MEMBER: p_id = self.party_service.player_party_map.get(player_id) party = self.party_service.parties.get(p_id) if p_id else None if not party or target_id not in party.members: return False, "GATE_2_RELATIONSHIP_UNAUTHORIZED: Không cùng tổ đội" elif target_type == TargetType.FRIEND: if not self.social_service.is_friend(player_id, target_id): return False, "GATE_2_RELATIONSHIP_UNAUTHORIZED: Không nằm trong danh sách bạn bè" elif target_type == TargetType.GUILDMATE: found = any(m.player_id == target_id for roster in self.guild_service.guilds.values() for m in roster) if not found: return False, "GATE_2_RELATIONSHIP_UNAUTHORIZED: Không cùng bang hội" # Gate 3: Zone Capacity z = self.zone_states.get(target_zone_id, {"player_count": 0, "max_capacity": 6, "is_active": True}) if z["player_count"] >= z["max_capacity"]: return False, "GATE_3_CAPACITY_EXCEEDED: Khu vực đã đầy người chơi" # Gate 4: Lifecycle if not z.get("is_active", True): return False, "GATE_4_LIFECYCLE_CLOSED: Khu vực đã đóng hoặc hết hạn" # Gate 5: Quarantine if z.get("is_boss_active", False): return False, "GATE_5_QUARANTINE_BOSS: Đang trong giao tranh Boss nguy cấp" if z.get("is_secret_active", False): return False, "GATE_5_QUARANTINE_SECRET: Mật thất thử thách đang diễn ra" return True, "SUCCESS" def start_teleport_channel(self, player_id: str, target_type: TargetType, target_id: str, target_zone_id: str, target_alive: bool = True) -> Tuple[bool, str]: ok, reason = self.validate_destination_5_gates(player_id, target_id, target_type, target_zone_id, target_alive) if not ok: return False, reason channel_id = f"chan_{player_id}_{int(time.time()*1000)}" self.active_channels[player_id] = { "channel_id": channel_id, "start_time": time.time(), "cast_time": 3.5, "target_id": target_id, "target_type": target_type, "target_zone_id": target_zone_id } return True, channel_id def on_player_moved(self, player_id: str, mag: float) -> Optional[InterruptReason]: if player_id in self.active_channels and mag > 0.05: del self.active_channels[player_id] return InterruptReason.INTERRUPT_MOVEMENT return None def on_player_damaged(self, player_id: str, damage: float) -> Optional[InterruptReason]: if player_id in self.active_channels and damage > 0.0: del self.active_channels[player_id] return InterruptReason.INTERRUPT_DAMAGE return None def complete_channel(self, player_id: str, elapsed_time: float) -> Tuple[bool, str]: chan = self.active_channels.get(player_id) if not chan: return False, "NO_ACTIVE_CHANNEL" if elapsed_time < chan["cast_time"]: return False, "CAST_INCOMPLETE" del self.active_channels[player_id] return True, chan["target_zone_id"] class TownPortalEngine: """PoE2 2-Way Town Portal Engine with 6 portal charges, Boss & Puzzle lockout.""" def __init__(self) -> None: self.portals: Dict[str, Dict[str, any]] = {} self.map_charges: Dict[str, int] = {} self.boss_active: Dict[str, bool] = {} self.secret_active: Dict[str, bool] = {} def set_encounter_state(self, instance_id: str, boss_active: bool = False, secret_active: bool = False) -> None: self.boss_active[instance_id] = boss_active self.secret_active[instance_id] = secret_active def check_quarantine_lock(self, zone_id: str, instance_id: str) -> Tuple[bool, str]: if self.boss_active.get(instance_id, False): return True, "QUARANTINE_BOSS_ENGAGED: Cấm mở cổng/trốn thoát khi Boss đang giao tranh!" if self.secret_active.get(instance_id, False): return True, "QUARANTINE_SECRET_CHAMBER: Mật thất đếm ngược đang kích hoạt, không thể mở cổng!" return False, "UNLOCKED" def request_town_portal(self, player_id: str, zone_id: str, instance_id: str, wx: float, wy: float, hub_zone: str = "zone_player_hideout") -> Tuple[bool, str, Optional[str]]: is_locked, err = self.check_quarantine_lock(zone_id, instance_id) if is_locked: return False, err, None portal_id = f"tp_{player_id}_{int(time.time()*1000)}" self.portals[portal_id] = { "player_id": player_id, "origin_zone": zone_id, "instance_id": instance_id, "origin_x": wx, "origin_y": wy, "hub_zone": hub_zone, "is_active": True } return True, "SUCCESS", portal_id def register_astral_map_device(self, instance_id: str, initial_charges: int = 6) -> None: self.map_charges[instance_id] = initial_charges def enter_map_portal(self, instance_id: str) -> Tuple[bool, str, int]: charges = self.map_charges.get(instance_id, 0) if charges <= 0: return False, "MAP_SEALED: Hết lượt cổng (0/6). Bản đồ đã phong tỏa vĩnh viễn!", 0 self.map_charges[instance_id] = charges - 1 return True, "ENTER_SUCCESS", self.map_charges[instance_id] def on_player_death(self, player_id: str, zone_id: str, instance_id: str, is_map_instance: bool = False) -> Tuple[bool, str]: if is_map_instance: charges = self.map_charges.get(instance_id, 0) if charges > 0: self.map_charges[instance_id] = charges - 1 return True, f"MAP_DEATH_CONSUMED_PORTAL: Còn lại {self.map_charges.get(instance_id, 0)} lượt" closed = 0 for pid, p in list(self.portals.items()): if p["player_id"] == player_id and p["is_active"]: p["is_active"] = False closed += 1 return True, f"WILDERNESS_DEATH_CLOSED_{closed}_PORTALS"