""" Bộ Kiểm Thử Toàn Diện Hệ Thống Agentic Self-Learning Engine (AutoPOE2 A3E) Kiểm chứng: LogWatcher, BlackboxRecorder, DPSAccumulator, RunTracker, LinUCBTacticalAdvisor, ReflexTuner, ReflectionCritic, và WhatIsBetterEvaluator. """ import os import shutil import sys import tempfile import time import unittest sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) from src.agentic_engine.bandit_advisor import LinUCBTacticalAdvisor from src.agentic_engine.blackbox import BlackboxRecorder from src.agentic_engine.dps_meter import DPSAccumulator from src.agentic_engine.reflection_critic import ReflectionCritic from src.agentic_engine.reflex_tuner import ReflexParameterTuner from src.agentic_engine.run_tracker import RunTracker from src.agentic_engine.what_is_better import ( ItemStats, WhatIsBetterEvaluator, ) from src.assistant_tool.core_shm_bridge import CoreTelemetrySnapshot, PlayerTelemetry from src.assistant_tool.log_watcher import LogWatcher class TestAgenticEngine(unittest.TestCase): def setUp(self): self.test_dir = tempfile.mkdtemp(prefix="poe2_agentic_test_") def tearDown(self): if os.path.exists(self.test_dir): shutil.rmtree(self.test_dir, ignore_errors=True) # 1. Kiểm thử LogWatcher bắt sự kiện tử trận def test_log_watcher_death_detection(self): lw = LogWatcher(log_path="dummy.txt") line_slain = "2026/09/07 19:09:17 12345678 [INFO Client 34188] : BEONGUSI has been slain." evt = lw.parse_line(line_slain) self.assertIsNotNone(evt) self.assertEqual(evt.event_type, "PLAYER_DEATH") self.assertEqual(evt.sender, "BEONGUSI") self.assertIn("has been slain", evt.message) line_suicide = "2026/09/07 19:10:00 12345678 [INFO Client 34188] : ShadowRunner has committed suicide." evt_suicide = lw.parse_line(line_suicide) self.assertIsNotNone(evt_suicide) self.assertEqual(evt_suicide.event_type, "PLAYER_DEATH") # 2. Kiểm thử BlackboxRecorder lưu trữ và xuất hộp đen def test_blackbox_recorder(self): bb = BlackboxRecorder(capacity=10, output_dir=self.test_dir) for i in range(15): snap = CoreTelemetrySnapshot( snapshot_id=i, timestamp_ms=1000 + i * 100, player=PlayerTelemetry(current_hp=1000 - i * 50, max_hp=1000, hp_percentage=(1000 - i * 50) / 10.0), monsters_count=5, area_name="The Riverbank", area_level=68 ) bb.record_snapshot(snap) # Buffer tối đa 10 ticks ticks = bb.get_recent_ticks() self.assertEqual(len(ticks), 10) self.assertEqual(ticks[-1]["snapshot_id"], 14) # Xuất hộp đen ra file JSON filepath = bb.export_blackbox(trigger_reason="TEST_BURST") self.assertTrue(os.path.exists(filepath)) self.assertIn("TEST_BURST", filepath.upper()) # 3. Kiểm thử DPSAccumulator đo lường Realized DPS def test_dps_accumulator(self): dps = DPSAccumulator(window_seconds=2.0) t0 = 100.0 # Tick 1: Quái A có 5,000 HP, Quái B có 3,000 HP monsters_t1 = [{"id": 1, "hp": 5000}, {"id": 2, "hp": 3000}] dps.update_monsters_snapshot(monsters_t1, current_time=t0) # Tick 2 (sau 1.0s): Quái A mất 2,000 HP, Quái B mất 1,000 HP -> Tổng sát thương = 3,000 monsters_t2 = [{"id": 1, "hp": 3000}, {"id": 2, "hp": 2000}] current_dps = dps.update_monsters_snapshot(monsters_t2, current_time=t0 + 1.0) self.assertGreater(current_dps, 0.0) self.assertEqual(dps.total_damage_dealt, 3000) self.assertEqual(dps.peak_dps, current_dps) # 4. Kiểm thử RunTracker quản lý map và clear time def test_run_tracker(self): hist_file = os.path.join(self.test_dir, "test_run_history.json") tracker = RunTracker(history_file=hist_file) # Bước vào Map Worlds Beach tracker.on_zone_change("MapWorldsBeach", area_level=75) self.assertIsNotNone(tracker.current_run) self.assertEqual(tracker.current_run.area_name, "MapWorldsBeach") # Ghi nhận tử vong và cập nhật stats tracker.record_death() tracker.update_run_stats(peak_dps=12500.0, kill_increment=45) self.assertEqual(tracker.current_run.deaths_count, 1) self.assertEqual(tracker.current_run.monsters_killed, 45) # Về Hideout -> Kết thúc map completed = tracker.on_zone_change("Hideout", area_level=1) self.assertIsNotNone(completed) self.assertEqual(completed.status, "COMPLETED") self.assertEqual(len(tracker.run_history), 1) summary = tracker.get_session_summary() self.assertEqual(summary["total_runs"], 1) self.assertEqual(summary["total_deaths"], 1) # 5. Kiểm thử Contextual Bandit LinUCB thích ứng online def test_contextual_bandit_linucb(self): model_file = os.path.join(self.test_dir, "bandit_test.json") bandit = LinUCBTacticalAdvisor(model_path=model_file, alpha=0.5) # Ngữ cảnh: HP tụt thấp 20%, nhiều quái (15 con) ctx = bandit.extract_context( hp_pct=20.0, es_pct=0.0, monster_count=15, boss_poise_pct=0.0, flask_charges_pct=10.0, map_tier=15, recent_dps=1200.0, damage_taken_rate=80.0 ) rec_action, score, all_scores = bandit.recommend_action(ctx) self.assertIn(rec_action, bandit.actions) self.assertIn("EMERGENCY_PORTAL", all_scores) # Phản hồi tiêu cực (Chết do đánh bừa khi máu thấp) reward_bad = bandit.compute_combat_reward("AGGRESSIVE_BURST", 0.0, 500.0, 0, is_player_slain=True) self.assertEqual(reward_bad, -15.0) bandit.update_feedback("AGGRESSIVE_BURST", ctx, reward_bad) # Phản hồi tích cực (Mở portal thoát thân an toàn) reward_good = bandit.compute_combat_reward("EMERGENCY_PORTAL", 25.0, 0.0, 0, is_player_slain=False) bandit.update_feedback("EMERGENCY_PORTAL", ctx, reward_good) # 6. Kiểm thử ReflexParameterTuner tự động siết an toàn def test_reflex_tuner(self): cfg_file = os.path.join(self.test_dir, "reflex_test.json") tuner = ReflexParameterTuner(config_path=cfg_file) self.assertEqual(tuner.flask_hp_threshold, 0.55) # Chết 2 lần -> Phải tự động siết an toàn (bơm máu sớm hơn, kiting xa hơn) tuner.record_death() params = tuner.record_death() self.assertGreater(params["flask_hp_threshold"], 0.55) self.assertGreater(params["kiting_distance"], 35.0) self.assertLess(params["burst_dodge_threshold"], 0.22) # 7. Kiểm thử ReflectionCritic chẩn đoán nguyên nhân tử vong def test_reflection_critic(self): mem_file = os.path.join(self.test_dir, "agent_memory_test.json") critic = ReflectionCritic(memory_file=mem_file) mock_ticks = [ {"snapshot_id": 1, "timestamp_ms": 1000, "hp": 4000, "max_hp": 4000, "monsters_count": 2}, {"snapshot_id": 2, "timestamp_ms": 2000, "hp": 0, "max_hp": 4000, "monsters_count": 2} ] lesson = critic.analyze_death_event( area_name="The Black Gate", area_level=80, killer_info="Jamanra, the Sandstorm King", blackbox_ticks=mock_ticks ) self.assertIsNotNone(lesson) self.assertEqual(lesson["local_diagnosis"]["root_cause"], "BURST_SLAM_DAMAGE") self.assertIn("SỐC SÁT THƯƠNG", lesson["local_diagnosis"]["summary"]) # 8. Kiểm thử WhatIsBetterEvaluator so sánh trang bị & hoạt động kinh tế def test_what_is_better_evaluator(self): evaluator = WhatIsBetterEvaluator() curr_gloves = ItemStats( name="Iron Gauntlets of Warding", slot="gloves", flat_phys=15.0, flat_cold=0.0, attack_speed_pct=5.0, max_life=40, fire_res=20, cold_res=20 ) cand_gloves = ItemStats( name="Spiked Fists of the Blizzard", slot="gloves", flat_phys=30.0, flat_cold=55.0, # Huge cold damage for Stonefist build attack_speed_pct=16.0, crit_multiplier_pct=25.0, max_life=85, fire_res=35, cold_res=35 ) result = evaluator.compare_items(curr_gloves, cand_gloves) self.assertEqual(result.winner, "CANDIDATE") self.assertEqual(result.decision, "EQUIP_UPGRADE") self.assertGreater(result.dps_delta_pct, 50.0) self.assertIn("ĐỒ MỚI TỐT HƠN VƯỢT TRỘI", result.headline_verdict) # Kiểm thử so sánh kinh tế econ_res = evaluator.compare_farming_activities(current_gold=30000, available_divine=10.0) self.assertEqual(econ_res["recommended_activity"], "SEKHEMA_CARRY") self.assertIn("SEKHEMA", econ_res["headline_verdict"].upper()) # 9. Kiểm thử Quy trình Đồ Chưa Giám Định (ItemLifecycleManager) def test_item_lifecycle_unidentified_pipeline(self): from src.agentic_engine.item_lifecycle import ( IdentificationChannel, ItemLifecycleManager, ) db_path = os.path.join(self.test_dir, "test_learning_db.json") manager = ItemLifecycleManager(learning_db_path=db_path) # 1. Sàng lọc tiền giám định phôi S-Tier (Spiked Fists iLvl 79 trong map có sẵn Scroll) triage_s = manager.triage_unidentified_item( base_name="Spiked Fists", item_level=79, rarity="Rare", slot="gloves", scrolls_available=25, in_combat_zone=True ) self.assertEqual(triage_s.recommended_channel, IdentificationChannel.FIELD_SCROLL_OF_WISDOM) self.assertGreater(triage_s.base_tier_score, 80.0) # 2. Sàng lọc phôi Rare thông thường -> Gom về The Hooded One triage_common = manager.triage_unidentified_item( base_name="Varnished Coat", item_level=70, rarity="Rare", slot="body_armour", scrolls_available=25, in_combat_zone=True ) self.assertEqual(triage_common.recommended_channel, IdentificationChannel.TOWN_THE_HOODED_ONE) # 3. Kiểm tra kích hoạt The Hooded One tại Hideout hooded_check_combat = manager.check_the_hooded_one_interaction( current_area_name="MapWorldsBeach", unidentified_items_in_inventory=6 ) self.assertFalse(hooded_check_combat["can_interact"]) hooded_check_town = manager.check_the_hooded_one_interaction( current_area_name="PersonalHideout", unidentified_items_in_inventory=6 ) self.assertTrue(hooded_check_town["can_interact"]) self.assertEqual(hooded_check_town["npc_name"], "The Hooded One") self.assertIn("Identify All", hooded_check_town["action_guidance"]) # 4. Hậu giám định: Giám định ra món đồ và tự học curr_gloves = ItemStats(name="Iron Gauntlets", slot="gloves", flat_phys=10.0, max_life=30) revealed_gloves = ItemStats(name="Spiked Fists of Winter", slot="gloves", flat_phys=28.0, flat_cold=45.0, max_life=75) comp_result = manager.process_identified_item( item_id="item_001", revealed_stats=revealed_gloves, current_equipped=curr_gloves ) self.assertEqual(comp_result.decision, "EQUIP_UPGRADE") # Kiểm tra tri thức học máy đã được cập nhật knowledge = manager.get_knowledge_summary() self.assertTrue(any(k["base_name"] == "Spiked Fists" for k in knowledge)) spiked_stats = next(k for k in knowledge if k["base_name"] == "Spiked Fists") self.assertEqual(spiked_stats["identified_count"], 1) self.assertEqual(spiked_stats["upgrades_found"], 1) # 10. Kiểm thử AgenticCoordinator Façade def test_agentic_coordinator(self): from src.agentic_engine.coordinator import AgenticCoordinator from src.assistant_tool.core_shm_bridge import CoreTelemetrySnapshot, PlayerTelemetry from src.assistant_tool.log_watcher import LogEvent coord = AgenticCoordinator(enabled=True) # 1. Telemetry Snapshot Event snap = CoreTelemetrySnapshot( is_online=True, snapshot_id=101, timestamp_ms=int(time.time() * 1000), player=PlayerTelemetry(current_hp=3200, max_hp=4000, hp_percentage=80.0), monsters_count=4, area_name="MapWorldsBeach", area_level=78 ) coord.on_telemetry_snapshot(snap) self.assertIn(coord.latest_tactical_action, coord.bandit_advisor.actions) # 2. Log Event: Zone Change zone_evt = LogEvent( event_type="ZONE_CHANGE", sender="System", message="Đã chuyển đến khu vực: PersonalHideout", timestamp="2026/09/07 19:15:00", raw_line="raw" ) coord.on_log_event(zone_evt) # 3. Log Event: Player Death death_evt = LogEvent( event_type="PLAYER_DEATH", sender="Jamanra", message="BEONGUSI has been slain.", timestamp="2026/09/07 19:16:00", raw_line="raw" ) coord.on_log_event(death_evt) # 4. Kiểm tra Dashboard State state = coord.get_dashboard_state() self.assertTrue(state["enabled"]) self.assertGreaterEqual(state["total_deaths"], 1) self.assertIn("Hậu kiểm tử vong", state["advice_headline"]) if __name__ == "__main__": unittest.main()