Merge pull request #389 from f3rs3n/fix/oci-diagnostic-wording

Clarify OCI import, adoption, recovery and removal diagnostics

.github/scripts/tests/test_oci_recovery_inventory_wording.py

@@ -0,0 +1,214 @@
+"""Offline checks of real OCI messages; never import administrative modules."""
+import ast
+import importlib.util
+import json
+from pathlib import Path
+import re
+import runpy
+import sys
+import tempfile
+from types import ModuleType, SimpleNamespace
+import unittest
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[3]
+MENU = ROOT / 'oci/src/proxmenux_oci/management.py'
+REMOVE = ROOT / 'oci/remote/oci_remove.py'
+PREVIEW_HOST = 'Host paths found in container configs or saved records (not targeted for removal):'
+EMPTY_HOST = 'No host directories found in the available container configs or saved records.'
+POST_HOST = 'Host directory listed in saved records (not targeted for removal):'
+UPDATE = ('All images are downloaded and verified first, and native backups are taken with the stack stopped. '
+          'Contracts are published after the whole set is checked. If a step fails, recovery is attempted '
+          'where needed; recovery can also fail.')
+PENDING = ('A coordinated operation has a saved journal. Continuing attempts to recover the previous stack '
+           'where needed, or finish cleanup for a completed operation. Recovery or cleanup can fail.')
+KEYS = (PREVIEW_HOST, EMPTY_HOST, POST_HOST, UPDATE, PENDING)
+TRANSACTION = ROOT / 'oci/remote/oci_stack_transaction.py'
+
+
+def extracted(path, name, scope):
+    node = next(n for n in ast.parse(path.read_text()).body
+                if isinstance(n, ast.FunctionDef) and n.name == name)
+    exec(compile(ast.Module(body=[node], type_ignores=[]), str(path), 'exec'), scope)
+    return scope[name]
+
+
+class InventoryMessages(unittest.TestCase):
+    def preview(self, configs, records, cache=None, translate=lambda text: text):
+        instances = ModuleType('oci_instances')
+        instances.ROOT = Path('/inert')
+        def read(root, vmid):
+            if vmid not in records:
+                raise OSError('unreadable record')
+            return records[vmid]
+        instances.read = read
+        remover = ModuleType('oci_remove')
+        remover.members_of = lambda root, vmid: (101, {}, [102, 101])
+        remover.guest_config = lambda vmid: configs.get(vmid)
+        remover.host_directories = lambda root, members: cache or []
+        remover.private_bridge = lambda primary: None
+        state = ModuleType('oci_installation_state')
+        state.parse_config = lambda raw: dict(line.split(': ', 1) for line in raw.decode().splitlines())
+        scope = {'sys': SimpleNamespace(path=[]), 'source_text': lambda text: text or '',
+                 'translate': translate, 're': re}
+        summary = extracted(MENU, '_removal_summary', scope)
+        with patch.dict(sys.modules, {'oci_instances': instances, 'oci_remove': remover,
+                                      'oci_installation_state': state}):
+            return summary(Path('/inert'), 101)
+
+    def test_missing_metadata_is_not_reported_as_no_host_usage(self):
+        text = self.preview({101: None, 102: None}, {})
+        self.assertIn(EMPTY_HOST, text)
+        self.assertNotIn('No host directory is used by this application.', text)
+
+    def test_discovered_paths_do_not_promise_exhaustive_preservation(self):
+        text = self.preview({101: b'mp0: /bind/current,mp=/data\n', 102: None}, {}, ['/bind/saved'])
+        self.assertIn(PREVIEW_HOST + '\n  /bind/current\n  /bind/saved', text)
+        self.assertNotIn('Host directories that are kept, with their content:', text)
+
+    def test_removal_notice_qualifies_saved_record_source(self):
+        events = []
+        scope = {'members_of': lambda root, vmid: (101, {}, [101]),
+                 'instances': SimpleNamespace(read=lambda root, vmid: {'installation_id': 'owned'},
+                    identity=lambda raw: 'owned', location=lambda root, vmid: Path('/inert/absent/record.json')),
+                 'guest_config': lambda vmid: b'description: owned',
+                 'host_directories': lambda root, members: ['/bind/saved'],
+                 'private_bridge': lambda primary: None,
+                 'run': lambda *args: events.append(('run', args)),
+                 'subprocess': SimpleNamespace(run=lambda *args, **kwargs: None),
+                 'Path': Path, 'shutil': SimpleNamespace(rmtree=lambda path: None),
+                 'image_cache': SimpleNamespace(prune=lambda root, lock: []),
+                 'translate': lambda text: text,
+                 'msg_info': lambda text: events.append(('info', text)),
+                 'msg_ok': lambda text: events.append(('ok', text)),
+                 'msg_warn': lambda text: events.append(('warn', text))}
+        remove = extracted(REMOVE, 'remove', scope)
+        remove(Path('/inert'), 101)
+        self.assertIn(('warn', POST_HOST + ' /bind/saved'), events)
+        self.assertIn(('run', ('pct', 'destroy', '101', '--purge', '1', '--destroy-unreferenced-disks', '1')), events)
+        scope['translate'] = lambda text: 'Tradotto: ' + text if text == POST_HOST else text
+        events.clear()
+        remove(Path('/inert'), 101)
+        self.assertIn(('warn', 'Tradotto: ' + POST_HOST + ' /bind/saved'), events)
+
+
+class RecoveryMessages(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.journal = Path(self.tmp.name) / 'journal.json'
+        self.members = [{'vmid': 101, 'deployment': {'mounts': []}}]
+        self.primary = {'stack': {'members': self.members}}
+        self.calls = []
+        instances = ModuleType('oci_instances')
+        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,
+                      '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}
+        self.ui = SimpleNamespace(
+            choose=lambda *args: 'update',
+            review=lambda message, *args, **kwargs: self.calls.append(('review', message)) or False,
+            confirm=lambda *args: self.fail('host confirmation must not run'),
+            message=lambda *args: self.fail('unexpected error'))
+
+    def manage(self):
+        with patch.dict(sys.modules, self.modules):
+            extracted(MENU, '_manage_stack', self.scope)(Path('/inert'), self.ui, {'vmid': 101})
+        return next(message for kind, message in self.calls if kind == 'review')
+
+    def test_update_preview_describes_recovery_attempt_not_guarantee(self):
+        message = self.manage()
+        self.assertIn(UPDATE, message)
+        self.assertNotIn('all members are recovered', message)
+
+    def test_pending_terminal_and_recovery_failed_states_share_bounded_wording(self):
+        for phase in ('committed', 'rolled-back', 'recovery-failed'):
+            with self.subTest(phase=phase):
+                self.calls.clear()
+                self.journal.write_text(json.dumps({'phase': phase, 'plan': {'members': self.members}}))
+                self.primary['pending_stack_transaction'] = str(self.journal)
+                self.assertEqual(self.manage(), PENDING)
+                self.assertFalse(any(isinstance(item[0], list) for item in self.calls))
+
+    def test_pending_confirmation_preserves_recover_command_in_each_state(self):
+        self.ui.review = lambda message, *args, **kwargs: True
+        for phase in ('committed', 'recovery-failed'):
+            with self.subTest(phase=phase):
+                self.calls.clear()
+                self.journal.write_text(json.dumps({'phase': phase, 'plan': {'members': self.members}}))
+                self.primary['pending_stack_transaction'] = str(self.journal)
+                with patch.dict(sys.modules, self.modules):
+                    extracted(MENU, '_manage_stack', self.scope)(Path('/inert'), self.ui, {'vmid': 101})
+                self.assertEqual(self.calls, [
+                    (['python3', '/inert/remote/oci_stack_native.py', '101', '--recover',
+                      '--acknowledge-external-data'], 'Recover OCI stack')])
+
+    def test_actual_transaction_distinguishes_terminal_cleanup_from_failed_restore(self):
+        events = []
+        scope = {'TERMINAL': {'committed', 'rolled-back'},
+                 'translate': lambda message: message,
+                 'member_tx': SimpleNamespace(log=lambda message: None),
+                 'write': lambda path, state: Path(path).write_text(json.dumps(state)),
+                 'msg_info': lambda message: None, 'msg_ok': lambda message: None,
+                 'member': lambda adapter, vmid: f'CT {vmid}'}
+        extracted(TRANSACTION, 'save', scope)
+        recover = extracted(TRANSACTION, 'recover_state', scope)
+        class Adapter:
+            def finalize(self, state): events.append('finalize')
+            def validate(self, plan): events.append('validate')
+            def restore(self, vmid, backup, txid):
+                events.append('restore')
+                raise RuntimeError('restore failed')
+            def stop(self, vmid): events.append('stop')
+        plan = {'start_order': [101], 'stop_order': [101]}
+        terminal = {'phase': 'committed', 'plan': plan}
+        self.assertIs(recover(self.journal, terminal, Adapter()), terminal)
+        self.assertEqual(events, ['finalize'])
+        events.clear()
+        failed = dict(terminal, phase='recovery-failed', id='tx', running={'101': True},
+                      stop_intent=True, replacement_intent=True, backups={'101': {'archive': 'a'}})
+        with self.assertRaisesRegex(RuntimeError, 'restore failed'):
+            recover(self.journal, failed, Adapter())
+        self.assertEqual(events, ['validate', 'stop', 'restore'])
+        self.assertEqual(json.loads(self.journal.read_text())['phase'], 'recovery-failed')
+
+    def test_new_keys_extract_and_shipped_fallback_and_synthetic_translation(self):
+        generator = runpy.run_path(str(ROOT / '.github/scripts/build_translation_cache.py'))
+        found = generator['extract_python_texts']([ROOT / 'oci/src', ROOT / 'oci/remote'])
+        self.assertEqual(len(KEYS), len(set(KEYS)))
+        self.assertTrue(set(KEYS) <= set(found), set(KEYS) - set(found))
+        spec = importlib.util.spec_from_file_location('oci_i18n', ROOT / 'oci/src/proxmenux_oci/i18n.py')
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        paths = sorted((ROOT / 'lang').glob('*.json'))
+        self.assertEqual(len(paths), 7)
+        for path in paths:
+            cache = json.loads(path.read_text())
+            setattr(module, '_language', path.stem)
+            setattr(module, '_cache', cache)
+            for key in KEYS:
+                self.assertEqual(module.translate(key), cache.get(key) or key, (path.name, key))
+            setattr(module, '_cache', {})
+            for key in KEYS:
+                self.assertEqual(module.translate(key), key, (path.name, key, 'synthetic missing'))
+        module._language, module._cache = 'it', {key: 'Tradotto: ' + key for key in KEYS}
+        self.assertEqual(module.translate(EMPTY_HOST), 'Tradotto: ' + EMPTY_HOST)
+        preview = InventoryMessages().preview({101: None, 102: None}, {}, translate=module.translate)
+        self.assertIn('Tradotto: ' + EMPTY_HOST, preview)
+        preview = InventoryMessages().preview({101: b'mp0: /bind,mp=/data'}, {},
+                                              translate=module.translate)
+        self.assertIn('Tradotto: ' + PREVIEW_HOST + '\n  /bind', preview)
+        # Exercise the actual management consumer, not just the provider.
+        self.scope['translate'] = module.translate
+        self.assertIn('Tradotto: ' + UPDATE, self.manage())
+        self.journal.write_text(json.dumps({'phase': 'committed', 'plan': {'members': self.members}}))
+        self.primary['pending_stack_transaction'] = str(self.journal)
+        self.calls.clear()
+        self.assertEqual(self.manage(), 'Tradotto: ' + PENDING)
+
+
+if __name__ == '__main__':
+    unittest.main()

.github/scripts/tests/test_oci_security_mount_wording.py

@@ -0,0 +1,124 @@
+"""Exercise the real OCI preview/import functions without importing host-management modules."""
+import ast
+import json
+from pathlib import Path
+import runpy
+import unittest
+from typing import Any
+
+import yaml
+
+ROOT = Path(__file__).resolve().parents[3]
+CUSTOM = ROOT / 'oci/src/proxmenux_oci/custom.py'
+ROWS = {
+    'It needs a privileged container, which is not isolated from the host.':
+        'This profile requires a privileged LXC, which reduces isolation from the host.',
+    'It asks for capabilities or a relaxed confinement profile.':
+        'A relaxed AppArmor or seccomp profile is requested; this may be optional.',
+    'The image expects files that are given to it one by one:':
+        'These container mount paths have an extension-like suffix:',
+    'ProxMenux attaches directories, not single files, so this image cannot be installed yet.':
+        'This import rejects these paths without checking whether they are files or directories.',
+}
+
+
+class ConversionError(Exception):
+    pass
+
+
+def extracted(name, **deps):
+    node = next(n for n in ast.parse(CUSTOM.read_text()).body
+                if isinstance(n, ast.FunctionDef) and n.name == name)
+    scope = {'Any': Any, 'Path': Path, 'yaml': yaml, 'ConversionError': ConversionError,
+             'translate': lambda text: text}
+    scope.update(deps)
+    exec(compile(ast.Module(body=[node], type_ignores=[]), str(CUSTOM), 'exec'), scope)
+    return scope[name]
+
+
+class SecurityMountWording(unittest.TestCase):
+    def test_security_warnings_match_their_guards_without_claiming_capabilities(self):
+        describe = extracted('describe', blocker_text=str)
+        template = {'container_contract': {'image': {'reference': 'test:latest'},
+                    'volumes': [], 'ports': [], 'environment': []},
+                    'proxmox': {'security_profile': {}}}
+        for profile, present in (
+            ({}, ()),
+            ({'optional_privileged_lxc': True}, ()),
+            ({'requires_privileged_lxc': True}, (ROWS['It needs a privileged container, which is not isolated from the host.'],)),
+            ({'requires_relaxed_confinement': True}, (ROWS['It asks for capabilities or a relaxed confinement profile.'],)),
+            ({'source_requests_relaxed_confinement': True}, (ROWS['It asks for capabilities or a relaxed confinement profile.'],)),
+            ({'requires_privileged_lxc': True, 'source_requests_relaxed_confinement': True},
+             (ROWS['It needs a privileged container, which is not isolated from the host.'],
+              ROWS['It asks for capabilities or a relaxed confinement profile.'])),
+        ):
+            with self.subTest(profile=profile):
+                template['proxmox']['security_profile'] = profile
+                text = describe(template)
+                for warning in (ROWS['It needs a privileged container, which is not isolated from the host.'],
+                                ROWS['It asks for capabilities or a relaxed confinement profile.']):
+                    if warning in present:
+                        self.assertIn('  ' + warning, text)
+                    else:
+                        self.assertNotIn(warning, text)
+        template['proxmox']['security_profile'] = {}
+        template['proxmox']['installer_profile'] = {'security': {'required_capabilities': ['NET_ADMIN']}}
+        text = describe(template)
+        self.assertNotIn(ROWS['It asks for capabilities or a relaxed confinement profile.'], text)
+
+    def test_suffix_hit_rejects_dotted_directory_without_claiming_file_type(self):
+        fn = extracted('template_from_compose',
+            _services=lambda d: d['services'], _check_mounts=lambda s: None,
+            normalize_app_id=lambda s: s, _published_port=lambda s: None,
+            _dump=lambda s: '', _keep_reference=lambda *a: None,
+            TIME_SETTINGS=('/etc/localtime', '/etc/timezone'),
+            convert_casaos_compose=lambda *args: {'container_contract': {
+                'image': {'reference': 'example:latest'},
+                'volumes': [{'container_path': '/srv/data.v2', 'installation_choice': ['managed-volume', 'host-bind']}]},
+                'proxmox': {}})
+        with self.assertRaises(ConversionError) as caught:
+            fn('services:\n  app:\n    image: example:latest\n')
+        self.assertEqual(str(caught.exception),
+            ROWS['The image expects files that are given to it one by one:'] + ' /srv/data.v2. ' +
+            ROWS['ProxMenux attaches directories, not single files, so this image cannot be installed yet.'])
+
+    def test_suffix_free_path_still_accepted(self):
+        fn = extracted('template_from_compose',
+            _services=lambda d: d['services'], _check_mounts=lambda s: None,
+            normalize_app_id=lambda s: s, _published_port=lambda s: None,
+            _dump=lambda s: '', _keep_reference=lambda *a: None,
+            TIME_SETTINGS=('/etc/localtime', '/etc/timezone'),
+            convert_casaos_compose=lambda *args: {'container_contract': {
+                'image': {'reference': 'example:latest'},
+                'volumes': [{'container_path': '/srv/config', 'installation_choice': ['managed-volume', 'host-bind']}]},
+                'proxmox': {}})
+        self.assertEqual(fn('services:\n  app:\n    image: example:latest\n')['container_contract']['volumes'][0]['container_path'], '/srv/config')
+
+    def test_all_four_keys_extract_and_fallback_in_seven_shipped_locales(self):
+        found = runpy.run_path(str(ROOT / '.github/scripts/build_translation_cache.py'))['extract_python_texts']([ROOT / 'oci/src', ROOT / 'oci/remote'])
+        self.assertEqual(len(ROWS), 4)
+        self.assertTrue(set(ROWS.values()) <= set(found), set(ROWS.values()) - set(found))
+        import importlib.util
+        spec = importlib.util.spec_from_file_location('oci_i18n', ROOT / 'oci/src/proxmenux_oci/i18n.py')
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        paths = sorted((ROOT / 'lang').glob('*.json'))
+        self.assertEqual(len(paths), 7)
+        for path in paths:
+            cache = json.loads(path.read_text())
+            setattr(module, '_language', path.stem)
+            setattr(module, '_cache', cache)
+            for new in ROWS.values():
+                self.assertEqual(module.translate(new), cache.get(new) or new, (path.name, new))
+            setattr(module, '_cache', {})
+            for new in ROWS.values():
+                self.assertEqual(module.translate(new), new, (path.name, new, 'synthetic missing'))
+        module._language, module._cache = 'it', {new: 'IT: ' + new for new in ROWS.values()}
+        self.assertIn('IT: ' + ROWS['It needs a privileged container, which is not isolated from the host.'],
+                      extracted('describe', translate=module.translate, blocker_text=str)({
+                          'container_contract': {'image': {'reference': 'test:latest'}, 'volumes': [], 'ports': [], 'environment': []},
+                          'proxmox': {'security_profile': {'requires_privileged_lxc': True}}}))
+
+
+if __name__ == '__main__':
+    unittest.main()

.github/scripts/tests/test_oci_wording.py

@@ -0,0 +1,181 @@
+"""Offline OCI wording seams; compile actual functions, never import operational modules."""
+import ast
+import importlib.util
+import io
+import json
+import runpy
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+from unittest import TestCase
+from unittest.mock import Mock
+
+ROOT = Path(__file__).resolve().parents[3]
+CUSTOM = ROOT / 'oci/src/proxmenux_oci/custom.py'
+TRANSACTION = ROOT / 'oci/remote/oci_instance_transaction.py'
+MENU = ROOT / 'oci/src/proxmenux_oci/management.py'
+CHOICE = 'Recover the previous installation'
+BEFORE = 'The operation stopped halfway. Choose "{choice}" for this container in the OCI management menu.'
+AFTER = 'The recovery did not complete. Review the log and choose "{choice}" again for this container in the OCI management menu.'
+NEW_KEYS = (
+    'The Compose file describes several services:',
+    'Only one service at a time can be installed this way.',
+    'No docker run command was given',
+    'Image selection was cancelled',
+    'Could not inspect the image in its registry:',
+    'Check the image reference and registry access. If the registry is unavailable, try again later. A private registry needs credentials, which are not supported yet.',
+    BEFORE, AFTER,
+)
+
+
+class ConversionError(Exception):
+    pass
+
+
+class UserCancelled(Exception):
+    pass
+
+
+def function(path, name, **dependencies):
+    node = next(n for n in ast.parse(path.read_text()).body
+                if isinstance(n, ast.FunctionDef) and n.name == name)
+    # Only read_definition's infrastructure import is omitted; keep the function intact.
+    if name == 'read_definition':
+        node.body = [n for n in node.body if not isinstance(n, ast.ImportFrom)]
+    scope = dict(translate=lambda text: text, ConversionError=ConversionError,
+                 UserCancelled=UserCancelled, Any=Any, yaml=__import__('yaml'), json=json,
+                 Path=Path, MENU_SIZE=(20, 80), MAX_COMPOSE_BYTES=256*1024)
+    scope.update(dependencies)
+    exec(compile(ast.fix_missing_locations(ast.Module(body=[node], type_ignores=[])),
+                 str(path), 'exec'), scope)
+    return scope[name]
+
+
+class OciWordingTests(TestCase):
+    def test_extractor_shipped_locales_and_synthetic_fallback(self):
+        generator = runpy.run_path(str(ROOT / '.github/scripts/build_translation_cache.py'))
+        extracted = generator['extract_python_texts']([ROOT / 'oci/src', ROOT / 'oci/remote'])
+        self.assertEqual(len(NEW_KEYS), len(set(NEW_KEYS)))
+        self.assertTrue(set(NEW_KEYS) <= set(extracted), set(NEW_KEYS) - set(extracted))
+        spec = importlib.util.spec_from_file_location('oci_i18n', ROOT / 'oci/src/proxmenux_oci/i18n.py')
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        paths = sorted((ROOT / 'lang').glob('*.json'))
+        self.assertEqual(len(paths), 7)
+        for path in paths:
+            cache = json.loads(path.read_text())
+            setattr(module, '_language', path.stem)
+            setattr(module, '_cache', cache)
+            for key in NEW_KEYS:
+                self.assertEqual(module.translate(key), cache.get(key) or key, (path.name, key))
+            setattr(module, '_cache', {})
+            for key in NEW_KEYS:
+                self.assertEqual(module.translate(key), key, (path.name, key, 'synthetic missing'))
+            setattr(module, '_cache', {NEW_KEYS[0]: 'Several services (translated):'})
+            self.assertEqual(module.translate(NEW_KEYS[0]), 'Several services (translated):')
+
+    def test_multiple_services_named_not_images(self):
+        template = function(CUSTOM, 'template_from_compose',
+            _services=lambda doc: doc['services'], _check_mounts=lambda _: None)
+        with self.assertRaises(ConversionError) as caught:
+            template(json.dumps({'services': {'alpha': {'image': 'same:latest'},
+                                              'beta': {'image': 'same:latest'}}}))
+        self.assertEqual(str(caught.exception),
+                         'The Compose file describes several services: alpha, beta. '
+                         'Only one service at a time can be installed this way.')
+
+    def test_empty_paste_is_specific_and_nonempty_unchanged(self):
+        for title, expected in ((None, 'No Compose file was given'),
+                                ('docker run command of the application', 'No docker run command was given')):
+            def paste(content):
+                return function(CUSTOM, '_read_pasted',
+                    console=SimpleNamespace(show_logo=lambda: None, msg_title=lambda *_: None,
+                                            msg_info2=lambda *_: None),
+                    sys=SimpleNamespace(stdin=io.StringIO(content)))
+            with self.assertRaises(UserCancelled) as caught:
+                paste('')(None, title)
+            self.assertEqual(str(caught.exception), expected)
+            self.assertEqual(paste('  literal input\n')(None, title), '  literal input\n')
+
+    def test_cancelled_image_menu(self):
+        with self.assertRaises(UserCancelled) as caught:
+            function(CUSTOM, 'read_definition')(SimpleNamespace(choose=lambda *_, **__: None))
+        self.assertEqual(str(caught.exception), 'Image selection was cancelled')
+
+    def test_registry_nonzero_and_success_command_unchanged(self):
+        result = SimpleNamespace(returncode=1, stdout='', stderr='connection timed out')
+        run = Mock(return_value=result)
+        report = function(CUSTOM, 'registry_report', subprocess=SimpleNamespace(run=run))
+        self.assertEqual(report('example:latest'),
+                         (False, 'Could not inspect the image in its registry: example:latest'))
+        run.assert_called_once_with(['skopeo', 'inspect', '--raw', 'docker://example:latest'],
+                                    capture_output=True, text=True, check=False, timeout=120)
+        run.reset_mock()
+        run.return_value = SimpleNamespace(returncode=0,
+            stdout='{"manifests":[{"platform":{"architecture":"amd64"}}]}', stderr='')
+        self.assertEqual(report('example:latest'), (True, 'The image is in its registry: amd64'))
+        self.assertEqual(run.call_count, 1)
+
+    def test_failed_registry_inspection_has_neutral_advice(self):
+        ui = SimpleNamespace(message=Mock())
+        fn = function(CUSTOM, 'explore', read_definition=lambda _: ('definition', 'source'),
+            ignored_settings=lambda _: [], template_from_compose=lambda _: {
+                'container_contract': {'image': {'reference': 'example:latest'}}},
+            describe=lambda _: 'SUMMARY', registry_report=lambda _: (False, 'INSPECTION FAILED'))
+        fn(ui)
+        self.assertEqual(ui.message.call_args.args[0], 'SUMMARY\n\nINSPECTION FAILED\n\n'
+            'Check the image reference and registry access. If the registry is unavailable, '
+            'try again later. A private registry needs credentials, which are not supported yet.')
+
+    def test_recovery_hint_quotes_translated_choice_in_all_locales(self):
+        # Pin the real translated management choice, not an English AST constant alone.
+        source = ast.parse(MENU.read_text())
+        self.assertTrue(any(isinstance(n, ast.Tuple) and len(n.elts) == 2
+            and isinstance(n.elts[0], ast.Constant) and n.elts[0].value == 'recover'
+            and isinstance(n.elts[1], ast.Call) and isinstance(n.elts[1].func, ast.Name)
+            and n.elts[1].func.id == 'translate' and n.elts[1].args
+            and isinstance(n.elts[1].args[0], ast.Constant)
+            and n.elts[1].args[0].value == CHOICE for n in ast.walk(source)))
+        spec = importlib.util.spec_from_file_location('oci_remote_ui', ROOT / 'oci/remote/oci_ui.py')
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        for path in sorted((ROOT / 'lang').glob('*.json')):
+            cache = json.loads(path.read_text())
+            setattr(module, '_language', path.stem)
+            setattr(module, '_cache', cache)
+            warnings = []
+            hint = function(TRANSACTION, 'recovery_hint', translate=module.translate,
+                            msg_warn=warnings.append)
+            hint()
+            hint(after_recovery=True)
+            label = module.translate(CHOICE)
+            self.assertNotEqual(label, CHOICE, path.name)
+            expected = [(cache.get(key) or key).replace('{choice}', label)
+                        if '{choice}' in (cache.get(key) or key) else key.replace('{choice}', label)
+                        for key in (BEFORE, AFTER)]
+            self.assertEqual(warnings, expected, path.name)
+            # An intentionally absent message key proves fallback independently of shipped catalogs.
+            setattr(module, '_cache', {CHOICE: label})
+            warnings.clear()
+            hint()
+            hint(after_recovery=True)
+            self.assertEqual(warnings, [BEFORE.replace('{choice}', label),
+                                        AFTER.replace('{choice}', label)], (path.name, 'synthetic missing'))
+
+    def test_recovery_hint_uses_translated_whole_messages_and_malformed_fallback(self):
+        spec = importlib.util.spec_from_file_location('oci_remote_ui', ROOT / 'oci/remote/oci_ui.py')
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        module._language = 'it'
+        for translation in ('Before: «{choice}».', 'Before: label missing.'):
+            module._cache = {CHOICE: 'Recupero', BEFORE: translation,
+                             AFTER: 'After: «{choice}».'}
+            warnings = []
+            hint = function(TRANSACTION, 'recovery_hint', translate=module.translate,
+                            msg_warn=warnings.append)
+            hint()
+            hint(after_recovery=True)
+            self.assertEqual(warnings, [
+                ('Before: «Recupero».' if '{choice}' in translation
+                 else BEFORE.replace('{choice}', 'Recupero')),
+                'After: «Recupero».'])

.github/workflows/test-i18n.yml

@@ -51,6 +51,8 @@ jobs:
         run: |
           test -x /bin/bash
           test -x /usr/bin/jq
+      - name: Install OCI wording test dependency
+        run: python3 -m pip install 'PyYAML>=6,<7'
       - name: Run complete offline script test suite
         env:
           PYTHONDONTWRITEBYTECODE: '1'

oci/remote/oci_instance_reconcile.py

@@ -44,12 +44,12 @@ def _mount(key, value, vmid):
                 'size_gb': None, 'backup': False, 'read_only': read_only,
                 'create_if_missing': False}
     if options.get('backup') != '1' or ':' not in source:
