"""WizardKit: Volume functions""" # vim: sts=2 sw=2 ts=2 import logging from wk import os as wk_os from wk.cfg.hw import VOLUME_FAILURE_THRESHOLD from wk.std import PLATFORM, bytes_to_string # STATIC VARIABLES LOG = logging.getLogger(__name__) DEV_INFO = '{dev}... {size}... {filesystem}' MOUNT_INFO = '... Mounted on {mountpoint}... ({used} used, {free} free) {msg}' # Functions def add_child_dev_line(test_obj, child) -> None: """Add child device line to test report.""" dev_info = DEV_INFO.format( dev=child['name'], size=bytes_to_string(child['size']), filesystem=child['fstype'], ) vol_full = False # Bail early if not child['mountpoint']: test_obj.report.append(dev_info) return # Get sizes percent_used = -1 size = child['size'] try: if PLATFORM == 'Darwin': size = child['TotalSize'] free = child['FreeSpace'] used = size - free elif PLATFORM == 'Linux': free = int(child['fsavail']) used = int(child['fsused']) percent_used = (used / size) * 100 except TypeError: # Bail early test_obj.report.append(f'{dev_info}... Mounted on {child["mountpoint"]}') return # Check for failures if percent_used >= VOLUME_FAILURE_THRESHOLD: test_obj.failed = True vol_full = True # Mount / Size info mount_size_info = MOUNT_INFO.format( mountpoint=child['mountpoint'], used=bytes_to_string(used), free=bytes_to_string(free), msg=f'[{percent_used:0.0f}% full]' if vol_full else '', ) # Done test_obj.report.append(f'{dev_info}{mount_size_info}') def add_root_dev_line(test_obj) -> None: """Add root device line to test report.""" dev = test_obj.dev dev_info= DEV_INFO.format( dev=dev.name, size=bytes_to_string(dev.size), filesystem=dev.filesystem, ) vol_full = False # Get sizes percent_used = 0 size = dev.size try: if PLATFORM == 'Darwin': size = dev.raw_details['TotalSize'] free = dev.raw_details['FreeSpace'] used = size - free elif PLATFORM == 'Linux': free = int(dev.raw_details['fsavail']) used = int(dev.raw_details['fsused']) percent_used = (used / size) * 100 except TypeError: # Bail early test_obj.report.append( f'{dev_info}... Mounted on {dev.raw_details["mountpoint"]}', ) return # Check for failures if percent_used >= VOLUME_FAILURE_THRESHOLD: test_obj.failed = True vol_full = True # Mount / Size info mount_size_info = MOUNT_INFO.format( mountpoint=dev.raw_details['mountpoint'], used=bytes_to_string(used), free=bytes_to_string(free), msg=f'[{percent_used:0.0f}% full]' if vol_full else '', ) # Done test_obj.report.append(f'{dev_info}{mount_size_info}') def check_volume_utilization(test_obj) -> None: """Check volume utilization using OS specific methods.""" dev = test_obj.dev # Mount all volumes (read only if possible) mount_all_volumes(test_obj.dev) # Add root dev to report add_root_dev_line(test_obj) # Add children dev.update_details(skip_children=False) for child in dev.children: add_child_dev_line(test_obj, child) # Update test object if test_obj.failed: test_obj.dev.add_note('Full volume(s) detected', color='YELLOW') test_obj.passed = False test_obj.set_status('Failed') else: test_obj.passed = True test_obj.set_status('Passed') def mount_all_volumes(dev) -> None: """Mount all volumes for dev using.""" if PLATFORM == 'Darwin': wk_os.mac.mount_disk(device_path=dev.path) elif PLATFORM == 'Linux': wk_os.linux.mount_volumes( device_path=dev.path, read_write=False, scan_corestorage=not any(t.failed for t in dev.tests), ) if __name__ == '__main__': print("This file is not meant to be called directly.")