"""Tests for bridge client helper functions. Tests observation_to_dict and commands_to_proto conversion functions using mock protobuf objects. """ import pytest from openra_env.server.bridge_client import commands_to_proto, observation_to_dict from openra_env.generated import rl_bridge_pb2 class TestCommandsToProto: def test_no_op(self): result = commands_to_proto([{"action": "no_op"}]) assert len(result.commands) == 1 assert result.commands[0].action == rl_bridge_pb2.NO_OP def test_move_command(self): result = commands_to_proto([ {"action": "move", "actor_id": 42, "target_x": 100, "target_y": 200} ]) cmd = result.commands[0] assert cmd.action == rl_bridge_pb2.MOVE assert cmd.actor_id == 42 assert cmd.target_x == 100 assert cmd.target_y == 200 def test_attack_command(self): result = commands_to_proto([ {"action": "attack", "actor_id": 10, "target_actor_id": 99} ]) cmd = result.commands[0] assert cmd.action == rl_bridge_pb2.ATTACK assert cmd.actor_id == 10 assert cmd.target_actor_id == 99 def test_build_command(self): result = commands_to_proto([ {"action": "build", "item_type": "powr"} ]) cmd = result.commands[0] assert cmd.action == rl_bridge_pb2.BUILD assert cmd.item_type == "powr" def test_queued_flag(self): result = commands_to_proto([ {"action": "move", "actor_id": 1, "target_x": 10, "target_y": 20, "queued": True} ]) assert result.commands[0].queued is True def test_multiple_commands(self): result = commands_to_proto([ {"action": "move", "actor_id": 1, "target_x": 10, "target_y": 20}, {"action": "attack", "actor_id": 2, "target_actor_id": 50}, {"action": "build", "item_type": "barr"}, ]) assert len(result.commands) == 3 assert result.commands[0].action == rl_bridge_pb2.MOVE assert result.commands[1].action == rl_bridge_pb2.ATTACK assert result.commands[2].action == rl_bridge_pb2.BUILD def test_unknown_action_defaults_to_noop(self): result = commands_to_proto([{"action": "invalid_action"}]) assert result.commands[0].action == rl_bridge_pb2.NO_OP def test_missing_action_defaults_to_noop(self): result = commands_to_proto([{}]) assert result.commands[0].action == rl_bridge_pb2.NO_OP def test_all_action_types(self): action_types = [ ("no_op", rl_bridge_pb2.NO_OP), ("move", rl_bridge_pb2.MOVE), ("attack_move", rl_bridge_pb2.ATTACK_MOVE), ("attack", rl_bridge_pb2.ATTACK), ("stop", rl_bridge_pb2.STOP), ("harvest", rl_bridge_pb2.HARVEST), ("build", rl_bridge_pb2.BUILD), ("train", rl_bridge_pb2.TRAIN), ("deploy", rl_bridge_pb2.DEPLOY), ("sell", rl_bridge_pb2.SELL), ("repair", rl_bridge_pb2.REPAIR), ("place_building", rl_bridge_pb2.PLACE_BUILDING), ("cancel_production", rl_bridge_pb2.CANCEL_PRODUCTION), ("set_rally_point", rl_bridge_pb2.SET_RALLY_POINT), ] for action_str, expected_enum in action_types: result = commands_to_proto([{"action": action_str}]) assert result.commands[0].action == expected_enum, f"Failed for {action_str}" def test_empty_list(self): result = commands_to_proto([]) assert len(result.commands) == 0 def test_default_values_for_missing_fields(self): result = commands_to_proto([{"action": "move"}]) cmd = result.commands[0] assert cmd.actor_id == 0 assert cmd.target_actor_id == 0 assert cmd.target_x == 0 assert cmd.target_y == 0 assert cmd.item_type == "" assert cmd.queued is False class TestObservationToDict: def _make_observation(self, **kwargs): """Create a protobuf GameObservation with given fields.""" obs = rl_bridge_pb2.GameObservation() obs.tick = kwargs.get("tick", 0) obs.done = kwargs.get("done", False) obs.result = kwargs.get("result", "") obs.reward = kwargs.get("reward", 0.0) if "economy" in kwargs: eco = kwargs["economy"] obs.economy.cash = eco.get("cash", 0) obs.economy.ore = eco.get("ore", 0) obs.economy.power_provided = eco.get("power_provided", 0) obs.economy.power_drained = eco.get("power_drained", 0) obs.economy.resource_capacity = eco.get("resource_capacity", 0) obs.economy.harvester_count = eco.get("harvester_count", 0) if "military" in kwargs: mil = kwargs["military"] obs.military.units_killed = mil.get("units_killed", 0) obs.military.units_lost = mil.get("units_lost", 0) obs.military.buildings_killed = mil.get("buildings_killed", 0) obs.military.buildings_lost = mil.get("buildings_lost", 0) obs.military.army_value = mil.get("army_value", 0) obs.military.active_unit_count = mil.get("active_unit_count", 0) if "map_info" in kwargs: mi = kwargs["map_info"] obs.map_info.width = mi.get("width", 0) obs.map_info.height = mi.get("height", 0) obs.map_info.map_name = mi.get("map_name", "") for u in kwargs.get("units", []): unit = obs.units.add() unit.actor_id = u.get("actor_id", 0) unit.type = u.get("type", "") unit.pos_x = u.get("pos_x", 0) unit.pos_y = u.get("pos_y", 0) unit.cell_x = u.get("cell_x", 0) unit.cell_y = u.get("cell_y", 0) unit.hp_percent = u.get("hp_percent", 1.0) unit.is_idle = u.get("is_idle", True) unit.current_activity = u.get("current_activity", "") unit.owner = u.get("owner", "") unit.can_attack = u.get("can_attack", False) for b in kwargs.get("buildings", []): bldg = obs.buildings.add() bldg.actor_id = b.get("actor_id", 0) bldg.type = b.get("type", "") bldg.pos_x = b.get("pos_x", 0) bldg.pos_y = b.get("pos_y", 0) bldg.hp_percent = b.get("hp_percent", 1.0) bldg.owner = b.get("owner", "") bldg.is_producing = b.get("is_producing", False) bldg.production_progress = b.get("production_progress", 0.0) bldg.producing_item = b.get("producing_item", "") bldg.is_powered = b.get("is_powered", True) for p in kwargs.get("production", []): prod = obs.production.add() prod.queue_type = p.get("queue_type", "") prod.item = p.get("item", "") prod.progress = p.get("progress", 0.0) prod.remaining_ticks = p.get("remaining_ticks", 0) prod.remaining_cost = p.get("remaining_cost", 0) prod.paused = p.get("paused", False) for ap in kwargs.get("available_production", []): obs.available_production.append(ap) return obs def test_basic_fields(self): obs = self._make_observation(tick=42, done=True, result="win", reward=1.5) d = observation_to_dict(obs) assert d["tick"] == 42 assert d["done"] is True assert d["result"] == "win" assert d["reward"] == 1.5 def test_economy(self): obs = self._make_observation( economy={"cash": 5000, "power_provided": 100, "power_drained": 60, "harvester_count": 2} ) d = observation_to_dict(obs) assert d["economy"]["cash"] == 5000 assert d["economy"]["power_provided"] == 100 assert d["economy"]["power_drained"] == 60 assert d["economy"]["harvester_count"] == 2 def test_military(self): obs = self._make_observation( military={"units_killed": 3, "units_lost": 1, "army_value": 5000} ) d = observation_to_dict(obs) assert d["military"]["units_killed"] == 3 assert d["military"]["units_lost"] == 1 assert d["military"]["army_value"] == 5000 def test_units(self): obs = self._make_observation( units=[ {"actor_id": 1, "type": "e1", "pos_x": 100, "pos_y": 200, "hp_percent": 0.75, "can_attack": True}, {"actor_id": 2, "type": "1tnk", "is_idle": False, "current_activity": "Move"}, ] ) d = observation_to_dict(obs) assert len(d["units"]) == 2 assert d["units"][0]["actor_id"] == 1 assert d["units"][0]["type"] == "e1" assert d["units"][0]["hp_percent"] == pytest.approx(0.75) assert d["units"][0]["can_attack"] is True assert d["units"][1]["is_idle"] is False assert d["units"][1]["current_activity"] == "Move" def test_buildings(self): obs = self._make_observation( buildings=[ {"actor_id": 10, "type": "powr", "is_powered": True}, {"actor_id": 20, "type": "barr", "is_producing": True, "producing_item": "e1"}, ] ) d = observation_to_dict(obs) assert len(d["buildings"]) == 2 assert d["buildings"][0]["type"] == "powr" assert d["buildings"][1]["is_producing"] is True assert d["buildings"][1]["producing_item"] == "e1" def test_production(self): obs = self._make_observation( production=[{"queue_type": "Infantry", "item": "e1", "progress": 0.5, "remaining_ticks": 100}] ) d = observation_to_dict(obs) assert len(d["production"]) == 1 assert d["production"][0]["queue_type"] == "Infantry" assert d["production"][0]["progress"] == pytest.approx(0.5) def test_visible_enemies(self): obs = self._make_observation() enemy = obs.visible_enemies.add() enemy.actor_id = 99 enemy.type = "2tnk" enemy.owner = "Enemy" d = observation_to_dict(obs) assert len(d["visible_enemies"]) == 1 assert d["visible_enemies"][0]["actor_id"] == 99 assert d["visible_enemies"][0]["owner"] == "Enemy" def test_map_info(self): obs = self._make_observation(map_info={"width": 128, "height": 128, "map_name": "Test Map"}) d = observation_to_dict(obs) assert d["map_info"]["width"] == 128 assert d["map_info"]["height"] == 128 assert d["map_info"]["map_name"] == "Test Map" def test_available_production(self): obs = self._make_observation(available_production=["e1", "e3", "1tnk"]) d = observation_to_dict(obs) assert d["available_production"] == ["e1", "e3", "1tnk"] def test_empty_observation(self): obs = self._make_observation() d = observation_to_dict(obs) assert d["tick"] == 0 assert d["units"] == [] assert d["buildings"] == [] assert d["production"] == [] assert d["visible_enemies"] == [] assert d["done"] is False assert d["result"] == ""