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

#1947: remove storing deleted files in a list and banish them to the shadow realm instead

Browse Source
This commit is contained in:
Czar.Echavez
2023-10-24 11:53:21 +01:00
parent 1e66795aff
commit c4b43c479e
3 changed files with 122 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,6 +1,5 @@
from __future__ import annotations from __future__ import annotations
import copy
import hashlib import hashlib
import json import json
import math import math
@@ -53,14 +52,17 @@ class FileSystemItemHealthStatus(Enum):
GOOD = 0 GOOD = 0
"""File/Folder is OK.""" """File/Folder is OK."""
QUARANTINED = 1 COMPROMISED = 1
"""File/Folder is quarantined.""" """File/Folder is quarantined."""
CORRUPTED = 2 CORRUPT = 2
"""File/Folder is corrupted.""" """File/Folder is corrupted."""
DELETED = 3 RESTORING = 3
"""File/Folder is deleted.""" """File/Folder is in the process of being restored."""
REPAIRING = 3
"""File/Folder is in the process of being repaired."""
class FileSystemItemABC(SimComponent): class FileSystemItemABC(SimComponent):
@@ -128,9 +130,7 @@ class FileSystemItemABC(SimComponent):
Return False if corruption is detected, otherwise True Return False if corruption is detected, otherwise True
""" """
# cannot check hash if deleted pass
if self.health_status == FileSystemItemHealthStatus.DELETED:
return False
@abstractmethod @abstractmethod
def repair(self) -> bool: def repair(self) -> bool:
@@ -139,9 +139,7 @@ class FileSystemItemABC(SimComponent):
True if successfully repaired. False otherwise. True if successfully repaired. False otherwise.
""" """
# cannot repair if deleted pass
if self.health_status == FileSystemItemHealthStatus.DELETED:
return False
@abstractmethod @abstractmethod
def corrupt(self) -> bool: def corrupt(self) -> bool:
@@ -150,17 +148,7 @@ class FileSystemItemABC(SimComponent):
True if successfully corrupted. False otherwise. True if successfully corrupted. False otherwise.
""" """
# cannot corrupt if deleted pass
if self.health_status == FileSystemItemHealthStatus.DELETED:
return False
def delete(self) -> None:
"""
Delete the FileSystemItem.
True if successfully deleted. False otherwise.
"""
self.health_status = FileSystemItemHealthStatus.DELETED
def restore(self) -> None: def restore(self) -> None:
"""Restore the file/folder to the state before it got ruined.""" """Restore the file/folder to the state before it got ruined."""
@@ -176,8 +164,6 @@ class FileSystem(SimComponent):
sys_log: SysLog sys_log: SysLog
sim_root: Path sim_root: Path
deleted_folders: Dict[str, Folder] = {}
def __init__(self, **kwargs): def __init__(self, **kwargs):
super().__init__(**kwargs) super().__init__(**kwargs)
# Ensure a default root folder # Ensure a default root folder
@@ -187,6 +173,11 @@ class FileSystem(SimComponent):
def _init_request_manager(self) -> RequestManager: def _init_request_manager(self) -> RequestManager:
rm = super()._init_request_manager() rm = super()._init_request_manager()
rm.add_request(
name="delete",
request_type=RequestType(func=lambda request, context: self.delete_folder_by_id(folder_uuid=request[0])),
)
self._folder_request_manager = RequestManager() 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))
@@ -271,18 +262,25 @@ class FileSystem(SimComponent):
return return
folder = self._folders_by_name.get(folder_name) folder = self._folders_by_name.get(folder_name)
if folder: if folder:
# add to deleted list # remove from folder list
folder.delete()
self.deleted_folders[folder.uuid] = folder
# remove from normal list
self.folders.pop(folder.uuid) self.folders.pop(folder.uuid)
self._folders_by_name.pop(folder.name) self._folders_by_name.pop(folder.name)
self.sys_log.info(f"Deleted folder /{folder.name} and its contents") self.sys_log.info(f"Deleted folder /{folder.name} and its contents")
self._folder_request_manager.remove_request(folder.uuid) self._folder_request_manager.remove_request(folder.uuid)
folder.remove_all_files()
else: else:
_LOGGER.debug(f"Cannot delete folder as it does not exist: {folder_name}") _LOGGER.debug(f"Cannot delete folder as it does not exist: {folder_name}")
def delete_folder_by_id(self, folder_uuid: str):
"""
Deletes a folder via its uuid.
:param: folder_uuid: UUID of the folder to delete
"""
folder = self.get_folder_by_id(folder_uuid=folder_uuid)
self.delete_folder(folder_name=folder.name)
def restore_folder(self, folder_id: str): def restore_folder(self, folder_id: str):
"""TODO.""" """TODO."""
pass pass
@@ -423,19 +421,13 @@ class FileSystem(SimComponent):
""" """
return self._folders_by_name.get(folder_name) return self._folders_by_name.get(folder_name)
def get_folder_by_id(self, folder_uuid: str, show_deleted: bool = False) -> Optional[Folder]: def get_folder_by_id(self, folder_uuid: str) -> Optional[Folder]:
""" """
Get a folder by its uuid if it exists. Get a folder by its uuid if it exists.
:param: folder_uuid: The folder uuid. :param: folder_uuid: The folder uuid.
:param: show_deleted: show deleted folders
:return: The matching Folder. :return: The matching Folder.
""" """
deleted_folder = self.deleted_folders.get(folder_uuid)
if show_deleted and deleted_folder:
return deleted_folder
return self.folders.get(folder_uuid) return self.folders.get(folder_uuid)
@@ -449,12 +441,17 @@ class Folder(FileSystemItemABC):
_files_by_name: Dict[str, File] = {} _files_by_name: Dict[str, File] = {}
"Files by their name as <file name>.<file type>." "Files by their name as <file name>.<file type>."
deleted_files: Dict[str, File] = {}
"List of files that have been deleted."
scan_duration: int = -1 scan_duration: int = -1
"How many timesteps to complete a scan." "How many timesteps to complete a scan."
def _init_request_manager(self) -> RequestManager:
rm = super()._init_request_manager()
rm.add_request(
name="delete",
request_type=RequestType(func=lambda request, context: self.remove_file_by_id(file_uuid=request[0])),
)
return rm
def describe_state(self) -> Dict: def describe_state(self) -> Dict:
""" """
Produce a dictionary describing the current state of this object. Produce a dictionary describing the current state of this object.
@@ -463,7 +460,6 @@ class Folder(FileSystemItemABC):
""" """
state = super().describe_state() state = super().describe_state()
state["files"] = {file.name: file.describe_state() for uuid, file in self.files.items()} state["files"] = {file.name: file.describe_state() for uuid, file in self.files.items()}
state["deleted_files"] = {file.name: file.describe_state() for uuid, file in self.deleted_files.items()}
return state return state
def show(self, markdown: bool = False): def show(self, markdown: bool = False):
@@ -508,8 +504,8 @@ class Folder(FileSystemItemABC):
# scan one file per timestep # scan one file per timestep
file = self.get_file_by_id(file_uuid=list(self.files)[self.scan_duration - 1]) file = self.get_file_by_id(file_uuid=list(self.files)[self.scan_duration - 1])
file.scan() file.scan()
if file.visible_health_status == FileSystemItemHealthStatus.CORRUPTED: if file.visible_health_status == FileSystemItemHealthStatus.CORRUPT:
self.visible_health_status = FileSystemItemHealthStatus.CORRUPTED self.visible_health_status = FileSystemItemHealthStatus.CORRUPT
self.scan_duration -= 1 self.scan_duration -= 1
def get_file(self, file_name: str) -> Optional[File]: def get_file(self, file_name: str) -> Optional[File]:
@@ -524,19 +520,13 @@ class Folder(FileSystemItemABC):
# TODO: Increment read count? # TODO: Increment read count?
return self._files_by_name.get(file_name) return self._files_by_name.get(file_name)
def get_file_by_id(self, file_uuid: str, show_deleted: bool = False) -> File: def get_file_by_id(self, file_uuid: str) -> File:
""" """
Get a file by its uuid. Get a file by its uuid.
:param: file_uuid: The file uuid. :param: file_uuid: The file uuid.
:param: show_deleted: show deleted files
:return: The matching File. :return: The matching File.
""" """
deleted_file = self.deleted_files.get(file_uuid)
if show_deleted and deleted_file:
return deleted_file
return self.files.get(file_uuid) return self.files.get(file_uuid)
def add_file(self, file: File): def add_file(self, file: File):
@@ -571,16 +561,25 @@ class Folder(FileSystemItemABC):
raise Exception(f"Invalid file: {file}") raise Exception(f"Invalid file: {file}")
if self.files.get(file.uuid): if self.files.get(file.uuid):
# add to deleted list
file.delete()
self.deleted_files[file.uuid] = file
# remove from normal file list
self.files.pop(file.uuid) self.files.pop(file.uuid)
self._files_by_name.pop(file.name) self._files_by_name.pop(file.name)
else: else:
_LOGGER.debug(f"File with UUID {file.uuid} was not found.") _LOGGER.debug(f"File with UUID {file.uuid} was not found.")
def remove_file_by_id(self, file_uuid: str):
"""
Remove a file using id.
:param: file_uuid: The UUID of the file to remove.
"""
file = self.get_file_by_id(file_uuid=file_uuid)
self.remove_file(file=file)
def remove_all_files(self):
"""Removes all the files in the folder."""
self.files = {}
self._files_by_name = {}
def restore_file(self, file: Optional[File]): def restore_file(self, file: Optional[File]):
""" """
Restores a file. Restores a file.
@@ -593,19 +592,15 @@ class Folder(FileSystemItemABC):
def quarantine(self): def quarantine(self):
"""Quarantines the File System Folder.""" """Quarantines the File System Folder."""
if self.health_status != FileSystemItemHealthStatus.QUARANTINED: pass
self.health_status = FileSystemItemHealthStatus.QUARANTINED
self.fs.sys_log.info(f"Quarantined folder ./{self.name}")
def unquarantine(self): def unquarantine(self):
"""Unquarantine of the File System Folder.""" """Unquarantine of the File System Folder."""
if self.health_status == FileSystemItemHealthStatus.QUARANTINED: pass
self.health_status = FileSystemItemHealthStatus.GOOD
self.fs.sys_log.info(f"Quarantined folder ./{self.name}")
def quarantine_status(self) -> bool: def quarantine_status(self) -> bool:
"""Returns true if the folder is being quarantined.""" """Returns true if the folder is being quarantined."""
return self.health_status == FileSystemItemHealthStatus.QUARANTINED pass
def scan(self) -> None: def scan(self) -> None:
"""Update Folder visible status.""" """Update Folder visible status."""
@@ -657,7 +652,7 @@ class Folder(FileSystemItemABC):
repaired = file.repair() repaired = file.repair()
# set file status to good if corrupt # set file status to good if corrupt
if self.health_status == FileSystemItemHealthStatus.CORRUPTED: if self.health_status == FileSystemItemHealthStatus.CORRUPT:
self.health_status = FileSystemItemHealthStatus.GOOD self.health_status = FileSystemItemHealthStatus.GOOD
repaired = True repaired = True
@@ -668,21 +663,8 @@ class Folder(FileSystemItemABC):
"""TODO.""" """TODO."""
pass pass
def delete(self):
"""Deletes the files within the folder and then deletes the folder."""
super().delete()
# iterate through the files in the folder
files = copy.copy(self.files)
for file_id in files:
file = self.get_file_by_id(file_uuid=file_id)
file.delete()
self.remove_file(file)
self.fs.sys_log.info(f"Deleted folder {self.name} (id: {self.uuid})")
def corrupt(self) -> bool: def corrupt(self) -> bool:
"""Corrupt a File by setting the folder and containing files status to FileSystemItemStatus.CORRUPTED.""" """Corrupt a File by setting the folder and containing files status to FileSystemItemStatus.CORRUPT."""
super().corrupt() super().corrupt()
corrupted = False corrupted = False
@@ -694,7 +676,7 @@ class Folder(FileSystemItemABC):
# set file status to corrupt if good # set file status to corrupt if good
if self.health_status == FileSystemItemHealthStatus.GOOD: if self.health_status == FileSystemItemHealthStatus.GOOD:
self.health_status = FileSystemItemHealthStatus.CORRUPTED self.health_status = FileSystemItemHealthStatus.CORRUPT
corrupted = True corrupted = True
self.fs.sys_log.info(f"Corrupted folder {self.name} (id: {self.uuid})") self.fs.sys_log.info(f"Corrupted folder {self.name} (id: {self.uuid})")
@@ -830,19 +812,12 @@ class File(FileSystemItemABC):
return True return True
def delete(self) -> None:
"""Deletes the file."""
super().delete()
path = self.folder.name + "/" + self.name
self.folder.fs.sys_log.info(f"Deleting file {self.sim_path if self.sim_path else path}")
def repair(self) -> bool: def repair(self) -> bool:
"""Repair a corrupted File by setting the status to FileSystemItemStatus.GOOD.""" """Repair a corrupted File by setting the status to FileSystemItemStatus.GOOD."""
super().repair() super().repair()
# set file status to good if corrupt # set file status to good if corrupt
if self.health_status == FileSystemItemHealthStatus.CORRUPTED: if self.health_status == FileSystemItemHealthStatus.CORRUPT:
self.health_status = FileSystemItemHealthStatus.GOOD self.health_status = FileSystemItemHealthStatus.GOOD
path = self.folder.name + "/" + self.name path = self.folder.name + "/" + self.name
@@ -854,14 +829,14 @@ class File(FileSystemItemABC):
pass pass
def corrupt(self) -> bool: def corrupt(self) -> bool:
"""Corrupt a File by setting the status to FileSystemItemStatus.CORRUPTED.""" """Corrupt a File by setting the status to FileSystemItemStatus.CORRUPT."""
super().corrupt() super().corrupt()
corrupted = False corrupted = False
# set file status to good if corrupt # set file status to good if corrupt
if self.health_status == FileSystemItemHealthStatus.GOOD: if self.health_status == FileSystemItemHealthStatus.GOOD:
self.health_status = FileSystemItemHealthStatus.CORRUPTED self.health_status = FileSystemItemHealthStatus.CORRUPT
corrupted = True corrupted = True
path = self.folder.name + "/" + self.name path = self.folder.name + "/" + self.name

