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

#1947: added test for agent actions + clearing up the implementation of the request managers for filesystem

Browse Source
This commit is contained in:
Czar.Echavez
2023-10-20 12:58:58 +01:00
parent 0edb9b46a7
commit ffc4711afb
6 changed files with 266 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

118
src/primaite/simulator/file_system/file_system.py
View File

@@ -95,6 +95,18 @@ class FileSystemItemABC(SimComponent):
state["previous_hash"] = self.previous_hash
return state
def _init_request_manager(self) -> RequestManager:
rm = super()._init_request_manager()
rm.add_request(name="scan", request_type=RequestType(func=lambda request, context: self.scan()))
rm.add_request(name="checkhash", request_type=RequestType(func=lambda request, context: self.check_hash()))
rm.add_request(name="repair", request_type=RequestType(func=lambda request, context: self.repair()))
rm.add_request(name="restore", request_type=RequestType(func=lambda request, context: self.restore()))
rm.add_request(name="corrupt", request_type=RequestType(func=lambda request, context: self.corrupt()))
return rm
@property
def size_str(self) -> str:
"""
@@ -181,110 +193,14 @@ class FileSystem(SimComponent):
def _init_request_manager(self) -> RequestManager:
rm = super()._init_request_manager()
self._folder_request_manager = self._init_folder_request_manager()
self._folder_request_manager = RequestManager()
rm.add_request("folder", RequestType(func=self._folder_request_manager))
self._file_request_manager = self._init_file_request_manager()
self._file_request_manager = RequestManager()
rm.add_request("file", RequestType(func=self._file_request_manager))
return rm
def _init_folder_request_manager(self) -> RequestManager:
rm = RequestManager()
rm.add_request(
"scan",
RequestType(
func=lambda request, context: self.get_folder_by_id(folder_uuid=request[0], show_deleted=True).scan()
),
)
rm.add_request(
"checkhash",
RequestType(func=lambda request, context: self.get_folder_by_id(folder_uuid=request[0]).check_hash()),
)
rm.add_request(
"repair", RequestType(func=lambda request, context: self.get_folder_by_id(folder_uuid=request[0]).repair())
)
rm.add_request(
"corrupt",
RequestType(func=lambda request, context: self.get_folder_by_id(folder_uuid=request[0]).corrupt()),
)
rm.add_request(
"delete", RequestType(func=lambda request, context: self.get_folder_by_id(folder_uuid=request[0]).delete())
)
rm.add_request(
"restore",
RequestType(
func=lambda request, context: self.get_folder_by_id(folder_uuid=request[0], show_deleted=True).restore()
),
)
return rm
def _init_file_request_manager(self) -> RequestManager:
rm = RequestManager()
rm.add_request(
"scan",
RequestType(
func=lambda request, context: self.get_folder_by_id(folder_uuid=request[0], show_deleted=True)
.get_file_by_id(file_uuid=request[1], show_deleted=True)
.scan()
),
)
rm.add_request(
"checkhash",
RequestType(
func=lambda request, context: self.get_folder_by_id(folder_uuid=request[0])
.get_file_by_id(file_uuid=request[1])
.check_hash()
),
)
rm.add_request(
"repair",
RequestType(
func=lambda request, context: self.get_folder_by_id(folder_uuid=request[0])
.get_file_by_id(file_uuid=request[1])
.repair()
),
)
rm.add_request(
"corrupt",
RequestType(
func=lambda request, context: self.get_folder_by_id(folder_uuid=request[0])
.get_file_by_id(file_uuid=request[1])
.corrupt()
),
)
rm.add_request(
"delete",
RequestType(
func=lambda request, context: self.get_folder_by_id(folder_uuid=request[0])
.get_file_by_id(file_uuid=request[1])
.delete()
),
)
rm.add_request(
"restore",
RequestType(
func=lambda request, context: self.get_folder_by_id(folder_uuid=request[0], show_deleted=True)
.get_file_by_id(file_uuid=request[1], show_deleted=True)
.restore()
),
)
return rm
@property
def size(self) -> int:
"""
@@ -345,7 +261,9 @@ class FileSystem(SimComponent):
self.folders[folder.uuid] = folder
self._folders_by_name[folder.name] = folder
self.sys_log.info(f"Created folder /{folder.name}")
self._folder_request_manager.add_request(folder.uuid, RequestType(func=folder._request_manager))
self._folder_request_manager.add_request(
name=folder.uuid, request_type=RequestType(func=folder._request_manager)
)
return folder
def delete_folder(self, folder_name: str):
@@ -413,7 +331,7 @@ class FileSystem(SimComponent):
)
folder.add_file(file)
self.sys_log.info(f"Created file /{file.path}")
self._file_request_manager.add_request(file.uuid, RequestType(func=file._request_manager))
self._file_request_manager.add_request(name=file.uuid, request_type=RequestType(func=file._request_manager))
return file
def get_file(self, folder_name: str, file_name: str) -> Optional[File]:

18
tests/conftest.py
View File

@@ -4,7 +4,7 @@ import shutil
import tempfile
from datetime import datetime
from pathlib import Path
from typing import Union
from typing import Any, Union
from unittest.mock import patch
import pytest
@@ -14,7 +14,9 @@ from primaite.environment.primaite_env import Primaite
from primaite.primaite_session import PrimaiteSession
from primaite.simulator.network.container import Network
from primaite.simulator.network.networks import arcd_uc2_network
from primaite.simulator.network.transmission.transport_layer import Port
from primaite.simulator.system.core.sys_log import SysLog
from primaite.simulator.system.services.service import Service
from tests.mock_and_patch.get_session_path_mock import get_temp_session_path
ACTION_SPACE_NODE_VALUES = 1
@@ -27,11 +29,25 @@ from primaite.simulator.file_system.file_system import FileSystem
from primaite.simulator.network.hardware.base import Node
class TestService(Service):
"""Test Service class"""
def receive(self, payload: Any, session_id: str, **kwargs) -> bool:
pass
@pytest.fixture(scope="function")
def uc2_network() -> Network:
return arcd_uc2_network()
@pytest.fixture(scope="function")
def service(file_system) -> TestService:
return TestService(
name="TestService", port=Port.ARP, file_system=file_system, sys_log=SysLog(hostname="test_service")
)
@pytest.fixture(scope="function")
def file_system() -> FileSystem:
return Node(hostname="fs_node").file_system

128
tests/unit_tests/_primaite/_simulator/_file_system/test_file_system_actions.py Normal file
View File

@@ -0,0 +1,128 @@
from typing import Tuple
import pytest
from primaite.simulator.file_system.file_system import File, FileSystem, FileSystemItemStatus, Folder
@pytest.fixture(scope="function")
def populated_file_system(file_system) -> Tuple[FileSystem, Folder, File]:
"""Test that an agent can request a file scan."""
folder = file_system.create_folder(folder_name="test_folder")
file = file_system.create_file(folder_name="test_folder", file_name="test_file.txt")
return file_system, folder, file
def test_file_scan_request(populated_file_system):
"""Test that an agent can request a file scan."""
fs, folder, file = populated_file_system
file.corrupt()
assert file.status == FileSystemItemStatus.CORRUPTED
assert file.visible_status == FileSystemItemStatus.GOOD
fs.apply_request(request=["file", file.uuid, "scan"])
assert file.status == FileSystemItemStatus.CORRUPTED
assert file.visible_status == FileSystemItemStatus.CORRUPTED
def test_folder_scan_request(populated_file_system):
"""Test that an agent can request a folder scan."""
fs, folder, file = populated_file_system
folder.corrupt()
assert folder.status == FileSystemItemStatus.CORRUPTED
assert file.status == FileSystemItemStatus.CORRUPTED
assert folder.visible_status == FileSystemItemStatus.GOOD
assert file.visible_status == FileSystemItemStatus.GOOD
fs.apply_request(request=["folder", folder.uuid, "scan"])
assert folder.status == FileSystemItemStatus.CORRUPTED
assert file.status == FileSystemItemStatus.CORRUPTED
assert folder.visible_status == FileSystemItemStatus.CORRUPTED
assert file.visible_status == FileSystemItemStatus.CORRUPTED
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"])
assert file.status == FileSystemItemStatus.GOOD
file.sim_size = 0
fs.apply_request(request=["file", file.uuid, "checkhash"])
assert file.status == FileSystemItemStatus.CORRUPTED
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"])
assert folder.status == FileSystemItemStatus.GOOD
file.sim_size = 0
fs.apply_request(request=["folder", folder.uuid, "checkhash"])
assert folder.status == FileSystemItemStatus.CORRUPTED
def test_file_repair_request(populated_file_system):
"""Test that an agent can request a file repair."""
fs, folder, file = populated_file_system
file.corrupt()
assert file.status == FileSystemItemStatus.CORRUPTED
fs.apply_request(request=["file", file.uuid, "repair"])
assert file.status == FileSystemItemStatus.GOOD
def test_folder_repair_request(populated_file_system):
"""Test that an agent can request a folder repair."""
fs, folder, file = populated_file_system
folder.corrupt()
assert file.status == FileSystemItemStatus.CORRUPTED
assert folder.status == FileSystemItemStatus.CORRUPTED
fs.apply_request(request=["folder", folder.uuid, "repair"])
assert file.status == FileSystemItemStatus.GOOD
assert folder.status == FileSystemItemStatus.GOOD
def test_file_restore_request(populated_file_system):
pass
def test_folder_restore_request(populated_file_system):
pass
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"])
assert file.status == FileSystemItemStatus.CORRUPTED
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"])
assert file.status == FileSystemItemStatus.CORRUPTED
assert folder.status == FileSystemItemStatus.CORRUPTED
def test_file_delete_request(populated_file_system):
pass
def test_folder_delete_request(populated_file_system):
pass