-        raise ValueError(f'{key}: {translate("Only backed-up Proxmox volumes can be adopted")}')
+        raise ValueError(f'{key}: {translate("Proxmox volume adoption requires backup=1 and a volume ID")}')
     storage, volume = source.split(':', 1)
     if not re.fullmatch(r'[A-Za-z0-9_-]+', storage) or not volume:
         raise ValueError(f'{key}: {translate("Invalid Proxmox volume ID")}')
     if not re.search(rf'(?:^|/)(?:vm|subvol)-{vmid}-disk-[0-9]+(?:\.|$)', volume):
-        raise ValueError(f'{key}: {translate("The disk does not belong to this CT; automatic adoption is unsafe")}')
+        raise ValueError(f'{key}: {translate("The volume ID does not contain a disk name for this CT; automatic adoption is unsafe")}')
     return {'type': 'managed-volume', 'container_path': target, 'source': storage,
             'size_gb': _managed_size(options.get('size')), 'backup': True,
             'read_only': read_only}

oci/remote/oci_instance_transaction.py

@@ -200,9 +200,14 @@ def pending_journal():
 
 def recovery_hint(after_recovery=False):
     if after_recovery:
-        msg_warn(translate('The recovery did not complete. Review the log and choose "Recover" again for this container in the OCI management menu.'))
+        source = 'The recovery did not complete. Review the log and choose "{choice}" again for this container in the OCI management menu.'
+        message = translate('The recovery did not complete. Review the log and choose "{choice}" again for this container in the OCI management menu.')
     else:
