Merged PR 606: Regression Testing Fixes

## Summary
Changes to PrimAITE to resolve bugs found during regression testing.

- Addressed hardcoding of "abstract_agent" as the name used in logging. This should now use the agent's `ref` from config.
- Fixed some failing terminal tests that were attempting to pass through a remote IP for actions where this wasn't needed.
- Corrected some old uses of `self.hostname` instead of `self.config.hostname` following refactoring.
- Removal of `node_types_map` when calling `network.show` to resolve issues with extensability.

## Test process
Tests pass when ran locally.

## Checklist
- [ ] PR is linked to a **work item**
- [ ] **acceptance criteria** of linked ticket are met
- [ ] performed **self-review** of the code
- [ ] written **tests** for any new functionality added with this PR
- [ ] updated the **documentation** if this PR changes or adds functionality
- [ ] written/updated **design docs** if this PR implements new functionality
- [ ] updated the **change log**
- [ ] ran **pre-commit** checks for code style
- [ ] attended to any **TO-DOs** left in the code
This commit is contained in:
Charlie Crane
2025-02-18 10:33:50 +00:00
committed by Marek Wolan
6 changed files with 39 additions and 48 deletions

View File

@@ -70,7 +70,7 @@ class AbstractAgent(BaseModel, ABC):
config: ConfigSchema = Field(default_factory=lambda: AbstractAgent.ConfigSchema()) config: ConfigSchema = Field(default_factory=lambda: AbstractAgent.ConfigSchema())
logger: AgentLog = AgentLog(agent_name="Abstract_Agent") logger: AgentLog = None
history: List[AgentHistoryItem] = [] history: List[AgentHistoryItem] = []
action_manager: ActionManager = Field(default_factory=lambda: ActionManager()) action_manager: ActionManager = Field(default_factory=lambda: ActionManager())
@@ -79,6 +79,11 @@ class AbstractAgent(BaseModel, ABC):
_registry: ClassVar[Dict[str, Type[AbstractAgent]]] = {} _registry: ClassVar[Dict[str, Type[AbstractAgent]]] = {}
def __init__(self, **kwargs):
"""Initialise and setup agent logger."""
super().__init__(**kwargs)
self.logger: AgentLog = AgentLog(agent_name=kwargs["config"]["ref"])
def __init_subclass__(cls, discriminator: Optional[str] = None, **kwargs: Any) -> None: def __init_subclass__(cls, discriminator: Optional[str] = None, **kwargs: Any) -> None:
super().__init_subclass__(**kwargs) super().__init_subclass__(**kwargs)
if discriminator is None: if discriminator is None:

View File

@@ -163,16 +163,6 @@ class Network(SimComponent):
:param links: Include link details in the output. Defaults to True. :param links: Include link details in the output. Defaults to True.
:param markdown: Use Markdown style in table output. Defaults to False. :param markdown: Use Markdown style in table output. Defaults to False.
""" """
nodes_type_map = {
"Router": self.router_nodes,
"Firewall": self.firewall_nodes,
"Switch": self.switch_nodes,
"Server": self.server_nodes,
"Computer": self.computer_nodes,
"Printer": self.printer_nodes,
"Wireless Router": self.wireless_router_nodes,
}
if nodes: if nodes:
table = PrettyTable(["Node", "Type", "Operating State"]) table = PrettyTable(["Node", "Type", "Operating State"])
if markdown: if markdown:
@@ -189,21 +179,20 @@ class Network(SimComponent):
table.set_style(MARKDOWN) table.set_style(MARKDOWN)
table.align = "l" table.align = "l"
table.title = "IP Addresses" table.title = "IP Addresses"
for nodes in nodes_type_map.values(): for node in self.nodes.values():
for node in nodes: for i, port in node.network_interface.items():
for i, port in node.network_interface.items(): if hasattr(port, "ip_address"):
if hasattr(port, "ip_address"): if port.ip_address != IPv4Address("127.0.0.1"):
if port.ip_address != IPv4Address("127.0.0.1"): port_str = port.port_name if port.port_name else port.port_num
port_str = port.port_name if port.port_name else port.port_num table.add_row(
table.add_row( [
[ node.config.hostname,
node.config.hostname, port_str,
port_str, port.ip_address,
port.ip_address, port.subnet_mask,
port.subnet_mask, node.config.default_gateway,
node.config.default_gateway, ]
] )
)
print(table) print(table)
if links: if links:
@@ -215,22 +204,21 @@ class Network(SimComponent):
table.align = "l" table.align = "l"
table.title = "Links" table.title = "Links"
links = list(self.links.values()) links = list(self.links.values())
for nodes in nodes_type_map.values(): for node in self.nodes.values():
for node in nodes: for link in links[::-1]:
for link in links[::-1]: if node in [link.endpoint_a.parent, link.endpoint_b.parent]:
if node in [link.endpoint_a.parent, link.endpoint_b.parent]: table.add_row(
table.add_row( [
[ link.endpoint_a.parent.config.hostname,
link.endpoint_a.parent.config.hostname, str(link.endpoint_a),
str(link.endpoint_a), link.endpoint_b.parent.config.hostname,
link.endpoint_b.parent.config.hostname, str(link.endpoint_b),
str(link.endpoint_b), link.is_up,
link.is_up, link.bandwidth,
link.bandwidth, link.current_load_percent,
link.current_load_percent, ]
] )
) links.remove(link)
links.remove(link)
print(table) print(table)
def clear_links(self): def clear_links(self):

