#1714: set up files

This commit is contained in:
Czar Echavez
2023-08-01 16:18:49 +01:00
parent 020682f546
commit ea8c65a17e
16 changed files with 223 additions and 4 deletions

View File

@@ -86,5 +86,5 @@ stages:
displayName: 'Perform PrimAITE Setup' displayName: 'Perform PrimAITE Setup'
- script: | - script: |
pytest -n 4 pytest -n auto
displayName: 'Run tests' displayName: 'Run tests'

View File

@@ -6,7 +6,7 @@ from pydantic import BaseModel
class SimComponent(BaseModel): class SimComponent(BaseModel):
"""Extension of pydantic BaseModel with additional methods that must be defined by all classes in the simulator.""" """Extension of pydantic BaseModel with additional methods that must be defined by all classes in the simulator."""
@abstractmethod @abstractmethod
def describe_state(self) -> Dict: def describe_state(self) -> Dict:

View File

@@ -0,0 +1,77 @@
from typing import Dict, List, Union
from primaite.simulator.core import SimComponent
from primaite.simulator.file_system.file_system_file import FileSystemFile
from primaite.simulator.file_system.file_system_folder import FileSystemFolder
class FileSystem(SimComponent):
"""Class that contains all the simulation File System."""
files: List[FileSystemFile]
"""List containing all the files in the file system."""
folders: List[FileSystemFolder]
"""List containing all the folders in the file system."""
def describe_state(self) -> Dict:
"""
Get the current state of the FileSystem as a dict.
:return: A dict containing the current state of the FileSystemFile.
"""
pass
def create_file(self):
"""Creates a FileSystemFile and adds it to the list of files."""
pass
def create_folder(self):
"""Creates a FileSystemFolder and adds it to the list of folders."""
pass
def delete_file(self, file_item: str):
"""
Deletes a file and removes it from the files list.
:param file_item: The UUID of the file item to delete
:type file_item: str
"""
self.files = list(filter(lambda x: (x.get_item_uuid() != file_item), self.files))
def delete_folder(self, file_item: str):
"""
Deletes a folder, removes it frdom the folders list and removes any child folders and files.
:param file_item: The UUID of the file item to delete
:type file_item: str
"""
self.files = list(filter(lambda x: (x.get_item_parent() != file_item), self.files))
self.folders = list(filter(lambda x: (x.get_item_uuid() != file_item), self.folders))
def move_file_item(self, file_item: str, target_directory: str):
"""
Check to see if the file_item and target_directory exists then moves the item by changing its parent item uuid.
:param file_item: The UUID of the file item to move
:type file_item: str
:param target_directory: The UUID of the directory the item should be moved into
:type target_directory: str
"""
item = self._file_item_exists(file_item)
if item and any(f for f in self.folders if f.get_item_uuid() == target_directory):
item.move(target_directory)
def _file_item_exists(self, file_item_uuid: str) -> Union[FileSystemFile, FileSystemFolder, None]:
"""Returns true if the file or folder UUID exists."""
item = next((x for x in self.files if x.get_item_uuid() == file_item_uuid), None)
if item:
return item
next((x for x in self.folders if x.get_item_uuid() == file_item_uuid), None)
if item:
return item
raise Exception(f"No file or folder found with id: {file_item_uuid}")

View File

@@ -0,0 +1,34 @@
from typing import Dict
from primaite.simulator.file_system.file_system_file_type import FileSystemFileType
from primaite.simulator.file_system.file_system_item_abc import FileSystemItemABC
class FileSystemFile(FileSystemItemABC):
"""Class that represents a file in the simulation."""
_file_type: FileSystemFileType
"""The type of the FileSystemFile"""
def get_file_type(self) -> FileSystemFileType:
"""Returns the FileSystemFileType of the file."""
return self._file_type
def move(self, target_directory: str):
"""
Changes the parent_item of the FileSystemFile.
Essentially simulates the file system item being moved from folder to folder
:param target_directory: The UUID of the directory the file system item should be moved to
:type target_directory: str
"""
super().move(target_directory)
def describe_state(self) -> Dict:
"""
Get the current state of the FileSystemFile as a dict.
:return: A dict containing the current state of the FileSystemFile.
"""
pass

