157 lines
4.1 KiB
Python
157 lines
4.1 KiB
Python
"""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, VOLUME_SIZE_THRESHOLD
|
|
from wk.std import PLATFORM, GenericError, 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'])
|
|
if free == 0:
|
|
raise GenericError('Most likely a false positive.')
|
|
percent_used = (used / size) * 100
|
|
except (GenericError, TypeError):
|
|
# Bail early
|
|
test_obj.report.append(f'{dev_info}... Mounted on {child["mountpoint"]}')
|
|
return
|
|
|
|
# Check for failures
|
|
if (percent_used >= VOLUME_FAILURE_THRESHOLD
|
|
and size >= VOLUME_SIZE_THRESHOLD * 1024**3):
|
|
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 = -1
|
|
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'])
|
|
if free == 0:
|
|
raise GenericError('Most likely a false positive.')
|
|
percent_used = (used / size) * 100
|
|
except (GenericError, 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':
|
|
# NOTE: Disabled due to a bug they should already be mounted by macOS
|
|
#wk_os.mac.mount_disk(device_path=dev.path)
|
|
pass
|
|
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.")
|