-        msg_warn(translate('The operation stopped halfway. Choose "Recover" for this container in the OCI management menu to restore the previous installation.'))
+        source = 'The operation stopped halfway. Choose "{choice}" for this container in the OCI management menu.'
+        message = translate('The operation stopped halfway. Choose "{choice}" for this container in the OCI management menu.')
+    # A malformed cache entry must not leave the user without the menu choice.
+    msg_warn((message if '{choice}' in message else source).replace(
+        '{choice}', translate('Recover the previous installation')))
 
 
 def recover_untouched(root, journal):

oci/remote/oci_remove.py

@@ -140,7 +140,7 @@ def remove(root, vmid):
     for path, size in image_cache.prune(root, lock=False):
         msg_ok(f"{translate('Unused image removed from the cache:')} {path.name}")
     for path in kept:
-        msg_warn(f"{translate('Host directory kept, with its content:')} {path}")
+        msg_warn(f"{translate('Host directory listed in saved records (not targeted for removal):')} {path}")
 
 
 def main():

oci/src/proxmenux_oci/custom.py

@@ -114,8 +114,8 @@ def template_from_compose(text: str, title: str | None = None) -> dict[str, Any]
     services = _services(compose)
     if len(services) > 1:
         raise ConversionError(
-            f"{translate('The Compose file describes several images:')} {', '.join(services)}. "
-            f"{translate('Only one image at a time can be installed this way.')}")
+            f"{translate('The Compose file describes several services:')} {', '.join(services)}. "
+            f"{translate('Only one service at a time can be installed this way.')}")
     name = next(iter(services))
     service = services[name] or {}
     if not service.get('image'):