49
tests/unit_tests/_primaite/_simulator/_file_system/test_file_system.py
View File

@@ -44,7 +44,6 @@ def test_delete_file(file_system):
file_system.delete_file(folder_name="root", file_name="test_file.txt") file_system.delete_file(folder_name="root", file_name="test_file.txt")
assert len(file_system.folders) == 1 assert len(file_system.folders) == 1
assert len(file_system.get_folder("root").files) == 0 assert len(file_system.get_folder("root").files) == 0
assert len(file_system.get_folder("root").deleted_files) == 1
def test_delete_non_existent_file(file_system): def test_delete_non_existent_file(file_system):
@@ -70,7 +69,6 @@ def test_delete_folder(file_system):
file_system.delete_folder(folder_name="test_folder") file_system.delete_folder(folder_name="test_folder")
assert len(file_system.folders) == 1 assert len(file_system.folders) == 1
assert len(file_system.deleted_folders) == 1
def test_deleting_a_non_existent_folder(file_system): def test_deleting_a_non_existent_folder(file_system):
@@ -97,13 +95,11 @@ def test_move_file(file_system):
original_uuid = file.uuid original_uuid = file.uuid
assert len(file_system.get_folder("src_folder").files) == 1 assert len(file_system.get_folder("src_folder").files) == 1
assert len(file_system.get_folder("src_folder").deleted_files) == 0
assert len(file_system.get_folder("dst_folder").files) == 0 assert len(file_system.get_folder("dst_folder").files) == 0
file_system.move_file(src_folder_name="src_folder", src_file_name="test_file.txt", dst_folder_name="dst_folder") file_system.move_file(src_folder_name="src_folder", src_file_name="test_file.txt", dst_folder_name="dst_folder")
assert len(file_system.get_folder("src_folder").files) == 0 assert len(file_system.get_folder("src_folder").files) == 0
assert len(file_system.get_folder("src_folder").deleted_files) == 1
assert len(file_system.get_folder("dst_folder").files) == 1 assert len(file_system.get_folder("dst_folder").files) == 1
assert file_system.get_file("dst_folder", "test_file.txt").uuid == original_uuid assert file_system.get_file("dst_folder", "test_file.txt").uuid == original_uuid
@@ -126,6 +122,7 @@ def test_copy_file(file_system):
assert file_system.get_file("dst_folder", "test_file.txt").uuid != original_uuid assert file_system.get_file("dst_folder", "test_file.txt").uuid != original_uuid
@pytest.mark.skip(reason="Implementation for quarantine not needed yet")
def test_folder_quarantine_state(file_system): def test_folder_quarantine_state(file_system):
"""Tests the changing of folder quarantine status.""" """Tests the changing of folder quarantine status."""
folder = file_system.get_folder("root") folder = file_system.get_folder("root")
@@ -147,7 +144,7 @@ def test_file_corrupt_repair(file_system):
file.corrupt() file.corrupt()
assert folder.health_status == FileSystemItemHealthStatus.GOOD assert folder.health_status == FileSystemItemHealthStatus.GOOD
assert file.health_status == FileSystemItemHealthStatus.CORRUPTED assert file.health_status == FileSystemItemHealthStatus.CORRUPT
file.repair() file.repair()
@@ -163,8 +160,8 @@ def test_folder_corrupt_repair(file_system):
folder.corrupt() folder.corrupt()
file = folder.get_file(file_name="test_file.txt") file = folder.get_file(file_name="test_file.txt")
assert folder.health_status == FileSystemItemHealthStatus.CORRUPTED assert folder.health_status == FileSystemItemHealthStatus.CORRUPT
assert file.health_status == FileSystemItemHealthStatus.CORRUPTED assert file.health_status == FileSystemItemHealthStatus.CORRUPT
folder.repair() folder.repair()
@@ -183,13 +180,13 @@ def test_file_scan(file_system):
file.corrupt() file.corrupt()
assert file.health_status == FileSystemItemHealthStatus.CORRUPTED assert file.health_status == FileSystemItemHealthStatus.CORRUPT
assert file.visible_health_status == FileSystemItemHealthStatus.GOOD assert file.visible_health_status == FileSystemItemHealthStatus.GOOD
file.scan() file.scan()
assert file.health_status == FileSystemItemHealthStatus.CORRUPTED assert file.health_status == FileSystemItemHealthStatus.CORRUPT
assert file.visible_health_status == FileSystemItemHealthStatus.CORRUPTED assert file.visible_health_status == FileSystemItemHealthStatus.CORRUPT
def test_folder_scan(file_system): def test_folder_scan(file_system):
@@ -208,7 +205,7 @@ def test_folder_scan(file_system):
folder.corrupt() folder.corrupt()
assert folder.health_status == FileSystemItemHealthStatus.CORRUPTED assert folder.health_status == FileSystemItemHealthStatus.CORRUPT
assert folder.visible_health_status == FileSystemItemHealthStatus.GOOD assert folder.visible_health_status == FileSystemItemHealthStatus.GOOD
assert file1.visible_health_status == FileSystemItemHealthStatus.GOOD assert file1.visible_health_status == FileSystemItemHealthStatus.GOOD
assert file2.visible_health_status == FileSystemItemHealthStatus.GOOD assert file2.visible_health_status == FileSystemItemHealthStatus.GOOD
@@ -217,24 +214,24 @@ def test_folder_scan(file_system):
folder.apply_timestep(timestep=0) folder.apply_timestep(timestep=0)
assert folder.health_status == FileSystemItemHealthStatus.CORRUPTED assert folder.health_status == FileSystemItemHealthStatus.CORRUPT
assert folder.visible_health_status == FileSystemItemHealthStatus.CORRUPTED assert folder.visible_health_status == FileSystemItemHealthStatus.CORRUPT
assert file1.visible_health_status == FileSystemItemHealthStatus.CORRUPTED assert file1.visible_health_status == FileSystemItemHealthStatus.CORRUPT
assert file2.visible_health_status == FileSystemItemHealthStatus.GOOD assert file2.visible_health_status == FileSystemItemHealthStatus.GOOD
folder.apply_timestep(timestep=1) folder.apply_timestep(timestep=1)
assert folder.health_status == FileSystemItemHealthStatus.CORRUPTED assert folder.health_status == FileSystemItemHealthStatus.CORRUPT
assert folder.visible_health_status == FileSystemItemHealthStatus.CORRUPTED assert folder.visible_health_status == FileSystemItemHealthStatus.CORRUPT
assert file1.visible_health_status == FileSystemItemHealthStatus.CORRUPTED assert file1.visible_health_status == FileSystemItemHealthStatus.CORRUPT
assert file2.visible_health_status == FileSystemItemHealthStatus.CORRUPTED assert file2.visible_health_status == FileSystemItemHealthStatus.CORRUPT
folder.apply_timestep(timestep=2) folder.apply_timestep(timestep=2)
assert folder.health_status == FileSystemItemHealthStatus.CORRUPTED assert folder.health_status == FileSystemItemHealthStatus.CORRUPT
assert folder.visible_health_status == FileSystemItemHealthStatus.CORRUPTED assert folder.visible_health_status == FileSystemItemHealthStatus.CORRUPT
assert file1.visible_health_status == FileSystemItemHealthStatus.CORRUPTED assert file1.visible_health_status == FileSystemItemHealthStatus.CORRUPT
assert file2.visible_health_status == FileSystemItemHealthStatus.CORRUPTED assert file2.visible_health_status == FileSystemItemHealthStatus.CORRUPT
def test_simulated_file_check_hash(file_system): def test_simulated_file_check_hash(file_system):
@@ -245,7 +242,7 @@ def test_simulated_file_check_hash(file_system):
# change simulated file size # change simulated file size
file.sim_size = 0 file.sim_size = 0
assert file.check_hash() is False assert file.check_hash() is False
assert file.health_status == FileSystemItemHealthStatus.CORRUPTED assert file.health_status == FileSystemItemHealthStatus.CORRUPT
def test_real_file_check_hash(file_system): def test_real_file_check_hash(file_system):
@@ -258,7 +255,7 @@ def test_real_file_check_hash(file_system):
f.write("get hacked scrub lol xD\n") f.write("get hacked scrub lol xD\n")
assert file.check_hash() is False assert file.check_hash() is False
assert file.health_status == FileSystemItemHealthStatus.CORRUPTED assert file.health_status == FileSystemItemHealthStatus.CORRUPT
def test_simulated_folder_check_hash(file_system): def test_simulated_folder_check_hash(file_system):
@@ -271,7 +268,7 @@ def test_simulated_folder_check_hash(file_system):
file = folder.get_file(file_name="test_file.txt") file = folder.get_file(file_name="test_file.txt")
file.sim_size = 0 file.sim_size = 0
assert folder.check_hash() is False assert folder.check_hash() is False
assert folder.health_status == FileSystemItemHealthStatus.CORRUPTED assert folder.health_status == FileSystemItemHealthStatus.CORRUPT
def test_real_folder_check_hash(file_system): def test_real_folder_check_hash(file_system):
@@ -288,7 +285,7 @@ def test_real_folder_check_hash(file_system):
f.write("get hacked scrub lol xD\n") f.write("get hacked scrub lol xD\n")
assert folder.check_hash() is False assert folder.check_hash() is False
assert folder.health_status == FileSystemItemHealthStatus.CORRUPTED assert folder.health_status == FileSystemItemHealthStatus.CORRUPT
@pytest.mark.skip(reason="Skipping until we tackle serialisation") @pytest.mark.skip(reason="Skipping until we tackle serialisation")

