Skip to content
174 changes: 137 additions & 37 deletions src/integrationtest/resource_validation.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,89 +16,189 @@
#
# import integrationtest.resource_validation as resource_validation
# resval = resource_validation.ResourceValidator()
# resval.require_cpu_count(64)
# resval.require_free_memory_gb(28)
# resval.cpu_count_needs(32, 64)
# resval.free_memory_needs(28)
# # set other minimum values, if desired
# resval_debug_string = resval.get_debug_string()
# print(f"{resval_debug_string}")
# # then, in one of the pytest "tests"
# if not resval.this_computer_has_sufficient_resources:
# resval_full_report = resval.get_insufficient_resources_report()
# if not resval.required_resources_are_present
# resval_full_report = resval.get_required_resources_report()
# print(f"{resval_full_report}")
# resval_summary_report = resval.get_insufficient_resources_summary()
# resval_summary_report = resval.get_required_resources_summary()
# pytest.skip(f"{resval_summary_report}")
# if not resval.recommended_resources_are_present
# resval_full_report = resval.get_recommended_resources_report()
# print(f"{resval_full_report}")

import os
import psutil
import shutil

class ResourceValidator:
def __init__(self):
self.this_computer_has_sufficient_resources = True

# initialization of constants
hostname = os.uname().nodename
self.report_header = f"This computer ({hostname}) does not have enough resources to run this test."
self.required_resource_report_header = f"This computer ({hostname}) does not have enough resources to run this test."
self.recommended_resource_report_header = f"This computer ({hostname}) does not have the recommended amount of resources to run this test."
self.report_indentation = " *"

# set the initial state of the status variables to indicate sufficient resources
# users check these attributes directly to confirm if resources are available
self.this_computer_has_sufficient_resources = True
self.required_resources_are_present = True
self.recommended_resources_are_present = True

# debug and report strings start out empty and are filled in, as needed
self.debug_string = ""
self.report_string = ""
self.required_resource_report_string = ""
self.recommended_resource_report_string = ""

# this attribute is provided to users for whatever extra disk-space-checking they may want to do.
# it gets filled in when free_disk_space_needs() is called.
self.free_disk_space_gb = -1

# deprecated method, will be removed someday, simply calls newer method
def require_cpu_count(self, minimum_cpu_count):
self.cpu_count_needs(required_count=minimum_cpu_count)

# method to specify the number of CPUs that is needed
def cpu_count_needs(self, required_count=-1, recommended_count=-1):
cpu_count = os.cpu_count()
self.debug_string += f"\nDEBUG: CPU count is {cpu_count}, minimum required number is {minimum_cpu_count}."
if cpu_count < minimum_cpu_count:
self.debug_string += f"\nDEBUG: CPU count is {cpu_count}"
if required_count >= 0:
self.debug_string += f", required number is {required_count}"
if recommended_count >= 0:
self.debug_string += f", recommended number is {recommended_count}"
if cpu_count < required_count:
self.this_computer_has_sufficient_resources = False
if len(self.report_string) == 0:
self.report_string = self.report_header
self.report_string += f"\n{self.report_indentation} CPU count is {cpu_count}, minimum CPU count is {minimum_cpu_count}."
self.required_resources_are_present = False
if len(self.required_resource_report_string) == 0:
self.required_resource_report_string = self.required_resource_report_header
self.required_resource_report_string += f"\n{self.report_indentation} CPU count is {cpu_count}, required CPU count is {required_count}."
if cpu_count < recommended_count:
self.recommended_resources_are_present = False
if len(self.recommended_resource_report_string) == 0:
self.recommended_resource_report_string = self.recommended_resource_report_header
self.recommended_resource_report_string += f"\n{self.report_indentation} CPU count is {cpu_count}, recommended CPU count is {recommended_count}."

# deprecated method, will be removed someday, simply calls newer method
def require_free_memory_gb(self, minimum_free_memory):
self.free_memory_needs(required_free_memory=minimum_free_memory)

# method to specify the free memory that is needed, in units of GB
def free_memory_needs(self, required_free_memory=-1, recommended_free_memory=-1):
mem_obj = psutil.virtual_memory()
free_mem = round((mem_obj.available / (1024 * 1024 * 1024)), 2)
self.debug_string += f"\nDEBUG: Free memory is {free_mem} GB, minimum required amount is {minimum_free_memory}."
if free_mem < minimum_free_memory:
self.debug_string += f"\nDEBUG: Free memory is {free_mem} GB"
if required_free_memory >= 0:
self.debug_string += f", required amount is {required_free_memory}"
if recommended_free_memory >= 0:
self.debug_string += f", recommended amount is {recommended_free_memory}"
if free_mem < required_free_memory:
self.this_computer_has_sufficient_resources = False
if len(self.report_string) == 0:
self.report_string = self.report_header
self.report_string += f"\n{self.report_indentation} Free memory is {free_mem} GB, minimum amount is {minimum_free_memory}."
self.required_resources_are_present = False
if len(self.required_resource_report_string) == 0:
self.required_resource_report_string = self.required_resource_report_header
self.required_resource_report_string += f"\n{self.report_indentation} Free memory is {free_mem} GB, required amount is {required_free_memory}."
if free_mem < recommended_free_memory:
self.recommended_resources_are_present = False
if len(self.recommended_resource_report_string) == 0:
self.recommended_resource_report_string = self.recommended_resource_report_header
self.recommended_resource_report_string += f"\n{self.report_indentation} Free memory is {free_mem} GB, recommended amount is {recommended_free_memory}."

# deprecated method, will be removed someday, simply calls newer method
def require_total_memory_gb(self, minimum_total_memory):
self.total_memory_needs(required_total_memory=minimum_total_memory)

# method to specify the total memory that is needed, in units of GB
def total_memory_needs(self, required_total_memory=-1, recommended_total_memory=-1):
mem_obj = psutil.virtual_memory()
total_mem = round((mem_obj.total / (1024 * 1024 * 1024)), 2)
self.debug_string += f"\nDEBUG: Total memory is {total_mem} GB, minimum required amount is {minimum_total_memory}."
if total_mem < minimum_total_memory:
self.debug_string += f"\nDEBUG: Total memory is {total_mem} GB"
if required_total_memory >= 0:
self.debug_string += f", required amount is {required_total_memory}"
if recommended_total_memory >= 0:
self.debug_string += f", recommended amount is {recommended_total_memory}"
if total_mem < required_total_memory:
self.this_computer_has_sufficient_resources = False
if len(self.report_string) == 0:
self.report_string = self.report_header
self.report_string += f"\n{self.report_indentation} Total memory is {total_mem} GB, minimum amount is {minimum_total_memory}."
self.required_resources_are_present = False
if len(self.required_resource_report_string) == 0:
self.required_resource_report_string = self.required_resource_report_header
self.required_resource_report_string += f"\n{self.report_indentation} Total memory is {total_mem} GB, required amount is {required_total_memory}."
if total_mem < recommended_total_memory:
self.recommended_resources_are_present = False
if len(self.recommended_resource_report_string) == 0:
self.recommended_resource_report_string = self.recommended_resource_report_header
self.recommended_resource_report_string += f"\n{self.report_indentation} Total memory is {total_mem} GB, recommended amount is {recommended_total_memory}."

# deprecated method, will be removed someday, simply calls newer method
def require_free_disk_space_gb(self, path_of_interest, minimum_free_disk_space):
self.free_disk_space_needs(path_of_interest, required_free_disk_space=minimum_free_disk_space)

