"""Selective OCI edits stay in a draft until reviewed, then use one transaction.""" import copy import json from pathlib import Path import stat import sys import unittest from unittest.mock import Mock, patch ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / 'src')) sys.path.insert(0, str(ROOT / 'remote')) from proxmenux_oci import management, recreation from proxmenux_oci.ui import BackRequested, UserCancelled def record(): return {'vmid': 115, 'installation_id': 'test-installation', 'status': 'installed', 'observed': {'config_sha256': 'native-config-fingerprint'}, 'template': {'id': 'example-app', 'catalog_ui': {'title': 'Example'}, 'container_contract': {'image': {'reference': 'example/app:latest'}}, 'proxmox': {'installer_profile': {}}}, 'deployment': {'vmid': 115, 'hostname': 'example', 'resources': {'cores': 2, 'memory_mb': 512}, 'network': {'bridge': 'vmbr0', 'ipv4': 'dhcp', 'gateway': None}, 'rootfs': {'storage': 'local-zfs', 'size_gb': 8}, 'mounts': [ {'container_path': '/data', 'source': 'local-zfs', 'type': 'managed-volume', 'size_gb': 8, 'backup': True}], 'devices': [], 'security': {'unprivileged': True}, 'environment': [{'name': 'PASSWORD', 'value': 'private-test-secret', 'sensitive': True}]}} class MenuUI: def __init__(self, choices, answers=(), reviews=(True,), confirms=()): self.choices, self.answers = iter(choices), iter(answers) self.reviews, self.confirms = iter(reviews), iter(confirms) self.events = [] self.back_enabled = False def choose(self, text, options, default=None, **kwargs): self.events.append(('menu', text, dict(options))) value = next(self.choices) if value is not None: assert value in dict(options), value return value def ask(self, text, default=None, **kwargs): self.events.append(('ask', text, default)) value = next(self.answers) if isinstance(value, Exception): raise value return value def confirm(self, text, default=False): self.events.append(('confirm', text)) return next(self.confirms) def review(self, text, *args, **kwargs): self.events.append(('review', text)) return next(self.reviews) def message(self, text, *args): self.events.append(('message', text)) @patch('proxmenux_oci.i18n.language', return_value='en') class ModifyMenuTests(unittest.TestCase): def test_ram_only_does_not_ask_about_cpu_network_paths_devices_or_catalog(self, _language): original = record() saved = copy.deepcopy(original) ui = MenuUI(['memory', 'apply'], ['768']) with patch.object(recreation, 'refresh_template') as refresh: result = recreation.edit_recreation(original, ui) self.assertEqual(result['candidate']['deployment']['resources'], {'cores': 2, 'memory_mb': 768}) self.assertEqual([e[1] for e in ui.events if e[0] == 'ask'], ['RAM in MiB']) self.assertEqual(result['base_config_sha256'], 'native-config-fingerprint') self.assertEqual(original, saved) refresh.assert_not_called() self.assertFalse(ui.back_enabled) def test_several_sections_and_revisits_keep_the_current_draft(self, _language): ui = MenuUI(['memory', 'cores', 'memory', 'apply'], ['768', '3', '1024']) result = recreation.edit_recreation(record(), ui) self.assertEqual(result['candidate']['deployment']['resources'], {'cores': 3, 'memory_mb': 1024}) self.assertEqual([e[2] for e in ui.events if e[0] == 'ask'], ['512', '2', '768']) self.assertEqual(sum(e[0] == 'review' for e in ui.events), 1) def test_declining_review_returns_to_menu_without_losing_edits(self, _language): ui = MenuUI(['cores', 'apply', 'memory', 'apply'], ['3', '768'], reviews=[False, True]) result = recreation.edit_recreation(record(), ui) self.assertEqual(result['candidate']['deployment']['resources'], {'cores': 3, 'memory_mb': 768}) self.assertEqual(sum(e[0] == 'review' for e in ui.events), 2) def test_unchanged_apply_does_not_propose_recreating_the_guest(self, _language): ui = MenuUI(['memory', 'apply'], ['512']) self.assertIsNone(recreation.edit_recreation(record(), ui)) self.assertFalse(any(e[0] == 'review' for e in ui.events)) self.assertIn(('message', 'Nothing was changed.'), ui.events) def test_discard_leaves_the_saved_record_untouched(self, _language): original = record() saved = copy.deepcopy(original) ui = MenuUI(['memory', 'discard'], ['768'], confirms=[True]) self.assertIsNone(recreation.edit_recreation(original, ui)) self.assertEqual(original, saved) self.assertFalse(any(e[0] == 'review' for e in ui.events)) def test_declining_discard_preserves_the_draft(self, _language): ui = MenuUI(['memory', None, 'apply'], ['768'], confirms=[False]) result = recreation.edit_recreation(record(), ui) self.assertEqual(result['candidate']['deployment']['resources']['memory_mb'], 768) def test_cancel_without_changes_asks_nothing_else(self, _language): ui = MenuUI([None]) self.assertIsNone(recreation.edit_recreation(record(), ui)) self.assertEqual(len(ui.events), 1) def test_invalid_ram_is_retried_without_restarting_other_sections(self, _language): ui = MenuUI(['cores', 'memory', 'apply'], ['3', 'invalid', '64', '768']) result = recreation.edit_recreation(record(), ui) self.assertEqual(result['candidate']['deployment']['resources'], {'cores': 3, 'memory_mb': 768}) self.assertEqual(sum(e[0] == 'message' for e in ui.events), 2) self.assertEqual(sum(e[:2] == ('ask', 'Cores') for e in ui.events), 1) def test_back_from_first_field_returns_to_menu_and_keeps_other_edits(self, _language): ui = MenuUI(['cores', 'memory', 'apply'], ['3', BackRequested()]) result = recreation.edit_recreation(record(), ui) self.assertEqual(result['candidate']['deployment']['resources'], {'cores': 3, 'memory_mb': 512}) self.assertFalse(ui.back_enabled) def test_cancelling_a_partly_edited_section_rolls_back_only_that_section(self, _language): real_edit = recreation.edit_setting def edit(candidate, ui, setting): if setting == 'environment': candidate['deployment']['environment'][0]['value'] = 'partial-value' raise UserCancelled() real_edit(candidate, ui, setting) ui = MenuUI(['cores', 'environment', 'apply'], ['3']) with patch.object(recreation, 'edit_setting', side_effect=edit): result = recreation.edit_recreation(record(), ui) self.assertEqual(result['candidate']['deployment']['resources']['cores'], 3) self.assertEqual(result['candidate']['deployment']['environment'], record()['deployment']['environment']) def test_menu_and_review_do_not_expose_a_saved_secret(self, _language): ui = MenuUI(['cores', 'apply'], ['3']) recreation.edit_recreation(record(), ui) self.assertNotIn('private-test-secret', json.dumps(ui.events)) def test_extra_paths_counts_only_paths_added_by_hand(self, _language): candidate = record() self.assertTrue(dict(recreation.modification_options(candidate))['paths'].rstrip().endswith(' 0')) candidate['deployment']['mounts'].append({'container_path': '/media-extra', 'type': 'host-bind', 'source': '/mnt/media', 'custom': True}) self.assertTrue(dict(recreation.modification_options(candidate))['paths'].rstrip().endswith(' 1')) def test_optional_categories_follow_the_application_profile(self, _language): candidate = record() options = dict(recreation.modification_options(candidate)) self.assertTrue({'cores', 'memory', 'network', 'paths', 'devices', 'environment', 'catalog', 'apply'} <= options.keys()) self.assertFalse({'shares', 'acceleration', 'detector'} & options.keys()) candidate['template']['proxmox']['installer_profile'] = { 'shared_directories': {'base_path': '/shares'}, 'object_detector': {'devices': []}, 'hardware_acceleration': {'profiles': [{'id': 'intel', 'label': 'Intel GPU'}]}} candidate['deployment']['hardware_profile'] = 'intel' options = dict(recreation.modification_options(candidate)) self.assertTrue({'shares', 'acceleration', 'detector'} <= options.keys()) self.assertIn('Intel GPU', options['acceleration']) class ModifyDispatchTests(unittest.TestCase): def call_manager(self, proposal, lifecycle, ui=None, offer=None): import oci_instances import oci_instance_reconcile saved = record() row = {'vmid': 115, 'reason': 'matched', 'stack': False, 'pending': False, 'status': 'installed'} with patch.object(management.images, 'offer_removal', offer or Mock()), \ patch.object(management, 'carry_records', return_value={}), \ patch.object(management, 'check_selected', return_value=row), \ patch.object(management, '_work_backup_ready', return_value=True), \ patch.object(oci_instances, 'read', return_value=saved), \ patch.object(oci_instances, 'command', return_value='hostname: example'), \ patch.object(oci_instance_reconcile, 'propose', return_value=None), \ patch.object(recreation, 'edit_recreation', return_value=proposal), \ patch.object(management, '_run_lifecycle', side_effect=lifecycle): return management.manage_instance(ROOT, ui or Mock(), row, action='modify') def test_legacy_none_result_does_not_start_a_lifecycle_transaction(self): lifecycle = Mock() self.assertFalse(self.call_manager(None, lifecycle)) lifecycle.assert_not_called() def test_apply_sends_one_private_proposal_to_the_existing_backend(self): proposal = {'operation': 'recreate', 'candidate': record(), 'base_config_sha256': 'fingerprint'} seen = [] def lifecycle(command, title): path = Path(command[command.index('--proposal') + 1]) seen.append(json.loads(path.read_text())) self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600) self.assertNotIn('private-test-secret', ' '.join(command)) self.assertIn('oci_update_current.py', command[1]) self.assertIn('--acknowledge-external-data', command) return True self.assertTrue(self.call_manager(proposal, lifecycle)) self.assertEqual(seen, [proposal]) def test_the_image_question_only_follows_a_recreated_container(self): proposal = {'operation': 'recreate', 'candidate': record(), 'base_config_sha256': 'fingerprint'} ui = Mock(unattended=False) for recreated in (False, True): with self.subTest(recreated=recreated): def lifecycle(command, title): if recreated: Path(management.os.environ['OCI_RECREATED_MARKER']).touch() return True offer = Mock() self.assertTrue(self.call_manager(proposal, lifecycle, ui, offer)) self.assertEqual(offer.called, recreated) self.assertNotIn('OCI_RECREATED_MARKER', management.os.environ) if __name__ == '__main__': unittest.main()