Merged PR 458: Carry over airspace hotfixes from internal
## Summary Carried over hit fixes from internal that backtracked on the complex channel width stuff for now and focussed on getting a stable data rate baked in for each frequency. Implemented overriding of frequency max capacities on the airspace. updated documentation to reflect the changes in airspace.py. ## Test process - Original tests still work - Tested reading the frequency capacity overrides from config file - Tested that setting the frequency override to 0.0 blocks the channel ## Checklist - [X] PR is linked to a **work item** - [X] **acceptance criteria** of linked ticket are met - [X] performed **self-review** of the code - [X] written **tests** for any new functionality added with this PR - [X] updated the **documentation** if this PR changes or adds functionality - [ ] written/updated **design docs** if this PR implements new functionality - [X] updated the **change log** - [X] ran **pre-commit** checks for code style - [X] attended to any **TO-DOs** left in the code Related work items: #2745
This commit is contained in:
Christopher McCarthy
@@ -1,106 +0,0 @@
|
||||
# © 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."
|
||||
44
tests/integration_tests/network/test_airspace_config.py
Normal file
44
tests/integration_tests/network/test_airspace_config.py
Normal file
@@ -0,0 +1,44 @@
|
||||
# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
|
||||
import yaml
|
||||
|
||||
from primaite.game.game import PrimaiteGame
|
||||
from primaite.simulator.network.airspace import AirSpaceFrequency
|
||||
from tests import TEST_ASSETS_ROOT
|
||||
|
||||
|
||||
def test_override_freq_max_capacity_mbps():
|
||||
config_path = TEST_ASSETS_ROOT / "configs" / "wireless_wan_network_config_freq_max_override.yaml"
|
||||
|
||||
with open(config_path, "r") as f:
|
||||
config_dict = yaml.safe_load(f)
|
||||
network = PrimaiteGame.from_config(cfg=config_dict).simulation.network
|
||||
|
||||
assert network.airspace.get_frequency_max_capacity_mbps(AirSpaceFrequency.WIFI_2_4) == 123.45
|
||||
assert network.airspace.get_frequency_max_capacity_mbps(AirSpaceFrequency.WIFI_5) == 0.0
|
||||
|
||||
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 be able to ping PC B"
|
||||
assert pc_b.ping(pc_a.network_interface[1].ip_address), "PC B should be able to ping PC A"
|
||||
|
||||
network.airspace.show()
|
||||
|
||||
|
||||
def test_override_freq_max_capacity_mbps_blocked():
|
||||
config_path = TEST_ASSETS_ROOT / "configs" / "wireless_wan_network_config_freq_max_override_blocked.yaml"
|
||||
|
||||
with open(config_path, "r") as f:
|
||||
config_dict = yaml.safe_load(f)
|
||||
network = PrimaiteGame.from_config(cfg=config_dict).simulation.network
|
||||
|
||||
assert network.airspace.get_frequency_max_capacity_mbps(AirSpaceFrequency.WIFI_2_4) == 0.0
|
||||
assert network.airspace.get_frequency_max_capacity_mbps(AirSpaceFrequency.WIFI_5) == 0.0
|
||||
|
||||
pc_a = network.get_node_by_hostname("pc_a")
|
||||
pc_b = network.get_node_by_hostname("pc_b")
|
||||
|
||||
assert not pc_a.ping(pc_b.network_interface[1].ip_address), "PC A should not be able to ping PC B"
|
||||
assert not pc_b.ping(pc_a.network_interface[1].ip_address), "PC B should not be able to ping PC A"
|
||||
|
||||
network.airspace.show()
|
||||
@@ -40,30 +40,6 @@ def test_wireless_link_loading(wireless_wan_network):
|
||||
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",
|
||||
|
||||
Reference in New Issue
Block a user