# method to specify the free disk space that is needed, in units of GB
def free_disk_space_needs(self, path_of_interest, required_free_disk_space=-1, recommended_free_disk_space=-1):
disk_space = shutil.disk_usage(path_of_interest)
self.free_disk_space_gb = disk_space.free / (1024 * 1024 * 1024)
self.debug_string += f"\nDEBUG: Free disk space on \"{path_of_interest}\" is {self.free_disk_space_gb} GB, minimum required amount is {minimum_free_disk_space}."
if self.free_disk_space_gb < minimum_free_disk_space:
self.debug_string += f"\nDEBUG: Free disk space on \"{path_of_interest}\" is {self.free_disk_space_gb} GB"
if required_free_disk_space >= 0:
self.debug_string += f", required amount is {required_free_disk_space}"
if recommended_free_disk_space >= 0:
self.debug_string += f", recommended amount is {recommended_free_disk_space}"
if self.free_disk_space_gb < required_free_disk_space:
self.this_computer_has_sufficient_resources = False
if len(self.report_string) == 0:
self.report_string = self.report_header
self.report_string += f"\n{self.report_indentation} Free disk space on \"{path_of_interest}\" is {self.free_disk_space_gb} GB, minimum amount is {minimum_free_disk_space}."
self.required_resources_are_present = False
if len(self.required_resource_report_string) == 0:
self.required_resource_report_string = self.required_resource_report_header
self.required_resource_report_string += f"\n{self.report_indentation} Free disk space on \"{path_of_interest}\" is {self.free_disk_space_gb} GB, required amount is {required_free_disk_space}."
if self.free_disk_space_gb < recommended_free_disk_space:
self.recommended_resources_are_present = False
if len(self.recommended_resource_report_string) == 0:
self.recommended_resource_report_string = self.recommended_resource_report_header
self.recommended_resource_report_string += f"\n{self.report_indentation} Free disk space on \"{path_of_interest}\" is {self.free_disk_space_gb} GB, recommended amount is {recommended_free_disk_space}."

# deprecated method, will be removed someday, simply calls newer method
def require_total_disk_space_gb(self, path_of_interest, minimum_total_disk_space):
self.total_disk_space_needs(path_of_interest, required_total_disk_space=minimum_total_disk_space)

# method to specify the total disk space that is needed, in units of GB
def total_disk_space_needs(self, path_of_interest, required_total_disk_space=-1, recommended_total_disk_space=-1):
disk_space = shutil.disk_usage(path_of_interest)
total_disk_space = disk_space.total / (1024 * 1024 * 1024)
self.debug_string += f"\nDEBUG: Total disk space on \"{path_of_interest}\" is {total_disk_space} GB, minimum required amount is {minimum_total_disk_space}."
if total_disk_space < minimum_total_disk_space:
total_disk_space_gb = disk_space.total / (1024 * 1024 * 1024)
self.debug_string += f"\nDEBUG: Total disk space on \"{path_of_interest}\" is {total_disk_space_gb} GB"
if required_total_disk_space >= 0:
self.debug_string += f", required amount is {required_total_disk_space}"
if recommended_total_disk_space >= 0:
self.debug_string += f", recommended amount is {recommended_total_disk_space}"
if total_disk_space_gb < required_total_disk_space:
self.this_computer_has_sufficient_resources = False
if len(self.report_string) == 0:
self.report_string = self.report_header
self.report_string += f"\n{self.report_indentation} Total disk space on \"{path_of_interest}\" is {total_disk_space} GB, minimum amount is {minimum_total_disk_space}."
self.required_resources_are_present = False
if len(self.required_resource_report_string) == 0:
self.required_resource_report_string = self.required_resource_report_header
self.required_resource_report_string += f"\n{self.report_indentation} Total disk space on \"{path_of_interest}\" is {total_disk_space_gb} GB, required amount is {required_total_disk_space}."
if total_disk_space_gb < recommended_total_disk_space:
self.recommended_resources_are_present = False
if len(self.recommended_resource_report_string) == 0:
self.recommended_resource_report_string = self.recommended_resource_report_header
self.recommended_resource_report_string += f"\n{self.report_indentation} Total disk space on \"{path_of_interest}\" is {total_disk_space_gb} GB, recommended amount is {recommended_total_disk_space}."

def get_debug_string(self):
return self.debug_string

def get_insufficient_resources_report(self):
return self.report_string
return self.required_resource_report_string

def get_insufficient_resources_summary(self):
return self.report_header
return self.required_resource_report_header

def get_required_resources_report(self):
return self.required_resource_report_string

def get_required_resources_summary(self):
return self.required_resource_report_header

def get_recommended_resources_report(self):
return self.recommended_resource_report_string

def get_recommended_resources_summary(self):
return self.recommended_resource_report_header