View File

@@ -0,0 +1,7 @@
from enum import Enum
class FileSystemFileType(str, Enum):
"""Enum used to determine the FileSystemFile type."""
TBD = "TBD"

View File

@@ -0,0 +1,41 @@
from typing import Dict
from primaite.simulator.file_system.file_system_item_abc import FileSystemItemABC
class FileSystemFolder(FileSystemItemABC):
"""Simulation FileSystemFolder."""
_is_quarantined: bool
"""Flag that marks the folder as quarantined if true."""
def quarantine(self):
"""Quarantines the File System Folder."""
self._is_quarantined = True
def end_quarantine(self):
"""Ends the quarantine of the File System Folder."""
self._is_quarantined = False
def quarantine_status(self) -> bool:
"""Returns true if the folder is being quarantined."""
return self._is_quarantined
def move(self, target_directory: str):
"""
Changes the parent_item of the file system item.
Essentially simulates the file system item being moved from folder to folder
:param target_directory: The UUID of the directory the file system item should be moved to
:type target_directory: str
"""
super().move(target_directory)
def describe_state(self) -> Dict:
"""
Get the current state of the FileSystemFolder as a dict.
:return: A dict containing the current state of the FileSystemFile.
"""
pass

View File

@@ -0,0 +1,60 @@
from abc import ABC, abstractmethod
from uuid import uuid4
from primaite.simulator.core import SimComponent
class FileSystemItemABC(SimComponent, ABC):
"""Abstract Base class for any file system items e.g. files and folders."""
_uuid: str
"""Unique identifier for the FileSystemItem"""
_parent_item: str
"""UUID of the parent FileSystemItem"""
_item_size: float
"""Disk size of the FileSystemItem"""
def __init__(self, parent_item: str, item_size: float):
"""
Abstract base class used by FileSystem items.
:param parent_item: The UUID of the FileSystemItem parent
:type parent_item: str
:param item_size: The size of the FileSystemItem
:type item_size: float
"""
super().__init__()
# generate random uuid for file system item
self._uuid = str(uuid4())
self._parent_item = parent_item
self._item_size = item_size
def get_item_uuid(self) -> str:
"""Returns the file system item UUID."""
return self._uuid
def get_item_parent(self) -> str:
"""Returns the UUID of the item's parent."""
return self._parent_item
def get_item_size(self) -> float:
"""Returns the item size."""
return self._item_size
@abstractmethod
def move(self, target_directory: str):
"""
Changes the parent_item of the file system item.
Essentially simulates the file system item being moved from folder to folder
:param target_directory: The UUID of the directory the file system item should be moved to
:type target_directory: str
"""
self._parent_item = target_directory

View File

@@ -96,7 +96,7 @@ def temp_primaite_session(request):
""" """
training_config_path = request.param[0] training_config_path = request.param[0]
lay_down_config_path = request.param[1] lay_down_config_path = request.param[1]
with patch("primaite.agents.agent_abc.get_session_path", get_temp_session_path) as mck: with patch("_primaite.agents.agent_abc.get_session_path", get_temp_session_path) as mck:
mck.session_timestamp = datetime.now() mck.session_timestamp = datetime.now()
return TempPrimaiteSession(training_config_path, lay_down_config_path) return TempPrimaiteSession(training_config_path, lay_down_config_path)
@@ -112,7 +112,7 @@ def temp_session_path() -> Path:
session_timestamp = datetime.now() session_timestamp = datetime.now()
date_dir = session_timestamp.strftime("%Y-%m-%d") date_dir = session_timestamp.strftime("%Y-%m-%d")
session_path = session_timestamp.strftime("%Y-%m-%d_%H-%M-%S") session_path = session_timestamp.strftime("%Y-%m-%d_%H-%M-%S")
session_path = Path(tempfile.gettempdir()) / "primaite" / date_dir / session_path session_path = Path(tempfile.gettempdir()) / "_primaite" / date_dir / session_path
session_path.mkdir(exist_ok=True, parents=True) session_path.mkdir(exist_ok=True, parents=True)
return session_path return session_path