#!/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_preserves_warnings_and_refuses_overwrite(self): path = self.root / 'report.csv' disks.export_csv(path, [], ['node offline']) self.assertIn('node offline', path.read_text()) with self.assertRaises(FileExistsError): disks.export_csv(path, [], []) 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, warnings) 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.assertEqual(exported[0]['size_is_approximate'], 'True') 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()