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

#2248 - synced wth dev

Browse Source
This commit is contained in:
Chris McCarthy
2024-02-08 15:27:02 +00:00
parent 411f0a320f cfc254753f
commit 114fb8c077
35 changed files with 1688 additions and 1035 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
tests/unit_tests/_primaite/_game/__init__.py Normal file
View File

0
tests/unit_tests/_primaite/_game/_agent/__init__.py Normal file
View File

90
tests/unit_tests/_primaite/_game/_agent/test_actions.py Normal file
View File

@@ -0,0 +1,90 @@
from unittest.mock import Mock
import pytest
from primaite.game.agent.actions import (
ActionManager,
DoNothingAction,
NodeServiceDisableAction,
NodeServiceEnableAction,
NodeServicePauseAction,
NodeServiceRestartAction,
NodeServiceResumeAction,
NodeServiceScanAction,
NodeServiceStartAction,
NodeServiceStopAction,
)
def test_do_nothing_action_form_request():
"""Test that the DoNothingAction can form a request and that it is correct."""
manager = Mock()
action = DoNothingAction(manager=manager)
request = action.form_request()
assert request == ["do_nothing"]
@pytest.mark.parametrize(
"action_class, action_verb",
[
(NodeServiceScanAction, "scan"),
(NodeServiceStopAction, "stop"),
(NodeServiceStartAction, "start"),
(NodeServicePauseAction, "pause"),
(NodeServiceResumeAction, "resume"),
(NodeServiceRestartAction, "restart"),
(NodeServiceDisableAction, "disable"),
(NodeServiceEnableAction, "enable"),
],
) # flake8: noqa
@pytest.mark.parametrize(
"node_name, service_name, expect_to_do_nothing",
[
("pc_1", "chrome", False),
(None, "chrome", True),
("pc_1", None, True),
(None, None, True),
],
) # flake8: noqa
def test_service_action_form_request(node_name, service_name, expect_to_do_nothing, action_class, action_verb):
"""Test that the ServiceScanAction can form a request and that it is correct."""
manager: ActionManager = Mock()
manager.get_node_name_by_idx.return_value = node_name
manager.get_service_name_by_idx.return_value = service_name
action = action_class(manager=manager, num_nodes=1, num_services=1)
request = action.form_request(node_id=0, service_id=0)
if expect_to_do_nothing:
assert request == ["do_nothing"]
else:
assert request == ["network", "node", node_name, "service", service_name, action_verb]
@pytest.mark.parametrize(
"node_name, service_name, expect_to_do_nothing",
[
("pc_1", "chrome", False),
(None, "chrome", True),
("pc_1", None, True),
(None, None, True),
],
) # flake8: noqa
def test_service_scan_form_request(node_name, service_name, expect_to_do_nothing):
"""Test that the ServiceScanAction can form a request and that it is correct."""
manager: ActionManager = Mock()
manager.get_node_name_by_idx.return_value = node_name
manager.get_service_name_by_idx.return_value = service_name
action = NodeServiceScanAction(manager=manager, num_nodes=1, num_services=1)
request = action.form_request(node_id=0, service_id=0)
if expect_to_do_nothing:
assert request == ["do_nothing"]
else:
assert request == ["network", "node", node_name, "service", service_name, "scan"]

18
tests/unit_tests/_primaite/_simulator/_file_system/test_file_actions.py
View File

