Skip to content
Snippets Groups Projects
Commit 1149a2eb authored by Justus Müller's avatar Justus Müller :bee:
Browse files

fix: Rewrite to raw pattern

parent 1c6331e6
Branches
Tags
No related merge requests found
......@@ -18,7 +18,6 @@ if __name__ == "__main__":
slack_test: bool = True
drive_letter: chr = ''
format_drive: bool = False
check_cluster_overlap: bool = False
max_memory: int = 1_024 # 1 GB
for argument in arguments.keys():
......@@ -50,14 +49,6 @@ if __name__ == "__main__":
else:
raise ValueError("Unknown format state.")
case("check_cluster_overlap"):
if (arguments.get("check_cluster_overlap").lower() == "false"):
check_cluster_overlap = False
elif (arguments.get("check_cluster_overlap").lower() == "true"):
check_cluster_overlap = True
else:
raise ValueError("Unknown cluster overlap check state.")
case("max_memory"): # max_memory is passed in MiB
max_memory = int(arguments.get("max_memory"))
......@@ -83,7 +74,6 @@ if __name__ == "__main__":
test_orchistrator_arguments: dict = {
"hash": hash_test,
"slack": slack_test,
"check_cluster_overlap": check_cluster_overlap,
}
TestOrchistrator.get_instance().generate_tests(**test_orchistrator_arguments)
......
......@@ -120,6 +120,7 @@ class DriveHandle:
# all extents are checked
# return final clusters
progressbar.finish()
return cluster_numbers
def __check_cluster_file_association(self, cluster_number: int, file_name: str) -> bool:
......@@ -142,27 +143,35 @@ class DriveHandle:
return True
return False
def write_pattern(self, path: str) -> None:
def write_pattern(self) -> None:
"""
Writes the pattern file to the disk
Parameters:
symbol (chr): The symbol the pattern file shall be filled with
path (str): The path to the pattern file
"""
print(colored(f"[DriveHandle] Writing a file ({path}) with a rolling pattern to fill the disk.", "light_grey"))
pattern_start_symbol: chr = 'A'
iteration: int = 0
(_, _, cluster_size) = self.get_allocation_unit_sizes()
progressbar = Spinner('[DriveHandle] Writing the pattern file')
with open(path, "wb") as file_descriptor:
while os.path.getsize(path) + cluster_size < self.get_free_drive_space():
current_pattern_symbol: chr = chr(ord(pattern_start_symbol) + iteration)
file_descriptor.write(current_pattern_symbol.encode() * cluster_size)
iteration = (iteration + 1) % 26
print(colored(f"[DriveHandle] Writing a pattern to the raw disk.", "light_grey"))
pattern_symbol: bytes = b'A' * 512
progressbar = Spinner('[DriveHandle] Writing the pattern')
volume_path = subprocess.check_output(['powershell.exe', f'(Get-Volume -DriveLetter {self._drive_letter}).UniqueID'])
raw_disk = open(volume_path, 'wb')
try:
while True:
raw_disk.write(pattern_symbol)
progressbar.next()
except OSError as error:
print(error)
raw_disk.close()
finally:
raw_disk.close()
progressbar.finish()
print(colored("[DriveHandle] Completed writing the pattern file.", "light_grey"))
print(colored("[DriveHandle] Completed writing the pattern.", "light_grey"))
print(colored("[DriveHandle] Quick-formatting the volume."))
os.system(f"format {self._drive_letter}: /FS:NTFS /A:{DriveHandle._DEFAULT_CLUSTER_SIZE} /q /Y")
print(colored("[DriveHandle] Quick-formatted the volume. Disk is now filled with the pattern."))
def create_overwrite_file(self, symbol: chr, path: str) -> None:
"""
......
......@@ -49,7 +49,7 @@ class TestOrchistrator:
"""
print(colored("[TestOrchistrator] Generating test instances.", "light_grey"))
if (kwargs.get("slack")):
self._tests.append(SlackTest(self._drive_handle, bool(kwargs.get("check_cluster_overlap"))))
self._tests.append(SlackTest(self._drive_handle)))
print(colored("[TestOrchistrator] Generated slack test instance.", "light_grey"))
if (kwargs.get("hash")):
self._tests.append(HashTest(self._drive_handle))
......
......@@ -7,15 +7,13 @@ from termcolor import colored
class SlackTest(DecisionTest):
_drive_handle: DriveHandle = None
_check_cluster_overlap: bool = True
@property
def WEIGHT(self):
return 2
def __init__(self, drive_handle: DriveHandle, check_cluster_overlap: bool) -> None:
def __init__(self, drive_handle: DriveHandle) -> None:
self._drive_handle = drive_handle
self._check_cluster_overlap = check_cluster_overlap
def determine_SSD_indication(self) -> bool:
"""
......@@ -28,36 +26,23 @@ class SlackTest(DecisionTest):
"""
print(colored("[SlackTest] Running the slack test!", 'light_blue'))
# create rolling pattern
pattern_file_path: str = f'{self._drive_handle.get_path()}transsd-slacktest-pattern'
self._drive_handle.write_pattern(pattern_file_path)
# create pattern
self.drive_handle.write_pattern()
pattern_file_clusters: list = []
if self._check_cluster_overlap:
pattern_file_clusters = self._drive_handle.get_clusters_for_file(pattern_file_path)
print(colored(f"[SlackTest] First cluster of the pattern file ({pattern_file_path}) is {pattern_file_clusters[0]}", "light_grey"))
print(colored(f"[SlackTest] Last cluster of the pattern file ({pattern_file_path}) is {pattern_file_clusters[-1]}", "light_grey"))
# delete file
self._drive_handle.delete_file(pattern_file_path)
print("[SlackTest] Waiting 60s for OS to clean up the disk...")
time.sleep(60)
print(colored("[SlackTest] File should be removed from disk without remainder.", "light_grey"))
self.drive_handle.flush_cache()
print("[SlackTest] Waiting 120s for OS to flush the cache...")
time.sleep(120)
# create overwriting file
overwrite_file_path: str = f'{self._drive_handle.get_path()}transsd-slacktest-overwrite'
self._drive_handle.create_overwrite_file('!', overwrite_file_path)
overwrite_file_clusters: list = self._drive_handle.get_clusters_for_file(overwrite_file_path)
overwrite_file_path: str = f'{self.drive_handle.get_path()}transsd-slacktest-overwrite'
self.drive_handle.create_overwrite_file('!', overwrite_file_path)
overwrite_file_clusters: list = self.drive_handle.get_clusters_for_file(overwrite_file_path)
print(colored(f"[SlackTest] The cluster numbers associated with the overwrite file ({overwrite_file_path}) are {overwrite_file_clusters}", "light_grey"))
overwrite_file_last_cluster: int = self._drive_handle.read_cluster(overwrite_file_clusters[-1])
if self._check_cluster_overlap:
if overwrite_file_clusters[-1] in pattern_file_clusters:
print(colored("[SlackTest] The last cluster of the overwrite file is part of the pattern file.", "green"))
else:
print(colored("[SlackTest] The last cluster of the overwrite file is disjoint with these of the pattern file.", "red"))
# If both clusters are not part, this test is useless
raise RuntimeError("SlackTest could not be performed.")
self.drive_handle.flush_cache()
print("[SlackTest] Waiting 120s for OS to flush the cache...")
time.sleep(120)
# analyse the slack space
(sector_size, sectors_per_cluster, _) = self._drive_handle.get_allocation_unit_sizes()
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment