David Schroeder
125 lines
6.0 KiB
Python
125 lines
6.0 KiB
Python
#!/usr/bin/env python3
|
|
import contextlib
|
|
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_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()
|