Merge https://dev.azure.com/ma-dev-uk/PrimAITE/_git/PrimAITE into feature/2716-Red-Agent-Logging

tests/integration_tests/network/test_airspace_capacity_configuration.py

@@ -0,0 +1,106 @@
+# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
+import yaml
+
+from primaite.game.game import PrimaiteGame
+from primaite.simulator.network.airspace import (
+    AirspaceEnvironmentType,
+    AirSpaceFrequency,
+    calculate_total_channel_capacity,
+    ChannelWidth,
+)
+from primaite.simulator.network.hardware.nodes.network.wireless_router import WirelessRouter
+from tests import TEST_ASSETS_ROOT
+
+
+def test_wireless_wan_wifi_5_80_channel_width_urban():
+    config_path = TEST_ASSETS_ROOT / "configs" / "wireless_wan_wifi_5_80_channel_width_urban.yaml"
+
+    with open(config_path, "r") as f:
+        config_dict = yaml.safe_load(f)
+    network = PrimaiteGame.from_config(cfg=config_dict).simulation.network
+
+    airspace = network.airspace
+
+    assert airspace.airspace_environment_type == AirspaceEnvironmentType.URBAN
+
+    router_1: WirelessRouter = network.get_node_by_hostname("router_1")
+    router_2: WirelessRouter = network.get_node_by_hostname("router_2")
+
+    expected_speed = calculate_total_channel_capacity(
+        channel_width=ChannelWidth.WIDTH_80_MHZ,
+        frequency=AirSpaceFrequency.WIFI_5,
+        environment_type=AirspaceEnvironmentType.URBAN,
+    )
+
+    assert router_1.wireless_access_point.speed == expected_speed
+    assert router_2.wireless_access_point.speed == expected_speed
+
+    pc_a = network.get_node_by_hostname("pc_a")
+    pc_b = network.get_node_by_hostname("pc_b")
+
+    assert pc_a.ping(pc_a.default_gateway), "PC A should ping its default gateway successfully."
+    assert pc_b.ping(pc_b.default_gateway), "PC B should ping its default gateway successfully."
+
+    assert pc_a.ping(pc_b.network_interface[1].ip_address), "PC A should ping PC B across routers successfully."
+    assert pc_b.ping(pc_a.network_interface[1].ip_address), "PC B should ping PC A across routers successfully."
+
+
+def test_wireless_wan_wifi_5_80_channel_width_blocked():
+    config_path = TEST_ASSETS_ROOT / "configs" / "wireless_wan_wifi_5_80_channel_width_blocked.yaml"
+
+    with open(config_path, "r") as f:
+        config_dict = yaml.safe_load(f)
+    network = PrimaiteGame.from_config(cfg=config_dict).simulation.network
+
+    airspace = network.airspace
+
+    assert airspace.airspace_environment_type == AirspaceEnvironmentType.BLOCKED
+
+    router_1: WirelessRouter = network.get_node_by_hostname("router_1")
+    router_2: WirelessRouter = network.get_node_by_hostname("router_2")
+
+    expected_speed = calculate_total_channel_capacity(
+        channel_width=ChannelWidth.WIDTH_80_MHZ,
+        frequency=AirSpaceFrequency.WIFI_5,
+        environment_type=AirspaceEnvironmentType.BLOCKED,
+    )
+
+    assert router_1.wireless_access_point.speed == expected_speed
+    assert router_2.wireless_access_point.speed == expected_speed
+
+    pc_a = network.get_node_by_hostname("pc_a")
+    pc_b = network.get_node_by_hostname("pc_b")
+
+    assert pc_a.ping(pc_a.default_gateway), "PC A should ping its default gateway successfully."
+    assert pc_b.ping(pc_b.default_gateway), "PC B should ping its default gateway successfully."
+
+    assert not pc_a.ping(pc_b.network_interface[1].ip_address), "PC A should ping PC B across routers unsuccessfully."
+    assert not pc_b.ping(pc_a.network_interface[1].ip_address), "PC B should ping PC A across routers unsuccessfully."
+
+
+def test_wireless_wan_blocking_and_unblocking_airspace():
+    config_path = TEST_ASSETS_ROOT / "configs" / "wireless_wan_wifi_5_80_channel_width_urban.yaml"
+
+    with open(config_path, "r") as f:
+        config_dict = yaml.safe_load(f)
+    network = PrimaiteGame.from_config(cfg=config_dict).simulation.network
+
+    airspace = network.airspace
+
+    assert airspace.airspace_environment_type == AirspaceEnvironmentType.URBAN
+
+    pc_a = network.get_node_by_hostname("pc_a")
+    pc_b = network.get_node_by_hostname("pc_b")
+
+    assert pc_a.ping(pc_b.network_interface[1].ip_address), "PC A should ping PC B across routers successfully."
+    assert pc_b.ping(pc_a.network_interface[1].ip_address), "PC B should ping PC A across routers successfully."
+
+    airspace.airspace_environment_type = AirspaceEnvironmentType.BLOCKED
+
+    assert not pc_a.ping(pc_b.network_interface[1].ip_address), "PC A should ping PC B across routers unsuccessfully."
+    assert not pc_b.ping(pc_a.network_interface[1].ip_address), "PC B should ping PC A across routers unsuccessfully."
+
+    airspace.airspace_environment_type = AirspaceEnvironmentType.URBAN
+
+    assert pc_a.ping(pc_b.network_interface[1].ip_address), "PC A should ping PC B across routers successfully."
+    assert pc_b.ping(pc_a.network_interface[1].ip_address), "PC B should ping PC A across routers successfully."

