Mirrors/ProxMenux
1
0
Fork 0
mirror of https://github.com/MacRimi/ProxMenux.git synced 2026-11-01 18:32:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: clarify OCI update dialog and distinguish cancellation

Browse Source
This commit is contained in:
VAIO73
2026-10-21 07:04:41 +00:00
parent 43041846eb
commit 7213fb2553
13 changed files with 357 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/scripts/tests/test_oci_recovery_inventory_wording.py
+1 -1
View File
@@ -116,7 +116,7 @@ class RecoveryMessages(unittest.TestCase):
instances.ROOT = Path('/inert')
instances.read = lambda root, vmid: self.primary
self.modules = {'oci_instances': instances, 'oci_stack_replay': ModuleType('oci_stack_replay')}
self.scope = {'sys': SimpleNamespace(path=[], executable='python3'), 'Path': Path,
self.scope = {'sys': SimpleNamespace(path=[], executable='python3'), 'Path': Path, 'CANCELLED': None,
'translate': lambda text: text, 'images': SimpleNamespace(offer_removal=lambda *args: None),
'_remove': lambda *args: self.fail('remove must not run'),
'_run_lifecycle': lambda command, title: self.calls.append((command, title)) or False}
.github/scripts/tests/test_oci_update_cancellation.py
+166
View File
@@ -0,0 +1,166 @@
"""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 UpdateCancellationTests(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)
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()
.github/scripts/tests/test_oci_update_preservation_wording.py
+7 -4
View File
@@ -24,9 +24,12 @@ OLD_GUARD = 'The current template changes the image or identity; an explicit mig
GUARD = 'The current template changes the template identity or image repository; an explicit migration is required'
OLD_PREVIEW = ('The saved image channel is checked for a newer image. If replacement is needed, the CT is stopped and '
'a native backup is verified before its root is replaced. Host directories are outside that backup.')
PREVIEW = ('If the image channel has a new version, the CT is stopped, backed up and verified, and replaced with '
'the new image to update the container. If the update fails after the container was changed, the backup '
'is restored. An update that stops halfway is recovered from the OCI management menu.')
PREVIEW = ('If a newer application image is available, the container is stopped and backed up, and the backup '
'integrity is verified. The container is then updated using the new image.\n\n'
'If the update fails after the container was changed, ProxMenux attempts to restore the previous '
'installation from the backup. If the operation remains incomplete, select the application in OCI '
'management and choose "Recover the previous installation".\n\n'
'Host directories mounted in the container are not restored by the container backup.')
def extracted(path, name, scope):
@@ -130,7 +133,7 @@ class UpdateWording(unittest.TestCase):
reconcile = ModuleType('oci_instance_reconcile')
reconcile.propose = lambda *args: None
scope = {'sys': SimpleNamespace(path=[], executable='python3'),
'os': os, 'tempfile': tempfile, 'Path': Path,
'os': os, 'tempfile': tempfile, 'Path': Path, 'CANCELLED': None,
'translate': lambda text: text, 'check_selected': lambda project, row: row,
'_selected_configuration': lambda vmid: instances.command('pct', 'config', str(vmid)),
'images': SimpleNamespace(offer_removal=lambda *args: None),
.github/scripts/tests/test_outcome_promises_wording.py
+5 -1
View File
@@ -20,7 +20,11 @@ class OutcomePromises(TestCase):
texts = oci_menu_texts()
self.assertEqual([text for text in texts if 'If anything fails' in text], [])
previews = [text for text in texts if 'stops halfway is recovered from the OCI management menu' in text]
self.assertEqual(len(previews), 3)
self.assertEqual(len(previews), 2) # Coordinated stack update/recreation previews.
update = next(text for text in texts if text.startswith('If a newer application image is available'))
self.assertIn('attempts to restore the previous installation', update)
self.assertIn('If the operation remains incomplete', update)
self.assertIn('choose "Recover the previous installation"', update)
def test_the_recovery_preview_says_when_each_step_happens(self):
preview = next(text for text in oci_menu_texts() if 'previous native backup' in text)