mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-11-01 10:22:08 +00:00
169 lines
8.6 KiB
Python
169 lines
8.6 KiB
Python
"""Real native OCI menu/CLI paths with all external effects replaced."""
|
|
from contextlib import redirect_stdout
|
|
import copy
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
sys.path[:0] = [str(ROOT / 'oci/src'), str(ROOT / 'oci/remote')]
|
|
from proxmenux_oci import cli, i18n, management as menu, ui as native_ui, updates
|
|
import oci_instances as instances
|
|
import oci_instance_reconcile as reconcile
|
|
|
|
|
|
class CancellationFixture(unittest.TestCase):
|
|
def setUp(self):
|
|
self.record = {'status': 'installed', 'vmid': 110, 'deployment': {}, 'template': {}}
|
|
self.original = copy.deepcopy(self.record)
|
|
self.row = dict(vmid=110, reason='matched', stack=False, pending=False, status='installed',
|
|
title='Fixture application', hostname='fixture', image='fixture:latest')
|
|
self.ui = Mock()
|
|
self.ui.unattended = False
|
|
self.ui.review.return_value = False
|
|
for module, name, options in [
|
|
(i18n, '_language', dict(new='en')), (i18n, '_cache', dict(new={})),
|
|
(menu.os, 'geteuid', dict(return_value=0)), (menu.shutil, 'which', dict(return_value='/fixture/pct')),
|
|
(native_ui, 'interactive_ui', dict(return_value=self.ui)),
|
|
(menu, 'recover_automatically', dict(return_value=None)),
|
|
(menu, 'carry_records', dict(return_value={})),
|
|
(menu, 'check_selected', dict(side_effect=lambda project, row: row)),
|
|
(menu, 'saved_inventory', dict(return_value=[self.row])),
|
|
(menu, '_selected_configuration', dict(return_value=b'fixture configuration')),
|
|
(instances, 'read', dict(return_value=self.record)),
|
|
(reconcile, 'propose', dict(return_value=None)),
|
|
]:
|
|
item = patch.object(module, name, **options)
|
|
item.start()
|
|
self.addCleanup(item.stop)
|
|
self.lifecycle = self.guard(menu, '_run_lifecycle')
|
|
self.backup = self.guard(menu, '_work_backup_ready')
|
|
self.write = self.guard(instances, 'write')
|
|
self.command = self.guard(instances, 'command')
|
|
self.commit = self.guard(reconcile, 'commit')
|
|
|
|
def guard(self, module, name):
|
|
item = patch.object(module, name, side_effect=AssertionError('Forbidden effect: ' + name))
|
|
mock = item.start()
|
|
self.addCleanup(item.stop)
|
|
return mock
|
|
|
|
def no_effects(self):
|
|
for mock in [self.lifecycle, self.backup, self.write, self.command, self.commit]:
|
|
mock.assert_not_called()
|
|
self.assertEqual(self.record, self.original)
|
|
|
|
|
|
class UpdateCancellationTests(CancellationFixture):
|
|
def test_no_is_a_distinct_falsy_result_before_backup_or_lifecycle(self):
|
|
result = menu.manage_instance(ROOT / 'oci', self.ui, self.row, action='update')
|
|
self.assertIs(result, menu.CANCELLED)
|
|
self.assertFalse(result)
|
|
self.no_effects()
|
|
|
|
def test_real_cli_no_returns_cancel_code_instead_of_failure_or_success(self):
|
|
with patch.object(cli, 'Catalog'), redirect_stdout(io.StringIO()) as output:
|
|
result = cli.main(['manage', '110', '--action', 'update'])
|
|
self.assertEqual(result, 6)
|
|
self.assertEqual(output.getvalue(), '') # The terminal renders the localized outcome once.
|
|
self.no_effects()
|
|
|
|
def test_no_in_stack_confirmation_keeps_every_member_unchanged(self):
|
|
self.row['stack'] = True
|
|
self.record['stack'] = {'members': [{'vmid': 110}, {'vmid': 111}]}
|
|
self.original = copy.deepcopy(self.record)
|
|
with redirect_stdout(io.StringIO()):
|
|
result = menu.direct_management(ROOT / 'oci', 110, 'update')
|
|
self.assertEqual(result, 6)
|
|
self.no_effects()
|
|
|
|
def test_native_menu_returns_to_its_list_after_no(self):
|
|
self.ui.choose.side_effect = ['110', 'update', None]
|
|
with patch.object(menu, '_clean_orphans'), patch.object(menu, 'offer_recovery'), \
|
|
patch.object(menu, '_records_unread', return_value=[]):
|
|
menu.interactive_management(ROOT / 'oci', self.ui)
|
|
self.assertEqual(self.ui.choose.call_count, 3)
|
|
self.no_effects()
|
|
|
|
def test_batch_no_does_not_start_the_next_application(self):
|
|
available = [{'vmid': 110, 'title': 'First'}, {'vmid': 111, 'title': 'Second'}]
|
|
rows = [self.row, {**self.row, 'vmid': 111}]
|
|
self.ui.checklist.return_value = ['all']
|
|
self.ui.review.side_effect = [True, False] # Batch review, first application's confirmation.
|
|
with patch.object(updates, 'UpdateCheck') as checker, patch.object(menu, 'saved_inventory', return_value=rows), \
|
|
patch.object(menu, 'check_selected', wraps=lambda project, row: row) as selected:
|
|
checker.return_value.status.return_value = {'available': available}
|
|
self.assertFalse(updates.select_updates(ROOT / 'oci', self.ui, available))
|
|
self.assertEqual([call.args[1]['vmid'] for call in selected.call_args_list], [110])
|
|
self.no_effects()
|
|
|
|
def test_accepted_update_failure_keeps_failure_code_one(self):
|
|
self.ui.review.return_value = True
|
|
self.backup.side_effect = None
|
|
self.backup.return_value = False
|
|
self.assertEqual(menu.direct_management(ROOT / 'oci', 110, 'update'), 1)
|
|
self.lifecycle.assert_not_called()
|
|
self.write.assert_not_called()
|
|
|
|
def test_accepted_update_preserves_success_and_existing_lifecycle_command(self):
|
|
self.ui.review.return_value = True
|
|
self.backup.side_effect = None
|
|
self.backup.return_value = True
|
|
self.lifecycle.side_effect = None
|
|
self.lifecycle.return_value = True
|
|
self.assertEqual(menu.direct_management(ROOT / 'oci', 110, 'update'), 0)
|
|
self.lifecycle.assert_called_once()
|
|
command = self.lifecycle.call_args.args[0]
|
|
self.assertEqual(command[:3], [sys.executable, str(ROOT / 'oci/remote/oci_update_current.py'), '110'])
|
|
self.assertIn('--acknowledge-external-data', command)
|
|
self.write.assert_not_called()
|
|
|
|
def test_actual_lifecycle_failure_remains_failure_after_positive_confirmation(self):
|
|
self.ui.review.return_value = True
|
|
self.backup.side_effect = None
|
|
self.backup.return_value = True
|
|
self.lifecycle.side_effect = None
|
|
self.lifecycle.return_value = False
|
|
self.assertEqual(menu.direct_management(ROOT / 'oci', 110, 'update'), 1)
|
|
self.lifecycle.assert_called_once()
|
|
self.write.assert_not_called()
|
|
|
|
def test_unattended_positive_default_still_reaches_existing_update(self):
|
|
self.backup.side_effect = None
|
|
self.backup.return_value = True
|
|
self.lifecycle.side_effect = None
|
|
self.lifecycle.return_value = True
|
|
self.assertEqual(menu.direct_management(ROOT / 'oci', 110, 'update', unattended=True), 0)
|
|
self.lifecycle.assert_called_once()
|
|
self.write.assert_not_called()
|
|
|
|
def test_existing_unattended_review_age_busy_and_missing_codes_are_preserved(self):
|
|
with patch.object(reconcile, 'propose', return_value={'details': ['external change']}), redirect_stdout(io.StringIO()):
|
|
self.assertEqual(menu.direct_management(ROOT / 'oci', 110, 'update', unattended=True), 4)
|
|
with patch.object(menu, '_image_too_recent', return_value=True), redirect_stdout(io.StringIO()):
|
|
self.assertEqual(menu.direct_management(ROOT / 'oci', 110, 'update', unattended=True, min_image_age_days=3), 5)
|
|
with patch.object(menu, 'saved_inventory', return_value=[]):
|
|
self.assertEqual(menu.direct_management(ROOT / 'oci', 110, 'update'), 2)
|
|
with patch.object(menu, 'recover_automatically', side_effect=BlockingIOError()):
|
|
self.assertEqual(menu.direct_management(ROOT / 'oci', 110, 'update'), 3)
|
|
self.no_effects()
|
|
|
|
def test_slovak_dialog_and_recovery_choice_have_the_approved_meaning(self):
|
|
cache = json.loads((ROOT / 'lang/sk.json').read_text())
|
|
with patch.object(i18n, '_language', 'sk'), patch.object(i18n, '_cache', cache):
|
|
menu.manage_instance(ROOT / 'oci', self.ui, self.row, action='update')
|
|
self.assertEqual(i18n.translate('Recover the previous installation'), 'Obnoviť predchádzajúcu inštaláciu')
|
|
text = self.ui.review.call_args.args[0]
|
|
for part in ['novší obraz aplikácie', 'overí sa jej integrita', 'pokúsi obnoviť', 'Pripojené adresáre hostiteľa']:
|
|
self.assertIn(part, text)
|
|
self.assertNotIn('kanál obrazov', text)
|
|
self.assertEqual(self.ui.review.call_args.kwargs['question'], 'Spustiť aktualizáciu?')
|
|
self.no_effects()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|