@@ -158,8 +158,8 @@ def template_from_compose(text: str, title: str | None = None) -> dict[str, Any]
              if Path(volume['container_path']).suffix]
     if files:
         raise ConversionError(
-            f"{translate('The image expects files that are given to it one by one:')} {', '.join(files)}. "
-            f"{translate('ProxMenux attaches directories, not single files, so this image cannot be installed yet.')}")
+            f"{translate('These container mount paths have an extension-like suffix:')} {', '.join(files)}. "
+            f"{translate('This import rejects these paths without checking whether they are files or directories.')}")
     return template
 
 
@@ -316,7 +316,8 @@ def _read_pasted(ui, title: str | None = None) -> str:
     print()
     text = sys.stdin.read(MAX_COMPOSE_BYTES + 1)
     if not text.strip():
-        raise UserCancelled(translate('No Compose file was given'))
+        raise UserCancelled(translate('No docker run command was given') if title is not None
+                            else translate('No Compose file was given'))
     return text
 
 
@@ -331,7 +332,7 @@ def read_definition(ui) -> tuple[str, str]:
         ('image', translate('Only the image reference, with no Compose file')),
     ], 'paste', title=translate('Image that is not in the catalog'), size=MENU_SIZE)
     if source is None:
-        raise UserCancelled(translate('No image was given'))
+        raise UserCancelled(translate('Image selection was cancelled'))
     if source == 'paste':
         return _read_pasted(ui), translate('pasted Compose file')
     if source == 'file':