62
tests/unit_tests/_primaite/_simulator/_file_system/test_file_system_actions.py
View File

@@ -19,13 +19,13 @@ def test_file_scan_request(populated_file_system):
fs, folder, file = populated_file_system fs, folder, file = populated_file_system
file.corrupt() file.corrupt()
assert file.health_status == FileSystemItemHealthStatus.CORRUPTED assert file.health_status == FileSystemItemHealthStatus.CORRUPT
assert file.visible_health_status == FileSystemItemHealthStatus.GOOD assert file.visible_health_status == FileSystemItemHealthStatus.GOOD
fs.apply_request(request=["file", file.uuid, "scan"]) fs.apply_request(request=["file", file.uuid, "scan"])
assert file.health_status == FileSystemItemHealthStatus.CORRUPTED assert file.health_status == FileSystemItemHealthStatus.CORRUPT
assert file.visible_health_status == FileSystemItemHealthStatus.CORRUPTED assert file.visible_health_status == FileSystemItemHealthStatus.CORRUPT
def test_folder_scan_request(populated_file_system): def test_folder_scan_request(populated_file_system):
@@ -37,7 +37,7 @@ def test_folder_scan_request(populated_file_system):
file2: File = folder.get_file_by_id(file_uuid=list(folder.files)[0]) file2: File = folder.get_file_by_id(file_uuid=list(folder.files)[0])
folder.corrupt() folder.corrupt()
assert folder.health_status == FileSystemItemHealthStatus.CORRUPTED assert folder.health_status == FileSystemItemHealthStatus.CORRUPT
assert folder.visible_health_status == FileSystemItemHealthStatus.GOOD assert folder.visible_health_status == FileSystemItemHealthStatus.GOOD
assert file1.visible_health_status == FileSystemItemHealthStatus.GOOD assert file1.visible_health_status == FileSystemItemHealthStatus.GOOD
assert file2.visible_health_status == FileSystemItemHealthStatus.GOOD assert file2.visible_health_status == FileSystemItemHealthStatus.GOOD
@@ -46,24 +46,24 @@ def test_folder_scan_request(populated_file_system):
folder.apply_timestep(timestep=0) folder.apply_timestep(timestep=0)
assert folder.health_status == FileSystemItemHealthStatus.CORRUPTED assert folder.health_status == FileSystemItemHealthStatus.CORRUPT
assert folder.visible_health_status == FileSystemItemHealthStatus.CORRUPTED assert folder.visible_health_status == FileSystemItemHealthStatus.CORRUPT
assert file1.visible_health_status == FileSystemItemHealthStatus.CORRUPTED assert file1.visible_health_status == FileSystemItemHealthStatus.CORRUPT
assert file2.visible_health_status == FileSystemItemHealthStatus.GOOD assert file2.visible_health_status == FileSystemItemHealthStatus.GOOD
folder.apply_timestep(timestep=1) folder.apply_timestep(timestep=1)
assert folder.health_status == FileSystemItemHealthStatus.CORRUPTED assert folder.health_status == FileSystemItemHealthStatus.CORRUPT
assert folder.visible_health_status == FileSystemItemHealthStatus.CORRUPTED assert folder.visible_health_status == FileSystemItemHealthStatus.CORRUPT
assert file1.visible_health_status == FileSystemItemHealthStatus.CORRUPTED assert file1.visible_health_status == FileSystemItemHealthStatus.CORRUPT
assert file2.visible_health_status == FileSystemItemHealthStatus.CORRUPTED assert file2.visible_health_status == FileSystemItemHealthStatus.CORRUPT
folder.apply_timestep(timestep=2) folder.apply_timestep(timestep=2)
assert folder.health_status == FileSystemItemHealthStatus.CORRUPTED assert folder.health_status == FileSystemItemHealthStatus.CORRUPT
assert folder.visible_health_status == FileSystemItemHealthStatus.CORRUPTED assert folder.visible_health_status == FileSystemItemHealthStatus.CORRUPT
assert file1.visible_health_status == FileSystemItemHealthStatus.CORRUPTED assert file1.visible_health_status == FileSystemItemHealthStatus.CORRUPT
assert file2.visible_health_status == FileSystemItemHealthStatus.CORRUPTED assert file2.visible_health_status == FileSystemItemHealthStatus.CORRUPT
def test_file_checkhash_request(populated_file_system): def test_file_checkhash_request(populated_file_system):
@@ -77,7 +77,7 @@ def test_file_checkhash_request(populated_file_system):
fs.apply_request(request=["file", file.uuid, "checkhash"]) fs.apply_request(request=["file", file.uuid, "checkhash"])
assert file.health_status == FileSystemItemHealthStatus.CORRUPTED assert file.health_status == FileSystemItemHealthStatus.CORRUPT
def test_folder_checkhash_request(populated_file_system): def test_folder_checkhash_request(populated_file_system):
@@ -90,7 +90,7 @@ def test_folder_checkhash_request(populated_file_system):
file.sim_size = 0 file.sim_size = 0
fs.apply_request(request=["folder", folder.uuid, "checkhash"]) fs.apply_request(request=["folder", folder.uuid, "checkhash"])
assert folder.health_status == FileSystemItemHealthStatus.CORRUPTED assert folder.health_status == FileSystemItemHealthStatus.CORRUPT
def test_file_repair_request(populated_file_system): def test_file_repair_request(populated_file_system):
@@ -98,7 +98,7 @@ def test_file_repair_request(populated_file_system):
fs, folder, file = populated_file_system fs, folder, file = populated_file_system
file.corrupt() file.corrupt()
assert file.health_status == FileSystemItemHealthStatus.CORRUPTED assert file.health_status == FileSystemItemHealthStatus.CORRUPT
fs.apply_request(request=["file", file.uuid, "repair"]) fs.apply_request(request=["file", file.uuid, "repair"])
assert file.health_status == FileSystemItemHealthStatus.GOOD assert file.health_status == FileSystemItemHealthStatus.GOOD
@@ -109,8 +109,8 @@ def test_folder_repair_request(populated_file_system):
fs, folder, file = populated_file_system fs, folder, file = populated_file_system
folder.corrupt() folder.corrupt()
assert file.health_status == FileSystemItemHealthStatus.CORRUPTED assert file.health_status == FileSystemItemHealthStatus.CORRUPT
assert folder.health_status == FileSystemItemHealthStatus.CORRUPTED assert folder.health_status == FileSystemItemHealthStatus.CORRUPT
fs.apply_request(request=["folder", folder.uuid, "repair"]) fs.apply_request(request=["folder", folder.uuid, "repair"])
assert file.health_status == FileSystemItemHealthStatus.GOOD assert file.health_status == FileSystemItemHealthStatus.GOOD
@@ -129,20 +129,32 @@ def test_file_corrupt_request(populated_file_system):
"""Test that an agent can request a file corruption.""" """Test that an agent can request a file corruption."""
fs, folder, file = populated_file_system fs, folder, file = populated_file_system
fs.apply_request(request=["file", file.uuid, "corrupt"]) fs.apply_request(request=["file", file.uuid, "corrupt"])
assert file.health_status == FileSystemItemHealthStatus.CORRUPTED assert file.health_status == FileSystemItemHealthStatus.CORRUPT
def test_folder_corrupt_request(populated_file_system): def test_folder_corrupt_request(populated_file_system):
"""Test that an agent can request a folder corruption.""" """Test that an agent can request a folder corruption."""
fs, folder, file = populated_file_system fs, folder, file = populated_file_system
fs.apply_request(request=["folder", folder.uuid, "corrupt"]) fs.apply_request(request=["folder", folder.uuid, "corrupt"])
assert file.health_status == FileSystemItemHealthStatus.CORRUPTED assert file.health_status == FileSystemItemHealthStatus.CORRUPT
assert folder.health_status == FileSystemItemHealthStatus.CORRUPTED assert folder.health_status == FileSystemItemHealthStatus.CORRUPT
def test_file_delete_request(populated_file_system): def test_file_delete_request(populated_file_system):
pass """Test that an agent can request a file deletion."""
fs, folder, file = populated_file_system
assert folder.get_file_by_id(file_uuid=file.uuid) is not None
fs.apply_request(request=["folder", folder.uuid, "delete", file.uuid])
assert folder.get_file_by_id(file_uuid=file.uuid) is None
def test_folder_delete_request(populated_file_system): def test_folder_delete_request(populated_file_system):
pass """Test that an agent can request a folder deletion."""
fs, folder, file = populated_file_system
assert folder.get_file_by_id(file_uuid=file.uuid) is not None
assert fs.get_folder_by_id(folder_uuid=folder.uuid) is not None
fs.apply_request(request=["delete", folder.uuid])
assert fs.get_folder_by_id(folder_uuid=folder.uuid) is None
assert folder.get_file_by_id(file_uuid=file.uuid) is None