@@ -25,7 +25,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.uuid, "scan"])
fs.apply_request(request=["file", file.name, "scan"])
assert file.health_status == FileSystemItemHealthStatus.CORRUPT
assert file.visible_health_status == FileSystemItemHealthStatus.CORRUPT
@@ -35,12 +35,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.uuid, "checkhash"])
fs.apply_request(request=["file", file.name, "checkhash"])
assert file.health_status == FileSystemItemHealthStatus.GOOD
file.sim_size = 0
fs.apply_request(request=["file", file.uuid, "checkhash"])
fs.apply_request(request=["file", file.name, "checkhash"])
assert file.health_status == FileSystemItemHealthStatus.CORRUPT
@@ -52,7 +52,7 @@ def test_file_repair_request(populated_file_system):
file.corrupt()
assert file.health_status == FileSystemItemHealthStatus.CORRUPT
fs.apply_request(request=["file", file.uuid, "repair"])
fs.apply_request(request=["file", file.name, "repair"])
assert file.health_status == FileSystemItemHealthStatus.GOOD
@@ -69,7 +69,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.uuid, "corrupt"])
fs.apply_request(request=["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.uuid, file.uuid])
@@ -79,7 +79,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.uuid, "corrupt"])
fs.apply_request(request=["file", file.name, "corrupt"])
assert file.health_status == FileSystemItemHealthStatus.CORRUPT
@@ -88,7 +88,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.uuid, "corrupt"])
fs.apply_request(request=["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
@@ -98,8 +98,8 @@ def test_deleted_file_cannot_be_interacted_with(populated_file_system):
fs.apply_request(request=["delete", "file", folder.uuid, file.uuid])
assert fs.get_file(folder_name=folder.name, file_name=file.name) is None
fs.apply_request(request=["file", file.uuid, "repair"])
fs.apply_request(request=["file", file.uuid, "scan"])
fs.apply_request(request=["file", file.name, "repair"])
fs.apply_request(request=["file", file.name, "scan"])
file = folder.deleted_files.get(file.uuid)

16
tests/unit_tests/_primaite/_simulator/_file_system/test_folder_actions.py
View File

@@ -31,7 +31,7 @@ def test_folder_scan_request(populated_file_system):
assert file1.visible_health_status == FileSystemItemHealthStatus.GOOD
assert file2.visible_health_status == FileSystemItemHealthStatus.GOOD
fs.apply_request(request=["folder", folder.uuid, "scan"])
fs.apply_request(request=["folder", folder.name, "scan"])
folder.apply_timestep(timestep=0)
@@ -53,12 +53,12 @@ def test_folder_checkhash_request(populated_file_system):
"""Test that an agent can request a folder hash check."""
fs, folder, file = populated_file_system
fs.apply_request(request=["folder", folder.uuid, "checkhash"])
fs.apply_request(request=["folder", folder.name, "checkhash"])
assert folder.health_status == FileSystemItemHealthStatus.GOOD
file.sim_size = 0
fs.apply_request(request=["folder", folder.uuid, "checkhash"])
fs.apply_request(request=["folder", folder.name, "checkhash"])
assert folder.health_status == FileSystemItemHealthStatus.CORRUPT
@@ -70,7 +70,7 @@ def test_folder_repair_request(populated_file_system):
assert file.health_status == FileSystemItemHealthStatus.CORRUPT
assert folder.health_status == FileSystemItemHealthStatus.CORRUPT
fs.apply_request(request=["folder", folder.uuid, "repair"])
fs.apply_request(request=["folder", folder.name, "repair"])
assert file.health_status == FileSystemItemHealthStatus.GOOD
assert folder.health_status == FileSystemItemHealthStatus.GOOD
@@ -116,7 +116,7 @@ def test_folder_restore_request(populated_file_system):
assert fs.get_file_by_id(folder_uuid=folder.uuid, file_uuid=file.uuid, include_deleted=True).deleted is False
# corrupt folder
fs.apply_request(request=["folder", folder.uuid, "corrupt"])
fs.apply_request(request=["folder", folder.name, "corrupt"])
assert fs.get_folder(folder_name=folder.name).health_status == FileSystemItemHealthStatus.CORRUPT
assert fs.get_file(folder_name=folder.name, file_name=file.name).health_status == FileSystemItemHealthStatus.CORRUPT
@@ -143,7 +143,7 @@ def test_folder_restore_request(populated_file_system):
def test_folder_corrupt_request(populated_file_system):
"""Test that an agent can request a folder corruption."""
fs, folder, file = populated_file_system
fs.apply_request(request=["folder", folder.uuid, "corrupt"])
fs.apply_request(request=["folder", folder.name, "corrupt"])
assert file.health_status == FileSystemItemHealthStatus.CORRUPT
assert folder.health_status == FileSystemItemHealthStatus.CORRUPT
@@ -153,13 +153,13 @@ 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.uuid, "corrupt"])
fs.apply_request(request=["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.uuid])
assert fs.get_file(folder_name=folder.name, file_name=file.name) is None
fs.apply_request(request=["file", file.uuid, "repair"])
fs.apply_request(request=["file", file.name, "repair"])
deleted_folder = fs.deleted_folders.get(folder.uuid)
deleted_file = deleted_folder.deleted_files.get(file.uuid)

3
tests/unit_tests/_primaite/_simulator/_network/_hardware/nodes/test_switch.py
View File

@@ -6,7 +6,8 @@ from primaite.simulator.network.hardware.nodes.network.switch import Switch
@pytest.fixture(scope="function")
def switch() -> Switch:
switch: Switch = Switch(hostname="switch_1", num_ports=8, operating_state=NodeOperatingState.ON)
switch: Switch = Switch(hostname="switch_1", num_ports=8, start_up_duration=0)
switch.power_on()
switch.show()
return switch

7
tests/unit_tests/_primaite/_simulator/_system/_applications/_red_applications/test_dos_bot.py
View File

@@ -12,12 +12,9 @@ from primaite.simulator.system.applications.red_applications.dos_bot import DoSA
@pytest.fixture(scope="function")
def dos_bot() -> DoSBot:
computer = Computer(
hostname="compromised_pc",
ip_address="192.168.0.1",
subnet_mask="255.255.255.0",
operating_state=NodeOperatingState.ON,
hostname="compromised_pc", ip_address="192.168.0.1", subnet_mask="255.255.255.0", start_up_duration=0
)
computer.power_on()
computer.software_manager.install(DoSBot)
dos_bot: DoSBot = computer.software_manager.software.get("DoSBot")

3
tests/unit_tests/_primaite/_simulator/_system/_applications/test_web_browser.py
View File

@@ -32,8 +32,9 @@ def test_create_web_client():
ip_address="192.168.1.11",
subnet_mask="255.255.255.0",
default_gateway="192.168.1.1",
operating_state=NodeOperatingState.ON,
start_up_duration=0,
)
computer.power_on()
# Web Browser should be pre-installed in computer
web_browser: WebBrowser = computer.software_manager.software.get("WebBrowser")
assert web_browser.name is "WebBrowser"

3
tests/unit_tests/_primaite/_simulator/_system/_services/test_dns_server.py
View File

@@ -20,8 +20,9 @@ def dns_server() -> Node:
ip_address="192.168.1.10",
subnet_mask="255.255.255.0",
default_gateway="192.168.1.1",
operating_state=NodeOperatingState.ON,
start_up_duration=0,
)
node.power_on()
node.software_manager.install(software_class=DNSServer)
return node

3
tests/unit_tests/_primaite/_simulator/_system/_services/test_ftp_client.py
View File

@@ -20,8 +20,9 @@ def ftp_client() -> Node:
ip_address="192.168.1.11",
subnet_mask="255.255.255.0",
default_gateway="192.168.1.1",
operating_state=NodeOperatingState.ON,
start_up_duration=0,
)
node.power_on()
return node

3
tests/unit_tests/_primaite/_simulator/_system/_services/test_ftp_server.py
View File

@@ -18,8 +18,9 @@ def ftp_server() -> Node:
ip_address="192.168.1.10",
subnet_mask="255.255.255.0",
default_gateway="192.168.1.1",
operating_state=NodeOperatingState.ON,
start_up_duration=0,
)
node.power_on()
node.software_manager.install(software_class=FTPServer)
return node

3
tests/unit_tests/_primaite/_simulator/_system/_services/test_web_server.py
View File

@@ -20,8 +20,9 @@ def web_server() -> Server:
ip_address="192.168.1.10",
subnet_mask="255.255.255.0",
default_gateway="192.168.1.1",
operating_state=NodeOperatingState.ON,
start_up_duration=0,
)
node.power_on()
node.software_manager.install(WebServer)
node.software_manager.software.get("WebServer").start()
return node