@@ -415,7 +416,7 @@ def registry_report(reference: str) -> tuple[bool, str]:
     result = subprocess.run(['skopeo', 'inspect', '--raw', f'docker://{reference}'],
                             capture_output=True, text=True, check=False, timeout=120)
     if result.returncode != 0:
-        return False, f"{translate('The image was not found in its registry, or it is private:')} {reference}"
+        return False, f"{translate('Could not inspect the image in its registry:')} {reference}"
     try:
         document = json.loads(result.stdout)
     except ValueError:
@@ -493,12 +494,12 @@ def describe(template: dict[str, Any]) -> str:
                   for item in devices]
     warnings = []
     if security.get('requires_privileged_lxc'):
-        warnings.append(translate('It needs a privileged container, which is not isolated from the host.'))
+        warnings.append(translate('This profile requires a privileged LXC, which reduces isolation from the host.'))
     elif security.get('source_requests_privileged_lxc') or security.get('optional_privileged_lxc'):
         warnings.append(translate('Its Compose file asks for privileged mode; the container is created '
                                   'unprivileged and that mode is only offered as an option.'))
     if security.get('requires_relaxed_confinement') or security.get('source_requests_relaxed_confinement'):
-        warnings.append(translate('It asks for capabilities or a relaxed confinement profile.'))
+        warnings.append(translate('A relaxed AppArmor or seccomp profile is requested; this may be optional.'))
     if security.get('requires_host_pid_namespace'):
         warnings.append(translate('It asks to see the processes of the host.'))
     if warnings:
