""" Unit tests for FreeExile Character Stat Aggregator, Formula Persistence, and Server Engine Integration. Verifies Path of Exile modifier mathematics, tag filtering, conditional modifiers, AST generation, SQLite persistence, and ServerEngineLoop registration. Strictly decoupled, non-trivial test cases with genuine mathematical validations. """ import os import sys import json import tempfile import unittest 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 MockItem: def __init__(self, name: str, item_id: str, affixes: list, is_2h: bool = False, slot: str = "MAIN_HAND", item_type: ItemType = ItemType.WEAPON): self.name = name self.item_id = item_id self.affixes = affixes self.item_type = item_type self.slot = slot self.metadata = {"is_two_handed": is_2h, "grip": "TWO_HANDED" if is_2h else "ONE_HANDED"} class TestCharacterStatAggregator(unittest.TestCase): def setUp(self) -> None: self.aggregator = CharacterStatAggregator() def test_flat_modifiers_only(self) -> None: """Verifies base 50.0 attack + 30.0 flat yields exactly 80.0 without scaling.""" mods = [StatModifier("attack_damage", ModifierType.FLAT, 30.0, source="gem")] ctx = EvaluationContext(active_tags=frozenset({"physical"})) stats = self.aggregator.calculate_stats(custom_modifiers=mods, context=ctx) stats_clean = self.aggregator.calculate_stats(custom_modifiers=mods, context=EvaluationContext()) self.assertEqual(stats_clean.attack_damage, 80.0) def test_increased_and_reduced_modifiers(self) -> None: """Verifies additive (1.0 + (Inc - Red)/100.0) percentage scaling.""" mods = [ StatModifier("attack_damage", ModifierType.INCREASED, 40.0, source="ring"), StatModifier("attack_damage", ModifierType.REDUCED, 10.0, source="curse"), ] stats = self.aggregator.calculate_stats(custom_modifiers=mods, context=EvaluationContext()) self.assertEqual(stats.attack_damage, 65.0) def test_more_and_less_multipliers(self) -> None: """Verifies multiplicative compound scaling: base * (1 + More/100) * (1 - Less/100).""" mods = [ StatModifier("attack_damage", ModifierType.MORE, 20.0, source="support_gem"), StatModifier("attack_damage", ModifierType.LESS, 10.0, source="debuff"), ] stats = self.aggregator.calculate_stats(custom_modifiers=mods, context=EvaluationContext()) self.assertEqual(stats.attack_damage, 54.0) def test_full_poe_formula_combination(self) -> None: """Verifies (Base + Flat) * (1 + Inc/100) * More = (50 + 30) * 1.60 * 1.50 = 192.0.""" mods = [ StatModifier("attack_damage", ModifierType.FLAT, 30.0, source="sword"), StatModifier("attack_damage", ModifierType.INCREASED, 50.0, source="belt"), StatModifier("attack_damage", ModifierType.MORE, 50.0, source="two_hand_grip"), ] ctx = EvaluationContext(active_tags=frozenset({"physical", "melee", "attack"})) stats = self.aggregator.calculate_stats(custom_modifiers=mods, context=ctx) self.assertEqual(stats.attack_damage, 192.0) def test_tag_based_filtering(self) -> None: """Verifies modifiers are only evaluated when context tags satisfy required tags.""" mods = [ StatModifier("attack_damage", ModifierType.FLAT, 40.0, tags=frozenset({"fire"}), source="fire_gem"), StatModifier("attack_damage", ModifierType.FLAT, 20.0, tags=frozenset({"cold"}), source="cold_gem"), ] ctx_fire = EvaluationContext(active_tags=frozenset({"fire"})) stats_fire = self.aggregator.calculate_stats(custom_modifiers=mods, context=ctx_fire) self.assertEqual(stats_fire.attack_damage, 90.0) ctx_cold = EvaluationContext(active_tags=frozenset({"cold"})) stats_cold = self.aggregator.calculate_stats(custom_modifiers=mods, context=ctx_cold) self.assertEqual(stats_cold.attack_damage, 70.0) def test_conditional_modifiers_on_low_health(self) -> None: """Verifies conditional modifier triggers when on_low_health and is skipped otherwise.""" mods = [ StatModifier("attack_damage", ModifierType.MORE, 30.0, condition="on_low_health", source="low_hp_node") ] ctx_low = EvaluationContext(current_hp_ratio=0.20) stats_low = self.aggregator.calculate_stats(custom_modifiers=mods, context=ctx_low) self.assertEqual(stats_low.attack_damage, 65.0) ctx_full = EvaluationContext(current_hp_ratio=1.0) stats_full = self.aggregator.calculate_stats(custom_modifiers=mods, context=ctx_full) self.assertEqual(stats_full.attack_damage, 50.0) def test_mock_player_items_and_passives_ingestion(self) -> None: """Verifies aggregator ingests equipment items and Meridian bonuses cleanly.""" affix_hp = Affix(name="Thần Thể", affix_type=AffixType.PREFIX, stat_key="max_hp", min_val=80, max_val=80, current_val=80) affix_phys = Affix(name="Toái Cốt", affix_type=AffixType.PREFIX, stat_key="phys_dmg", min_val=35, max_val=35, current_val=35) item = MockItem("Thanh Long Kiếm", "wpn_sword_1", [affix_hp, affix_phys]) meridian = MeridianStatBonus(hp=150, dps=20.0, dps_mult=0.10, crit_rate=0.04, resist=15.0) stats = self.aggregator.calculate_stats(inventory=[item], meridian_bonus=meridian, context=EvaluationContext()) self.assertEqual(stats.max_hp, 1280.0) self.assertEqual(stats.attack_damage, 115.5) self.assertAlmostEqual(stats.crit_chance, 0.14, places=4) self.assertEqual(stats.resistances["hoa"], 15.0) def test_weapon_mechanics_two_handed_and_dual_wield(self) -> None: """Verifies 2H weapon grants +50% More damage and Dual-Wield grants +10% More APS & +15% block.""" item_2h = MockItem("Cự Đại Cổ Đao 2H", "wpn_2h_blade", [], is_2h=True, slot="MAIN_HAND") stats_2h = self.aggregator.calculate_stats(inventory=[item_2h]) self.assertEqual(stats_2h.attack_damage, 75.0) item_dw_1 = MockItem("Đoản Kiếm 1", "wpn_1h_1", [], is_2h=False, slot="MAIN_HAND") item_dw_2 = MockItem("Đoản Kiếm 2", "wpn_1h_2", [], is_2h=False, slot="OFF_HAND") stats_dw = self.aggregator.calculate_stats(inventory=[item_dw_1, item_dw_2]) self.assertGreater(stats_dw.all_stats["attack_speed"], 1.15) self.assertEqual(stats_dw.all_stats.get("block_chance", 0.0), 15.0) def test_ast_hierarchy_and_json_serialization(self) -> None: """Verifies AST contains expected node types and serializes to valid JSON.""" mods = [ StatModifier("attack_damage", ModifierType.FLAT, 20.0, source="flat_src"), StatModifier("attack_damage", ModifierType.INCREASED, 15.0, source="inc_src"), StatModifier("attack_damage", ModifierType.MORE, 10.0, source="more_src"), ] calc_service = FormulaPersistenceService(":memory:") agg = CharacterStatAggregator(persistence_service=calc_service) agg.calculate_stats(custom_modifiers=mods, player_id="player_ast_test") history = calc_service.get_player_calculations("player_ast_test", limit=1) self.assertEqual(len(history), 1) ast_data = history[0]["formula_ast"] self.assertIn("attack_damage", ast_data) atk_ast = ast_data["attack_damage"] self.assertEqual(atk_ast["base"]["ast_type"], "Constant") self.assertEqual(atk_ast["flat"]["ast_type"], "Sum") self.assertEqual(atk_ast["scale"]["ast_type"], "ScaleFactor") self.assertEqual(atk_ast["more"]["ast_type"], "Product") serialized = json.dumps(ast_data) self.assertIsInstance(json.loads(serialized), dict) def test_sqlite_persistence_and_queries(self) -> None: """Verifies disk-backed SQLite database persistence, retrieval, and schema integrity.""" with tempfile.TemporaryDirectory() as tmp_dir: db_path = os.path.join(tmp_dir, "test_formulas.db") service = FormulaPersistenceService(db_path=db_path) row_id = service.save_calculation( player_id="user_123", calculation_id="calc_123", formula_ast={"hp": {"value": 1200.0}}, final_stats={"max_hp": 1200.0, "attack_damage": 75.0}, context_tags=["melee"], ) self.assertGreater(row_id, 0) rec = service.get_calculation("calc_123") self.assertIsNotNone(rec) self.assertEqual(rec["player_id"], "user_123") self.assertEqual(rec["final_stats"]["max_hp"], 1200.0) player_history = service.get_player_calculations("user_123") self.assertEqual(len(player_history), 1) self.assertEqual(player_history[0]["calculation_id"], "calc_123") service.close() def test_server_engine_loop_registration_with_aggregated_stats(self) -> None: """Verifies ServerEngineLoop registers CombatActor with calculated stats > 50.0.""" agg = CharacterStatAggregator() engine = ServerEngineLoop(cell_size=64.0, stat_aggregator=agg) custom_stats = AggregatedCharacterStats( attack_damage=140.0, max_hp=1500.0, crit_chance=0.25, crit_multiplier=2.10, move_speed=12.0, resistances={"hoa": 45.0, "thuy": 30.0}, ) actor = engine.register_player( entity_id=1, initial_x=10.0, initial_y=10.0, aggregated_stats=custom_stats ) self.assertEqual(actor.base_attack, 140.0) self.assertEqual(actor.max_hp, 1500.0) self.assertEqual(actor.crit_chance, 0.25) self.assertEqual(actor.crit_multiplier, 2.10) self.assertTrue(actor.is_player) self.assertEqual(actor.resistances[FiveElements.HOA], 45.0) player_char = engine.movement_authority.players[1] self.assertEqual(player_char.move_speed, 12.0) def test_server_engine_loop_backward_compatibility(self) -> None: """Verifies legacy register_player without stats maintains 50.0 attack and 1000.0 HP.""" engine = ServerEngineLoop(cell_size=64.0) actor = engine.register_player(entity_id=99, initial_x=0.0, initial_y=0.0) self.assertEqual(actor.base_attack, 50.0) self.assertEqual(actor.max_hp, 1000.0) self.assertEqual(actor.crit_chance, 0.05) self.assertEqual(actor.crit_multiplier, 1.50) self.assertEqual(engine.movement_authority.players[99].move_speed, 6.0) if __name__ == "__main__": unittest.main()