oxbrbtx/PrimAITE
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

#2887 - Test changes to correct NodeOperatingState is correct per passed config.

Browse Source
This commit is contained in:
Charlie Crane
2025-01-28 09:37:58 +00:00
parent 0570ab984d
commit e1f2f73db0
4 changed files with 9 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/primaite/simulator/network/hardware/base.py
View File

@@ -1574,6 +1574,7 @@ class Node(SimComponent, ABC):
msg = f"Configuration contains an invalid Node type: {config['type']}"
return ValueError(msg)
obj = cls(config=cls.ConfigSchema(**config))
obj.operating_state = NodeOperatingState.ON if not (p := config.get("operating_state")) else NodeOperatingState[p.upper()]
return obj
def __init_subclass__(cls, identifier: Optional[str] = None, **kwargs: Any) -> None:

2
tests/integration_tests/game_layer/observations/test_link_observations.py
View File

@@ -57,7 +57,7 @@ def test_link_observation():
net = Network()
sim = Simulation(network=net)
switch: Switch = Switch.from_config(
config={"type": "switch", "hostname": "switch", "num_ports": 5, "operating_state": NodeOperatingState.ON}
config={"type": "switch", "hostname": "switch", "num_ports": 5, "operating_state": "ON"}
)
computer_1: Computer = Computer.from_config(
config={

4
tests/integration_tests/game_layer/observations/test_router_observation.py
View File

@@ -17,7 +17,7 @@ def test_router_observation():
"""Test adding/removing acl rules and enabling/disabling ports."""
net = Network()
router = Router.from_config(
config={"type": "router", "hostname": "router", "num_ports": 5, "operating_state": NodeOperatingState.ON}
config={"type": "router", "hostname": "router", "num_ports": 5, "operating_state": "ON"}
)
ports = [PortObservation(where=["NICs", i]) for i in range(1, 6)]
@@ -92,7 +92,7 @@ def test_router_observation():
# connect a switch to the router and check that only the correct port is updated
switch: Switch = Switch.from_config(
config={"type": "switch", "hostname": "switch", "num_ports": 1, "operating_state": NodeOperatingState.ON}
config={"type": "switch", "hostname": "switch", "num_ports": 1, "operating_state": "ON"}
)
link = net.connect(router.network_interface[1], switch.network_interface[1])
assert router.network_interface[1].enabled

10
tests/integration_tests/system/test_database_on_node.py
View File

@@ -20,11 +20,11 @@ from primaite.simulator.system.software import SoftwareHealthState
@pytest.fixture(scope="function")
def peer_to_peer() -> Tuple[Computer, Computer]:
network = Network()
node_a = Computer(hostname="node_a", ip_address="192.168.0.10", subnet_mask="255.255.255.0", start_up_duration=0)
node_a: Computer = Computer.from_config(config={"type":"computer", "hostname":"node_a", "ip_address":"192.168.0.10", "subnet_mask":"255.255.255.0", "start_up_duration":0})
node_a.power_on()
node_a.software_manager.get_open_ports()
node_b = Computer(hostname="node_b", ip_address="192.168.0.11", subnet_mask="255.255.255.0", start_up_duration=0)
node_b: Computer = Computer.from_config(config={"type":"computer", "hostname":"node_b", "ip_address":"192.168.0.11", "subnet_mask":"255.255.255.0", "start_up_duration":0})
node_b.power_on()
network.connect(node_a.network_interface[1], node_b.network_interface[1])
@@ -412,8 +412,8 @@ def test_database_service_can_terminate_connection(peer_to_peer):
def test_client_connection_terminate_does_not_terminate_another_clients_connection():
network = Network()
db_server = Server(
hostname="db_client", ip_address="192.168.0.11", subnet_mask="255.255.255.0", start_up_duration=0
db_server: Server = Server.from_config(config={"type":"server",
"hostname":"db_client", "ip_address":"192.168.0.11", "subnet_mask":"255.255.255.0", "start_up_duration":0}
)
db_server.power_on()
@@ -465,6 +465,6 @@ def test_client_connection_terminate_does_not_terminate_another_clients_connecti
def test_database_server_install_ftp_client():
server = Server(hostname="db_server", ip_address="192.168.1.2", subnet_mask="255.255.255.0", start_up_duration=0)
server: Server = Server.from_config(config={"type":"server", "hostname":"db_server", "ip_address":"192.168.1.2", "subnet_mask":"255.255.255.0", "start_up_duration":0})
server.software_manager.install(DatabaseService)
assert server.software_manager.software.get("FTPClient")