From 6e7fef6747198c9e7a084e3930bb6b75da9b1f19 Mon Sep 17 00:00:00 2001 From: Jacob Boddey Date: Mon, 11 Mar 2024 15:10:18 +0000 Subject: [PATCH 1/2] Remove report limit for now --- .../python/src/test_orc/test_orchestrator.py | 98 +++++++++---------- 1 file changed, 49 insertions(+), 49 deletions(-) diff --git a/framework/python/src/test_orc/test_orchestrator.py b/framework/python/src/test_orc/test_orchestrator.py index 6711584be..e6ddd584b 100644 --- a/framework/python/src/test_orc/test_orchestrator.py +++ b/framework/python/src/test_orc/test_orchestrator.py @@ -127,8 +127,8 @@ def run_test_modules(self): # Archive the runtime directory self._zip_results(timestamp_dir) - LOGGER.debug("Cleaning old test results...") - self._cleanup_old_test_results(device) + #LOGGER.debug("Cleaning old test results...") + #self._cleanup_old_test_results(device) LOGGER.debug("Old test results cleaned") @@ -182,53 +182,53 @@ def _calculate_result(self): result = "Non-Compliant" return result - def _cleanup_old_test_results(self, device): - - if device.max_device_reports is not None: - max_device_reports = device.max_device_reports - else: - max_device_reports = self._session.get_max_device_reports() - - completed_results_dir = os.path.join( - self._root_path, - LOCAL_DEVICE_REPORTS.replace("{device_folder}", device.device_folder)) - - completed_tests = os.listdir(completed_results_dir) - cur_test_count = len(completed_tests) - if cur_test_count > max_device_reports: - LOGGER.debug("Current device has more than max tests results allowed: " + - str(cur_test_count) + ">" + str(max_device_reports)) - - # Find and delete the oldest test - oldest_test = self._find_oldest_test(completed_results_dir) - if oldest_test is not None: - LOGGER.debug("Oldest test found, removing: " + str(oldest_test[1])) - shutil.rmtree(oldest_test[1], ignore_errors=True) - - # Remove oldest test from session - oldest_timestamp = oldest_test[0] - self.get_session().get_target_device().remove_report(oldest_timestamp) - - # Confirm the delete was succesful - new_test_count = len(os.listdir(completed_results_dir)) - if (new_test_count != cur_test_count - and new_test_count > max_device_reports): - # Continue cleaning up until we're under the max - self._cleanup_old_test_results(device) - - def _find_oldest_test(self, completed_tests_dir): - oldest_timestamp = None - oldest_directory = None - for completed_test in os.listdir(completed_tests_dir): - timestamp = datetime.strptime(str(completed_test), "%Y-%m-%dT%H:%M:%S") - if oldest_timestamp is None or timestamp < oldest_timestamp: - oldest_timestamp = timestamp - oldest_directory = completed_test - if oldest_directory: - return oldest_timestamp, os.path.join(completed_tests_dir, - oldest_directory) - else: - return None + # def _cleanup_old_test_results(self, device): + + # if device.max_device_reports is not None: + # max_device_reports = device.max_device_reports + # else: + # max_device_reports = self._session.get_max_device_reports() + + # completed_results_dir = os.path.join( + # self._root_path, + # LOCAL_DEVICE_REPORTS.replace("{device_folder}", device.device_folder)) + + # completed_tests = os.listdir(completed_results_dir) + # cur_test_count = len(completed_tests) + # if cur_test_count > max_device_reports: + # LOGGER.debug("Current device has more than max tests results allowed: " + + # str(cur_test_count) + ">" + str(max_device_reports)) + + # # Find and delete the oldest test + # oldest_test = self._find_oldest_test(completed_results_dir) + # if oldest_test is not None: + # LOGGER.debug("Oldest test found, removing: " + str(oldest_test[1])) + # shutil.rmtree(oldest_test[1], ignore_errors=True) + + # # Remove oldest test from session + # oldest_timestamp = oldest_test[0] + # self.get_session().get_target_device().remove_report(oldest_timestamp) + + # # Confirm the delete was succesful + # new_test_count = len(os.listdir(completed_results_dir)) + # if (new_test_count != cur_test_count + # and new_test_count > max_device_reports): + # # Continue cleaning up until we're under the max + # self._cleanup_old_test_results(device) + + # def _find_oldest_test(self, completed_tests_dir): + # oldest_timestamp = None + # oldest_directory = None + # for completed_test in os.listdir(completed_tests_dir): + # timestamp = datetime.strptime(str(completed_test), "%Y-%m-%dT%H:%M:%S") + # if oldest_timestamp is None or timestamp < oldest_timestamp: + # oldest_timestamp = timestamp + # oldest_directory = completed_test + # if oldest_directory: + # return oldest_timestamp, os.path.join(completed_tests_dir, + # oldest_directory) + # else: + # return None def _timestamp_results(self, device): From 98c11bd49f9f13a5a08ab534384999caff7e8db5 Mon Sep 17 00:00:00 2001 From: jhughesbiot Date: Thu, 21 Mar 2024 11:43:35 -0600 Subject: [PATCH 2/2] Re-enable test limit Fix timestamp comparison in device for session report removal --- framework/python/src/common/device.py | 14 ++- .../python/src/test_orc/test_orchestrator.py | 98 +++++++++---------- 2 files changed, 55 insertions(+), 57 deletions(-) diff --git a/framework/python/src/common/device.py b/framework/python/src/common/device.py index 979344eb4..144161d8a 100644 --- a/framework/python/src/common/device.py +++ b/framework/python/src/common/device.py @@ -17,6 +17,7 @@ from typing import Dict, List from dataclasses import dataclass, field from common.testreport import TestReport +from datetime import datetime @dataclass class Device(): @@ -40,16 +41,13 @@ def add_report(self, report): def get_reports(self): return self.reports - def remove_report(self, timestamp): - - remove_report_target = None + def remove_report(self, timestamp: datetime): + # remove_report_target = None for report in self.reports: report_timestamp = report.get_started().strftime('%Y-%m-%dT%H:%M:%S') - if report_timestamp == timestamp: - remove_report_target = report - - if remove_report_target is not None: - self.reports.remove(remove_report_target) + if report.get_started() == timestamp: + self.reports.remove(report) + break def to_dict(self): """Returns the device as a python dictionary. This is used for the diff --git a/framework/python/src/test_orc/test_orchestrator.py b/framework/python/src/test_orc/test_orchestrator.py index f817c597d..d0ca63b24 100644 --- a/framework/python/src/test_orc/test_orchestrator.py +++ b/framework/python/src/test_orc/test_orchestrator.py @@ -127,8 +127,8 @@ def run_test_modules(self): # Archive the runtime directory self._zip_results(timestamp_dir) - #LOGGER.debug("Cleaning old test results...") - #self._cleanup_old_test_results(device) + LOGGER.debug("Cleaning old test results...") + self._cleanup_old_test_results(device) LOGGER.debug("Old test results cleaned") @@ -182,53 +182,53 @@ def _calculate_result(self): result = "Non-Compliant" return result - # def _cleanup_old_test_results(self, device): - - # if device.max_device_reports is not None: - # max_device_reports = device.max_device_reports - # else: - # max_device_reports = self._session.get_max_device_reports() - - # completed_results_dir = os.path.join( - # self._root_path, - # LOCAL_DEVICE_REPORTS.replace("{device_folder}", device.device_folder)) - - # completed_tests = os.listdir(completed_results_dir) - # cur_test_count = len(completed_tests) - # if cur_test_count > max_device_reports: - # LOGGER.debug("Current device has more than max tests results allowed: " + - # str(cur_test_count) + ">" + str(max_device_reports)) - - # # Find and delete the oldest test - # oldest_test = self._find_oldest_test(completed_results_dir) - # if oldest_test is not None: - # LOGGER.debug("Oldest test found, removing: " + str(oldest_test[1])) - # shutil.rmtree(oldest_test[1], ignore_errors=True) - - # # Remove oldest test from session - # oldest_timestamp = oldest_test[0] - # self.get_session().get_target_device().remove_report(oldest_timestamp) - - # # Confirm the delete was succesful - # new_test_count = len(os.listdir(completed_results_dir)) - # if (new_test_count != cur_test_count - # and new_test_count > max_device_reports): - # # Continue cleaning up until we're under the max - # self._cleanup_old_test_results(device) - - # def _find_oldest_test(self, completed_tests_dir): - # oldest_timestamp = None - # oldest_directory = None - # for completed_test in os.listdir(completed_tests_dir): - # timestamp = datetime.strptime(str(completed_test), "%Y-%m-%dT%H:%M:%S") - # if oldest_timestamp is None or timestamp < oldest_timestamp: - # oldest_timestamp = timestamp - # oldest_directory = completed_test - # if oldest_directory: - # return oldest_timestamp, os.path.join(completed_tests_dir, - # oldest_directory) - # else: - # return None + def _cleanup_old_test_results(self, device): + + if device.max_device_reports is not None: + max_device_reports = device.max_device_reports + else: + max_device_reports = self._session.get_max_device_reports() + + completed_results_dir = os.path.join( + self._root_path, + LOCAL_DEVICE_REPORTS.replace("{device_folder}", device.device_folder)) + + completed_tests = os.listdir(completed_results_dir) + cur_test_count = len(completed_tests) + if cur_test_count > max_device_reports: + LOGGER.debug("Current device has more than max tests results allowed: " + + str(cur_test_count) + ">" + str(max_device_reports)) + + # Find and delete the oldest test + oldest_test = self._find_oldest_test(completed_results_dir) + if oldest_test is not None: + LOGGER.debug("Oldest test found, removing: " + str(oldest_test[1])) + shutil.rmtree(oldest_test[1], ignore_errors=True) + + # Remove oldest test from session + oldest_timestamp = oldest_test[0] + self.get_session().get_target_device().remove_report(oldest_timestamp) + + # Confirm the delete was succesful + new_test_count = len(os.listdir(completed_results_dir)) + if (new_test_count != cur_test_count + and new_test_count > max_device_reports): + # Continue cleaning up until we're under the max + self._cleanup_old_test_results(device) + + def _find_oldest_test(self, completed_tests_dir): + oldest_timestamp = None + oldest_directory = None + for completed_test in os.listdir(completed_tests_dir): + timestamp = datetime.strptime(str(completed_test), "%Y-%m-%dT%H:%M:%S") + if oldest_timestamp is None or timestamp < oldest_timestamp: + oldest_timestamp = timestamp + oldest_directory = completed_test + if oldest_directory: + return oldest_timestamp, os.path.join(completed_tests_dir, + oldest_directory) + else: + return None def _timestamp_results(self, device):