""" Tier 2 E2E Tests: Boundary, Corner, and Precision Edge Cases. Tests precision float distances, timers, zero/max thresholds, and race conditions. """ from __future__ import annotations import math import pytest from tests.e2e_social_party.test_harness import ( GroundLootDrop, GuildRosterService, InterruptReason, LootAllocationMode, PartyMemberState, PartyScalingCalculator, PartyService, SocialService, TargetType, TeleportService, TownPortalEngine, ) @pytest.fixture def test_env(): party_svc = PartyService() social_svc = SocialService() guild_svc = GuildRosterService() teleport_svc = TeleportService(party_svc, social_svc, guild_svc) portal_engine = TownPortalEngine() return party_svc, social_svc, guild_svc, teleport_svc, portal_engine class TestTier2BoundaryAndCornerCases: """Tier 2: Boundary & Corner Cases.""" def test_party_max_six_players_boundary(self, test_env) -> None: party_svc, _, _, _, _ = test_env p = party_svc.create_party("p1", "Leader", 30) # Add 5 more players -> total 6 for i in range(2, 7): assert party_svc.invite_player(p.party_id, "p1", f"p{i}") is True assert party_svc.accept_invite(p.party_id, f"p{i}", f"Member{i}", 30) is True assert len(p.members) == 6 def test_party_seventh_player_rejected(self, test_env) -> None: party_svc, _, _, _, _ = test_env p = party_svc.create_party("p1", "Leader", 30) for i in range(2, 7): party_svc.invite_player(p.party_id, "p1", f"p{i}") party_svc.accept_invite(p.party_id, f"p{i}", f"Member{i}", 30) # Attempt to invite 7th player invited = party_svc.invite_player(p.party_id, "p1", "p7") assert invited is False assert len(p.members) == 6 def test_nearby_radius_precision_boundary_15m_vs_15_01m(self, test_env) -> None: party_svc, _, _, _, _ = test_env p = party_svc.create_party("p1", "Leader", 30) party_svc.invite_player(p.party_id, "p1", "p2") party_svc.accept_invite(p.party_id, "p2", "EdgeCase", 30) # Exact 15.000m p.members["p1"].x, p.members["p1"].y = 0.0, 0.0 p.members["p2"].x, p.members["p2"].y = 15.000, 0.0 assert len(party_svc.get_nearby_members("p1", radius=15.0)) == 2 # Exactly 15.010m (outside) p.members["p2"].x, p.members["p2"].y = 15.010, 0.0 assert len(party_svc.get_nearby_members("p1", radius=15.0)) == 1 def test_short_allocation_expiry_precision_4_99s_vs_5_00s(self) -> None: drop = GroundLootDrop("d_edge", "item_gem", "p1", drop_timestamp=100.0, allocation_mode=LootAllocationMode.SHORT) # Non-allocated player at 4.99s -> REJECTED assert drop.can_player_pickup("p2", current_time=104.99) is False # Non-allocated player at exactly 5.00s -> ACCEPTED assert drop.can_player_pickup("p2", current_time=105.00) is True def test_map_device_portal_depletion_to_zero(self, test_env) -> None: _, _, _, _, portal_engine = test_env portal_engine.register_astral_map_device("hardcore_map", initial_charges=1) ok1, _, remaining1 = portal_engine.enter_map_portal("hardcore_map") assert ok1 is True and remaining1 == 0 ok2, msg, remaining2 = portal_engine.enter_map_portal("hardcore_map") assert ok2 is False and remaining2 == 0 and "MAP_SEALED" in msg def test_teleport_damage_interrupt_at_instant_zero_point_zero_one(self, test_env) -> None: party_svc, _, _, teleport_svc, _ = test_env p = party_svc.create_party("p1", "Caster", 30) party_svc.invite_player(p.party_id, "p1", "p2") party_svc.accept_invite(p.party_id, "p2", "Target", 30) teleport_svc.start_teleport_channel("p1", TargetType.PARTY_MEMBER, "p2", "zone_wild") # Hit immediately after cast start reason = teleport_svc.on_player_damaged("p1", damage=0.1) assert reason == InterruptReason.INTERRUPT_DAMAGE assert "p1" not in teleport_svc.active_channels def test_teleport_movement_threshold_boundary_0_04_vs_0_051(self, test_env) -> None: party_svc, _, _, teleport_svc, _ = test_env p = party_svc.create_party("p1", "Caster", 30) party_svc.invite_player(p.party_id, "p1", "p2") party_svc.accept_invite(p.party_id, "p2", "Target", 30) teleport_svc.start_teleport_channel("p1", TargetType.PARTY_MEMBER, "p2", "zone_wild") # Jitter below threshold (0.04 <= 0.05) -> No cancel assert teleport_svc.on_player_moved("p1", mag=0.04) is None assert "p1" in teleport_svc.active_channels # Movement exceeding threshold (0.051 > 0.05) -> Cancel assert teleport_svc.on_player_moved("p1", mag=0.051) == InterruptReason.INTERRUPT_MOVEMENT assert "p1" not in teleport_svc.active_channels def test_boss_alive_vs_dead_quarantine_transition(self, test_env) -> None: _, _, _, _, portal_engine = test_env portal_engine.set_encounter_state("arena_inst", boss_active=True) # Boss is active -> portal rejected ok_alive, msg_alive, _ = portal_engine.request_town_portal("hero", "zone_arena", "arena_inst", 0.0, 0.0) assert ok_alive is False and "QUARANTINE_BOSS" in msg_alive # Boss transitions to dead -> portal immediately allowed portal_engine.set_encounter_state("arena_inst", boss_active=False) ok_dead, _, p_id = portal_engine.request_town_portal("hero", "zone_arena", "arena_inst", 0.0, 0.0) assert ok_dead is True and p_id is not None def test_offline_friend_teleport_rejected(self, test_env) -> None: _, social_svc, _, teleport_svc, _ = test_env social_svc.send_friend_request("hero", "sleepy") social_svc.accept_friend_request("sleepy", "hero") social_svc.update_presence("sleepy", "Sleepy", 50, False, "offline", "Offline") # Target liveness / offline gate check ok, reason = teleport_svc.validate_destination_5_gates("hero", "sleepy", TargetType.FRIEND, "offline", target_alive=False) assert ok is False and "GATE_1_TARGET_DEAD" in reason def test_mutual_block_rejection(self, test_env) -> None: party_svc, social_svc, _, _, _ = test_env social_svc.block_player("user_a", "user_b") social_svc.block_player("user_b", "user_a") # Neither can invite nor request friendship assert social_svc.send_friend_request("user_a", "user_b") is False assert social_svc.send_friend_request("user_b", "user_a") is False assert social_svc.is_blocked("user_a", "user_b") is True assert social_svc.is_blocked("user_b", "user_a") is True def test_empty_party_nearby_lookup_returns_empty(self, test_env) -> None: party_svc, _, _, _, _ = test_env assert party_svc.get_nearby_members("nobody", radius=15.0) == []