orphan disk search update

This commit is contained in:
David Schroeder
2026-09-22 22:35:01 -05:00
parent e34f9d832e
commit e262260358
4 changed files with 53 additions and 6 deletions
+4
View File
@@ -93,6 +93,10 @@ Changes to guest configurations during the scan require a rerun. Scan when migra
replication, and disk operations are idle because the inventory is not atomic.
CSV export contains disk rows only, omits coverage warnings, and refuses to
overwrite an existing file.
Export paths without a `.csv` extension receive it automatically. Folder paths
use `tapm-orphan-disks-YYYY-MM-DD-HHMMSS.csv` (local host time), with a numbered
suffix if that generated name already exists. A trailing `/` marks a folder
path and creates it if needed. The final saved path is displayed after export.
Exit status is 0 for a completed scan, 2 for incomplete coverage or changed
configurations, and 1 for a fatal scan/export error.
+1 -1
View File
@@ -3,7 +3,7 @@
action="${1:-}"
FOLDER='/opt/idssys/ta-proxmenu'
VERS='2026.9.22-10'
VERS='2026.9.22-11'
noupdate=' '
+25 -5
View File
@@ -2,6 +2,7 @@
"""Read-only Proxmox storage/reference inventory; no third-party dependencies."""
import argparse
import csv
from datetime import datetime
import json
import math
from pathlib import Path
@@ -198,17 +199,36 @@ def render(rows, warnings, coverage):
def export_csv(path, rows):
destination = Path(path).expanduser()
folder = destination.is_dir() or str(path).endswith('/')
if folder:
destination.mkdir(parents=True, exist_ok=True)
destination /= datetime.now().strftime('tapm-orphan-disks-%Y-%m-%d-%H%M%S.csv')
elif destination.suffix.lower() != '.csv':
destination = Path(str(destination) + '.csv')
# Exclusive creation avoids overwriting existing files, including symlink targets.
with open(path, 'x', newline='') as handle:
base = destination
suffix = 0
while True:
try:
handle = destination.open('x', newline='')
break
except FileExistsError:
if not folder:
raise
suffix += 1
destination = base.with_name(f'{base.stem}-{suffix}.csv')
with handle:
writer = csv.DictWriter(handle, fieldnames=FIELDS + ['disk_size'])
writer.writeheader()
for row in rows:
writer.writerow(dict(row, disk_size=format_size(row['size_bytes'], row.get('size_is_approximate', False))))
return destination
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--csv', metavar='PATH', help='export to a new CSV file')
parser.add_argument('--csv', metavar='PATH', help='export to a CSV file or folder (.csv is added if missing)')
parser.add_argument('--interactive', action='store_true', help='offer CSV export after the report')
# The TAPM preloader pads its forwarded arguments with empty strings.
args = parser.parse_args([arg for arg in sys.argv[1:] if arg])
@@ -217,10 +237,10 @@ def main():
render(rows, warnings, coverage)
destination = args.csv
if args.interactive and sys.stdin.isatty() and not destination:
destination = input('\nCSV export path (Enter to skip): ').strip()
destination = input('\nCSV export file or folder (Enter to skip): ').strip()
if destination:
export_csv(destination, rows)
print('CSV saved: ' + destination)
destination = export_csv(destination, rows)
print(f'CSV saved: {destination}')
return 2 if warnings else 0
except (OSError, RuntimeError, ValueError, subprocess.TimeoutExpired) as exc:
print(f'Disk report failed: {exc}', file=sys.stderr)
+23
View File
@@ -90,6 +90,29 @@ class ReportTests(unittest.TestCase):
with self.assertRaises(FileExistsError):
disks.export_csv(path, [])
def test_csv_filename_extensions(self):
for name, expected in [('report', 'report.csv'), ('uppercase.CSV', 'uppercase.CSV'),
('report.txt', 'report.txt.csv')]:
with self.subTest(name=name):
saved = disks.export_csv(self.root / name, [])
self.assertEqual(saved, self.root / expected)
self.assertTrue(saved.is_file())
def test_csv_folder_default_and_collisions(self):
from datetime import datetime
from unittest.mock import patch
with patch.object(disks, 'datetime') as clock:
clock.now.return_value = datetime(2026, 9, 22, 14, 30, 59)
first = disks.export_csv(self.root, [])
original = first.read_text()
second = disks.export_csv(str(self.root) + '/', [])
self.assertEqual(first.name, 'tapm-orphan-disks-2026-09-22-143059.csv')
self.assertEqual(second.name, 'tapm-orphan-disks-2026-09-22-143059-1.csv')
self.assertEqual(first.read_text(), original)
nested = disks.export_csv(str(self.root / 'new-folder') + '/', [])
self.assertTrue(nested.is_file())
self.assertEqual(nested.parent, self.root / 'new-folder')
def test_disk_sizes_in_display_and_csv(self):
for index, (size, expected) in enumerate([
(4 * 1024 ** 2, '4.00 MiB'),