tests/integration_tests/network/test_bandwidth_load_checks_before_transmission.py

@@ -0,0 +1,138 @@
+# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
+from primaite.simulator.file_system.file_type import FileType
+from primaite.simulator.network.hardware.nodes.network.router import ACLAction
+from primaite.simulator.system.services.ftp.ftp_client import FTPClient
+from primaite.simulator.system.services.ftp.ftp_server import FTPServer
+from tests.integration_tests.network.test_wireless_router import wireless_wan_network
+from tests.integration_tests.system.test_ftp_client_server import ftp_client_and_ftp_server
+
+
+def test_wireless_link_loading(wireless_wan_network):
+    client, server, router_1, router_2 = wireless_wan_network
+
+    # Configure Router 1 ACLs
+    router_1.acl.add_rule(action=ACLAction.PERMIT, position=1)
+
+    # Configure Router 2 ACLs
+    router_2.acl.add_rule(action=ACLAction.PERMIT, position=1)
+
+    airspace = router_1.airspace
+
+    client.software_manager.install(FTPClient)
+    ftp_client: FTPClient = client.software_manager.software.get("FTPClient")
+    ftp_client.start()
+
+    server.software_manager.install(FTPServer)
+    ftp_server: FTPServer = server.software_manager.software.get("FTPServer")
+    ftp_server.start()
+
+    client.file_system.create_file(file_name="mixtape", size=10 * 10**6, file_type=FileType.MP3, folder_name="music")
+
+    assert ftp_client.send_file(
+        src_file_name="mixtape.mp3",
+        src_folder_name="music",
+        dest_ip_address=server.network_interface[1].ip_address,
+        dest_file_name="mixtape.mp3",
+        dest_folder_name="music",
+    )
+
+    # Reset the physical links between the host nodes and the routers
+    client.network_interface[1]._connected_link.pre_timestep(1)
+    server.network_interface[1]._connected_link.pre_timestep(1)
+
+    assert ftp_client.send_file(
+        src_file_name="mixtape.mp3",
+        src_folder_name="music",
+        dest_ip_address=server.network_interface[1].ip_address,
+        dest_file_name="mixtape1.mp3",
+        dest_folder_name="music",
+    )
+
+    # Reset the physical links between the host nodes and the routers
+    client.network_interface[1]._connected_link.pre_timestep(1)
+    server.network_interface[1]._connected_link.pre_timestep(1)
+
+    assert ftp_client.send_file(
+        src_file_name="mixtape.mp3",
+        src_folder_name="music",
+        dest_ip_address=server.network_interface[1].ip_address,
+        dest_file_name="mixtape2.mp3",
+        dest_folder_name="music",
+    )
+
+    # Reset the physical links between the host nodes and the routers
+    client.network_interface[1]._connected_link.pre_timestep(1)
+    server.network_interface[1]._connected_link.pre_timestep(1)
+
+    assert not ftp_client.send_file(
+        src_file_name="mixtape.mp3",
+        src_folder_name="music",
+        dest_ip_address=server.network_interface[1].ip_address,
+        dest_file_name="mixtape3.mp3",
+        dest_folder_name="music",
+    )
+
+    # Reset the physical links between the host nodes and the routers
+    client.network_interface[1]._connected_link.pre_timestep(1)
+    server.network_interface[1]._connected_link.pre_timestep(1)
+
+    airspace.reset_bandwidth_load()
+
+    assert ftp_client.send_file(
+        src_file_name="mixtape.mp3",
+        src_folder_name="music",
+        dest_ip_address=server.network_interface[1].ip_address,
+        dest_file_name="mixtape3.mp3",
+        dest_folder_name="music",
+    )
+
+
+def test_wired_link_loading(ftp_client_and_ftp_server):
+    ftp_client, computer, ftp_server, server = ftp_client_and_ftp_server
+
+    link = computer.network_interface[1]._connected_link  # noqa
+
+    assert link.is_up
+
+    link.pre_timestep(1)
+
+    computer.file_system.create_file(
+        file_name="mixtape", size=10 * 10**6, file_type=FileType.MP3, folder_name="music"
+    )
+    link_load = link.current_load
+    assert link_load == 0.0
+
+    assert ftp_client.send_file(
+        src_file_name="mixtape.mp3",
+        src_folder_name="music",
+        dest_ip_address=server.network_interface[1].ip_address,
+        dest_file_name="mixtape.mp3",
+        dest_folder_name="music",
+    )
+
+    new_link_load = link.current_load
+    assert new_link_load > link_load
+
+    assert not ftp_client.send_file(
+        src_file_name="mixtape.mp3",
+        src_folder_name="music",
+        dest_ip_address=server.network_interface[1].ip_address,
+        dest_file_name="mixtape1.mp3",
+        dest_folder_name="music",
+    )
+
+    link.pre_timestep(2)
+
+    link_load = link.current_load
+    assert link_load == 0.0
+
+    assert ftp_client.send_file(
+        src_file_name="mixtape.mp3",
+        src_folder_name="music",
+        dest_ip_address=server.network_interface[1].ip_address,
+        dest_file_name="mixtape1.mp3",
+        dest_folder_name="music",
+    )
+
+    new_link_load = link.current_load
+    assert new_link_load > link_load