David Schroeder
204 lines
10 KiB
Python
204 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
import contextlib
|
|
import csv
|
|
import importlib.util
|
|
import io
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
|
|
spec = importlib.util.spec_from_file_location('disks', Path(__file__).resolve().parents[1] / 'inc/orphan-disks.py')
|
|
disks = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(disks)
|
|
|
|
|
|
class ReportTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
for node in ('a', 'b'):
|
|
for kind in ('qemu-server', 'lxc'):
|
|
(self.root / node / kind).mkdir(parents=True)
|
|
(self.root / 'a/qemu-server/100.conf').write_text(
|
|
'name: app-server\n'
|
|
'scsi0: local:vm-100-disk-0,size=10G\n'
|
|
'unused0: local:vm-100-disk-1\n'
|
|
'efidisk0: shared:vm-100-efi\n'
|
|
'tpmstate0: shared:vm-100-tpm\n'
|
|
'[snap]\nscsi0: local:vm-100-old\n'
|
|
'[PENDING]\nscsi1: local:vm-100-pending\n')
|
|
(self.root / 'b/lxc/200.conf').write_text('hostname: fileserver\nrootfs: local:subvol-200-disk-0\nmp0: shared:subvol-200-disk-1,mp=/data\n')
|
|
self.nodes = [{'node': n, 'status': 'online'} for n in ('a', 'b')]
|
|
self.guests = [{'node': 'a', 'type': 'qemu', 'vmid': 100}, {'node': 'b', 'type': 'lxc', 'vmid': 200}]
|
|
self.refs, self.errors, _ = disks.references(self.root, self.nodes, self.guests)
|
|
|
|
def result(self, vol, node='a', shared=False, incomplete=False):
|
|
return disks.classify({'volid': vol}, node, shared, self.refs, incomplete)
|
|
|
|
def test_current_snapshots_pending_and_containers_are_protected(self):
|
|
self.assertEqual(self.errors, [])
|
|
for vol in ('vm-100-disk-0', 'vm-100-old', 'vm-100-pending'):
|
|
self.assertIsNone(self.result('local:' + vol))
|
|
for vol in ('vm-100-efi', 'vm-100-tpm', 'subvol-200-disk-1'):
|
|
self.assertIsNone(self.result('shared:' + vol, shared=True))
|
|
self.assertIsNone(self.result('local:subvol-200-disk-0', node='b'))
|
|
|
|
def test_candidates_unused_replication_and_backing(self):
|
|
self.assertEqual(self.result('local:vm-999-disk-0')[0], 'Orphan candidate')
|
|
self.assertEqual(self.result('local:vm-100-disk-1'), ('Unused disk', 'app-server (VM 100)'))
|
|
self.assertIn('a/app-server (VM 100)', self.result('local:vm-100-disk-0', node='b')[1])
|
|
self.assertIn('b/fileserver (LXC 200)', self.result('local:subvol-200-disk-0')[1])
|
|
self.assertEqual(self.result('local:vm-100-disk-0', node='b')[0], 'Needs review')
|
|
self.assertEqual(self.result('local:base-900-disk-0')[0], 'Needs review')
|
|
self.assertEqual(self.result('local:999/base-999-disk-0.qcow2')[0], 'Needs review')
|
|
self.assertEqual(self.result('local:vm-999-disk-0', incomplete=True)[0], 'Needs review')
|
|
|
|
def test_missing_config_is_incomplete(self):
|
|
(self.root / 'a/qemu-server/100.conf').unlink()
|
|
_, errors, _ = disks.references(self.root, self.nodes, self.guests)
|
|
self.assertTrue(errors)
|
|
|
|
def test_scan_deduplicates_shared_but_not_local_and_reports_failures(self):
|
|
calls = []
|
|
def get(path, *options):
|
|
calls.append(path)
|
|
if path == '/nodes':
|
|
return self.nodes
|
|
if path == '/cluster/resources':
|
|
return self.guests
|
|
if path.endswith('/storage'):
|
|
return [{'storage': sid, 'shared': shared, 'active': 1, 'content': 'images,rootdir'}
|
|
for sid, shared in [('shared', 1), ('local', 0), ('broken', 0)]]
|
|
if '/broken/' in path:
|
|
raise RuntimeError('unavailable')
|
|
sid = path.split('/')[-2]
|
|
return [{'volid': sid + ':vm-999-disk-0', 'content': 'images', 'size': 1024},
|
|
{'volid': sid + ':iso/test.iso', 'content': 'iso'}]
|
|
with contextlib.redirect_stderr(io.StringIO()):
|
|
rows, warnings, coverage = disks.scan(self.root, get)
|
|
self.assertEqual(len(rows), 3)
|
|
self.assertEqual(len(warnings), 2)
|
|
self.assertEqual(len(coverage), 3)
|
|
self.assertNotIn('/nodes/b/storage/shared/content', calls)
|
|
|
|
def test_csv_omits_warnings_and_refuses_overwrite(self):
|
|
path = self.root / 'report.csv'
|
|
disks.export_csv(path, [])
|
|
self.assertNotIn('coverage_warnings', path.read_text())
|
|
self.assertEqual(len(path.read_text().splitlines()), 1)
|
|
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'),
|
|
(100 * 1024 ** 3, '100.00 GiB'),
|
|
(2 * 1024 ** 4, '2.00 TiB'),
|
|
(0, '0.00 B'), (None, 'Unknown')]):
|
|
with self.subTest(size=size):
|
|
row = dict(zip(disks.FIELDS, ['a', 'local', 'local:vm-100-disk-1',
|
|
size, 'Unused disk', 'app-server (VM 100)']))
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output):
|
|
disks.render([row], [], [])
|
|
self.assertIn('Disk size: ' + expected, output.getvalue())
|
|
path = self.root / f'sizes-{index}.csv'
|
|
disks.export_csv(path, [row])
|
|
with path.open(newline='') as handle:
|
|
exported = next(csv.DictReader(handle))
|
|
self.assertEqual(exported['disk_size'], expected)
|
|
self.assertEqual(exported['size_bytes'], '' if size is None else str(size))
|
|
|
|
def test_approximate_storage_sizes_match_proxmox_display(self):
|
|
approximate = round(25.91 * 1024 ** 3)
|
|
volumes = [
|
|
{'volid': 'local:vm-100-disk-1', 'approximate-size': approximate},
|
|
{'volid': 'local:vm-999-exact', 'size': 30 * 1024 ** 3, 'approximate-size': approximate},
|
|
{'volid': 'local:vm-999-zero', 'size': 0, 'approximate-size': approximate},
|
|
{'volid': 'local:vm-999-null', 'size': None, 'approximate-size': approximate},
|
|
{'volid': 'local:vm-999-unknown'},
|
|
]
|
|
def get(path, *options):
|
|
if path == '/nodes':
|
|
return self.nodes
|
|
if path == '/cluster/resources':
|
|
return self.guests
|
|
if path.endswith('/storage'):
|
|
return [{'storage': 'local', 'content': 'images', 'active': 1}] if '/a/' in path else []
|
|
return [dict(volume, content='images') for volume in volumes]
|
|
with contextlib.redirect_stderr(io.StringIO()):
|
|
rows, warnings, coverage = disks.scan(self.root, get)
|
|
self.assertEqual(warnings, [])
|
|
self.assertEqual([row['size_is_approximate'] for row in rows], [True, False, False, True, False])
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output):
|
|
disks.render(rows, warnings, coverage)
|
|
self.assertIn('Disk size: ~25.91 GiB', output.getvalue())
|
|
path = self.root / 'approximate.csv'
|
|
disks.export_csv(path, rows)
|
|
with path.open(newline='') as handle:
|
|
exported = list(csv.DictReader(handle))
|
|
self.assertEqual([row['disk_size'] for row in exported],
|
|
['~25.91 GiB', '30.00 GiB', '0.00 B', '~25.91 GiB', 'Unknown'])
|
|
self.assertEqual(exported[0]['size_bytes'], str(approximate))
|
|
self.assertNotIn('size_is_approximate', exported[0])
|
|
|
|
def test_offline_node_and_config_change(self):
|
|
def get(path, *options):
|
|
if path == '/nodes':
|
|
return [self.nodes[0], {'node': 'b', 'status': 'offline'}]
|
|
if path == '/cluster/resources':
|
|
return self.guests
|
|
if path.endswith('/storage'):
|
|
return [{'storage': 'local', 'content': 'images', 'active': 1}]
|
|
(self.root / 'a/qemu-server/100.conf').write_text('scsi0: local:vm-999-disk-0\n')
|
|
return [{'volid': 'local:vm-999-disk-0', 'content': 'images'}]
|
|
with contextlib.redirect_stderr(io.StringIO()):
|
|
rows, warnings, _ = disks.scan(self.root, get)
|
|
self.assertEqual(rows[0]['status'], 'Needs review')
|
|
self.assertTrue(any('offline' in warning for warning in warnings))
|
|
self.assertTrue(any('changed' in warning for warning in warnings))
|
|
|
|
def test_launcher_padded_arguments(self):
|
|
from unittest.mock import patch
|
|
with patch.object(disks.sys, 'argv', ['orphan-disks.py', '', '', '']), \
|
|
patch.object(disks, 'scan', return_value=([], [], [])), \
|
|
contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertEqual(disks.main(), 0)
|
|
|
|
def test_api_is_read_only(self):
|
|
from unittest.mock import patch
|
|
with patch.object(disks.subprocess, 'run') as run:
|
|
run.return_value.returncode = 0
|
|
run.return_value.stdout = '[]'
|
|
disks.api('/nodes')
|
|
self.assertEqual(run.call_args.args[0], ['pvesh', 'get', '/nodes', '--output-format', 'json'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|