diff --git a/src/primaite/simulator/core.py b/src/primaite/simulator/core.py index 8d8425ec..70485af5 100644 --- a/src/primaite/simulator/core.py +++ b/src/primaite/simulator/core.py @@ -34,6 +34,20 @@ class RequestPermissionValidator(BaseModel): """Message that is reported when a request is rejected by this validator.""" return "request rejected" + def __add__(self, other: "RequestPermissionValidator") -> "_CombinedValidator": + return _CombinedValidator(validators=[self, other]) + + +class _CombinedValidator(RequestPermissionValidator): + validators: List[RequestPermissionValidator] = [] + + def __call__(self, request, context) -> bool: + return all(x(request, context) for x in self.validators) + + @property + def fail_message(self): + return f"One of the following conditions are not met: {[v.fail_message for v in self.validators]}" + class AllowAllValidator(RequestPermissionValidator): """Always allows the request.""" diff --git a/src/primaite/simulator/file_system/file_system.py b/src/primaite/simulator/file_system/file_system.py index 456b800c..2162915f 100644 --- a/src/primaite/simulator/file_system/file_system.py +++ b/src/primaite/simulator/file_system/file_system.py @@ -6,8 +6,8 @@ from typing import Any, Dict, List, Optional from prettytable import MARKDOWN, PrettyTable -from primaite.interface.request import RequestResponse -from primaite.simulator.core import RequestManager, RequestType, SimComponent +from primaite.interface.request import RequestFormat, RequestResponse +from primaite.simulator.core import RequestManager, RequestPermissionValidator, RequestType, SimComponent from primaite.simulator.file_system.file import File from primaite.simulator.file_system.file_type import FileType from primaite.simulator.file_system.folder import Folder @@ -42,6 +42,10 @@ class FileSystem(SimComponent): More information in user guide and docstring for SimComponent._init_request_manager. """ + self._folder_exists = FileSystem._FolderExistsValidator(file_system=self) + self._folder_not_deleted = FileSystem._FolderNotDeletedValidator(file_system=self) + self._file_exists = FileSystem._FileExistsValidator(file_system=self) + rm = super()._init_request_manager() self._delete_manager = RequestManager() @@ -50,13 +54,15 @@ class FileSystem(SimComponent): request_type=RequestType( func=lambda request, context: RequestResponse.from_bool( self.delete_file(folder_name=request[0], file_name=request[1]) - ) + ), + validator=self._file_exists, ), ) self._delete_manager.add_request( name="folder", request_type=RequestType( - func=lambda request, context: RequestResponse.from_bool(self.delete_folder(folder_name=request[0])) + func=lambda request, context: RequestResponse.from_bool(self.delete_folder(folder_name=request[0])), + validator=self._folder_exists, ), ) rm.add_request( @@ -144,10 +150,13 @@ class FileSystem(SimComponent): ) self._folder_request_manager = RequestManager() - rm.add_request("folder", RequestType(func=self._folder_request_manager)) + rm.add_request( + "folder", + RequestType(func=self._folder_request_manager, validator=self._folder_exists + self._folder_not_deleted), + ) self._file_request_manager = RequestManager() - rm.add_request("file", RequestType(func=self._file_request_manager)) + rm.add_request("file", RequestType(func=self._file_request_manager, validator=self._file_exists)) return rm @@ -626,3 +635,62 @@ class FileSystem(SimComponent): self.sys_log.error(f"Unable to access file that does not exist. (file name: {file_name})") return False + + class _FolderExistsValidator(RequestPermissionValidator): + """ + When requests come in, this validator will only let them through if the Folder exists. + + Actions cannot be performed on a non-existent folder. + """ + + file_system: FileSystem + """Save a reference to the FileSystem instance.""" + + def __call__(self, request: RequestFormat, context: Dict) -> bool: + """Returns True if folder exists.""" + return self.file_system.get_folder(folder_name=request[0]) is not None + + @property + def fail_message(self) -> str: + """Message that is reported when a request is rejected by this validator.""" + return "Cannot perform request on folder because it does not exist." + + class _FolderNotDeletedValidator(RequestPermissionValidator): + """ + When requests come in, this validator will only let them through if the Folder has not been deleted. + + Actions cannot be performed on a deleted folder. + """ + + file_system: FileSystem + """Save a reference to the FileSystem instance.""" + + def __call__(self, request: RequestFormat, context: Dict) -> bool: + """Returns True if folder exists and is not deleted.""" + # get folder + folder = self.file_system.get_folder(folder_name=request[0], include_deleted=True) + return folder is not None and not folder.deleted + + @property + def fail_message(self) -> str: + """Message that is reported when a request is rejected by this validator.""" + return "Cannot perform request on folder because it is deleted." + + class _FileExistsValidator(RequestPermissionValidator): + """ + When requests come in, this validator will only let them through if the File exists. + + Actions cannot be performed on a non-existent file. + """ + + file_system: FileSystem + """Save a reference to the FileSystem instance.""" + + def __call__(self, request: RequestFormat, context: Dict) -> bool: + """Returns True if file exists.""" + return self.file_system.get_file(folder_name=request[0], file_name=request[1]) is not None + + @property + def fail_message(self) -> str: + """Message that is reported when a request is rejected by this validator.""" + return "Cannot perform request on a file that does not exist." diff --git a/src/primaite/simulator/file_system/folder.py b/src/primaite/simulator/file_system/folder.py index dd2a4c70..c98e4492 100644 --- a/src/primaite/simulator/file_system/folder.py +++ b/src/primaite/simulator/file_system/folder.py @@ -6,8 +6,8 @@ from typing import Dict, Optional from prettytable import MARKDOWN, PrettyTable -from primaite.interface.request import RequestResponse -from primaite.simulator.core import RequestManager, RequestType +from primaite.interface.request import RequestFormat, RequestResponse +from primaite.simulator.core import RequestManager, RequestPermissionValidator, RequestType from primaite.simulator.file_system.file import File from primaite.simulator.file_system.file_system_item_abc import FileSystemItemABC, FileSystemItemHealthStatus @@ -55,6 +55,9 @@ class Folder(FileSystemItemABC): More information in user guide and docstring for SimComponent._init_request_manager. """ + self._file_exists = Folder._FileExistsValidator(folder=self) + self._file_not_deleted = Folder._FileNotDeletedValidator(folder=self) + rm = super()._init_request_manager() rm.add_request( name="delete", @@ -65,7 +68,9 @@ class Folder(FileSystemItemABC): self._file_request_manager = RequestManager() rm.add_request( name="file", - request_type=RequestType(func=self._file_request_manager), + request_type=RequestType( + func=self._file_request_manager, validator=self._file_exists + self._file_not_deleted + ), ) return rm @@ -469,3 +474,42 @@ class Folder(FileSystemItemABC): self.deleted = True return True + + class _FileExistsValidator(RequestPermissionValidator): + """ + When requests come in, this validator will only let them through if the File exists. + + Actions cannot be performed on a non-existent file. + """ + + folder: Folder + """Save a reference to the Folder instance.""" + + def __call__(self, request: RequestFormat, context: Dict) -> bool: + """Returns True if file exists.""" + return self.folder.get_file(file_name=request[0]) is not None + + @property + def fail_message(self) -> str: + """Message that is reported when a request is rejected by this validator.""" + return "Cannot perform request on a file that does not exist." + + class _FileNotDeletedValidator(RequestPermissionValidator): + """ + When requests come in, this validator will only let them through if the File is not deleted. + + Actions cannot be performed on a deleted file. + """ + + folder: Folder + """Save a reference to the Folder instance.""" + + def __call__(self, request: RequestFormat, context: Dict) -> bool: + """Returns True if file exists and is not deleted.""" + file = self.folder.get_file(file_name=request[0]) + return file is not None and not file.deleted + + @property + def fail_message(self) -> str: + """Message that is reported when a request is rejected by this validator.""" + return "Cannot perform request on a file that is deleted." diff --git a/tests/integration_tests/game_layer/actions/test_file_request_permission.py b/tests/integration_tests/game_layer/actions/test_file_request_permission.py new file mode 100644 index 00000000..1c143aed --- /dev/null +++ b/tests/integration_tests/game_layer/actions/test_file_request_permission.py @@ -0,0 +1,159 @@ +# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK +import uuid +from typing import Tuple + +import pytest + +from primaite.game.agent.interface import ProxyAgent +from primaite.game.game import PrimaiteGame +from primaite.simulator.file_system.file_system_item_abc import FileSystemItemHealthStatus +from primaite.simulator.network.hardware.nodes.host.computer import Computer + + +@pytest.fixture +def game_and_agent_fixture(game_and_agent): + """Create a game with a simple agent that can be controlled by the tests.""" + game, agent = game_and_agent + + client_1: Computer = game.simulation.network.get_node_by_hostname("client_1") + client_1.start_up_duration = 3 + + return (game, agent) + + +def test_create_file(game_and_agent_fixture: Tuple[PrimaiteGame, ProxyAgent]): + """Test that the validator allows a files to be created.""" + game, agent = game_and_agent_fixture + + client_1 = game.simulation.network.get_node_by_hostname("client_1") + + random_folder = str(uuid.uuid4()) + random_file = str(uuid.uuid4()) + + assert client_1.file_system.get_file(folder_name=random_folder, file_name=random_file) is None + + action = ( + "NODE_FILE_CREATE", + {"node_id": 0, "folder_name": random_folder, "file_name": random_file}, + ) + agent.store_action(action) + game.step() + + assert client_1.file_system.get_file(folder_name=random_folder, file_name=random_file) is not None + + +def test_file_delete_action(game_and_agent_fixture: Tuple[PrimaiteGame, ProxyAgent]): + """Test that the validator allows a file to be deleted.""" + game, agent = game_and_agent_fixture + + client_1 = game.simulation.network.get_node_by_hostname("client_1") + file = client_1.file_system.get_file(folder_name="downloads", file_name="cat.png") + assert file.deleted is False + + action = ( + "NODE_FILE_DELETE", + {"node_id": 0, "folder_id": 0, "file_id": 0}, + ) + agent.store_action(action) + game.step() + + assert file.deleted + + +def test_file_scan_action(game_and_agent_fixture: Tuple[PrimaiteGame, ProxyAgent]): + """Test that the validator allows a file to be scanned.""" + game, agent = game_and_agent_fixture + + client_1 = game.simulation.network.get_node_by_hostname("client_1") + file = client_1.file_system.get_file(folder_name="downloads", file_name="cat.png") + + file.corrupt() + assert file.health_status == FileSystemItemHealthStatus.CORRUPT + assert file.visible_health_status == FileSystemItemHealthStatus.GOOD + + action = ( + "NODE_FILE_SCAN", + {"node_id": 0, "folder_id": 0, "file_id": 0}, + ) + agent.store_action(action) + game.step() + + assert file.health_status == FileSystemItemHealthStatus.CORRUPT + assert file.visible_health_status == FileSystemItemHealthStatus.CORRUPT + + +def test_file_repair_action(game_and_agent_fixture: Tuple[PrimaiteGame, ProxyAgent]): + """Test that the validator allows a folder to be created.""" + game, agent = game_and_agent_fixture + + client_1 = game.simulation.network.get_node_by_hostname("client_1") + file = client_1.file_system.get_file(folder_name="downloads", file_name="cat.png") + + file.corrupt() + assert file.health_status == FileSystemItemHealthStatus.CORRUPT + + action = ( + "NODE_FILE_REPAIR", + {"node_id": 0, "folder_id": 0, "file_id": 0}, + ) + agent.store_action(action) + game.step() + + assert file.health_status == FileSystemItemHealthStatus.GOOD + + +def test_file_restore_action(game_and_agent_fixture: Tuple[PrimaiteGame, ProxyAgent]): + """Test that the validator allows a file to be restored.""" + game, agent = game_and_agent_fixture + + client_1 = game.simulation.network.get_node_by_hostname("client_1") + file = client_1.file_system.get_file(folder_name="downloads", file_name="cat.png") + + file.corrupt() + assert file.health_status == FileSystemItemHealthStatus.CORRUPT + + action = ( + "NODE_FILE_RESTORE", + {"node_id": 0, "folder_id": 0, "file_id": 0}, + ) + agent.store_action(action) + game.step() + + assert file.health_status == FileSystemItemHealthStatus.GOOD + + +def test_file_corrupt_action(game_and_agent_fixture: Tuple[PrimaiteGame, ProxyAgent]): + """Test that the validator allows a file to be corrupted.""" + game, agent = game_and_agent_fixture + + client_1 = game.simulation.network.get_node_by_hostname("client_1") + file = client_1.file_system.get_file(folder_name="downloads", file_name="cat.png") + + assert file.health_status == FileSystemItemHealthStatus.GOOD + + action = ( + "NODE_FILE_CORRUPT", + {"node_id": 0, "folder_id": 0, "file_id": 0}, + ) + agent.store_action(action) + game.step() + + assert file.health_status == FileSystemItemHealthStatus.CORRUPT + + +def test_file_access_action(game_and_agent_fixture: Tuple[PrimaiteGame, ProxyAgent]): + """Test that the validator allows a file to be accessed.""" + game, agent = game_and_agent_fixture + + client_1 = game.simulation.network.get_node_by_hostname("client_1") + file = client_1.file_system.get_file(folder_name="downloads", file_name="cat.png") + assert file.num_access == 0 + + action = ( + "NODE_FILE_ACCESS", + {"node_id": 0, "folder_name": file.folder_name, "file_name": file.name}, + ) + agent.store_action(action) + game.step() + + assert file.num_access == 1 diff --git a/tests/integration_tests/game_layer/actions/test_folder_request_permission.py b/tests/integration_tests/game_layer/actions/test_folder_request_permission.py new file mode 100644 index 00000000..e5e0806a --- /dev/null +++ b/tests/integration_tests/game_layer/actions/test_folder_request_permission.py @@ -0,0 +1,123 @@ +# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK +import uuid +from typing import Tuple + +import pytest + +from primaite.game.agent.interface import ProxyAgent +from primaite.game.game import PrimaiteGame +from primaite.simulator.file_system.file_system_item_abc import FileSystemItemHealthStatus +from primaite.simulator.network.hardware.nodes.host.computer import Computer + + +@pytest.fixture +def game_and_agent_fixture(game_and_agent): + """Create a game with a simple agent that can be controlled by the tests.""" + game, agent = game_and_agent + + client_1: Computer = game.simulation.network.get_node_by_hostname("client_1") + client_1.start_up_duration = 3 + + return (game, agent) + + +def test_create_folder(game_and_agent_fixture: Tuple[PrimaiteGame, ProxyAgent]): + """Test that the validator allows a folder to be created.""" + game, agent = game_and_agent_fixture + + client_1 = game.simulation.network.get_node_by_hostname("client_1") + + random_folder = str(uuid.uuid4()) + + assert client_1.file_system.get_folder(folder_name=random_folder) is None + + action = ( + "NODE_FOLDER_CREATE", + { + "node_id": 0, + "folder_name": random_folder, + }, + ) + agent.store_action(action) + game.step() + + assert client_1.file_system.get_folder(folder_name=random_folder) is not None + + +def test_folder_scan_action(game_and_agent_fixture: Tuple[PrimaiteGame, ProxyAgent]): + """Test to make sure that the validator checks if the folder exists before scanning.""" + game, agent = game_and_agent_fixture + + client_1 = game.simulation.network.get_node_by_hostname("client_1") + + folder = client_1.file_system.get_folder(folder_name="downloads") + assert folder.health_status == FileSystemItemHealthStatus.GOOD + assert folder.visible_health_status == FileSystemItemHealthStatus.GOOD + + folder.corrupt() + + assert folder.health_status == FileSystemItemHealthStatus.CORRUPT + assert folder.visible_health_status == FileSystemItemHealthStatus.GOOD + + action = ( + "NODE_FOLDER_SCAN", + { + "node_id": 0, # client_1, + "folder_id": 0, # downloads + }, + ) + agent.store_action(action) + game.step() + + for i in range(folder.scan_duration + 1): + game.step() + + assert folder.health_status == FileSystemItemHealthStatus.CORRUPT + assert folder.visible_health_status == FileSystemItemHealthStatus.CORRUPT + + +def test_folder_repair_action(game_and_agent_fixture: Tuple[PrimaiteGame, ProxyAgent]): + """Test to make sure that the validator checks if the folder exists before repairing.""" + game, agent = game_and_agent_fixture + + client_1 = game.simulation.network.get_node_by_hostname("client_1") + + folder = client_1.file_system.get_folder(folder_name="downloads") + folder.corrupt() + assert folder.health_status == FileSystemItemHealthStatus.CORRUPT + + action = ( + "NODE_FOLDER_REPAIR", + { + "node_id": 0, # client_1, + "folder_id": 0, # downloads + }, + ) + agent.store_action(action) + game.step() + + assert folder.health_status == FileSystemItemHealthStatus.GOOD + + +def test_folder_restore_action(game_and_agent_fixture: Tuple[PrimaiteGame, ProxyAgent]): + """Test to make sure that the validator checks if the folder exists before restoring.""" + game, agent = game_and_agent_fixture + + client_1 = game.simulation.network.get_node_by_hostname("client_1") + + folder = client_1.file_system.get_folder(folder_name="downloads") + folder.corrupt() + + assert folder.health_status == FileSystemItemHealthStatus.CORRUPT + + action = ( + "NODE_FOLDER_RESTORE", + { + "node_id": 0, # client_1, + "folder_id": 0, # downloads + }, + ) + agent.store_action(action) + game.step() + + assert folder.health_status == FileSystemItemHealthStatus.RESTORING diff --git a/tests/integration_tests/game_layer/actions/test_nic_request_permission.py b/tests/integration_tests/game_layer/actions/test_nic_request_permission.py index 4c1619e7..d796b75e 100644 --- a/tests/integration_tests/game_layer/actions/test_nic_request_permission.py +++ b/tests/integration_tests/game_layer/actions/test_nic_request_permission.py @@ -6,8 +6,6 @@ import pytest from primaite.game.agent.interface import ProxyAgent from primaite.game.game import PrimaiteGame from primaite.simulator.network.hardware.nodes.host.computer import Computer -from primaite.simulator.network.hardware.nodes.host.server import Server -from primaite.simulator.system.services.service import ServiceOperatingState @pytest.fixture diff --git a/tests/integration_tests/game_layer/observations/test_nic_observations.py b/tests/integration_tests/game_layer/observations/test_nic_observations.py index f1a8ea92..88dd2bd5 100644 --- a/tests/integration_tests/game_layer/observations/test_nic_observations.py +++ b/tests/integration_tests/game_layer/observations/test_nic_observations.py @@ -155,7 +155,7 @@ def test_nic_monitored_traffic(simulation): assert traffic_obs["icmp"]["outbound"] == 0 # send a ping - pc.ping(target_ip_address=pc2.network_interface[1].ip_address) + assert pc.ping(target_ip_address=pc2.network_interface[1].ip_address) traffic_obs = nic_obs.observe(simulation.describe_state()).get("TRAFFIC") assert traffic_obs["icmp"]["inbound"] == 1 @@ -178,7 +178,7 @@ def test_nic_monitored_traffic(simulation): traffic_obs = nic_obs.observe(simulation.describe_state()).get("TRAFFIC") assert traffic_obs["icmp"]["inbound"] == 0 assert traffic_obs["icmp"]["outbound"] == 0 - assert traffic_obs["tcp"][53]["inbound"] == 0 + assert traffic_obs["tcp"][53]["inbound"] == 1 assert traffic_obs["tcp"][53]["outbound"] == 1 # getting a webpage sent a dns request out simulation.pre_timestep(2) # apply timestep to whole sim diff --git a/tests/integration_tests/system/test_nmap.py b/tests/integration_tests/system/test_nmap.py index bbfa4f43..2b8691cc 100644 --- a/tests/integration_tests/system/test_nmap.py +++ b/tests/integration_tests/system/test_nmap.py @@ -101,6 +101,7 @@ def test_port_scan_full_subnet_all_ports_and_protocols(example_network): actual_result = client_1_nmap.port_scan( target_ip_address=IPv4Network("192.168.10.0/24"), + target_port=[Port.ARP, Port.HTTP, Port.FTP, Port.DNS, Port.NTP], ) expected_result = { diff --git a/tests/unit_tests/_primaite/_simulator/_file_system/test_file_actions.py b/tests/unit_tests/_primaite/_simulator/_file_system/test_file_actions.py index 295bca08..594c7afe 100644 --- a/tests/unit_tests/_primaite/_simulator/_file_system/test_file_actions.py +++ b/tests/unit_tests/_primaite/_simulator/_file_system/test_file_actions.py @@ -26,7 +26,7 @@ def test_file_scan_request(populated_file_system): assert file.health_status == FileSystemItemHealthStatus.CORRUPT assert file.visible_health_status == FileSystemItemHealthStatus.GOOD - fs.apply_request(request=["file", file.name, "scan"]) + fs.apply_request(request=["folder", folder.name, "file", file.name, "scan"]) assert file.health_status == FileSystemItemHealthStatus.CORRUPT assert file.visible_health_status == FileSystemItemHealthStatus.CORRUPT @@ -37,12 +37,12 @@ def test_file_checkhash_request(populated_file_system): """Test that an agent can request a file hash check.""" fs, folder, file = populated_file_system - fs.apply_request(request=["file", file.name, "checkhash"]) + fs.apply_request(request=["folder", folder.name, "file", file.name, "checkhash"]) assert file.health_status == FileSystemItemHealthStatus.GOOD file.sim_size = 0 - fs.apply_request(request=["file", file.name, "checkhash"]) + fs.apply_request(request=["folder", folder.name, "file", file.name, "checkhash"]) assert file.health_status == FileSystemItemHealthStatus.CORRUPT @@ -54,7 +54,7 @@ def test_file_repair_request(populated_file_system): file.corrupt() assert file.health_status == FileSystemItemHealthStatus.CORRUPT - fs.apply_request(request=["file", file.name, "repair"]) + fs.apply_request(request=["folder", folder.name, "file", file.name, "repair"]) assert file.health_status == FileSystemItemHealthStatus.GOOD @@ -71,7 +71,7 @@ def test_file_restore_request(populated_file_system): assert fs.get_file(folder_name=folder.name, file_name=file.name) is not None assert fs.get_file(folder_name=folder.name, file_name=file.name).deleted is False - fs.apply_request(request=["file", file.name, "corrupt"]) + fs.apply_request(request=["folder", folder.name, "file", file.name, "corrupt"]) assert fs.get_file(folder_name=folder.name, file_name=file.name).health_status == FileSystemItemHealthStatus.CORRUPT fs.apply_request(request=["restore", "file", folder.name, file.name]) @@ -81,7 +81,7 @@ def test_file_restore_request(populated_file_system): def test_file_corrupt_request(populated_file_system): """Test that an agent can request a file corruption.""" fs, folder, file = populated_file_system - fs.apply_request(request=["file", file.name, "corrupt"]) + fs.apply_request(request=["folder", folder.name, "file", file.name, "corrupt"]) assert file.health_status == FileSystemItemHealthStatus.CORRUPT @@ -90,7 +90,7 @@ def test_deleted_file_cannot_be_interacted_with(populated_file_system): fs, folder, file = populated_file_system assert fs.get_file(folder_name=folder.name, file_name=file.name) is not None - fs.apply_request(request=["file", file.name, "corrupt"]) + fs.apply_request(request=["folder", folder.name, "file", file.name, "corrupt"]) assert fs.get_file(folder_name=folder.name, file_name=file.name).health_status == FileSystemItemHealthStatus.CORRUPT assert ( fs.get_file(folder_name=folder.name, file_name=file.name).visible_health_status diff --git a/tests/unit_tests/_primaite/_simulator/_file_system/test_file_system_actions.py b/tests/unit_tests/_primaite/_simulator/_file_system/test_file_system_actions.py index 209668c4..7d022ea4 100644 --- a/tests/unit_tests/_primaite/_simulator/_file_system/test_file_system_actions.py +++ b/tests/unit_tests/_primaite/_simulator/_file_system/test_file_system_actions.py @@ -39,3 +39,39 @@ def test_folder_delete_request(populated_file_system): assert fs.get_file_by_id(folder_uuid=folder.uuid, file_uuid=file.uuid) is None fs.show(full=True) + + +def test_folder_exists_request_validator(populated_file_system): + """Tests that the _FolderExistsValidator works as intended.""" + fs, folder, file = populated_file_system + validator = FileSystem._FolderExistsValidator(file_system=fs) + + assert validator(request=["test_folder"], context={}) # test_folder exists + assert validator(request=["fake_folder"], context={}) is False # fake_folder does not exist + + assert validator.fail_message == "Cannot perform request on folder because it does not exist." + + +def test_file_exists_request_validator(populated_file_system): + """Tests that the _FolderExistsValidator works as intended.""" + fs, folder, file = populated_file_system + validator = FileSystem._FileExistsValidator(file_system=fs) + + assert validator(request=["test_folder", "test_file.txt"], context={}) # test_file.txt exists + assert validator(request=["test_folder", "fake_file.txt"], context={}) is False # fake_file.txt does not exist + + assert validator.fail_message == "Cannot perform request on a file that does not exist." + + +def test_folder_not_deleted_request_validator(populated_file_system): + """Tests that the _FolderExistsValidator works as intended.""" + fs, folder, file = populated_file_system + validator = FileSystem._FolderNotDeletedValidator(file_system=fs) + + assert validator(request=["test_folder"], context={}) # test_folder is not deleted + + fs.delete_folder(folder_name="test_folder") + + assert validator(request=["test_folder"], context={}) is False # test_folder is deleted + + assert validator.fail_message == "Cannot perform request on folder because it is deleted." diff --git a/tests/unit_tests/_primaite/_simulator/_file_system/test_folder_actions.py b/tests/unit_tests/_primaite/_simulator/_file_system/test_folder_actions.py index 40f1e78b..4a561b97 100644 --- a/tests/unit_tests/_primaite/_simulator/_file_system/test_folder_actions.py +++ b/tests/unit_tests/_primaite/_simulator/_file_system/test_folder_actions.py @@ -166,15 +166,40 @@ def test_deleted_folder_and_its_files_cannot_be_interacted_with(populated_file_s fs, folder, file = populated_file_system assert fs.get_file(folder_name=folder.name, file_name=file.name) is not None - fs.apply_request(request=["file", file.name, "corrupt"]) + fs.apply_request(request=["folder", folder.name, "file", file.name, "corrupt"]) assert fs.get_file(folder_name=folder.name, file_name=file.name).health_status == FileSystemItemHealthStatus.CORRUPT fs.apply_request(request=["delete", "folder", folder.name]) assert fs.get_file(folder_name=folder.name, file_name=file.name) is None - fs.apply_request(request=["file", file.name, "repair"]) + fs.apply_request(request=["folder", folder.name, "file", file.name, "repair"]) deleted_folder = fs.deleted_folders.get(folder.uuid) deleted_file = deleted_folder.deleted_files.get(file.uuid) assert deleted_file.health_status is not FileSystemItemHealthStatus.GOOD + + +def test_file_exists_request_validator(populated_file_system): + """Tests that the _FolderExistsValidator works as intended.""" + fs, folder, file = populated_file_system + validator = Folder._FileExistsValidator(folder=folder) + + assert validator(request=["test_file.txt"], context={}) # test_file.txt exists + assert validator(request=["fake_file.txt"], context={}) is False # fake_file.txt does not exist + + assert validator.fail_message == "Cannot perform request on a file that does not exist." + + +def test_file_not_deleted_request_validator(populated_file_system): + """Tests that the _FolderExistsValidator works as intended.""" + fs, folder, file = populated_file_system + validator = Folder._FileNotDeletedValidator(folder=folder) + + assert validator(request=["test_file.txt"], context={}) # test_file.txt is not deleted + + fs.delete_file(folder_name="test_folder", file_name="test_file.txt") + + assert validator(request=["fake_file.txt"], context={}) is False # test_file.txt is deleted + + assert validator.fail_message == "Cannot perform request on a file that is deleted." diff --git a/tests/unit_tests/_primaite/_simulator/_network/_hardware/test_network_interface_actions.py b/tests/unit_tests/_primaite/_simulator/_network/_hardware/test_network_interface_actions.py new file mode 100644 index 00000000..f35cf171 --- /dev/null +++ b/tests/unit_tests/_primaite/_simulator/_network/_hardware/test_network_interface_actions.py @@ -0,0 +1,34 @@ +# © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK +import pytest + +from primaite.simulator.network.hardware.base import NetworkInterface, Node +from primaite.simulator.network.hardware.nodes.host.computer import Computer + + +@pytest.fixture +def node() -> Node: + return Computer(hostname="test", ip_address="192.168.1.2", subnet_mask="255.255.255.0") + + +def test_nic_enabled_validator(node): + """Test the NetworkInterface enabled validator.""" + network_interface = node.network_interface[1] + validator = NetworkInterface._EnabledValidator(network_interface=network_interface) + + assert validator(request=[], context={}) is False # not enabled + + network_interface.enabled = True + + assert validator(request=[], context={}) # enabled + + +def test_nic_disabled_validator(node): + """Test the NetworkInterface enabled validator.""" + network_interface = node.network_interface[1] + validator = NetworkInterface._DisabledValidator(network_interface=network_interface) + + assert validator(request=[], context={}) # not enabled + + network_interface.enabled = True + + assert validator(request=[], context={}) is False # enabled diff --git a/tests/unit_tests/_primaite/_simulator/_network/_hardware/test_node_actions.py b/tests/unit_tests/_primaite/_simulator/_network/_hardware/test_node_actions.py index 57d6cecb..9b37ac80 100644 --- a/tests/unit_tests/_primaite/_simulator/_network/_hardware/test_node_actions.py +++ b/tests/unit_tests/_primaite/_simulator/_network/_hardware/test_node_actions.py @@ -155,3 +155,39 @@ def test_reset_node(node): assert node.operating_state == NodeOperatingState.BOOTING assert node.operating_state == NodeOperatingState.ON + + +def test_node_is_on_validator(node): + """Test that the node is on validator.""" + node.power_on() + + for i in range(node.start_up_duration + 1): + node.apply_timestep(i) + + validator = Node._NodeIsOnValidator(node=node) + + assert validator(request=[], context={}) + + node.power_off() + for i in range(node.shut_down_duration + 1): + node.apply_timestep(i) + + assert validator(request=[], context={}) is False + + +def test_node_is_off_validator(node): + """Test that the node is on validator.""" + node.power_on() + + for i in range(node.start_up_duration + 1): + node.apply_timestep(i) + + validator = Node._NodeIsOffValidator(node=node) + + assert validator(request=[], context={}) is False + + node.power_off() + for i in range(node.shut_down_duration + 1): + node.apply_timestep(i) + + assert validator(request=[], context={}) diff --git a/tests/unit_tests/_primaite/_simulator/_system/_applications/test_application_actions.py b/tests/unit_tests/_primaite/_simulator/_system/_applications/test_application_actions.py index be6c00e7..0e9c536c 100644 --- a/tests/unit_tests/_primaite/_simulator/_system/_applications/test_application_actions.py +++ b/tests/unit_tests/_primaite/_simulator/_system/_applications/test_application_actions.py @@ -1 +1,15 @@ # © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK +from primaite.simulator.system.applications.application import Application, ApplicationOperatingState + + +def test_application_state_validator(application): + """Test the application state validator.""" + validator = Application._StateValidator(application=application, state=ApplicationOperatingState.CLOSED) + assert validator(request=[], context={}) # application is closed + application.run() + assert validator(request=[], context={}) is False # application is running - expecting closed + + validator = Application._StateValidator(application=application, state=ApplicationOperatingState.RUNNING) + assert validator(request=[], context={}) # application is running + application.close() + assert validator(request=[], context={}) is False # application is closed - expecting running diff --git a/tests/unit_tests/_primaite/_simulator/_system/_services/test_service_actions.py b/tests/unit_tests/_primaite/_simulator/_system/_services/test_service_actions.py index 2d9a6c52..537beb8b 100644 --- a/tests/unit_tests/_primaite/_simulator/_system/_services/test_service_actions.py +++ b/tests/unit_tests/_primaite/_simulator/_system/_services/test_service_actions.py @@ -1,5 +1,5 @@ # © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK -from primaite.simulator.system.services.service import ServiceOperatingState +from primaite.simulator.system.services.service import Service, ServiceOperatingState from primaite.simulator.system.software import SoftwareHealthState @@ -92,3 +92,21 @@ def test_service_fix(service): assert service.health_state_actual == SoftwareHealthState.FIXING service.apply_timestep(2) assert service.health_state_actual == SoftwareHealthState.GOOD + + +def test_service_state_validator(service): + """Test the service state validator.""" + validator = Service._StateValidator(service=service, state=ServiceOperatingState.STOPPED) + assert validator(request=[], context={}) # service is stopped + service.start() + assert validator(request=[], context={}) is False # service is running - expecting stopped + + validator = Service._StateValidator(service=service, state=ServiceOperatingState.RUNNING) + assert validator(request=[], context={}) # service is running + service.pause() + assert validator(request=[], context={}) is False # service is paused - expecting running + + validator = Service._StateValidator(service=service, state=ServiceOperatingState.PAUSED) + assert validator(request=[], context={}) # service is paused + service.resume() + assert validator(request=[], context={}) is False # service is running - expecting paused