View File

@@ -1197,7 +1197,7 @@ class UserSessionManager(Service, discriminator="user-session-manager"):
"""Request should take the form [username, password, remote_ip_address].""" """Request should take the form [username, password, remote_ip_address]."""
username, password, remote_ip_address = request username, password, remote_ip_address = request
response = RequestResponse.from_bool(self.remote_login(username, password, remote_ip_address)) response = RequestResponse.from_bool(self.remote_login(username, password, remote_ip_address))
response.data = {"remote_hostname": self.parent.hostname, "username": username} response.data = {"remote_hostname": self.parent.config.hostname, "username": username}
return response return response
rm.add_request("remote_login", RequestType(func=_remote_login)) rm.add_request("remote_login", RequestType(func=_remote_login))
@@ -1230,7 +1230,7 @@ class UserSessionManager(Service, discriminator="user-session-manager"):
if markdown: if markdown:
table.set_style(MARKDOWN) table.set_style(MARKDOWN)
table.align = "l" table.align = "l"
table.title = f"{self.parent.hostname} User Sessions" table.title = f"{self.parent.config.hostname} User Sessions"
def _add_session_to_table(user_session: UserSession): def _add_session_to_table(user_session: UserSession):
""" """

View File

@@ -49,7 +49,7 @@ class GigaSwitch(NetworkNode, discriminator="gigaswitch"):
if markdown: if markdown:
table.set_style(MARKDOWN) table.set_style(MARKDOWN)
table.align = "l" table.align = "l"
table.title = f"{self.hostname} Switch Ports" table.title = f"{self.config.hostname} Switch Ports"
for port_num, port in self.network_interface.items(): for port_num, port in self.network_interface.items():
table.add_row([port_num, port.mac_address, port.speed, "Enabled" if port.enabled else "Disabled"]) table.add_row([port_num, port.mac_address, port.speed, "Enabled" if port.enabled else "Disabled"])
print(table) print(table)

View File

@@ -106,7 +106,6 @@ def test_remote_login_change_password(game_and_agent_fixture: Tuple[PrimaiteGame
"username": "user123", "username": "user123",
"current_password": "password", "current_password": "password",
"new_password": "different_password", "new_password": "different_password",
"remote_ip": str(server_1.network_interface[1].ip_address),
}, },
) )
agent.store_action(action) agent.store_action(action)
@@ -146,7 +145,6 @@ def test_change_password_logs_out_user(game_and_agent_fixture: Tuple[PrimaiteGam
"username": "user123", "username": "user123",
"current_password": "password", "current_password": "password",
"new_password": "different_password", "new_password": "different_password",
"remote_ip": str(server_1.network_interface[1].ip_address),
}, },
) )
agent.store_action(action) agent.store_action(action)

View File

@@ -4,7 +4,7 @@ from primaite.game.agent.scripted_agents.random_agent import RandomAgent
def test_creating_empty_agent(): def test_creating_empty_agent():
agent = RandomAgent() agent = RandomAgent(config={"ref": "Empty Agent"})
assert len(agent.action_manager.action_map) == 0 assert len(agent.action_manager.action_map) == 0
assert isinstance(agent.observation_manager.obs, NullObservation) assert isinstance(agent.observation_manager.obs, NullObservation)
assert len(agent.reward_function.reward_components) == 0 assert len(agent.reward_function.reward_components) == 0