""" Empirical Stress Test Oracle for CharacterStatAggregator, AST, Persistence & Engine Loop. Adversarial Challenger Suite: Exhaustive combinatorial testing of PoE math formulas, tag-filtering matrix, conditional thresholds, weapon grips, and SQLite audits. """ import os import sys import json import tempfile import unittest from types import SimpleNamespace sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../server"))) from server.stats.stat_types import ( ModifierType, StatModifier, EvaluationContext, AggregatedCharacterStats, ) from server.stats.formula_persistence import FormulaPersistenceService from server.stats.stat_aggregator import CharacterStatAggregator from server.world.server_engine_loop import ServerEngineLoop, FiveElements from server.world.meridian_types import MeridianStatBonus from server.world.primal_stones_crafting import Affix, AffixType from server.inventory.inventory_types import ItemType class TestChallengerStatAggregatorStress(unittest.TestCase): """Adversarial stress testing for CharacterStatAggregator mathematical correctness.""" def setUp(self) -> None: self.aggregator = CharacterStatAggregator() def test_stress_combinatorial_poe_formula_oracle(self) -> None: """Adversarial validation of the canonical PoE formula across multiple pools.""" # Case 1: Base 50, Flat +50, Inc +100%, Red -20%, More +50%, Less -25% # Result: (50 + 50) * (1.0 + (100 - 20)/100) * (1.0 + 0.50) * (1.0 - 0.25) # = 100 * 1.80 * 1.50 * 0.75 = 202.50 mods_1 = [ StatModifier("attack_damage", ModifierType.FLAT, 50.0), StatModifier("attack_damage", ModifierType.INCREASED, 100.0), StatModifier("attack_damage", ModifierType.REDUCED, 20.0), StatModifier("attack_damage", ModifierType.MORE, 50.0), StatModifier("attack_damage", ModifierType.LESS, 25.0), ] stats_1 = self.aggregator.calculate_stats(custom_modifiers=mods_1) self.assertEqual(stats_1.attack_damage, 202.50) # Case 2: Negative net increased clamped to 0.0 (PoE behavior) # Net Inc: 20% Inc - 140% Red = -120% -> Scale Factor = max(0.0, 1.0 - 1.2) = 0.0 mods_2 = [ StatModifier("attack_damage", ModifierType.INCREASED, 20.0), StatModifier("attack_damage", ModifierType.REDUCED, 140.0), ] stats_2 = self.aggregator.calculate_stats(custom_modifiers=mods_2) self.assertEqual(stats_2.attack_damage, 0.0) # Case 3: Compound More multipliers (3 distinct sources: +20%, +30%, +50%) # Expected: 50.0 * 1.20 * 1.30 * 1.50 = 117.0 mods_3 = [ StatModifier("attack_damage", ModifierType.MORE, 20.0, source="gem1"), StatModifier("attack_damage", ModifierType.MORE, 30.0, source="gem2"), StatModifier("attack_damage", ModifierType.MORE, 50.0, source="gem3"), ] stats_3 = self.aggregator.calculate_stats(custom_modifiers=mods_3) self.assertEqual(stats_3.attack_damage, 117.0) # Case 4: Compound Less multipliers (2 sources: -20%, -30%) # Expected: 50.0 * 0.80 * 0.70 = 28.0 mods_4 = [ StatModifier("attack_damage", ModifierType.LESS, 20.0, source="curse1"), StatModifier("attack_damage", ModifierType.LESS, 30.0, source="curse2"), ] stats_4 = self.aggregator.calculate_stats(custom_modifiers=mods_4) self.assertEqual(stats_4.attack_damage, 28.0) # Case 5: 100% Less multiplier drops stat to 0.0 mods_5 = [StatModifier("attack_damage", ModifierType.LESS, 100.0)] stats_5 = self.aggregator.calculate_stats(custom_modifiers=mods_5) self.assertEqual(stats_5.attack_damage, 0.0) def test_stress_tag_matrix_filtering(self) -> None: """Adversarial validation of tag matching (subsets, supersets, case-insensitivity).""" mods = [ StatModifier("attack_damage", ModifierType.FLAT, 30.0, tags=frozenset({"FIRE", "SPELL"})), StatModifier("attack_damage", ModifierType.FLAT, 20.0, tags=frozenset({"fire"})), StatModifier("attack_damage", ModifierType.FLAT, 10.0, tags=frozenset()), # Universal StatModifier("attack_damage", ModifierType.INCREASED, 50.0, tags=frozenset({"cold"})), ] # Scenario A: Context has {"fire"} -> matches mod 2 (20) and mod 3 (10), ignores mod 1 & mod 4 # Result: (50 + 20 + 10) * 1.0 = 80.0 ctx_a = EvaluationContext(active_tags=frozenset({"fire"})) stats_a = self.aggregator.calculate_stats(custom_modifiers=mods, context=ctx_a) self.assertEqual(stats_a.attack_damage, 80.0) # Scenario B: Context has {"fire", "spell", "area"} (superset) -> matches mod 1 (30), mod 2 (20), mod 3 (10) # Result: (50 + 30 + 20 + 10) * 1.0 = 110.0 ctx_b = EvaluationContext(active_tags=frozenset({"fire", "spell", "area"})) stats_b = self.aggregator.calculate_stats(custom_modifiers=mods, context=ctx_b) self.assertEqual(stats_b.attack_damage, 110.0) # Scenario C: Context has {"cold"} -> matches mod 3 (10) and mod 4 (50% inc) # Result: (50 + 10) * 1.50 = 90.0 ctx_c = EvaluationContext(active_tags=frozenset({"cold"})) stats_c = self.aggregator.calculate_stats(custom_modifiers=mods, context=ctx_c) self.assertEqual(stats_c.attack_damage, 90.0) # Scenario D: Empty context -> only mod 3 (universal) matches ctx_d = EvaluationContext() stats_d = self.aggregator.calculate_stats(custom_modifiers=mods, context=ctx_d) self.assertEqual(stats_d.attack_damage, 60.0) def test_stress_conditional_modifiers_edge_cases(self) -> None: """Adversarial probing of conditional triggers at exact mathematical thresholds.""" mods = [ StatModifier("attack_damage", ModifierType.MORE, 40.0, condition="on_low_health"), StatModifier("max_hp", ModifierType.INCREASED, 20.0, condition="on_full_health"), StatModifier("attack_damage", ModifierType.FLAT, 25.0, condition="wielding_sword"), ] # Exact threshold test: 0.35 is low life in PoE ctx_exact_low = EvaluationContext(current_hp_ratio=0.35) stats_low = self.aggregator.calculate_stats(custom_modifiers=mods, context=ctx_exact_low) self.assertEqual(stats_low.attack_damage, 70.0) # 50.0 * 1.40 = 70.0 # Just above threshold: 0.3501 is NOT low life ctx_above_low = EvaluationContext(current_hp_ratio=0.3501) stats_above = self.aggregator.calculate_stats(custom_modifiers=mods, context=ctx_above_low) self.assertEqual(stats_above.attack_damage, 50.0) # Full health boundary: 1.0 is full health, 0.999 is NOT ctx_full = EvaluationContext(current_hp_ratio=1.0) stats_full = self.aggregator.calculate_stats(custom_modifiers=mods, context=ctx_full) self.assertEqual(stats_full.max_hp, 1260.0) # (1000 + 50 STR) * 1.20 = 1260.0 ctx_not_full = EvaluationContext(current_hp_ratio=0.999) stats_not_full = self.aggregator.calculate_stats(custom_modifiers=mods, context=ctx_not_full) self.assertEqual(stats_not_full.max_hp, 1050.0) # (1000 + 50 STR) * 1.0 = 1050.0 # Custom named condition with case-insensitive string matching ctx_sword = EvaluationContext(conditions=frozenset({"WIELDING_SWORD"})) stats_sword = self.aggregator.calculate_stats(custom_modifiers=mods, context=ctx_sword) self.assertEqual(stats_sword.attack_damage, 75.0) # 50 + 25 = 75.0 def test_stress_weapon_grip_matrix(self) -> None: """Adversarial validation of 2H and Dual Wield detection logic.""" # Grip 1: Two-Handed weapon grants +50% More damage item_2h = SimpleNamespace( name="Huyết Trảm Cổ Đao 2H", item_id="wpn_greatsword", slot="MAIN_HAND", item_type=ItemType.WEAPON, affixes=[], metadata={"grip": "TWO_HANDED", "is_two_handed": True} ) stats_2h = self.aggregator.calculate_stats(inventory=[item_2h]) self.assertEqual(stats_2h.attack_damage, 75.0) # 50 * 1.50 = 75.0 # Grip 2: Dual Wield (1H Main + 1H Off) grants +10% More APS and +15% Block item_mh = SimpleNamespace( name="Tàn Sát Đao", item_id="wpn_1h_a", slot="MAIN_HAND", item_type=ItemType.WEAPON, affixes=[], metadata={} ) item_oh = SimpleNamespace( name="Phong Ma Kiếm", item_id="wpn_1h_b", slot="OFF_HAND", item_type=ItemType.WEAPON, affixes=[], metadata={} ) stats_dw = self.aggregator.calculate_stats(inventory=[item_mh, item_oh]) # Base APS = 1.0, DEX +50 -> +7.5% inc -> 1.075 * 1.10 (More) = 1.1825 -> rounded 1.18 self.assertAlmostEqual(stats_dw.all_stats["attack_speed"], 1.18, places=2) self.assertEqual(stats_dw.all_stats["block_chance"], 15.0) # Grip 3: 1H Weapon + Shield does NOT trigger Dual Wield item_shield = SimpleNamespace( name="Hắc Thiết Thuẫn", item_id="shield_iron", slot="OFF_HAND", item_type=ItemType.ARMOR, affixes=[], metadata={} ) stats_sword_shield = self.aggregator.calculate_stats(inventory=[item_mh, item_shield]) self.assertNotIn("block_chance", stats_sword_shield.all_stats) def test_stress_ast_and_sqlite_persistence(self) -> None: """Adversarial validation of AST node composition and SQLite WAL persistence.""" with tempfile.TemporaryDirectory() as tmp_dir: db_path = os.path.join(tmp_dir, "challenger_audit.db") persist_svc = FormulaPersistenceService(db_path=db_path) agg = CharacterStatAggregator(persistence_service=persist_svc) mods = [ StatModifier("attack_damage", ModifierType.FLAT, 25.0, source="flat_gem"), StatModifier("attack_damage", ModifierType.INCREASED, 30.0, source="passive_tree"), StatModifier("attack_damage", ModifierType.MORE, 20.0, source="support_gem"), ] final_stats = agg.calculate_stats( custom_modifiers=mods, player_id="player_stress_01" ) # (50 + 25) * 1.30 * 1.20 = 75 * 1.56 = 117.0 self.assertEqual(final_stats.attack_damage, 117.0) # Query persisted record from SQLite records = persist_svc.get_player_calculations("player_stress_01", limit=5) self.assertEqual(len(records), 1) rec = records[0] self.assertEqual(rec["player_id"], "player_stress_01") self.assertEqual(rec["final_stats"]["attack_damage"], 117.0) # Verify AST structural integrity ast = rec["formula_ast"]["attack_damage"] self.assertEqual(ast["base"]["value"], 50.0) self.assertEqual(ast["flat"]["total"], 25.0) self.assertEqual(ast["scale"]["multiplier"], 1.30) self.assertEqual(ast["more"]["multiplier"], 1.20) persist_svc.close() def test_stress_server_engine_loop_live_integration(self) -> None: """Adversarial validation of ServerEngineLoop player registration with full stats.""" persist_svc = FormulaPersistenceService(":memory:") agg = CharacterStatAggregator(persistence_service=persist_svc) loop = ServerEngineLoop(cell_size=64.0, stat_aggregator=agg) # Mock player equipment and passives affix_hp = Affix(name="Thần Thể", affix_type=AffixType.PREFIX, stat_key="max_hp", min_val=100, max_val=100, current_val=100) affix_atk = Affix(name="Toái Cốt", affix_type=AffixType.PREFIX, stat_key="phys_dmg", min_val=40, max_val=40, current_val=40) item = SimpleNamespace( name="Cự Kiếm 2H", item_id="wpn_2h", slot="MAIN_HAND", item_type=ItemType.WEAPON, affixes=[affix_hp, affix_atk], metadata={"is_two_handed": True, "grip": "TWO_HANDED"} ) meridian = MeridianStatBonus(hp=200, dps=30.0, dps_mult=0.20, crit_rate=0.05, resist=25.0) # Directly calculate stats for player calculated = agg.calculate_stats( inventory=[item], meridian_bonus=meridian, player_id="hero_999" ) # Max HP: 1000 (base) + 50 (STR) + 100 (affix) + 200 (meridian) = 1350.0 self.assertEqual(calculated.max_hp, 1350.0) # Attack: (50 base + 40 affix + 30 meridian) * (1 + 0.20 inc) * (1.50 2H more) = 120 * 1.20 * 1.50 = 216.0 self.assertEqual(calculated.attack_damage, 216.0) # Register player into live ServerEngineLoop with aggregated_stats actor = loop.register_player( entity_id=42, initial_x=15.0, initial_y=25.0, element=FiveElements.HOA, aggregated_stats=calculated ) self.assertEqual(actor.actor_id, 42) self.assertEqual(actor.max_hp, 1350.0) self.assertEqual(actor.current_hp, 1350.0) self.assertEqual(actor.base_attack, 216.0) self.assertTrue(actor.is_player) self.assertEqual(actor.resistances[FiveElements.HOA], 25.0) self.assertEqual(loop.movement_authority.players[42].move_speed, 6.0) def test_stress_extreme_float_precision_and_clamping(self) -> None: """Adversarial stress of float rounding, crit clamps [0, 1], and speed clamps >= 1.0.""" # Crit chance cannot exceed 1.0 (100%) or fall below 0.0 mods_high_crit = [StatModifier("crit_chance", ModifierType.FLAT, 5.0)] stats_high = self.aggregator.calculate_stats(custom_modifiers=mods_high_crit) self.assertEqual(stats_high.crit_chance, 1.0) mods_neg_crit = [StatModifier("crit_chance", ModifierType.FLAT, -1.0)] stats_neg = self.aggregator.calculate_stats(custom_modifiers=mods_neg_crit) self.assertEqual(stats_neg.crit_chance, 0.0) # Move speed cannot drop below 1.0 mods_slow = [StatModifier("move_speed", ModifierType.REDUCED, 200.0)] stats_slow = self.aggregator.calculate_stats(custom_modifiers=mods_slow) self.assertEqual(stats_slow.move_speed, 1.0) if __name__ == "__main__": unittest.main()