"""LiveAdapter 离线测试：用 SDK 真实模型伪造 Turn，验证正式世界适配层。 正式世界的事件名/字段和模拟器不同，这层翻译错了策略就是瞎的，但线上没法回归， 所以在这里离线锁住。 运行：.venv/bin/python tests/test_deploy_adapter.py """ import sys from pathlib import Path from uuid import UUID sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from arena_hero import Turn from arena_hero.actions import CancelMoveAction, CommandPlan, DropBeaconAction from arena_hero.enums import BeaconStatus, CoreState, PlayerStatus, UnitType from arena_hero.models import ( ChampionBeacon, CoreView, PlayerState, ResolutionEvent, TerrainView, UnitView, ) from deploy import LiveAdapter from strategies.heuristic import HeuristicStrategy, make_default_genes W_ID = UUID("00000000-0000-0000-0000-0000000000a1") C_ID = UUID("00000000-0000-0000-0000-0000000000c1") E_ID = UUID("00000000-0000-0000-0000-0000000000e1") def make_turn(*, tick=10, worker_pos=(5, 5), core_pos=(4, 5), events=(), beacon_status=None, carrier_id=None, obstacles=(), resources=()): objects = [ UnitView(kind="UNIT", id=W_ID, controlled=True, position=worker_pos, hp=2, unit_type=UnitType.WORKER, cargo=0), CoreView(kind="CORE", id=C_ID, controlled=True, owner_username="tester", position=core_pos, hp=5, shield=5, state=CoreState.NORMAL), ] if obstacles: objects.append(TerrainView(kind="OBSTACLE", positions=tuple(obstacles))) if resources: objects.append(TerrainView(kind="RESOURCE", positions=tuple(resources))) state = PlayerState( status=PlayerStatus.ACTIVE, respawn_at_tick=None, resources=7, population=1, # rules v0.14 / SDK 0.2.9：population_tier/upkeep_next_tick 已删除 champion_beacon=ChampionBeacon(position=(0, 0), status=beacon_status, carrier_id=carrier_id), objects=tuple(objects), events=tuple(events), ) return Turn(tick=tick, state=state, submitter=lambda *a, **k: None) def mk_adapter(): return LiveAdapter(HeuristicStrategy(genes=make_default_genes(), bounds=None)) # ---------------------------------------------------------------------- def test_move_failed_becomes_move_blocked(): """UNIT_MOVE_FAILED(actor_id) + 上一 tick 提交的方向 → MOVE_BLOCKED。""" a = mk_adapter() a._sent_dirs = {W_ID: "RIGHT"} # 模拟上一 tick 提交过 MOVE RIGHT ev = ResolutionEvent(event_id=UUID(int=1), tick=9, event_type="UNIT_MOVE_FAILED", reason_code="MOVE_BLOCKED_TERRAIN", actor_id=W_ID, target_id=None, position=(5, 5), values=None) out = a.translate_events(make_turn(events=[ev])) assert len(out) == 1, out assert out[0] == {"type": "MOVE_BLOCKED", "obj_id": W_ID, "direction": "RIGHT", "reason": "MOVE_BLOCKED_TERRAIN", "source_type": "UNIT_MOVE_FAILED", "blocked_cell": (6, 5)}, out[0] def test_move_failed_without_known_direction_is_skipped(): """方向不明时必须跳过，否则会把错误的格子学成障碍。""" a = mk_adapter() ev = ResolutionEvent(event_id=UUID(int=2), tick=9, event_type="UNIT_MOVE_FAILED", reason_code="MOVE_CONTESTED", actor_id=W_ID, target_id=None, position=(5, 5), values=None) out = a.translate_events(make_turn(events=[ev])) assert out == [], out def test_move_succeeded_becomes_moved(): a = mk_adapter() ev = ResolutionEvent(event_id=UUID(int=3), tick=9, event_type="UNIT_MOVE_SUCCEEDED", reason_code=None, actor_id=W_ID, target_id=None, position=(6, 5), values=None) out = a.translate_events(make_turn(events=[ev])) assert out[0]["type"] == "MOVED" and out[0]["obj_id"] == W_ID, out def test_core_move_failed_uses_migration_direction(): a = mk_adapter() a._core_dir = "UP" ev = ResolutionEvent(event_id=UUID(int=4), tick=9, event_type="CORE_MOVE_FAILED", reason_code="CORE_DESTINATION_TERRAIN_BLOCKED", actor_id=C_ID, target_id=None, position=(4, 5), values=None) out = a.translate_events(make_turn(events=[ev])) assert out[0] == {"type": "MOVE_BLOCKED", "obj_id": C_ID, "direction": "UP", "reason": "CORE_DESTINATION_TERRAIN_BLOCKED", "source_type": "CORE_MOVE_FAILED", "blocked_cell": (4, 4)}, out def test_carries_beacon_derived_from_carrier_id(): a = mk_adapter() turn = make_turn(beacon_status=BeaconStatus.CARRIED, carrier_id=W_ID) obs = a.build_observation(turn) assert obs.units[0]["carries_beacon"] is True turn2 = make_turn(beacon_status=BeaconStatus.CARRIED, carrier_id=E_ID) obs2 = a.build_observation(turn2) assert obs2.units[0]["carries_beacon"] is False def test_beacon_status_compares_as_plain_string(): """策略里是 obs.beacon["status"] == "GROUND"，StrEnum 必须能直接比。""" a = mk_adapter() obs = a.build_observation(make_turn(beacon_status=BeaconStatus.GROUND)) assert obs.beacon["status"] == "GROUND" obs2 = a.build_observation(make_turn()) assert obs2.beacon["status"] == "UNKNOWN" def test_v014_observation_and_status_have_no_upkeep_fields(): """Removed v0.13 maintenance fields must not leak into local adapters.""" a = mk_adapter() turn = make_turn() obs = a.build_observation(turn) assert not hasattr(obs, "population_tier") assert not hasattr(obs, "upkeep_next_tick") status = a.build_status(turn.tick, obs, {"units": {}, "core": None}) assert "upkeep_next" not in status def test_events_come_from_current_turn_not_stale_copy(): """turn.events 已经是上一 Tick 的结果，不能再延迟一拍。""" a = mk_adapter() a._sent_dirs = {W_ID: "RIGHT"} ev = ResolutionEvent(event_id=UUID(int=5), tick=9, event_type="UNIT_MOVE_FAILED", reason_code="X", actor_id=W_ID, target_id=None, position=(5, 5), values=None) obs = a.build_observation(make_turn(events=[ev])) assert obs.prev_events and obs.prev_events[0]["type"] == "MOVE_BLOCKED" def test_memory_learns_obstacle_end_to_end(): """整条链路：连续两次撞同一格 → 临时避让（temp_blocked）， 不学成永久障碍——MOVE_BLOCKED 无法区分地形与单位占位， 永久化会产生假障碍把可达资源围死。""" a = mk_adapter() for _ in range(2): a._sent_dirs = {W_ID: "RIGHT"} ev = ResolutionEvent(event_id=UUID(int=6), tick=9, event_type="UNIT_MOVE_FAILED", reason_code="MOVE_BLOCKED_TERRAIN", actor_id=W_ID, target_id=None, position=(5, 5), values=None) obs = a.build_observation(make_turn(events=[ev])) a.strategy.decide(obs) assert (6, 5) in a.strategy.mem.temp_blocked, a.strategy.mem.temp_blocked assert (6, 5) not in a.strategy.mem.obstacles, a.strategy.mem.obstacles def test_dead_own_unit_is_classified_from_previous_observation(): """The dead Unit is absent from the new state but remains ours for logging.""" from types import SimpleNamespace a = mk_adapter() a._known_own_ids = {W_ID, C_ID} messages = [] a._log_key_event = lambda ev, et, ids, obs=None: messages.append( a._fmt_key_event(ev, et, ids, obs)) death = ResolutionEvent( event_id=UUID(int=7), tick=10, event_type="UNIT_DAMAGED", reason_code="ATTACK", actor_id=None, target_id=W_ID, position=(5, 5), values={"damage": 1, "hp": 0}) empty_obs = SimpleNamespace(units=[], core={"uid": C_ID}, enemies=[]) a.translate_events(make_turn(events=[death]), obs=empty_obs) assert messages and "我方单位阵亡" in messages[0], messages assert a.recent_events[-1]["target_is_own"] is True def _run_all(): tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")] passed = 0 for t in tests: try: t() print(f"PASS {t.__name__}") passed += 1 except AssertionError as e: print(f"FAIL {t.__name__}: {e}") except Exception as e: print(f"ERROR {t.__name__}: {type(e).__name__}: {e}") print(f"{passed}/{len(tests)} passed") return passed == len(tests) def test_memory_from_dict_stationary_pos_tuple(): """回归：memory.json 恢复的 stationary.pos 必须是 tuple—— 否则 _pick_raid_target 返回 list → _raid_point → pf.find key 含 list 崩溃（2026-08-07 agent2 线上每 tick TypeError）。""" from strategies.base import Memory m = Memory.from_dict({ "stationary": [["core_abc", {"pos": [1, 2], "count": 3, "first": 10, "last": 20}]]}) st = m.stationary["core_abc"] assert isinstance(st["pos"], tuple), st["pos"] assert st["pos"] == (1, 2) def test_pathfinder_accepts_list_pos(): """回归：pf.find 必须接受 list 坐标（调用方可能传 list）。""" from strategies.base import Pathfinder pf = Pathfinder(is_obstacle=lambda x, y: False, bounds=(-64, 63, -64, 63)) path = pf.find([0, 0], [3, 0]) assert path and tuple(path[0]) == (1, 0) def test_legacy_result_cannot_override_frozen_v014_genes(): import json import tempfile from deploy import load_genes payload = {"history": [{"gen": 3, "best": 99.0, "genes": { "max_population": 19, "selfdestruct_pop": 19, "worker_ratio": 0.6, "obsolete_gene": 123}}]} with tempfile.NamedTemporaryFile("w", suffix=".json") as fh: json.dump(payload, fh) fh.flush() genes = load_genes(fh.name) assert genes["max_population"] == 32 assert genes["selfdestruct_pop"] == 999 assert genes["worker_ratio"] == 0.6 assert "obsolete_gene" not in genes def test_status_exposes_manual_core_control(): a = mk_adapter() a.strategy.set_manual_core_migration("RIGHT") turn = make_turn() obs = a.build_observation(turn) status = a.build_status(turn.tick, obs, {"units": {}, "core": None}) assert status["core_control"] == { "movement_mode": "migration", "mode": "fixed", "direction": "RIGHT", "heading": "RIGHT", "step_direction": None, "phase": "stationed", "active": False, "paused": False, "wait_reason": None, } def test_status_stationed_mode_blocks_migration(): a = mk_adapter() a.strategy.station_core() a.strategy.set_manual_core_migration("RIGHT") turn = make_turn() obs = a.build_observation(turn) status = a.build_status(turn.tick, obs, {"units": {}, "core": None}) assert status["core_control"]["movement_mode"] == "stationed" assert status["core_control"]["phase"] == "stationed" assert status["core_control"]["active"] is False assert status["core_control"]["paused"] is False def test_core_move_start_failed_uses_submitted_direction(): a = mk_adapter() a._core_dir = "UP" a._sent_core_dir = "RIGHT" ev = ResolutionEvent( event_id=UUID(int=8), tick=9, event_type="CORE_MOVE_START_FAILED", reason_code="CORE_DESTINATION_TERRAIN_BLOCKED", actor_id=C_ID, target_id=None, position=(4, 5), values=None) out = a.translate_events(make_turn(events=[ev])) assert out[0]["type"] == "MOVE_BLOCKED" assert out[0]["direction"] == "RIGHT" assert out[0]["blocked_cell"] == (5, 5) assert out[0]["source_type"] == "CORE_MOVE_START_FAILED" if __name__ == "__main__": sys.exit(0 if _run_all() else 1)