@@ -548,9 +549,8 @@ def explore(ui) -> None:
     if notes:
         summary += '\n\n' + '\n'.join(notes)
     if not available:
-        advice = translate('Some projects publish a Dockerfile and not an image: it has to be built '
-                           'and published to a registry before it can be installed this way. An image '
-                           'of a private registry needs credentials, which are not supported yet.')
+        advice = translate('Check the image reference and registry access. If the registry is unavailable, '
+                           'try again later. A private registry needs credentials, which are not supported yet.')
         ui.message(f'{summary}\n\n{advice}', title)
         return
     if not template.get('compatibility', {}).get('automatic_install_candidate'):

oci/src/proxmenux_oci/management.py

@@ -319,10 +319,10 @@ def _removal_summary(project, vmid):
     if bridge:
         text += ['', f"{translate('Private network of the application that is released:')} {bridge}"]
     if kept:
-        text += ['', translate('Host directories that are kept, with their content:'),
+        text += ['', translate('Host paths found in container configs or saved records (not targeted for removal):'),
                  *[f'  {path}' for path in kept]]
     else:
-        text += ['', translate('No host directory is used by this application.')]
+        text += ['', translate('No host directories found in the available container configs or saved records.')]
     return '\n'.join(text)
 
 
@@ -372,11 +372,11 @@ def _manage_stack(project, ui, row, action=None, lifecycle_args=()):
             return _remove(project, ui, primary_id)
         if not ui.review(f"{translate('All stack members are updated together. Main CT:')} {primary_id}, "
                 f"{translate('members:')} {len(members)}. "
-                f"{translate('All images are downloaded and verified first, and native backups are taken with the stack stopped. Contracts are published after the whole set is checked. If anything fails, all members are recovered.')}",
+                f"{translate('All images are downloaded and verified first, and native backups are taken with the stack stopped. Contracts are published after the whole set is checked. If a step fails, recovery is attempted where needed; recovery can also fail.')}",
                 translate('Update OCI stack'), question=translate('Update the whole stack?'), default=True):
             return False
     else:
-        if not ui.review(translate('A coordinated operation is pending. The whole previous stack will be recovered, not only the selected member. If the operation already finished, the cleanup of its markers is completed.'), translate('Recover OCI stack'),
+        if not ui.review(translate('A coordinated operation has a saved journal. Continuing attempts to recover the previous stack where needed, or finish cleanup for a completed operation. Recovery or cleanup can fail.'), translate('Recover OCI stack'),
                 question=translate('Recover or complete the operation?'), default=True):
             return False
         import json

oci/tests/test_adoption_wording_contract.py

@@ -0,0 +1,43 @@
+"""Inert, extracted producer checks for the two adoption rejection messages."""
+import ast
+from pathlib import Path
+import re
+import types
+import unittest
+
+SOURCE = Path(__file__).resolve().parents[1] / 'remote' / 'oci_instance_reconcile.py'
+
+
+def mount_reader():
+    tree = ast.parse(SOURCE.read_text())
+    funcs: list[ast.stmt] = [node for node in tree.body if isinstance(node, ast.FunctionDef)
+                             and node.name in ('_mount', '_managed_size')]
+    namespace = {'re': re, 'translate': lambda text: text,
+                 'host_mounts': types.SimpleNamespace(valid_path=lambda path: path,
+                                                      validate_source=lambda path: None)}
+    exec(compile(ast.Module(body=funcs, type_ignores=[]), str(SOURCE), 'exec'), namespace)
+    return namespace['_mount']
+
+
+class AdoptionWording(unittest.TestCase):
+    def test_backup_flag_is_a_requirement_not_a_prior_backup(self):
+        mount = mount_reader()
+        for value in ('local-lvm:vm-200-disk-3,mp=/data,size=8G,backup=0',
+                      'local-lvm:vm-200-disk-3,mp=/data,size=8G'):
+            with self.subTest(value=value), self.assertRaisesRegex(
+                    ValueError, 'Proxmox volume adoption requires backup=1 and a volume ID'):
+                mount('mp2', value, 200)
+        self.assertTrue(mount('mp2', 'local-lvm:vm-200-disk-3,mp=/data,size=8G,backup=1', 200)['backup'])
+
+    def test_disk_name_check_does_not_claim_to_prove_ownership(self):
+        mount = mount_reader()
+        with self.assertRaisesRegex(ValueError,
+                'The volume ID does not contain a disk name for this CT; automatic adoption is unsafe'):
+            mount('mp2', 'local-lvm:vm-201-disk-3,mp=/data,size=8G,backup=1', 200)
+        # A matching substring passes the actual guard; no ownership query is made.
+        result = mount('mp2', 'local-lvm:other/vm-200-disk-3.raw,mp=/data,size=8G,backup=1', 200)
+        self.assertEqual(result['type'], 'managed-volume')
+
+
+if __name__ == '__main__':
+    unittest.main()