23
tests/unit_tests/_primaite/_simulator/_network/_hardware/test_node_actions.py Normal file
View File

@@ -0,0 +1,23 @@
import pytest
from primaite.simulator.network.hardware.base import Node, NodeOperatingState
@pytest.fixture
def node() -> Node:
return Node(hostname="test")
def test_node_startup(node):
assert node.operating_state == NodeOperatingState.OFF
node.apply_request(["startup"])
assert node.operating_state == NodeOperatingState.ON
def test_node_shutdown(node):
assert node.operating_state == NodeOperatingState.OFF
node.apply_request(["startup"])
assert node.operating_state == NodeOperatingState.ON
node.apply_request(["shutdown"])
assert node.operating_state == NodeOperatingState.OFF

79
tests/unit_tests/_primaite/_simulator/_system/_services/test_service_actions.py Normal file
View File

@@ -0,0 +1,79 @@
from primaite.simulator.system.services.service import ServiceOperatingState
def test_service_scan(service):
"""Test that an agent can request a service scan."""
service.start()
assert service.operating_state == ServiceOperatingState.RUNNING
assert service.visible_operating_state == ServiceOperatingState.STOPPED
service.apply_request(["scan"])
assert service.operating_state == ServiceOperatingState.RUNNING
assert service.visible_operating_state == ServiceOperatingState.RUNNING
def test_service_stop(service):
"""Test that an agent can request to stop a service."""
service.start()
assert service.operating_state == ServiceOperatingState.RUNNING
service.apply_request(["stop"])
assert service.operating_state == ServiceOperatingState.STOPPED
def test_service_start(service):
"""Test that an agent can request to start a service."""
assert service.operating_state == ServiceOperatingState.STOPPED
service.apply_request(["start"])
assert service.operating_state == ServiceOperatingState.RUNNING
def test_service_pause(service):
"""Test that an agent can request to pause a service."""
service.start()
assert service.operating_state == ServiceOperatingState.RUNNING
service.apply_request(["pause"])
assert service.operating_state == ServiceOperatingState.PAUSED
def test_service_resume(service):
"""Test that an agent can request to resume a service."""
service.start()
assert service.operating_state == ServiceOperatingState.RUNNING
service.apply_request(["pause"])
assert service.operating_state == ServiceOperatingState.PAUSED
service.apply_request(["resume"])
assert service.operating_state == ServiceOperatingState.RUNNING
def test_service_restart(service):
"""Test that an agent can request to restart a service."""
service.start()
assert service.operating_state == ServiceOperatingState.RUNNING
service.apply_request(["restart"])
assert service.operating_state == ServiceOperatingState.RESTARTING
def test_service_disable(service):
"""Test that an agent can request to disable a service."""
service.start()
assert service.operating_state == ServiceOperatingState.RUNNING
service.apply_request(["disable"])
assert service.operating_state == ServiceOperatingState.DISABLED
def test_service_enable(service):
"""Test that an agent can request to enable a service."""
service.start()
assert service.operating_state == ServiceOperatingState.RUNNING
service.apply_request(["disable"])
assert service.operating_state == ServiceOperatingState.DISABLED
service.apply_request(["enable"])
assert service.operating_state == ServiceOperatingState.STOPPED

22
tests/unit_tests/_primaite/_simulator/_system/_services/test_services.py
View File

@@ -1,24 +1,4 @@
from typing import Any
import pytest
from primaite.simulator.network.transmission.transport_layer import Port
from primaite.simulator.system.core.sys_log import SysLog
from primaite.simulator.system.services.service import Service, ServiceOperatingState
class TestService(Service):
"""Test Service class"""
def receive(self, payload: Any, session_id: str, **kwargs) -> bool:
pass
@pytest.fixture(scope="function")
def service(file_system) -> TestService:
return TestService(
name="TestService", port=Port.ARP, file_system=file_system, sys_log=SysLog(hostname="test_service")
)
from primaite.simulator.system.services.service import ServiceOperatingState
def test_scan(service):