Merge commit 'c471bccdab95' into HEAD

# Conflicts:
#	.github/scripts/run_offline_qualified.py
This commit is contained in:
MacRimi
2026-10-21 23:14:28 +02:00
75 changed files with 6574 additions and 1500 deletions
+7
View File
@@ -13,6 +13,9 @@ import sys
import tempfile
PYTHON_FILES = (
"AppImage/scripts/tests/test_pbs_i18n.py",
"AppImage/scripts/tests/test_monitor_links_i18n.py",
"AppImage/scripts/tests/test_monitor_links.py",
"AppImage/scripts/tests/test_local_app_icon_isolation.py",
"AppImage/scripts/tests/test_local_app_icons.py",
"AppImage/scripts/tests/test_docker_app_discovery.py",
@@ -29,6 +32,10 @@ PYTHON_FILES = (
)
NODE_FILES = (
"tests/lxc_updates/test_app_editor_errors.cjs",
"tests/test_oci_update_cancellation.cjs",
"tests/test_host_backup_messages.cjs",
"tests/test_pbs_i18n.cjs",
"tests/test_monitor_links_i18n.cjs",
"tests/lxc_updates/test_local_app_icons_ui.cjs",
"tests/lxc_updates/test_docker_app_discovery_ui.cjs",
"tests/lxc_updates/test_native_app_discovery_ui.cjs",
@@ -0,0 +1,284 @@
"""Human diagnostics translate; IDs, JSON, error types and forbidden effects do not."""
from contextlib import ExitStack, nullcontext, redirect_stderr, redirect_stdout
import copy
import importlib.util
import io
import json
from pathlib import Path
import sys
import tempfile
import unittest
from unittest.mock import MagicMock, Mock, patch
import urllib.error
sys.path.insert(0, str(Path(__file__).resolve().parent))
import test_oci_complete_ui_sk as fixtures
from proxmenux_oci import casaos, cli, converter, i18n
from proxmenux_oci.catalog import Catalog
from proxmenux_oci.github_source import GitHubSource, SourceError
import oci_image_cache
import oci_dhcp_lease
import oci_rocm_check
import oci_ui
import oci_watchdog
import configure_jellyfin_encoding
class TechnicalMessages(unittest.TestCase):
setUp = fixtures.CompleteUI.setUp
patch = fixtures.CompleteUI.patch
def remote_language(self):
self.patch(oci_ui, '_language', 'sk')
self.patch(oci_ui, '_cache', self.sk)
def test_real_readme_errors_translate_only_at_the_display_boundary(self):
for text, expected in [('# App', 'README neobsahuje'),
('## docker-compose\nplain text', 'Sekcia docker-compose neobsahuje blok kódu'),
('## docker-compose\n```yaml\nservices: {}', 'Blok docker-compose nie je ukončený'),
('## docker-compose\n```yaml\nnot-compose\n```', 'Nájdený blok nevyzerá')]:
with self.subTest(text=text), self.assertRaises(converter.ConversionError) as caught:
converter.extract_compose(text)
raw = str(caught.exception)
self.assertIn(expected, i18n.diagnostic(caught.exception))
self.assertEqual(str(caught.exception), raw)
def test_schema_error_keeps_field_details_and_original_exception(self):
with self.assertRaises(converter.ConversionError) as caught:
self.catalog.validate({})
raw = str(caught.exception)
self.assertTrue(raw.startswith('La plantilla generada no cumple el esquema:'))
self.assertEqual(i18n.diagnostic(caught.exception), 'Vygenerovaná šablóna nezodpovedá schéme:' + raw.split(':', 1)[1])
def test_missing_schema_validator_is_distinct_from_invalid_schema_without_effects(self):
template = {}
read = self.patch(Path, 'read_text', side_effect=AssertionError('schema must not be read without a validator'))
with patch.dict(sys.modules, {'jsonschema': None}):
with self.assertRaises(converter.ConversionError) as caught:
self.catalog.validate(template)
raw = 'python3-jsonschema is required to generate templates'
self.assertEqual(str(caught.exception), raw)
self.assertEqual(i18n.diagnostic(caught.exception), 'Na generovanie šablón je potrebný python3-jsonschema')
self.assertEqual(str(caught.exception), raw)
self.assertIsNone(self.catalog.validate(template, required=False))
read.assert_not_called()
self.assertEqual(template, {})
for effect in self.effects:
effect.assert_not_called()
def test_real_casaos_error_preserves_the_protocol_and_localizes_the_message(self):
with self.assertRaises(converter.ConversionError) as caught:
casaos.parse_casaos_compose('[]')
self.assertEqual(str(caught.exception), 'The CasaOS document is not a Compose mapping')
self.assertEqual(i18n.diagnostic(caught.exception), 'Dokument Compose musí byť mapa kľúčov a hodnôt.')
def test_github_rate_limit_keeps_url_and_token_variable_without_a_network_call(self):
url = 'https://fiction.invalid/catalogue?ref=example'
error = urllib.error.HTTPError(url, 403, 'Forbidden', {'X-RateLimit-Remaining': '0'}, None)
request = self.patch(__import__('urllib.request', fromlist=['urlopen']), 'urlopen', side_effect=error)
with self.assertRaises(SourceError) as caught:
GitHubSource(token='fictional-token')._request(url)
request.assert_called_once()
text = i18n.diagnostic(caught.exception)
self.assertIn('GitHub vrátil HTTP 403', text)
self.assertIn(url, text)
self.assertIn('GITHUB_TOKEN', text)
self.assertNotIn('fictional-token', text)
def test_github_timeout_retains_external_reason_and_three_attempts(self):
request = self.patch(__import__('urllib.request', fromlist=['urlopen']), 'urlopen',
side_effect=urllib.error.URLError('fictional external transport reason'))
self.patch(__import__('time'), 'sleep')
with self.assertRaises(SourceError) as caught:
GitHubSource()._request('https://fiction.invalid/template')
self.assertEqual(request.call_count, 3)
self.assertIn('ani po 3 pokusoch: fictional external transport reason', i18n.diagnostic(caught.exception))
def test_unknown_diagnostics_and_machine_codes_pass_through_unchanged(self):
for value in ['service:db:healthcheck-format', 'external tool: failure at 0x1234',
'Check failed', 'Invalid cache', 'Ambiguous application', 'private-operation-id']:
self.assertEqual(i18n.diagnostic(value), value)
def test_cli_help_preserves_flags_actions_defaults_and_numeric_types(self):
argv = ['manage', '104', '--action', 'recreate', '--unattended', '--min-image-age-days', '7']
slovak = cli.build_parser()
self.assertIn('Vypísať aplikácie', slovak.format_help())
with patch.object(i18n, '_language', 'en'):
english = cli.build_parser()
self.assertIn('List applications', english.format_help())
self.assertEqual(vars(english.parse_args(argv)), vars(slovak.parse_args(argv)))
def test_cli_summary_preserves_image_reference_ports_mount_paths_and_status_code(self):
template = self.catalog.compose('tandoor')
saved = copy.deepcopy(template)
text = cli._template_summary_text(template)
self.assertIn('Obraz: ' + template['container_contract']['image']['reference'], text)
self.assertIn('Dátové zväzky:', text)
self.assertIn('Stav: ' + template['status'], text)
for mount in template['container_contract']['volumes']:
self.assertIn(mount['container_path'], text)
self.assertEqual(template, saved)
def test_install_cli_no_stays_successful_cancellation_without_dispatch(self):
template = self.catalog.compose('jellyfin')
saved = copy.deepcopy(template)
self.patch(cli.Catalog, 'compose', return_value=template)
self.patch(cli, 'build_deployment', return_value={'fiction': True})
self.patch(cli, '_deployment_summary_text', return_value='Fiktívny plán')
answer = self.patch(__import__('builtins'), 'input', return_value='n')
output = io.StringIO()
with redirect_stdout(output): self.assertEqual(cli.main(['install', 'jellyfin']), 0)
self.assertIn('Nainštalovať? [y/N]:', answer.call_args.args[0])
self.assertEqual(output.getvalue().count('Inštalácia bola zrušená.'), 1)
self.assertEqual(template, saved)
def test_blocked_install_remains_error_one_and_never_asks_to_install(self):
template = self.catalog.compose('jellyfin')
template['compatibility'].update(automatic_install_candidate=False, untranslated_blockers=['fiction-blocker-code'])
self.patch(cli.Catalog, 'compose', return_value=template)
answer = self.patch(__import__('builtins'), 'input', side_effect=AssertionError('blocked installation asked a question'))
output = io.StringIO()
with redirect_stdout(io.StringIO()), redirect_stderr(output):
self.assertEqual(cli.main(['install', 'jellyfin']), 1)
self.assertIn('Inštalácia je zablokovaná: fiction-blocker-code', output.getvalue())
answer.assert_not_called()
def test_watchdog_manual_stop_stays_stopped_without_start_or_notice(self):
self.remote_language()
self.patch(oci_watchdog, 'watched', return_value={777: 'Fiction'})
self.patch(oci_watchdog, 'is_running', return_value=False)
self.patch(oci_watchdog, 'is_busy', return_value=False)
self.patch(oci_watchdog, 'read_tasks', return_value=[('vzshutdown', '777', 75)])
start = self.patch(oci_watchdog, 'start', side_effect=AssertionError('manual stop was undone'))
notice = self.patch(oci_watchdog.oci_operation_notice, 'notify')
state = oci_watchdog.new_state(); state['seen'] = 50
output = io.StringIO()
with redirect_stdout(output): oci_watchdog.look({777: state}, 100)
self.assertTrue(state['settled'])
self.assertEqual(output.getvalue(), '')
start.assert_not_called(); notice.assert_not_called()
def test_watchdog_localization_does_not_change_retry_state_or_notification_payload(self):
self.patch(oci_watchdog, 'watched', return_value={777: 'Fiction'})
self.patch(oci_watchdog, 'is_running', return_value=False)
self.patch(oci_watchdog, 'is_busy', return_value=False)
self.patch(oci_watchdog, 'read_tasks', return_value=[])
start = self.patch(oci_watchdog, 'start', return_value=True)
notice = self.patch(oci_watchdog.oci_operation_notice, 'notify')
outcomes = []
for locale in ('en', 'sk'):
with patch.object(oci_ui, '_language', locale), patch.object(oci_ui, '_cache', self.sk):
state = oci_watchdog.new_state(); state['seen'] = 50
text = io.StringIO()
with redirect_stdout(text): oci_watchdog.look({777: state}, 100)
outcomes.append((copy.deepcopy(state), notice.call_args, start.call_args))
self.assertIn('CT 777 (Fiction)', text.getvalue())
if locale == 'sk': self.assertIn('spúšťa sa znova (pokus 1)', text.getvalue())
self.assertEqual(outcomes[0], outcomes[1])
def test_image_cache_json_contract_and_invalid_command_have_no_mutations(self):
self.remote_language()
self.patch(oci_image_cache.instances, 'locked', side_effect=lambda *a: nullcontext())
self.patch(oci_image_cache, 'archives_of', return_value={})
output = io.StringIO()
with redirect_stdout(output): self.assertEqual(oci_image_cache.main(['list', '777']), 0)
self.assertEqual(json.loads(output.getvalue()), {'archives': [], 'freed': 0})
error = io.StringIO()
with redirect_stderr(error): self.assertEqual(oci_image_cache.main(['invalid']), 2)
self.assertIn('Použitie: oci_image_cache.py [prune | list VMID... | remove VMID...]', error.getvalue())
def test_rocm_timeout_localizes_but_external_gpu_errors_are_preserved(self):
self.remote_language()
self.patch(sys, 'argv', ['oci_rocm_check.py', '777'])
for raw, expected in [('the GPU did not answer in time', 'GPU neodpovedalo včas'),
('external provider exception 0x123', 'external provider exception 0x123')]:
with self.subTest(raw=raw), patch.object(oci_rocm_check, 'check', side_effect=RuntimeError(raw)):
output = io.StringIO()
with redirect_stderr(output): self.assertEqual(oci_rocm_check.main(), 1)
self.assertEqual(output.getvalue().strip(), expected)
def test_jellyfin_completion_distinguishes_changed_from_already_applied(self):
self.remote_language()
self.patch(sys, 'argv', ['configure_jellyfin_encoding.py', '/fiction-root', '{}'])
for changed, expected in [(True, 'Nastavenia Jellyfin boli aktualizované:'),
(False, 'Nastavenia Jellyfin už boli použité:')]:
with self.subTest(changed=changed), patch.object(configure_jellyfin_encoding, 'update_encoding',
return_value=(Path('/fiction-root/config/encoding.xml'), changed)):
output = io.StringIO()
with redirect_stdout(output): configure_jellyfin_encoding.main()
self.assertEqual(output.getvalue().strip(), expected + ' /config/encoding.xml')
def test_dhcp_completion_keeps_interface_and_ip_version_without_repairing_a_guest(self):
self.remote_language()
self.patch(oci_dhcp_lease, 'LOCKS', MagicMock())
self.patch(oci_dhcp_lease.fcntl, 'flock')
repair = self.patch(oci_dhcp_lease, 'repair', return_value=[('eth0', 4)])
output = io.StringIO()
with redirect_stdout(output): oci_dhcp_lease.check(777, 0)
repair.assert_called_once_with(777, 0)
self.assertEqual(output.getvalue().strip(), 'CT 777: klient DHCP rozhrania eth0 (IPv4) bol znova spustený')
class StandalonePublisher(unittest.TestCase):
def test_mount_error_keeps_host_language_after_entering_the_container_root(self):
spec = importlib.util.spec_from_file_location('fiction_namespace_publisher',
fixtures.ROOT / 'oci/remote/rclone_mount_publish.py')
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
inside = False
def chroot(path):
nonlocal inside
inside = True
def read(path, **kwargs):
if inside:
raise FileNotFoundError('Host language catalog is outside the container root')
return ('{"language":"sk"}' if path.name == 'config.json' else
'{"clone source mount tree":"naklonovať strom zdrojového pripojenia"}')
libc = Mock(); libc.setns.return_value = 0; libc.syscall.return_value = -1
with ExitStack() as guards:
for target, name, options in [
(sys, 'argv', {'new': ['publisher.py', '777', '/fiction-source', '/fiction-target', 'ro']}),
(module.platform, 'machine', {'return_value': 'x86_64'}),
(Path, 'read_text', {'autospec': True, 'side_effect': read}),
(module.ctypes, 'CDLL', {'return_value': libc}),
(module.ctypes, 'get_errno', {'return_value': 13}),
(module.os, 'open', {'return_value': 42}),
(module.os, 'close', {}), (module.os, 'fchdir', {}),
(module.os, 'chroot', {'side_effect': chroot}), (module.os, 'chdir', {})]:
guards.enter_context(patch.object(target, name, **options))
with self.assertRaises(OSError) as caught:
module.main()
self.assertTrue(inside)
self.assertEqual(caught.exception.errno, 13)
self.assertIn('naklonovať strom zdrojového pripojenia', str(caught.exception))
self.assertEqual(libc.syscall.call_args.args[0], module.SYSCALLS['x86_64'][0])
def test_copied_helper_has_localized_errors_without_engine_import_or_mount_syscalls(self):
with tempfile.TemporaryDirectory(prefix='oci-publisher-language-') as directory:
root = Path(directory); (root / 'lang').mkdir()
(root / 'config.json').write_text('{"language":"sk"}')
(root / 'lang/sk.json').write_text(json.dumps({'Usage:': 'Použitie:', 'Unsupported host architecture:': 'Nepodporovaná architektúra hostiteľa:'}))
source = fixtures.ROOT / 'oci/remote/rclone_mount_publish.py'
copied = root / 'publisher.py'; copied.write_bytes(source.read_bytes())
spec = importlib.util.spec_from_file_location('fiction_standalone_publisher', copied)
module = importlib.util.module_from_spec(spec)
with patch.dict(sys.modules, {'oci_ui': None}), patch.dict(__import__('os').environ, {'PMX_BASE_DIR': str(root)}):
spec.loader.exec_module(module)
with patch.object(module.ctypes, 'CDLL', side_effect=AssertionError('mount syscall path entered')) as syscall:
with patch.object(sys, 'argv', ['publisher.py']), redirect_stderr(io.StringIO()) as output:
self.assertEqual(module.main(), 2)
self.assertIn('Použitie: publisher.py PID SOURCE TARGET rw|ro', output.getvalue())
with patch.object(sys, 'argv', ['publisher.py', '1', '/fiction-source', '/fiction-target', 'ro']), \
patch.object(module.platform, 'machine', return_value='fiction-architecture'), redirect_stderr(io.StringIO()) as output:
self.assertEqual(module.main(), 2)
self.assertIn('Nepodporovaná architektúra hostiteľa: fiction-architecture', output.getvalue())
syscall.assert_not_called()
(root / 'config.json').write_text('invalid json')
module._messages.cache_clear()
self.assertEqual(module.translate('Usage:'), 'Usage:')
(root / 'config.json').write_text('{"language":"en"}')
module._messages.cache_clear()
self.assertEqual(module.translate('Usage:'), 'Usage:')
if __name__ == '__main__': unittest.main()
@@ -0,0 +1,148 @@
"""Real catalogue/wizard rendering, with host and installation effects forbidden."""
import copy
from contextlib import ExitStack, redirect_stdout
import io
import json
import os
from pathlib import Path
import subprocess
import sys
import unittest
from unittest.mock import Mock, patch
ROOT = Path(os.environ.get('OCI_I18N_TEST_ROOT', Path(__file__).resolve().parents[3]))
sys.path[:0] = [str(ROOT / 'oci/src'), str(ROOT / 'oci/remote'), str(ROOT / 'oci/tests')]
from proxmenux_oci import cli, i18n, installer, stack
from proxmenux_oci.catalog import Catalog
from proxmenux_oci.converter import ConversionError
from proxmenux_oci.ui import UserCancelled
from test_advanced_flow_order import RecordingUI, addresses, storages
class CompleteUI(unittest.TestCase):
def setUp(self):
self.scope = ExitStack()
self.addCleanup(self.scope.close)
self.sk = json.loads((ROOT / 'lang/sk.json').read_text())
self.catalog = Catalog(ROOT / 'oci')
self.patch(i18n, '_language', 'sk')
self.patch(i18n, '_cache', self.sk)
self.patch(installer.host, 'storages', side_effect=storages)
self.patch(installer.host, 'bridges', return_value=[{'iface': 'vmbr0', 'cidr': '192.0.2.10/24'}])
self.patch(installer.host, 'timezone', return_value='Europe/Bratislava')
self.patch(installer.host, 'gpus', return_value={'intel': [], 'amd': [], 'nvidia': False})
self.patch(installer.access, 'ask_addresses', side_effect=addresses)
self.patch(installer.access, 'ask_ipv4', return_value=('dhcp', None))
self.effects = []
for module, name in [(cli, '_install'), (cli, '_rclone_mount'), (cli, 'run_remote_install'),
(installer, 'run_remote_install'), (cli.images, 'offer_removal'),
(subprocess, 'run'), (subprocess, 'Popen'), (subprocess, 'check_output'),
(Path, 'write_text'), (Path, 'write_bytes'), (Path, 'unlink')]:
self.effects.append(self.patch(module, name, side_effect=AssertionError('Forbidden effect: ' + name)))
self.addCleanup(lambda: [effect.assert_not_called() for effect in self.effects])
def patch(self, obj, name, *args, **kwargs):
return self.scope.enter_context(patch.object(obj, name, *args, **kwargs))
def test_search_no_match_uses_application_wording_and_never_installs(self):
ui = Mock()
ui.ask.return_value = 'fiction-no-such-app'
cli._search(self.catalog, ui, [])
self.assertEqual(ui.ask.call_args.args[0], 'Názov alebo časť popisu aplikácie')
self.assertIn('Žiadna zhoda aplikácií', ui.message.call_args.args[0])
ui.choose.assert_not_called()
def test_immich_detail_translates_service_roles_without_changing_template(self):
template = self.catalog.compose('immich')
before = copy.deepcopy(template)
text = cli._app_detail_text(self.catalog, self.catalog.find_item('immich'), template)
self.assertIn('Strojové učenie', text)
self.assertIn('Aplikácia', text)
self.assertIn('Oficiálny obraz', text)
self.assertNotIn('Machine learning', text)
self.assertEqual(template, before)
with patch.object(i18n, '_language', 'en'):
self.assertIn('Machine learning', cli._app_detail_text(self.catalog, self.catalog.find_item('immich'), template))
def test_missing_catalogue_entry_preserves_raw_error_and_localizes_cli(self):
for language, expected in [('sk', "Aplikácia 'fiction-missing' nie je v indexe"),
('en', "The application 'fiction-missing' is not in the index")]:
with self.subTest(language=language), patch.object(i18n, '_language', language):
with self.assertRaisesRegex(ConversionError, "The application 'fiction-missing' is not in the index"):
self.catalog.find_item('fiction-missing')
output = io.StringIO()
with redirect_stdout(io.StringIO()), patch.object(sys, 'stderr', output), patch.object(cli, 'PROJECT_ROOT', ROOT / 'oci'):
self.assertEqual(cli.main(['show', 'fiction-missing']), 1)
self.assertEqual(output.getvalue().strip(), 'ERROR: ' + expected)
def test_missing_template_uses_existing_translation_and_keeps_identifier(self):
self.patch(self.catalog, 'find_item', return_value={'id': 'fiction-missing'})
with self.assertRaises(ConversionError) as caught:
self.catalog.compose('fiction-missing')
self.assertEqual(str(caught.exception), "No template is available for 'fiction-missing'")
self.assertEqual(i18n.diagnostic(caught.exception), "Pre aplikáciu 'fiction-missing' nie je dostupná šablóna")
def test_detail_cancel_and_escape_do_not_enter_install_or_rclone(self):
for app in ('jellyfin', 'rclone', 'immich', 'suite-arr'):
for escape in (False, True):
with self.subTest(app=app, escape=escape):
ui = Mock()
ui.confirm.return_value = False
ui.detail_menu.return_value = None
if escape:
ui.detail_menu.side_effect = UserCancelled()
try:
cli._app_detail(self.catalog, ui, self.catalog.find_item(app))
except UserCancelled:
self.assertTrue(escape)
def test_special_wizard_plans_are_identical_in_english_and_slovak(self):
for app in ('immich', 'nextcloud-stack', 'paperless-ngx', 'tandoor'):
for mode in (installer.DEFAULT_MODE, installer.ADVANCED_MODE):
with self.subTest(app=app, mode=mode):
template = self.catalog.compose(app)
before = copy.deepcopy(template)
plan_sk = installer.build_deployment(template, RecordingUI(), mode)
with patch.object(i18n, '_language', 'en'):
plan_en = installer.build_deployment(template, RecordingUI(), mode)
self.assertEqual(plan_sk, plan_en)
self.assertEqual(template, before)
text = cli._deployment_summary_text(template, plan_sk)
self.assertNotIn('Consume/export:', text)
if app == 'paperless-ngx':
self.assertIn('Priečinky consume a export', text)
def test_romm_optional_credentials_are_localized_but_variable_names_stay(self):
ui = RecordingUI()
ui.confirm = lambda text, default=False, **kwargs: (ui.asked.append(text) or text.startswith('Nastaviť IGDB'))
ui.password = lambda text, **kwargs: (ui.asked.append(text) or 'fiction-secret')
ui.ask = lambda text, default=None, **kwargs: (ui.asked.append(text) or (default if default is not None else 'fiction-value'))
plan = installer.build_deployment(self.catalog.compose('romm'), ui, installer.ADVANCED_MODE)
self.assertTrue(any('Tajný kľúč klienta IGDB' == text for text in ui.asked))
self.assertFalse(any('IGDB Client Secret' == text for text in ui.asked))
server = next(service for service in plan['services'] if service['name'] == 'romm')
self.assertTrue(any(row['name'] == 'IGDB_CLIENT_SECRET' for row in server['deployment']['environment']))
def test_tandoor_warning_keeps_negative_requirement_and_first_run_account(self):
template = self.catalog.compose('tandoor')
warning = i18n.translate(template['proxmox']['security_profile']['warning'])
self.assertIn('nevyžaduje privilegovaný', warning)
credentials = template['first_run']['credentials']
self.assertTrue(any('Vygenerovaný účet správcu Tandoor' == i18n.translate(row['label']) for row in credentials))
def test_documented_commands_paths_and_literals_remain_copyable(self):
fragments = {
'rsnapshot starts with the default configuration, which backs up /data into /.snapshots. Edit /config/rsnapshot.conf inside the container to set your own backup points, snapshot root and retention intervals.': ['/data', '/.snapshots', '/config/rsnapshot.conf'],
'Each /request command needs a backend: add a [[backends]] block in the same file with the url and api_key of your Sonarr, Radarr or Seerr instance, then restart the container.': ['/request', '[[backends]]', 'url', 'api_key'],
'Named accounts need private mode. Stop the container, set public: false in /config/config.js, start it again and create each user with: pct exec <CTID> -- env THELOUNGE_HOME=/config s6-setuidgid abc thelounge add <user>': ['public: false', 'pct exec <CTID> -- env THELOUNGE_HOME=/config s6-setuidgid abc thelounge add <user>'],
'Certificate provider: empty for Let\'s Encrypt, zerossl for ZeroSSL': ["Let's Encrypt", 'zerossl', 'ZeroSSL'],
'Libraries: /data/media/movies, /data/media/series and /data/media/music. Select them in the media server.': ['/data/media/movies', '/data/media/series', '/data/media/music'],
}
for source, tokens in fragments.items():
with self.subTest(source=source):
for token in tokens:
self.assertIn(token, i18n.translate(source))
if __name__ == '__main__':
unittest.main()
@@ -0,0 +1,268 @@
"""Exercise the native custom-image and registry UI with the real EN/SK lookup."""
import json
from pathlib import Path
import runpy
import sys
import tempfile
from types import SimpleNamespace
import unittest
from unittest.mock import MagicMock, patch
ROOT = Path(__file__).resolve().parents[3]
sys.path.insert(0, str(ROOT / 'oci/src'))
from proxmenux_oci import custom, i18n, registry_accounts as accounts, registry_auth as auth
from proxmenux_oci.ui import UserCancelled
COMPOSE = 'services:\n app:\n image: ghcr.io/owner/app:latest\n'
class UI:
def __init__(self, choices=(), answers=(), confirms=()):
self.choices = iter(choices)
self.answers = iter(answers)
self.confirms = iter(confirms)
self.seen = []
def choose(self, text, options, default=None, **kwargs):
self.seen.append((text, options, kwargs))
return next(self.choices)
def ask(self, text, default=None, **kwargs):
self.seen.append(text)
return next(self.answers)
def password(self, text):
self.seen.append(text)
raise UserCancelled(text)
def confirm(self, text, default=False, **kwargs):
self.seen.append((text, default))
return next(self.confirms)
def review(self, text, title=None, **kwargs):
self.seen.append((text, title, kwargs))
return False
def message(self, text, title=None):
self.seen.append((text, title))
info = message
class CustomRegistryLocalizationTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.base = Path(self.temp.name)
(self.base / 'lang').mkdir()
(self.base / 'config.json').write_text('{"language":"sk"}')
(self.base / 'lang/sk.json').write_bytes((ROOT / 'lang/sk.json').read_bytes())
for name, value in [('BASE_DIR', self.base), ('_language', None), ('_cache', None)]:
item = patch.object(i18n, name, value)
item.start()
self.addCleanup(item.stop)
# Any unmocked subprocess, network fetch or installation is a failure.
self.run = self.guard(custom.subprocess, 'run')
self.network = self.guard(custom.urllib.request, 'urlopen')
self.install = self.guard(custom, 'install_template')
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):
self.run.assert_not_called()
self.network.assert_not_called()
self.install.assert_not_called()
def manager(self):
return SimpleNamespace(path=Path('/fixture/auth.json'),
accounts=MagicMock(return_value=[{'registry': 'ghcr.io', 'account': 'owner',
'method': 'Používateľské meno / token', 'status': 'Saved'}]),
login=MagicMock(), logout=MagicMock())
def menu(self, ui, manager):
with patch.object(accounts, 'require_root'):
accounts.registry_accounts_menu(ui, auth=manager)
def test_dynamic_blockers_are_extracted_and_rendered_in_slovak(self):
expected = {
'multi-service-compose': 'definuje viac než jeden obraz',
'native-multi-lxc-orchestrator-not-yet-implemented': 'definuje viac než jeden obraz',
'top-level-configs': 'používa nastavenia Compose configs, ktoré tu nemajú ekvivalent',
'devices-format': 'požadované zariadenia sú zadané v nepodporovanom formáte',
'healthcheck-format': 'kontrola stavu je zadaná v nepodporovanom formáte',
'stop-grace-period-format': 'časový limit zastavenia je zadaný v nepodporovanom formáte',
'shm-size-format': 'veľkosť zdieľanej pamäte je zadaná v nepodporovanom formáte',
'ulimits-format-or-resource': 'požadované limity systémových prostriedkov nemožno uplatniť',
'mem-limit-format-or-below-proxmox-minimum': 'požadovaný limit pamäte nemožno uplatniť',
'mem-limit-deploy-consistency-review': 'požaduje dva rozdielne limity pamäte',
}
extractor = runpy.run_path(str(ROOT / '.github/scripts/build_translation_cache.py'))
extracted = extractor['extract_python_texts']([ROOT / 'oci/src'])
self.assertTrue(set(custom.BLOCKER_TEXTS.values()) <= set(extracted))
for code, message in expected.items():
with self.subTest(code=code):
self.assertEqual(custom.blocker_text('service:app:' + code), message)
self.assertEqual(custom.blocker_text('unrecognized:raw-code'), 'unrecognized:raw-code')
self.no_effects()
def test_cancelled_source_menu_is_translated_without_reading_an_image(self):
ui = UI([None])
with self.assertRaises(UserCancelled) as caught:
custom.explore(ui)
self.assertEqual(str(caught.exception), 'Výber obrazu bol zrušený')
self.assertIn('Obraz, ktorý nie je v katalógu', repr(ui.seen))
self.assertIn('Len odkaz na obraz bez súboru Compose', repr(ui.seen))
self.no_effects()
def test_invalid_compose_inputs_reach_localized_error_before_registry_or_install(self):
cases = [
('services: {}', 'Súbor Compose neobsahuje žiadnu službu'),
('services: [', 'Súbor Compose nie je platný YAML:'),
(COMPOSE + ' environment: 42\n', 'Položka environment'),
(COMPOSE + ' volumes: {}\n', 'Položka volumes'),
(COMPOSE + ' ports: invalid\n', 'Položka ports'),
(COMPOSE + ' ports: [invalid]\n', 'Neplatný port'),
(COMPOSE + ' volumes: [relative:relative]\n', 'Neplatná cesta dátového úložiska'),
(COMPOSE + 'x-casaos:\n main: missing\n', 'Položka x-casaos.main'),
]
for text, expected in cases:
with self.subTest(text=text), patch.object(custom, 'read_definition', return_value=(text, 'fixture')), \
self.assertRaises(custom.ConversionError) as caught:
custom.explore(UI())
self.assertIn(expected, str(caught.exception))
self.no_effects()
def test_converter_raw_contract_and_english_ui_are_preserved(self):
bad = COMPOSE + ' ports: [invalid]\n'
with self.assertRaises(custom.ConversionError) as raw:
custom.template_from_compose(bad)
self.assertEqual(str(raw.exception), "Puerto no valido: 'invalid'")
with patch.object(i18n, '_language', 'en'), \
patch.object(custom, 'read_definition', return_value=(bad, 'fixture')), \
self.assertRaises(custom.ConversionError) as ui_error:
custom.explore(UI())
self.assertEqual(str(ui_error.exception), str(raw.exception))
self.no_effects()
def test_conversion_message_retains_literal_diagnostic_data_and_unknown_fallback(self):
detail = " {'target': '<path>{literal}\\á'}"
message = custom.conversion_message(custom.ConversionError('Ruta de volumen no valida:' + detail))
self.assertEqual(message, 'Neplatná cesta dátového úložiska v súbore Compose:' + detail)
unknown = 'unknown <diagnostic> {literal}'
self.assertEqual(custom.conversion_message(custom.ConversionError(unknown)), unknown)
self.no_effects()
def test_all_legacy_converter_diagnostics_have_slovak_ui_and_unchanged_english(self):
messages = [
'environment must be a list or a mapping', 'volumes must be a list', 'ports must be a list',
'The CasaOS Compose file has no services', 'The CasaOS document is not a Compose mapping',
'The Compose file has no x-casaos metadata', 'x-casaos.main no identifica un servicio valido',
'The main CasaOS service declares no image', 'Ruta de volumen no valida: /data',
'Puerto no valido: invalid', 'Cannot normalise the CasaOS identifier: ???',
'Docker Compose CasaOS no valido: fixture',
]
for text in messages:
with self.subTest(text=text):
self.assertNotEqual(custom.conversion_message(custom.ConversionError(text)), text)
with patch.object(i18n, '_language', 'en'):
self.assertEqual(custom.conversion_message(custom.ConversionError(text)), text)
self.no_effects()
def test_custom_review_cancel_keeps_installation_disabled(self):
ui = UI()
with patch.object(custom, 'read_definition', return_value=(COMPOSE, 'vloženého súboru Compose')), \
patch.object(custom, 'registry_report', return_value=(True, 'Dostupný obraz')):
custom.explore(ui)
self.assertIn('Nainštalovať tento obraz?', repr(ui.seen))
self.assertIn('Obraz:', repr(ui.seen))
self.assertIn('ghcr.io/owner/app:latest', repr(ui.seen))
self.no_effects()
def test_unavailable_image_explains_registry_accounts_without_authentication_or_install(self):
ui = UI()
with patch.object(custom, 'read_definition', return_value=(COMPOSE, 'fixture')), \
patch.object(custom, 'registry_report', return_value=(False, 'Nedostupný obraz')), \
patch.object(custom, '_registry_refused', return_value=False), \
patch.object(accounts, 'offer_sign_in') as offer:
custom.explore(ui)
self.assertIn('prihláste cez Účty registrov', repr(ui.seen))
offer.assert_not_called()
self.no_effects()
def test_registry_metadata_dialog_uses_image_and_digest_and_retains_values(self):
manager = self.manager()
ui = UI(['account-0', 'inspect', None], ['ghcr.io/owner/app:latest'])
metadata = dict(architecture='amd64', version='<v>{literal}', size=1024, digest='sha256:' + 'a'*64)
with patch.object(accounts, 'inspect_registry_image', return_value=metadata) as inspect:
self.menu(ui, manager)
inspect.assert_called_once_with('ghcr.io/owner/app:latest')
rendered = repr(ui.seen)
for value in ('Prístup k obrazu bol potvrdený.', 'Digest: sha256:', '<v>{literal}', 'Účty registrov'):
self.assertIn(value, rendered)
self.assertNotIn('obráz', rendered)
manager.login.assert_not_called()
manager.logout.assert_not_called()
self.no_effects()
def test_credential_help_retains_permissions_and_no_encryption_claim(self):
manager = self.manager()
ui = UI(['storage', None])
self.menu(ui, manager)
rendered = repr(ui.seen)
for value in ('root', '700', '600', 'base64', 'nie sú šifrované', 'každý uzol klastra', '/fixture/auth.json'):
self.assertIn(value, rendered)
manager.login.assert_not_called()
manager.logout.assert_not_called()
self.no_effects()
def test_cancelled_password_and_declined_logout_do_not_change_accounts(self):
for ui in (UI(['add', None], ['ghcr.io', 'owner']), UI(['account-0', 'logout', None], confirms=[False])):
manager = self.manager()
self.menu(ui, manager)
manager.login.assert_not_called()
manager.logout.assert_not_called()
self.no_effects()
def test_mismatched_registry_is_rejected_before_download(self):
manager = self.manager()
ui = UI(['account-0', 'download', None], ['quay.io/owner/app:latest'])
with patch.object(accounts, 'download_registry_image') as download:
self.menu(ui, manager)
download.assert_not_called()
self.assertIn('Zadajte obraz z tohto registra.', repr(ui.seen))
self.no_effects()
def test_declined_private_image_signin_does_not_even_construct_auth_manager(self):
ui = UI(confirms=[False])
with patch.object(accounts, 'RegistryAuth') as manager, patch.object(accounts, 'require_root') as root:
self.assertFalse(accounts.offer_sign_in(ui, 'ghcr.io/owner/app:latest'))
manager.assert_not_called()
root.assert_not_called()
self.assertIn('Prihlásiť sa teraz?', repr(ui.seen))
self.no_effects()
def test_registry_error_classes_hide_untrusted_remote_data_in_both_languages(self):
cases = [('status code: 429', 'obmedzuje počet požiadaviek'),
('requested access to the resource is denied', 'odmietol prístup'),
('manifest unknown', 'Obraz alebo tag'),
('unsupported manifest', 'metadáta obrazu'),
('docker-credential-helper', 'správu prihlasovacích údajov'),
('unknown failure', 'Obraz sa nepodarilo načítať')]
for language in ('sk', 'en'):
for error, expected in cases:
with self.subTest(language=language, error=error), patch.object(i18n, '_language', language):
message = auth.registry_error('registry.test:5000', error + ' REMOTE_PRIVATE_VALUE')
self.assertNotIn('REMOTE_PRIVATE_VALUE', message)
if language == 'sk':
self.assertIn(expected, message)
if 'credential' not in error:
self.assertIn('registry.test:5000', message)
self.no_effects()
if __name__ == '__main__':
unittest.main()
@@ -0,0 +1,176 @@
"""Cancel a real editor through Menu/CLI/Monitor contracts, never apply it."""
from contextlib import redirect_stdout
import copy
import io
import json
from pathlib import Path
import sys
from unittest.mock import patch
from test_oci_update_cancellation import CancellationFixture, ROOT
from proxmenux_oci import cli, management as menu, recreation
from proxmenux_oci.ui import UserCancelled
class ModifyCancellationTests(CancellationFixture):
def setUp(self):
super().setUp()
self.record.update({
'installation_id': 'fixture-oci-edit-cancel',
'observed': {'config_sha256': 'fixture-config-hash'},
'template': {'id': 'fixture-app', 'catalog_ui': {'title': 'Fixture'},
'container_contract': {'image': {'reference': 'example/fixture:latest'}}},
'deployment': {'vmid': 110, 'hostname': 'fixture',
'resources': {'cores': 2, 'memory_mb': 512},
'rootfs': {'storage': 'fixture-storage', 'size_gb': 8},
'mounts': [], 'devices': [], 'environment': [
{'name': 'PASSWORD', 'value': 'private-fixture-value', 'sensitive': True}],
'network': {'bridge': 'vmbr-fixture', 'ipv4': 'dhcp'}}})
self.original = copy.deepcopy(self.record)
self.ui.back_enabled = False
self.path_write = self.guard(Path, 'write_text')
self.path_bytes = self.guard(Path, 'write_bytes')
def script(self, choices, answers=(), confirms=(), reviews=()):
self.ui.choose.side_effect = choices
self.ui.ask.side_effect = answers
self.ui.confirm.side_effect = confirms
self.ui.review.side_effect = reviews
def no_effects(self):
super().no_effects()
self.path_write.assert_not_called()
self.path_bytes.assert_not_called()
self.assertFalse(self.ui.back_enabled)
def test_main_cancel_is_distinct_falsy_cancellation(self):
self.script([None])
self.assertIs(menu.manage_instance(ROOT/'oci', self.ui, self.row, action='modify'), menu.CANCELLED)
self.assertFalse(menu.CANCELLED)
self.no_effects()
def test_main_discard_without_edits_returns_cancel_code(self):
self.script(['discard'])
self.assertEqual(menu.direct_management(ROOT/'oci', 110, 'modify'), 6)
self.ui.confirm.assert_not_called()
self.no_effects()
def test_real_cli_cancel_returns_code_six_without_extra_console_output(self):
self.script([None])
with patch.object(cli, 'Catalog'), redirect_stdout(io.StringIO()) as output:
self.assertEqual(cli.main(['manage', '110', '--action', 'modify']), 6)
self.assertEqual(output.getvalue(), '')
self.no_effects()
def test_real_cli_discard_after_edits_never_calls_backup_or_lifecycle(self):
self.script(['cores', 'discard'], ['3'], [True])
with patch.object(cli, 'Catalog'), redirect_stdout(io.StringIO()):
self.assertEqual(cli.main(['manage', '110', '--action', 'modify']), 6)
self.ui.confirm.assert_called_once_with('Discard unapplied changes?', False)
self.no_effects()
def test_main_cancel_after_edits_requires_discard_confirmation(self):
self.script(['cores', None], ['3'], [True])
self.assertEqual(menu.direct_management(ROOT/'oci', 110, 'modify'), 6)
self.ui.confirm.assert_called_once_with('Discard unapplied changes?', False)
self.no_effects()
def test_declining_discard_preserves_draft_until_later_main_cancel(self):
self.script(['cores', 'discard', None], ['3'], [False, True])
self.assertEqual(menu.direct_management(ROOT/'oci', 110, 'modify'), 6)
last_options = dict(self.ui.choose.call_args.args[1])
self.assertTrue(last_options['cores'].rstrip().endswith('3'))
self.assertIn('Unapplied changes', self.ui.choose.call_args.args[0])
self.assertEqual(self.ui.confirm.call_count, 2)
self.no_effects()
def test_section_cancel_returns_to_editor_before_main_cancel(self):
self.script(['cores', None], [UserCancelled()])
self.assertEqual(menu.direct_management(ROOT/'oci', 110, 'modify'), 6)
self.assertEqual(self.ui.choose.call_count, 2)
self.assertIn('No pending changes', self.ui.choose.call_args.args[0])
self.ui.review.assert_not_called()
self.no_effects()
def test_zero_cores_rejected_then_section_and_main_cancel_are_safe(self):
self.script(['cores', None], ['0', UserCancelled()])
self.assertEqual(menu.direct_management(ROOT/'oci', 110, 'modify'), 6)
self.assertIn('Cores: minimum 1', self.ui.message.call_args.args[0])
self.assertIn('No pending changes', self.ui.choose.call_args.args[0])
self.no_effects()
def test_review_no_returns_to_editor_with_draft_then_discard_cancels(self):
self.script(['cores', 'apply', 'discard'], ['3'], [True], [False])
self.assertEqual(menu.direct_management(ROOT/'oci', 110, 'modify'), 6)
self.assertEqual(self.ui.choose.call_count, 3)
self.assertIn('Unapplied changes', self.ui.choose.call_args.args[0])
self.assertNotIn('private-fixture-value', self.ui.review.call_args.args[0])
self.no_effects()
def test_native_menu_returns_to_list_after_editor_cancel(self):
self.script(['110', 'modify', None, 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, 4)
self.no_effects()
def test_legacy_editor_callers_still_receive_none_on_cancel(self):
for choice in (None, 'discard'):
with self.subTest(choice=choice):
self.script([choice])
self.assertIsNone(recreation.edit_recreation(self.record, self.ui))
self.no_effects()
def test_unchanged_review_is_not_mislabeled_as_user_cancellation(self):
self.script(['apply'])
self.assertIsNone(recreation.edit_recreation(self.record, self.ui, cancelled_result=menu.CANCELLED))
self.ui.message.assert_called_once_with('Nothing was changed.', 'Modify OCI')
self.no_effects()
def test_real_editor_error_is_not_converted_to_cancellation(self):
self.script(['cores'], [RuntimeError('fixture editor failure')])
with self.assertRaisesRegex(RuntimeError, 'fixture editor failure'):
menu.direct_management(ROOT/'oci', 110, 'modify')
self.no_effects()
def test_real_cli_validation_failure_remains_failure_one(self):
self.script(['cores'], [cli.InstallError('fixture validation failure')])
with patch.object(cli, 'Catalog'), redirect_stdout(io.StringIO()):
self.assertEqual(cli.main(['manage', '110', '--action', 'modify']), 1)
self.no_effects()
def test_unattended_modify_is_rejected_before_opening_the_editor(self):
self.assertEqual(menu.direct_management(ROOT/'oci', 110, 'modify', unattended=True), 1)
self.ui.choose.assert_not_called()
self.no_effects()
def test_accepted_proposal_keeps_original_backend_and_private_payload(self):
self.script(['cores', 'apply'], ['3'], reviews=[True])
self.backup.side_effect = None
self.backup.return_value = True
def lifecycle(command, title):
self.assertEqual(command[:3], [sys.executable, str(ROOT/'oci/remote/oci_update_current.py'), '110'])
proposal_file = Path(command[command.index('--proposal')+1])
self.assertEqual(proposal_file.stat().st_mode & 0o777, 0o600)
proposal = json.loads(proposal_file.read_text())
self.assertEqual(proposal['operation'], 'recreate')
self.assertEqual(proposal['candidate']['deployment']['resources']['cores'], 3)
self.assertEqual(proposal['base_config_sha256'], 'fixture-config-hash')
self.assertNotIn('private-fixture-value', ' '.join(command))
return True
self.lifecycle.side_effect = lifecycle
self.assertEqual(menu.direct_management(ROOT/'oci', 110, 'modify'), 0)
self.backup.assert_called_once()
self.lifecycle.assert_called_once()
self.assertEqual(self.record, self.original)
self.write.assert_not_called()
def test_failed_backup_after_review_remains_failure_without_lifecycle(self):
self.script(['cores', 'apply'], ['3'], reviews=[True])
self.backup.side_effect = None
self.backup.return_value = False
self.assertEqual(menu.direct_management(ROOT/'oci', 110, 'modify'), 1)
self.lifecycle.assert_not_called()
self.write.assert_not_called()
self.assertEqual(self.record, self.original)
@@ -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}
@@ -0,0 +1,246 @@
"""Pending recovery through real Menu/CLI/Dialog paths; all host effects forbidden."""
from contextlib import redirect_stdout
import copy
import io
import json
from pathlib import Path
import subprocess
import sys
from types import SimpleNamespace
from unittest.mock import patch
from test_oci_update_cancellation import CancellationFixture, ROOT
from proxmenux_oci import cli, i18n, management as menu, ui as native_ui
import oci_instances as instances
import oci_instance_transaction as transaction
import oci_native_modify as native_modify
import oci_stack_transaction as stack_transaction
import oci_ui
ACTUAL_RECOVER = transaction.recover
class PendingRecoveryFixture(CancellationFixture):
def setUp(self):
super().setUp()
self.record.update(pending_transaction='/fiction-only/transaction.json')
self.row['pending'] = True
self.ui.choose.return_value = 'recover'
self.original = copy.deepcopy(self.record)
self.writes = [self.guard(Path, name) for name in ('write_text', 'write_bytes', 'unlink')]
self.processes = [self.guard(subprocess, name) for name in ('run', 'Popen', 'check_output')]
self.recovery = self.guard(transaction, 'recover')
self.native_recovery = self.guard(native_modify, 'recover')
self.stack_recovery = self.guard(stack_transaction, 'execute')
def stack(self):
self.row['stack'] = True
self.record['pending_stack_transaction'] = '/fiction-only/stack.json'
self.original = copy.deepcopy(self.record)
def no_effects(self):
super().no_effects()
for mock in [*self.writes, *self.processes, self.recovery, self.native_recovery, self.stack_recovery]:
mock.assert_not_called()
class PendingRecoveryTests(PendingRecoveryFixture):
def test_single_no_is_distinct_falsy_cancellation(self):
self.assertIs(menu.manage_instance(ROOT/'oci', self.ui, self.row), menu.CANCELLED)
self.assertFalse(menu.CANCELLED)
self.no_effects()
def test_single_menu_cancel_is_distinct_cancellation(self):
self.ui.choose.return_value = None
self.assertIs(menu.manage_instance(ROOT/'oci', self.ui, self.row), menu.CANCELLED)
self.ui.review.assert_not_called()
self.no_effects()
def test_real_cli_single_no_and_cancel_return_six_without_duplicate_output(self):
for choice in ('recover', None):
with self.subTest(choice=choice), patch.object(cli, 'Catalog'), redirect_stdout(io.StringIO()) as output:
self.ui.choose.return_value = choice
self.assertEqual(cli.main(['manage', '110', '--action', 'update']), 6)
self.assertEqual(output.getvalue(), '')
self.no_effects()
def test_real_cli_stack_no_returns_six_before_reading_the_journal(self):
self.stack()
with patch.object(cli, 'Catalog'), patch.object(Path, 'read_text', side_effect=AssertionError('Journal read after No')) as read:
self.assertEqual(cli.main(['manage', '110', '--action', 'update']), 6)
read.assert_not_called()
self.no_effects()
def test_actual_dialog_no_and_escape_cancel_single_and_stack(self):
dialog = native_ui.DialogUI()
for stacked in (False, True):
if stacked:
self.stack()
for code in (1, 255):
with self.subTest(stack=stacked, code=code):
answers = ([] if stacked else [subprocess.CompletedProcess('dialog', 0, stderr='2')])
answers.append(subprocess.CompletedProcess('dialog', code, stderr=''))
with patch.object(native_ui, 'interactive_ui', return_value=dialog), \
patch.object(dialog, '_run', side_effect=answers):
self.assertEqual(menu.direct_management(ROOT/'oci', 110, 'update'), 6)
self.no_effects()
def test_actual_dialog_cancel_and_escape_from_single_operation_menu(self):
dialog = native_ui.DialogUI()
for code in (1, 255):
with self.subTest(code=code), patch.object(native_ui, 'interactive_ui', return_value=dialog), \
patch.object(dialog, '_run', return_value=subprocess.CompletedProcess('dialog', code, stderr='')):
self.assertEqual(menu.direct_management(ROOT/'oci', 110, 'update'), 6)
self.no_effects()
def test_native_menu_returns_to_its_list_after_pending_single_or_stack_no(self):
for stacked in (False, True):
if stacked:
self.stack()
self.ui.choose.reset_mock()
self.ui.choose.side_effect = ['110', None] if stacked else ['110', 'recover', None]
with self.subTest(stack=stacked), 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, 2 if stacked else 3)
self.no_effects()
def test_confirmed_single_recovery_and_status_preserve_exact_commands_and_results(self):
self.ui.review.return_value = True
for action in ('recover', 'status'):
for success, code in ((True, 0), (False, 1)):
self.lifecycle.reset_mock()
self.lifecycle.side_effect = None
self.lifecycle.return_value = success
self.ui.choose.return_value = action
with self.subTest(action=action, result=success):
self.assertEqual(menu.direct_management(ROOT/'oci', 110, 'update'), code)
self.lifecycle.assert_called_once_with(
[sys.executable, str(ROOT/'oci/remote/oci_instance_transaction.py'), action, '110'],
'Recover OCI' if action == 'recover' else 'OCI management')
self.write.assert_not_called(); self.backup.assert_not_called()
self.assertEqual(self.record, self.original)
def test_confirmed_stack_recovery_preserves_command_and_failure_is_not_cancel(self):
self.stack()
self.ui.review.return_value = True
for success, code in ((True, 0), (False, 1)):
self.lifecycle.reset_mock()
self.lifecycle.side_effect = None
self.lifecycle.return_value = success
with self.subTest(result=success), patch.object(Path, 'read_text', return_value=json.dumps({'plan': {'members': [{'vmid': 110}, {'vmid': 111}]}})):
self.assertEqual(menu.direct_management(ROOT/'oci', 110, 'update'), code)
self.lifecycle.assert_called_once_with(
[sys.executable, str(ROOT/'oci/remote/oci_stack_native.py'), '110', '--recover', '--acknowledge-external-data'],
'Recover OCI stack')
self.write.assert_not_called(); self.backup.assert_not_called()
self.assertEqual(self.record, self.original)
def test_identity_failure_stays_failure_and_does_not_offer_recovery(self):
with patch.object(menu, 'check_selected', return_value={**self.row, 'reason': 'different-installation'}):
self.assertEqual(menu.direct_management(ROOT/'oci', 110, 'update'), 1)
self.ui.choose.assert_not_called(); self.ui.review.assert_not_called()
self.no_effects()
def test_unattended_does_not_turn_pending_menu_into_an_automatic_restore(self):
self.assertEqual(menu.direct_management(ROOT/'oci', 110, 'update', unattended=True), 1)
self.no_effects()
def test_registration_cli_retains_its_separate_entry_point(self):
with patch.object(cli, 'Catalog'), patch.object(menu, 'direct_recovery', return_value=0) as register, \
patch.object(menu, 'direct_management', side_effect=AssertionError('Registration routed to rollback')):
self.assertEqual(cli.main(['recover']), 0)
register.assert_called_once_with(ROOT/'oci')
self.no_effects()
class RecoverySlovakMeaning(PendingRecoveryFixture):
def test_actual_recovery_dialogs_preserve_host_data_and_bounded_recovery_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)
text = self.ui.review.call_args.args[0]
self.assertIn('Zdieľané hostiteľské adresáre sa nevrátia späť', text)
self.assertIn('po úspešnom vyčistení', text)
self.assertEqual(self.ui.choose.call_args.args[0], 'Prerušená operácia')
self.assertEqual(dict(self.ui.choose.call_args.args[1])['status'], 'Zobraziť stav')
self.stack()
menu.manage_instance(ROOT/'oci', self.ui, self.row)
self.assertEqual(self.ui.review.call_args.args[1], 'Obnoviť OCI aplikáciu s viacerými kontajnermi')
self.assertIn('môžu zlyhať', self.ui.review.call_args.args[0])
self.assertEqual(self.ui.review.call_args.kwargs['question'], 'Obnoviť alebo dokončiť operáciu?')
self.no_effects()
class RecoveryGuardMessages(PendingRecoveryFixture):
def setUp(self):
super().setUp()
cache = json.loads((ROOT/'lang/sk.json').read_text())
for name, value in (('_language', 'sk'), ('_cache', cache)):
p = patch.object(oci_ui, name, value)
p.start(); self.addCleanup(p.stop)
self.record.update(installation_id='fiction-installation', transaction_id='fiction-tx')
self.original = copy.deepcopy(self.record)
self.journal = Path(self.record['pending_transaction'])
self.state = {'id': 'fiction-tx', 'vmid': 110, 'phase': 'backup-ready',
'record': {'installation_id': 'fiction-installation'}, 'before_config': 'before',
'backup': None, 'backup_sha256': 'saved-sha'}
def rejected(self, text, *, node='fiction-host', changed_backup=False):
calls = []
def readonly(*args):
calls.append(args)
if args[0] == 'pvesh':
return json.dumps([{'type': 'lxc', 'vmid': 110, 'node': node}]).encode()
if args == ('pct', 'config', '110'):
return b'changed'
raise AssertionError('Unexpected recovery effect: ' + repr(args))
with patch.object(Path, 'read_text', return_value=json.dumps(self.state)), \
patch.object(Path, 'exists', return_value=True), patch.object(Path, 'is_file', return_value=True), \
patch.object(transaction.socket, 'gethostname', return_value='fiction-host'), \
patch.object(transaction.host_mounts, 'verify_sources'), patch.object(transaction.gpu_devices, 'verify'), \
patch.object(transaction, 'owned', return_value=b'fiction config'), \
patch.object(transaction, 'filehash', return_value='changed-sha' if changed_backup else 'saved-sha'), \
patch.object(transaction, 'run', side_effect=readonly), redirect_stdout(io.StringIO()):
# Only reads before the rejecting guard are supplied. Any next
# command or attempt to persist state fails this fixture.
with self.assertRaisesRegex(ValueError, text):
ACTUAL_RECOVER(ROOT/'oci', self.journal)
self.no_effects()
return calls
def test_foreign_transaction_is_rejected_before_probing_the_container(self):
self.state['record']['installation_id'] = 'different-installation'
self.assertEqual(self.rejected('Záznam nepatrí k tejto operácii'), [])
def test_missing_verified_backup_cannot_touch_a_changed_container(self):
self.assertEqual(len(self.rejected('Chýba overená záloha; zmenený kontajner zostáva nedotknutý')), 2)
def test_changed_backup_is_rejected_before_decompression_or_restore(self):
self.state['backup'] = '/fiction-only/backup.zst'
self.assertEqual(len(self.rejected('Záloha bola zmenená; obnova zablokovaná', changed_backup=True)), 2)
def test_guest_on_another_node_is_never_overwritten(self):
self.assertEqual(len(self.rejected('kontajner je na inom uzle; nie je prepísaný', node='another-node')), 1)
def test_incomplete_stack_recovery_state_fails_before_any_adapter_call(self):
adapter = SimpleNamespace(validate=self.recovery, stop=self.native_recovery,
restore=self.stack_recovery)
state = {'phase': 'recovery-failed', 'plan': {'start_order': [110, 111]},
'running': {'110': True}, 'stop_intent': True, 'replacement_intent': False}
before = copy.deepcopy(state)
with self.assertRaisesRegex(ValueError, 'Záznam operácie obsahuje neúplný stav obnovy'):
stack_transaction.recover_state(self.journal, state, adapter)
self.assertEqual(state, before)
self.no_effects()
def test_critical_safety_and_untouched_result_have_the_correct_meaning(self):
cache = json.loads((ROOT/'lang/sk.json').read_text())
self.assertEqual(cache['A host mount is not part of the journal; recovery blocked'],
'Pripojenie adresára hostiteľa nie je uvedené v zázname operácie; obnova je zablokovaná')
self.assertEqual(cache['Recovery completed. The container had not been modified yet.'],
'Obnova dokončená. Kontajner ešte nebol zmenený.')
self.assertEqual(cache['A backup was modified'], 'Záloha bola zmenená')
self.assertEqual(cache['Recovering the previous installation'], 'Obnovovanie predchádzajúcej inštalácie')
self.assertEqual(cache['Recover the previous installation'], 'Obnoviť predchádzajúcu inštaláciu')
self.no_effects()
@@ -0,0 +1,234 @@
"""Restored-app registration through real Menu/CLI/Dialog; no host effects."""
from contextlib import nullcontext, redirect_stdout
import copy
import io
import json
from pathlib import Path
import string
import subprocess
import tempfile
import unittest
from unittest.mock import patch
from test_oci_update_cancellation import CancellationFixture, ROOT
from proxmenux_oci import cli, i18n, management as menu, ui as native_ui
import oci_restore_recovery as restored
import oci_ui
class RegistrationFixture(CancellationFixture):
def setUp(self):
super().setUp()
self.found = [{'vmid':110, 'hostname':'fiction-primary'}]
self.original_found = copy.deepcopy(self.found)
self.ui.confirm.return_value = False
item = patch.object(menu, 'restored_applications', side_effect=lambda project:self.found)
self.pending = item.start();self.addCleanup(item.stop)
self.firewall = self.guard(menu, '_restored_firewall_rules')
self.backend = [self.guard(restored, name) for name in ('recover','register','run')]
self.files = [self.guard(Path, name) for name in ('write_text','write_bytes','unlink')]
self.processes = [self.guard(subprocess, name) for name in ('run','Popen','check_output')]
def members(self):
self.found.append({'vmid':111,'hostname':'fiction-dependent'})
self.original_found = copy.deepcopy(self.found)
def no_effects(self):
super().no_effects()
self.firewall.assert_not_called()
self.safe_backend()
def safe_backend(self):
for item in [*self.backend,*self.files,*self.processes]:item.assert_not_called()
self.assertEqual(self.found,self.original_found)
def accepted(self, answers, completed=True):
self.ui.confirm.side_effect=answers
self.firewall.side_effect=None
self.firewall.return_value=[{'vmid':row['vmid'],'port':61208 if row['vmid']==110 else 19999,
'source':'192.0.2.0/24'} for row in self.found]
self.lifecycle.side_effect=None;self.lifecycle.return_value=completed
self.pending.side_effect=lambda project:[] if completed and self.lifecycle.called else self.found
class RegistrationCancellationTests(RegistrationFixture):
def test_offer_no_is_explicit_falsy_cancellation(self):
self.assertIs(menu.offer_recovery(ROOT/'oci',self.ui),menu.CANCELLED)
self.assertFalse(menu.CANCELLED);self.no_effects()
def test_direct_single_no_finishes_before_a_third_pending_read(self):
self.assertEqual(menu.direct_recovery(ROOT/'oci'),6)
self.assertEqual(self.pending.call_count,2)
self.no_effects()
def test_real_cli_single_no_returns_six_without_output(self):
with patch.object(cli,'Catalog'),redirect_stdout(io.StringIO()) as output:
self.assertEqual(cli.main(['recover']),6)
self.assertEqual(output.getvalue(),'');self.no_effects()
def test_real_cli_multi_container_no_is_not_a_guest_action(self):
self.members()
with patch.object(cli,'Catalog'):
self.assertEqual(cli.main(['recover']),6)
text=self.ui.confirm.call_args.args[0]
self.assertIn('CT 110',text);self.assertIn('CT 111',text)
self.no_effects()
def test_actual_dialog_initial_no_and_escape_for_single_and_multi(self):
dialog=native_ui.DialogUI()
for multi in (False,True):
if multi:self.members()
for code in (1,255):
with self.subTest(multi=multi,code=code),patch.object(native_ui,'interactive_ui',return_value=dialog), \
patch.object(dialog,'_run',return_value=subprocess.CompletedProcess('dialog',code,stderr='')):
self.assertEqual(menu.direct_recovery(ROOT/'oci'),6)
self.no_effects()
def test_native_menu_continues_to_list_after_registration_no(self):
self.ui.choose.return_value=None
with patch.object(menu,'_clean_orphans'),patch.object(menu,'_records_unread',return_value=[]):
menu.interactive_management(ROOT/'oci',self.ui)
self.ui.choose.assert_called_once();self.no_effects()
def test_optional_firewall_and_start_no_registers_without_flags(self):
self.members();self.accepted([True,False,False,False])
self.assertEqual(menu.direct_recovery(ROOT/'oci'),0)
self.assertEqual(self.lifecycle.call_args.args[0],
[menu.sys.executable,str(ROOT/'oci/remote/oci_restore_recovery.py'),'recover'])
self.assertEqual(self.ui.confirm.call_count,4);self.safe_backend()
def test_single_optional_no_preserves_registration_without_start(self):
self.accepted([True,False,False])
self.assertEqual(menu.direct_recovery(ROOT/'oci'),0)
self.assertEqual(self.lifecycle.call_args.args[0],
[menu.sys.executable,str(ROOT/'oci/remote/oci_restore_recovery.py'),'recover'])
self.assertEqual(self.ui.confirm.call_count,3);self.safe_backend()
def test_only_one_confirmed_firewall_is_forwarded_and_no_start(self):
self.members();self.accepted([True,False,True,False])
self.assertEqual(menu.direct_recovery(ROOT/'oci'),0)
self.assertEqual(self.lifecycle.call_args.args[0][-2:],['--host-firewall','111'])
self.assertNotIn('--start',self.lifecycle.call_args.args[0]);self.safe_backend()
def test_explicit_start_yes_retains_flag_but_never_starts_any_guest_here(self):
self.accepted([True,False,True])
self.assertEqual(menu.direct_recovery(ROOT/'oci'),0)
self.assertEqual(self.lifecycle.call_args.args[0][-1],'--start');self.safe_backend()
def test_failed_registration_stays_failure_with_pending_records(self):
self.accepted([True,False,False],completed=False)
self.assertEqual(menu.direct_recovery(ROOT/'oci'),1)
self.safe_backend()
def test_empty_list_is_success_without_confirmation_or_backend(self):
self.found.clear();self.original_found=[]
self.assertEqual(menu.direct_recovery(ROOT/'oci'),0)
self.ui.confirm.assert_not_called();self.ui.message.assert_called_once();self.no_effects()
def test_root_identity_failure_never_probes_restored_apps(self):
with patch.object(menu.os,'geteuid',return_value=1000):
self.assertEqual(menu.direct_recovery(ROOT/'oci'),1)
self.pending.assert_not_called();self.ui.confirm.assert_not_called();self.no_effects()
def test_empty_offer_does_not_cancel_the_native_menu(self):
self.found.clear();self.original_found=[]
self.assertIsNone(menu.offer_recovery(ROOT/'oci',self.ui))
self.ui.confirm.assert_not_called();self.no_effects()
def test_empty_registration_state_is_rendered_in_slovak_without_recovery(self):
catalog=json.loads((ROOT/'lang/sk.json').read_text())
with patch.object(i18n,'_language','sk'),patch.object(i18n,'_cache',catalog):
self.found.clear();self.original_found=[]
self.assertEqual(menu.direct_recovery(ROOT/'oci'),0)
self.assertEqual(self.ui.message.call_args.args[0],
'Žiadna OCI aplikácia obnovená zo zálohy nečaká na opätovnú registráciu.')
self.no_effects()
def test_initial_single_and_multi_dialog_describes_registration_precheck(self):
catalog=json.loads((ROOT/'lang/sk.json').read_text())
for multi in (False,True):
if multi:self.members()
with self.subTest(multi=multi),patch.object(i18n,'_language','sk'),patch.object(i18n,'_cache',catalog):
self.assertEqual(menu.direct_recovery(ROOT/'oci'),6)
text=self.ui.confirm.call_args.args[0]
self.assertIn('Pred opätovnou registráciou',text)
self.assertNotIn('Obnova najprv',text)
self.assertIn('aplikácie sa potom nespúšťajú automaticky',text)
self.no_effects()
def test_real_backend_blocker_stays_error_and_uses_registration_wording(self):
catalog=json.loads((ROOT/'lang/sk.json').read_text())
plan={'title':'Fictional restored application','restored':[{'vmid':110}],
'blockers':['Fictional missing resource'],'notes':[]}
with patch.object(menu.sys,'argv',['fixture','recover','--root','/fiction-only']), \
patch.object(restored.instances,'locked',return_value=nullcontext()), \
patch.object(restored,'prepare',return_value=[plan]), \
patch.object(oci_ui,'_language','sk'),patch.object(oci_ui,'_cache',catalog), \
patch.object(restored,'msg_error') as error,patch.object(restored,'msg_info2'):
self.assertEqual(restored.main(),1)
error.assert_called_once_with('Túto aplikáciu zatiaľ nemožno znova zaregistrovať:')
self.no_effects()
def test_real_host_resource_check_reports_missing_data_without_creating_anything(self):
catalog=json.loads((ROOT/'lang/sk.json').read_text())
plan={'restored':[{'vmid':110,'config':'mp0: /fiction-only/data,mp=/data\n',
'copy':{'record':{'deployment':{},'observed':{}}}}], 'blockers':[],'notes':[]}
with patch.object(oci_ui,'_language','sk'),patch.object(oci_ui,'_cache',catalog), \
patch.object(restored.os.path,'isdir',return_value=False), \
patch.object(restored,'firewall_rule',return_value=None):
restored.check_host_resources(plan)
self.assertEqual(len(plan['blockers']),1)
self.assertIn('/fiction-only/data',plan['blockers'][0])
self.assertIn('registráciu zopakujte',plan['blockers'][0])
self.no_effects()
class RegistrationContractTests(unittest.TestCase):
def test_real_monitor_shell_recover_keeps_global_cli_contract_and_exit_six(self):
with tempfile.TemporaryDirectory() as folder:
base=Path(folder);(base/'scripts/oci').mkdir(parents=True)
(base/'utils.sh').write_text('load_language() { :; }\ninitialize_cache() { :; }\n')
(base/'scripts/oci/oci_manager_apps.sh').write_text('printf "%s\\n" "$@"\nexit 6\n')
script=(ROOT/'scripts/oci/manage_instance.sh').read_text()
script=script.replace('/usr/local/share/proxmenux',folder)
import os
result=subprocess.run(['bash','-c',script],capture_output=True,text=True,
env={**os.environ,'ACTION':'recover','VMID':'not-a-vmid'})
self.assertEqual(result.returncode,6);self.assertEqual(result.stdout,'recover\n')
self.assertEqual(result.stderr,'')
def test_curated_slovak_registration_words_and_format_fields(self):
catalog=json.loads((ROOT/'lang/sk.json').read_text())
expected={
'Recover restored OCI applications':'Registrácia',
'No restored OCI application is waiting to be recovered.':'registráciu',
'Recover them now?':'zaregistrovať',
'This application cannot be recovered yet:':'zaregistrovať',
'The application could not be recovered:':'zaregistrovať',
'Start the applications once they are registered? Answer No if the original containers are still running on another host: both would use the same addresses.':'vyberte Nie',
'it is locked or its disk could not be mounted':'uzamknutý',
'The local storage does not accept snippets':'local',
'it uses an NVIDIA GPU and this host has no working NVIDIA driver and Container Toolkit. Install them and run the recovery again.':'NVIDIA Container Toolkit',
'the NVIDIA hook of the container is not available:':'spúšťací skript',
'its Rclone mount is published again on this host at':'v ceste',
'Choose another bridge for the container in Proxmox (Network) and run the recovery again.':'registráciu',
'Mount or create it with its data and run the recovery again.':'registráciu',
'Connect it, or remove it from the container in Proxmox (Resources), and run the recovery again.':'registráciu',
}
fields=lambda text:sorted(f for _,f,_,_ in string.Formatter().parse(text) if f)
for key,word in expected.items():
with self.subTest(key=key):
self.assertIn(word,catalog[key]);self.assertEqual(fields(key),fields(catalog[key]))
def test_registration_web_text_is_distinct_from_previous_installation_recovery(self):
for locale in ('en','sk'):
data=json.loads((ROOT/'AppImage/messages'/locale/'common.json').read_text())['vmLxc']
with self.subTest(locale=locale):
self.assertEqual(data['ociUpdates']['registerRestored'],
'Znova zaregistrovať' if locale=='sk' else 'Register again')
self.assertEqual(data['ociUpdates']['recover'],'Obnoviť' if locale=='sk' else 'Recover')
self.assertIn('zaregistrovať' if locale=='sk' else 'Register again',data['updates']['ociRestoredBody'])
self.assertIn('Zaregistrovať' if locale=='sk' else 'Register',data['ociUpdates']['terminalTitleRecover'])
if __name__=='__main__':unittest.main()
@@ -0,0 +1,235 @@
"""Actual removal UI/guards in SK and EN; real host effects are forbidden."""
from contextlib import nullcontext
import copy
import json
from pathlib import Path
import subprocess
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 i18n, management
from proxmenux_oci.ui import UserCancelled
import oci_instances as instances
import oci_remove as removal
import oci_ui
class RemovalSlovakTests(unittest.TestCase):
def setUp(self):
cache = json.loads((ROOT / 'lang/sk.json').read_text())
for module in (i18n, oci_ui):
for name, value in (('_language', 'sk'), ('_cache', cache)):
p = patch.object(module, name, value)
p.start(); self.addCleanup(p.stop)
self.records = {}
self.make_records(2)
self.read = self.start(instances, 'read', side_effect=lambda root, vmid: self.records[vmid])
self.start(management, '_selected_configuration', side_effect=self.config)
self.bridge = self.start(removal, 'bridge_in_use', return_value=False)
self.ui = Mock()
self.ui.review.return_value = False
self.forbidden = []
for module, name in ((subprocess, 'run'), (subprocess, 'Popen'), (subprocess, 'check_output'),
(Path, 'write_text'), (Path, 'write_bytes'), (Path, 'unlink'),
(removal.shutil, 'rmtree'), (instances, 'write'), (instances, 'command'),
(management, '_run_lifecycle'), (removal, 'run'),
(removal, 'remove_host_state'), (removal.oci_console, 'remove_log'),
(removal, 'release_bridge'), (removal, 'remove_owned_host_firewall'),
(removal, 'release_shared_host_files'), (removal.work_backup, 'forget'),
(removal.image_cache, 'prune')):
self.forbidden.append(self.start(module, name, side_effect=AssertionError('Forbidden effect: ' + name)))
def start(self, module, name, **kwargs):
p = patch.object(module, name, **kwargs)
result = p.start(); self.addCleanup(p.stop)
return result
def make_records(self, count):
self.records.clear()
for offset in range(count):
vmid = 999990 + offset
self.records[vmid] = {'vmid': vmid, 'installation_id': 'fixture-' + str(vmid),
'template': {'catalog_ui': {'title': 'Fixture App'}},
'deployment': {'mounts': [{'type': 'host-bind', 'source': '/fixture/saved',
'container_path': '/saved'}]}}
if offset:
self.records[vmid]['stack_member'] = {'primary_vmid': 999990}
if count > 1:
self.records[999990]['stack'] = {'members': [{'vmid': vmid} for vmid in self.records],
'deployment': {'network': {'private_bridge': 'vmbr999', 'private_subnet': '10.77.99.0/24'}}}
self.records[999990]['deployment']['host_firewall'] = {'port': 8080}
self.original = copy.deepcopy(self.records)
def config(self, vmid):
return (f'hostname: fixture-{vmid}\nrootfs: fixture:vm-{vmid}-disk-0,size=8G\n'
f'mp0: fixture:vm-{vmid}-disk-1,mp=/data,size=16G\n'
'mp1: /fixture/shared,mp=/shared\n').encode()
def tearDown(self):
self.assertEqual(self.records, self.original)
for effect in self.forbidden:
effect.assert_not_called()
def test_single_summary_separates_deleted_disks_and_retained_host_paths(self):
self.make_records(1)
text = management._removal_summary(ROOT / 'oci', 999990)
self.assertIn('Kontajnery určené na odstránenie:', text)
self.assertIn('Údaje kontajnera určené na vymazanie:', text)
self.assertIn('/data (16G)', text)
kept = text.split('Cesty hostiteľa nájdené')[1]
self.assertIn('/fixture/shared', kept); self.assertIn('/fixture/saved', kept)
self.assertNotIn('/fixture/shared', text.split('Cesty hostiteľa nájdené')[0])
self.assertNotIn('súkromnej siete:', text)
def test_primary_summary_counts_are_grammatical_for_two_three_and_five(self):
for count in (2, 3, 5):
with self.subTest(count=count):
self.make_records(count)
text = management._removal_summary(ROOT / 'oci', 999990)
self.assertIn(f'Počet kontajnerov aplikácie Fixture App: {count}.', text)
self.assertIn('Všetky sú určené na odstránenie.', text)
for vmid in self.records:
self.assertIn(f'CT {vmid}', text)
def test_member_summary_warns_that_entire_application_is_selected(self):
text = management._removal_summary(ROOT / 'oci', 999991)
self.assertIn('CT 999991 patrí do aplikácie Fixture App (počet kontajnerov: 2).', text)
self.assertIn('odstránenie celej aplikácie', text)
self.assertNotIn('nádob', text)
def test_free_bridge_is_only_targeted_and_firewall_removal_is_conditional(self):
text = management._removal_summary(ROOT / 'oci', 999990)
self.assertIn('Súkromná sieť určená na uvoľnenie, ak ju nepoužíva iná VM alebo iný CT: vmbr999', text)
self.assertIn('Môže sa odstrániť aj zodpovedajúce spravované pravidlo firewallu hostiteľa.', text)
def test_shared_bridge_is_kept_and_not_reported_as_released(self):
self.bridge.return_value = True
text = management._removal_summary(ROOT / 'oci', 999990)
self.assertIn('Súkromná sieť zachovaná', text)
self.assertNotIn('určená na uvoľnenie', text)
def test_missing_config_still_discloses_saved_host_paths_without_deleting(self):
with patch.object(management, '_selected_configuration', side_effect=OSError('fixture missing')):
text = management._removal_summary(ROOT / 'oci', 999990)
self.assertIn('/fixture/saved', text)
self.assertIn('rootfs -', text)
def test_no_from_real_remove_review_never_calls_lifecycle(self):
self.assertFalse(management._remove(ROOT / 'oci', self.ui, 999991))
review = self.ui.review.call_args
self.assertFalse(review.kwargs['default'])
self.assertEqual(review.kwargs['question'],
'Odstrániť aplikáciu? Jej kontajnerové disky sa vymažú a obnoviť ich možno iba zo zálohy.')
def test_cancelled_real_remove_review_never_calls_lifecycle(self):
self.ui.review.side_effect = UserCancelled('fixture cancelled')
with self.assertRaises(UserCancelled):
management._remove(ROOT / 'oci', self.ui, 999990)
def test_preparation_error_does_not_show_confirmation_or_call_lifecycle(self):
with patch.object(removal, 'members_of', side_effect=ValueError('fixture record invalid')):
self.assertFalse(management._remove(ROOT / 'oci', self.ui, 999990))
self.ui.review.assert_not_called()
self.assertIn('Odstránenie nebolo možné pripraviť:', self.ui.message.call_args.args[0])
def test_pending_operation_error_precedes_configuration_probe_and_all_mutations(self):
self.make_records(1)
self.records[999990]['pending_transaction'] = {'fixture': True}
self.original = copy.deepcopy(self.records)
with patch.object(removal, 'guest_config') as config:
with self.assertRaisesRegex(ValueError, 'Operácia tejto inštalácie sa ešte nedokončila'):
removal.remove(Path('/fixture/oci'), 999990)
config.assert_not_called()
def test_container_on_other_node_errors_before_host_cleanup(self):
with patch.object(removal, 'guest_config', return_value=None), \
patch.object(removal, 'guest_node', return_value='fixture-other-node'):
with self.assertRaisesRegex(ValueError, 'Kontajner beží na inom uzle klastra'):
removal.remove(Path('/fixture/oci'), 999990)
def run_main(self, result=False, busy=False):
with patch.object(sys, 'argv', ['oci_remove.py', '999990']), \
patch.object(removal.os, 'geteuid', return_value=0), \
patch.object(instances, 'locked', side_effect=BlockingIOError if busy else None,
return_value=nullcontext()), \
patch.object(removal, 'remove', return_value=result) as mocked_remove, \
patch.object(removal, 'msg_ok') as ok, patch.object(removal, 'msg_error') as error:
code = removal.main()
return code, mocked_remove, ok, error
def test_busy_registry_error_does_not_enter_removal(self):
code, remove, ok, error = self.run_main(busy=True)
self.assertEqual(code, 1); remove.assert_not_called(); ok.assert_not_called()
self.assertEqual(error.call_args.args[0], 'Register inštancií používa iná operácia OCI')
def test_partial_completion_is_a_warning_to_review_not_unqualified_success(self):
code, remove, ok, error = self.run_main(result=True)
self.assertEqual(code, 0); error.assert_not_called()
self.assertEqual(ok.call_args.args[0],
'Príkaz na odstránenie sa dokončil; skontrolujte všetky upozornenia vyššie.')
self.assertNotEqual(ok.call_args.args[0], i18n.translate('The application was removed'))
def test_clean_completion_keeps_existing_success_contract(self):
code, remove, ok, error = self.run_main()
self.assertEqual(code, 0); error.assert_not_called()
self.assertEqual(ok.call_args.args[0], 'Aplikácia bola odstránená')
def test_english_primary_and_member_summaries_keep_identical_sentences(self):
with patch.object(i18n, '_language', 'en'):
primary = management._removal_summary(ROOT / 'oci', 999990)
member = management._removal_summary(ROOT / 'oci', 999991)
self.assertTrue(primary.startswith('Fixture App runs in 2 containers. All of them are targeted for removal.\n'))
self.assertTrue(member.startswith('CT 999991 is one of the 2 containers of Fixture App. It cannot be removed on its own,'))
def test_missing_translation_falls_back_to_the_same_english_summary(self):
with patch.object(i18n, '_cache', {}):
text = management._removal_summary(ROOT / 'oci', 999990)
self.assertTrue(text.startswith('Fixture App runs in 2 containers. All of them are targeted for removal.\n'))
def test_yes_only_builds_the_existing_command_when_lifecycle_is_inert(self):
self.ui.review.return_value = True
with patch.object(management, '_run_lifecycle', return_value=True) as inert:
self.assertTrue(management._remove(ROOT / 'oci', self.ui, 999990))
self.assertEqual(inert.call_args.args[0],
[sys.executable, str(ROOT / 'oci/remote/oci_remove.py'), '999990'])
def test_single_and_stack_menu_routes_use_the_translated_label_and_real_no_review(self):
for count in (1, 2):
with self.subTest(count=count):
self.make_records(count)
self.ui.reset_mock()
self.ui.choose.return_value = 'remove'
row = {'vmid': 999990, 'stack': count > 1, 'pending': False,
'status': 'installed', 'reason': 'matched'}
with patch.object(management, 'carry_records', return_value={}), \
patch.object(management, 'check_selected', return_value=row):
self.assertFalse(management.manage_instance(ROOT / 'oci', self.ui, row))
options = self.ui.choose.call_args.args[1]
self.assertIn(('remove', 'Odstrániť: zmazať aplikáciu a jej kontajnery'), options)
self.ui.review.assert_called_once()
self.assertFalse(self.ui.review.call_args.kwargs['default'])
def test_changed_identity_warning_does_not_claim_the_other_container_was_deleted(self):
self.make_records(1)
# Exercise the real backend branch; record cleanup is inert as well.
with patch.object(removal, 'guest_config', return_value=b'fixture'), \
patch.object(instances, 'identity', return_value='fixture-other-identity'), \
patch.object(removal, 'remove_owned_host_firewall', return_value=True), \
patch.object(removal, 'release_shared_host_files'), \
patch.object(removal.work_backup, 'forget'), \
patch.object(removal.image_cache, 'prune', return_value=[]), \
patch.object(Path, 'exists', return_value=False), \
patch.object(Path, 'is_dir', return_value=False), \
patch.object(removal, 'msg_warn') as warning, \
patch.object(removal, 'msg_ok') as ok, patch.object(removal, 'msg_info'):
self.assertTrue(removal.remove(Path('/fixture/oci'), 999990))
self.assertEqual(warning.call_args.args[0],
'VMID teraz patrí inému kontajneru; ten zostane nedotknutý: CT 999990')
self.assertNotIn('Odstránený kontajner:', str(ok.mock_calls))
if __name__ == '__main__':
unittest.main()
@@ -0,0 +1,239 @@
"""Exercise the multi-container editor in Slovak without applying a change."""
from contextlib import redirect_stderr
import copy
import io
import json
from pathlib import Path
import subprocess
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 host, i18n, installer, management, stack_recreation as editor
import oci_immich_recognition as recognition
import oci_instances as instances
import oci_stack_modify as modify
import oci_ui
CONFIG = {
'rootfs': 'fixture:vm-999991-disk-0,size=8G',
'mp0': 'fixture:vm-999991-disk-1,mp=/opt/recipes/mediafiles,size=8G',
'mp1': 'fixture:vm-999991-disk-2,mp=/extra,size=4G',
'mp2': '/fixture/shared,mp=/shared',
'dev0': 'path=/dev/ttyACM7,mode=0660',
}
class StackEditorSlovakTests(unittest.TestCase):
def setUp(self):
self.translations = json.loads((ROOT / 'lang/sk.json').read_text())
for module in (i18n, oci_ui):
for name, value in (('_language', 'sk'), ('_cache', self.translations)):
item = patch.object(module, name, value)
item.start()
self.addCleanup(item.stop)
self.member = {'vmid': 999991, 'stack_member': {'name': 'application'},
'deployment': {'replay_profile': {'adapter': 'install_tandoor_stack.sh',
'role': 'application'}}}
self.primary = {'vmid': 999990, 'stack': {'members': [self.member,
{'vmid': 999992, 'deployment': {'replay_profile': {'role': 'database'}}}]}}
self.original = copy.deepcopy(self.primary)
self.ui = Mock()
self.ui.choose.return_value = None
self.ui.confirm.return_value = False
self.ui.review.return_value = False
self.ui.checklist.side_effect = lambda text, options, defaults: list(defaults)
config = patch.object(editor, '_config', return_value=CONFIG)
config.start()
self.addCleanup(config.stop)
self.forbidden = []
for target, name in ((subprocess, 'run'), (subprocess, 'Popen'),
(subprocess, 'check_output'), (Path, 'write_text'),
(Path, 'write_bytes'), (instances, 'write'),
(instances, 'command'), (modify, 'apply'),
(recognition, 'apply'), (recognition, 'configure'),
(management, '_run_lifecycle')):
item = patch.object(target, name, side_effect=AssertionError('Forbidden effect: ' + name))
self.forbidden.append(item.start())
self.addCleanup(item.stop)
self.lifecycle = Mock(side_effect=AssertionError('Editor called lifecycle'))
self.forbidden.append(self.lifecycle)
def tearDown(self):
self.assertEqual(self.primary, self.original)
for callback in self.forbidden:
callback.assert_not_called()
def shown(self):
return '\n'.join(str(call) for call in self.ui.mock_calls)
def immich(self, acceleration='cpu'):
self.member['deployment']['replay_profile'] = {
'adapter': 'install_immich_stack.sh', 'role': 'machine-learning'}
self.member['deployment']['machine_learning'] = {'acceleration': acceleration}
self.original = copy.deepcopy(self.primary)
return self.member
def changes(self, **values):
return {'add_mounts': [], 'remove_mounts': [], 'add_devices': [],
'remove_devices': [], **values}
def test_editor_only_offers_extra_paths_and_leaves_database_out(self):
self.assertFalse(editor.modify_stack(ROOT / 'oci', self.ui, self.primary, self.lifecycle))
shown = self.shown()
self.assertIn('odznačené sa odstránia po potvrdení', shown)
self.assertIn('odznačené sa odpoja po potvrdení', shown)
options = self.ui.checklist.call_args_list[0].args[1]
self.assertEqual([path for path, label in options], ['/extra', '/shared'])
self.assertNotIn('zásobník', shown)
def test_declining_volume_deletion_keeps_it_and_only_detaches_host_path_in_draft(self):
self.ui.checklist.side_effect = [[], ['/dev/ttyACM7']]
self.assertFalse(editor.modify_stack(ROOT / 'oci', self.ui, self.primary, self.lifecycle))
warning = self.ui.confirm.call_args_list[0].args[0]
self.assertIn('trvalé úložisko aj všetky údaje v ňom', warning)
self.assertIn('/extra', warning)
summary = self.ui.review.call_args.args[0]
self.assertIn('- /shared', summary)
self.assertNotIn('- /extra', summary)
self.assertIn('Spustený kontajner sa reštartuje', summary)
self.assertIn('zastavený zostane zastavený', summary)
self.assertIn('Databáza ani ostatné kontajnery sa nemenia', summary)
def test_declined_review_describes_confirmed_volume_deletion_without_performing_it(self):
self.ui.checklist.side_effect = [[], []]
self.ui.confirm.side_effect = [True, False, False]
self.assertFalse(editor.modify_stack(ROOT / 'oci', self.ui, self.primary, self.lifecycle))
self.assertIn('- /extra', self.ui.review.call_args.args[0])
self.assertIn('vymažú aj údaje', self.ui.review.call_args.args[0])
self.assertEqual(self.ui.review.call_args.kwargs['question'], 'Použiť tieto zmeny?')
def test_member_selection_cancels_before_any_configuration_read(self):
other = copy.deepcopy(self.member)
other['vmid'] = 999993
self.primary['stack']['members'].append(other)
self.original = copy.deepcopy(self.primary)
with patch.object(editor, '_config') as read:
self.assertFalse(editor.modify_stack(ROOT / 'oci', self.ui, self.primary, self.lifecycle))
read.assert_not_called()
self.assertEqual(self.ui.choose.call_args.args[0], 'Kontajner, ktorý chcete upraviť')
def test_missing_member_error_uses_application_language(self):
self.primary['stack']['members'] = []
self.original = copy.deepcopy(self.primary)
self.assertFalse(editor.modify_stack(ROOT / 'oci', self.ui, self.primary, self.lifecycle))
self.assertIn('Pre túto aplikáciu nie je uložený zoznam kontajnerov.', self.shown())
self.assertIn('Správa OCI aplikácie s viacerými kontajnermi', self.shown())
def test_immich_action_menu_identifies_recognition_instead_of_video_transcoding(self):
self.immich()
self.assertFalse(editor.modify_stack(ROOT / 'oci', self.ui, self.primary, self.lifecycle))
self.assertIn('Vybrať CPU alebo GPU pre rozpoznávanie', self.shown())
self.assertIn('Upraviť pripojené cesty a zariadenia', self.shown())
def test_cpu_only_host_explains_why_no_gpu_choice_is_offered(self):
member = self.immich()
with patch.object(host, 'gpus', return_value={'intel': [], 'amd': [], 'nvidia': []}):
self.assertFalse(editor.change_recognition(ROOT / 'oci', self.ui, self.primary,
member, self.lifecycle))
self.assertIn('nenašla použiteľná grafická karta', self.shown())
self.assertIn('Rozpoznávanie zostáva na CPU.', self.shown())
self.ui.review.assert_not_called()
def test_gpu_review_explains_whole_application_update_and_kept_data_before_no(self):
member = self.immich()
self.ui.choose.return_value = 'openvino'
found = {'intel': ['/dev/dri/renderD128'], 'amd': [], 'nvidia': []}
with patch.object(host, 'gpus', return_value=found):
self.assertFalse(editor.change_recognition(ROOT / 'oci', self.ui, self.primary,
member, self.lifecycle))
choices = self.ui.choose.call_args.args[1]
self.assertEqual(choices, [('cpu', 'CPU'), ('openvino', 'Intel GPU')])
text = self.ui.review.call_args.args[0]
for fragment in ('CPU → Intel GPU', 'pamäť modelov, knižnica a databáza sa zachovajú',
'Celá aplikácia sa zastaví a aktualizuje', 'ProxMenux sa pokúsi obnoviť',
'ponuky správy OCI'):
self.assertIn(fragment, text)
self.assertEqual(self.ui.review.call_args.kwargs['question'],
'Použiť zvolený CPU/GPU pre rozpoznávanie?')
def test_unchanged_recognition_does_not_open_confirmation_or_apply(self):
member = self.immich()
self.ui.choose.return_value = 'cpu'
with patch.object(editor, 'recognition_choices', return_value=[('cpu', 'CPU', {}),
('cuda', 'NVIDIA GPU', {})]):
self.assertFalse(editor.change_recognition(ROOT / 'oci', self.ui, self.primary,
member, self.lifecycle))
self.ui.review.assert_not_called()
self.assertIn('Nič sa nezmenilo.', self.shown())
def test_experimental_rocm_can_be_declined_before_final_review(self):
member = self.immich()
self.ui.choose.return_value = 'rocm'
choices = [('cpu', 'CPU', {}), ('rocm', 'AMD GPU', {'experimental': True})]
with patch.object(editor, 'recognition_choices', return_value=choices), \
patch.object(host, 'amd_gpu_name', return_value='Fixture AMD'):
self.assertFalse(editor.change_recognition(ROOT / 'oci', self.ui, self.primary,
member, self.lifecycle))
text = self.ui.confirm.call_args.args[0]
self.assertIn('ROCm túto grafickú kartu oficiálne nepodporuje.', text)
self.assertIn('GPU prestať reagovať', text)
self.assertIn('bez preinštalovania', text)
self.assertIn('Použiť napriek tomu GPU s ROCm?', text)
self.assertFalse(self.ui.confirm.call_args.args[1])
self.ui.review.assert_not_called()
def test_backend_invalid_volume_size_preserves_path_and_fails_before_mutation(self):
mount = {'type': 'managed-volume', 'container_path': '/new-data',
'size_gb': 0, 'source': 'fixture'}
with patch.object(modify, 'entries', return_value={}):
with self.assertRaisesRegex(ValueError, 'Neplatná veľkosť trvalého úložiska: /new-data'):
modify.validate(999991, self.changes(add_mounts=[mount]))
def test_backend_unsupported_device_explains_separate_profile_before_mutation(self):
device = {'kind': 'character-device', 'host_path': '/dev/sda'}
with patch.object(modify, 'entries', return_value={}):
with self.assertRaisesRegex(ValueError, 'NVIDIA a celé stromy zariadení vyžadujú iný profil: /dev/sda'):
modify.validate(999991, self.changes(add_devices=[device]))
def test_root_error_is_localized_without_starting_the_operation(self):
with patch.object(modify.os, 'geteuid', return_value=1000), \
patch.object(sys, 'argv', ['fixture', '999991', '--changes', '/not-read.json']), \
redirect_stderr(io.StringIO()) as output:
with self.assertRaises(SystemExit) as result:
modify.main()
self.assertEqual(result.exception.code, 2)
self.assertIn('Vyžadujú sa oprávnenia používateľa root', output.getvalue())
self.assertNotIn('required', output.getvalue())
def test_invalid_recognition_profile_is_localized_without_any_gpu_probe(self):
with self.assertRaisesRegex(ValueError, 'Profil GPU pre Immich nebol overený'):
recognition.validate('unknown', None, None)
def test_recognition_errors_distinguish_unchanged_from_incomplete_operation(self):
incomplete = RuntimeError('fixture diagnostic')
incomplete.recognition_kept = False
for error, expected in ((ValueError('fixture diagnostic'), 'Spôsob rozpoznávania v Immichu sa nezmenil:'),
(incomplete, 'Zmenu CPU/GPU pre rozpoznávanie v Immichu sa nepodarilo dokončiť:')):
with self.subTest(error=type(error).__name__), \
patch.object(recognition.os, 'geteuid', return_value=0), \
patch.object(sys, 'argv', ['fixture', '999990', '--acceleration', 'cpu']), \
patch.object(recognition, 'change', side_effect=error), \
patch.object(recognition, 'msg_error') as message:
self.assertEqual(recognition.main(), 1)
self.assertEqual(message.call_args.args[0], expected + ' fixture diagnostic')
def test_english_and_missing_translation_fallback_preserve_source_text(self):
changes = self.changes(remove_mounts=['/extra'])
for language, cache in (('en', self.translations), ('sk', {})):
with self.subTest(language=language, empty_cache=not cache), \
patch.object(i18n, '_language', language), patch.object(i18n, '_cache', cache):
text = editor.summary(self.member, changes)
self.assertIn('A running container is restarted', text)
self.assertIn('- /extra', text)
if __name__ == '__main__':
unittest.main()
@@ -0,0 +1,196 @@
"""Coordinated update/recreate contracts; no host lifecycle is permitted."""
import copy
import json
from pathlib import Path
import subprocess
import sys
from unittest.mock import patch
from test_oci_update_cancellation import CancellationFixture, ROOT
from proxmenux_oci import cli, i18n, management as menu, ui as native_ui, updates
import oci_instances as instances
import oci_stack_plan as stack_plan
import oci_stack_transaction as transaction
import oci_ui
class StackFixture(CancellationFixture):
def setUp(self):
super().setUp()
self.row['stack'] = True
self.record.update(installation_id='fiction-main', stack={
'id': 'fiction-main', 'members': [
{'vmid': vmid, 'installation_id': identity, 'status': 'installed',
'stack_member': {'stack_id': 'fiction-main', 'primary_vmid': 110}}
for vmid, identity in [(110, 'fiction-main'), (111, 'fiction-dependent')]],
'deployment': {'services': [{'vmid': 111}, {'vmid': 110}]}})
self.original = copy.deepcopy(self.record)
self.extra_effects = [self.guard(Path, name) for name in ('write_text', 'write_bytes', 'unlink')]
self.extra_effects += [self.guard(subprocess, name) for name in ('run', 'Popen', 'check_output')]
self.extra_effects += [self.guard(transaction, 'execute'), self.guard(transaction, 'save'),
self.guard(menu.images, 'offer_removal')]
def no_effects(self):
super().no_effects()
for effect in self.extra_effects:
effect.assert_not_called()
class StackCancellationTests(StackFixture):
def test_update_and_recreate_no_are_falsy_distinct_cancellation(self):
for action in ('update', 'recreate'):
with self.subTest(action=action):
result = menu.manage_instance(ROOT/'oci', self.ui, self.row, action=action)
self.assertIs(result, menu.CANCELLED)
self.assertFalse(result)
self.no_effects()
def test_real_cli_cancel_returns_six_for_both_actions(self):
for action in ('update', 'recreate'):
with self.subTest(action=action), patch.object(cli, 'Catalog'):
self.assertEqual(cli.main(['manage', '110', '--action', action]), 6)
self.no_effects()
def test_actual_dialog_no_and_escape_return_six_before_any_effect(self):
dialog = native_ui.DialogUI()
for action in ('update', 'recreate'):
for code in (1, 255):
with self.subTest(action=action, code=code), \
patch.object(native_ui, 'interactive_ui', return_value=dialog), \
patch.object(dialog, '_run', return_value=subprocess.CompletedProcess('dialog', code, stderr='')):
self.assertEqual(menu.direct_management(ROOT/'oci', 110, action), 6)
self.no_effects()
def test_confirmation_defaults_and_member_count_are_preserved(self):
for action, default in [('update', True), ('recreate', False)]:
menu.manage_instance(ROOT/'oci', self.ui, self.row, action=action)
args, kwargs = self.ui.review.call_args
self.assertIn('2 containers', args[0])
self.assertIn('main CT: 110', args[0])
self.assertEqual(kwargs['default'], default)
self.no_effects()
def test_native_menu_returns_to_the_application_list_after_no(self):
for action in ('update', 'recreate'):
with self.subTest(action=action):
self.ui.choose.reset_mock()
self.ui.choose.side_effect = ['110', action, 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_cancel_never_checks_the_next_application(self):
available = [{'vmid': 110, 'title': 'First'}, {'vmid': 112, 'title': 'Second'}]
self.ui.checklist.return_value = ['all']
self.ui.review.side_effect = [True, False]
with patch.object(updates, 'UpdateCheck') as checker, \
patch.object(menu, 'saved_inventory', return_value=[self.row, {**self.row, 'vmid': 112}]), \
patch.object(menu, 'check_selected', side_effect=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_rejected_backup_remains_failure_one_without_dispatch(self):
self.ui.review.return_value = True
self.backup.side_effect = None
self.backup.return_value = False
for action in ('update', 'recreate'):
self.assertEqual(menu.direct_management(ROOT/'oci', 110, action), 1)
self.lifecycle.assert_not_called()
self.assertEqual(self.record, self.original)
def test_positive_confirmation_preserves_success_failure_and_exact_commands(self):
self.ui.review.return_value = True
self.backup.side_effect = None
self.backup.return_value = True
self.lifecycle.side_effect = None
for action in ('update', 'recreate'):
for result in (False, True):
self.lifecycle.reset_mock()
self.lifecycle.return_value = result
with patch.object(menu.images, 'offer_removal') as offer:
self.assertEqual(menu.direct_management(ROOT/'oci', 110, action,
lifecycle_args=('--keep-backup',)), 0 if result else 1)
expected = [sys.executable, str(ROOT/'oci/remote/oci_stack_native.py'), '110', '--keep-backup']
if action == 'recreate':
expected += ['--operation', 'recreate']
self.assertEqual(self.lifecycle.call_args.args[0], expected+['--acknowledge-external-data'])
self.assertEqual(offer.call_count, int(result))
self.assertEqual(self.record, self.original)
self.write.assert_not_called()
self.command.assert_not_called()
def test_acknowledgement_is_not_duplicated(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
menu.direct_management(ROOT/'oci', 110, 'recreate', lifecycle_args=('--acknowledge-external-data',))
self.assertEqual(self.lifecycle.call_args.args[0].count('--acknowledge-external-data'), 1)
class StackSlovakTests(StackFixture):
def setUp(self):
super().setUp()
self.catalog = json.loads((ROOT/'lang/sk.json').read_text())
for module in (i18n, oci_ui):
for name, value in [('_language', 'sk'), ('_cache', self.catalog)]:
item = patch.object(module, name, value)
item.start()
self.addCleanup(item.stop)
def test_questions_and_titles_name_the_whole_application(self):
for action, question in [('update', 'Aktualizovať celú aplikáciu?'),
('recreate', 'Znovu vytvoriť všetky kontajnery aplikácie?')]:
menu.manage_instance(ROOT/'oci', self.ui, self.row, action=action)
self.assertEqual(self.ui.review.call_args.kwargs['question'], question)
self.assertIn('aplikáciu s viacerými kontajnermi', self.ui.review.call_args.args[1])
self.no_effects()
def test_recreation_dialog_explains_saved_images_and_restoring_backups(self):
menu.manage_instance(ROOT/'oci', self.ui, self.row, action='recreate')
text = self.ui.review.call_args.args[0]
self.assertIn('uložených identifikátorov obrazov', text)
self.assertIn('2', text)
self.assertIn('110', text)
self.assertIn('zálohy sa obnovia', text)
self.assertNotIn('Stack', text)
self.no_effects()
def test_shared_missing_primary_error_in_plan_and_publish_before_any_effect(self):
primary = copy.deepcopy(self.record)
primary['vmid'] = 112
with self.assertRaisesRegex(ValueError, '^Chýba hlavný kontajner aplikácie$'):
stack_plan.build(primary, {}, {}, 'update')
with self.assertRaisesRegex(ValueError, '^Chýba hlavný kontajner aplikácie$'):
instances.publish_stack(Path('/fiction-only'), 110, {}, {}, [])
self.no_effects()
def test_missing_member_error_keeps_volume_verification_requirement(self):
records = {row['vmid']: row for row in self.record['stack']['members']}
with self.assertRaisesRegex(ValueError, 'znova ich vytvorte až po overení ich dátových zväzkov'):
stack_plan.build(self.record, records, {110: 'fiction-main'}, 'update')
self.no_effects()
def test_pending_and_conflicting_member_errors_keep_the_identifier(self):
records = {row['vmid']: copy.deepcopy(row) for row in self.record['stack']['members']}
inventory = {key: row['installation_id'] for key, row in records.items()}
records[111]['status'] = 'pending'
with self.assertRaisesRegex(ValueError, '^Nedokončená operácia existuje pre CT 111$'):
stack_plan.build(self.record, records, inventory, 'update')
records[111]['status'] = 'installed'
inventory[111] = 'different-installation'
with self.assertRaisesRegex(ValueError, '^Iná inštancia už používa VMID 111$'):
stack_plan.build(self.record, records, inventory, 'update')
self.no_effects()
def test_plan_failure_is_read_only_and_distinguishes_invalid_identity(self):
primary = copy.deepcopy(self.record)
primary['stack']['id'] = 'different-stack'
with self.assertRaisesRegex(ValueError, '^Nekonzistentná identita aplikácie s viacerými kontajnermi$'):
stack_plan.build(primary, {}, {}, 'recreate')
self.no_effects()
@@ -0,0 +1,210 @@
"""Slovak storage questions through real builders and UI; installation forbidden."""
from contextlib import redirect_stdout
import copy
import io
import json
from pathlib import Path
import subprocess
import sys
from types import SimpleNamespace
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, custom, custom_mounts, extra_devices, i18n, installer, ui as native_ui, updates
from proxmenux_oci.catalog import Catalog
class Recorder:
def __init__(self, answers=None):
self.answers=answers or {}
self.asked=[]
self.messages=[]
def _answer(self,text,default):
self.asked.append(text)
return self.answers.get(text,default)
def ask(self,text,default=None,**kwargs):return self._answer(text,default if default is not None else '')
def password(self,text,**kwargs):return self._answer(text,'fictional-value-for-tests-000000')
def choose(self,text,options,default=None,**kwargs):return self._answer(text,default)
def confirm(self,text,default=False,**kwargs):return self._answer(text,default)
def checklist(self,text,options,defaults=None,**kwargs):return self._answer(text,defaults or [])
def review(self,*args,**kwargs):return False
def info(self,text):pass
def message(self,text,*args):self.messages.append(text)
class StorageFixture(unittest.TestCase):
def setUp(self):
self.catalog=Catalog(ROOT/'oci')
self.sk=json.loads((ROOT/'lang/sk.json').read_text())
self.forbidden=[]
self.patch(i18n,'_language','sk')
self.patch(i18n,'_cache',self.sk)
self.patch(installer.host,'storages',side_effect=lambda content: [
{'storage':'local-lvm','type':'lvmthin','avail':100*2**30},
{'storage':'fiction-storage','type':'dir','avail':200*2**30}] if content=='rootdir' else [
{'storage':'local','type':'dir','avail':100*2**30}])
self.patch(installer.host,'bridges',return_value=[{'iface':'vmbr0','cidr':'192.0.2.10/24'}])
self.patch(installer.host,'timezone',return_value='Europe/Bratislava')
self.patch(installer.host,'gpus',return_value={'intel':[],'amd':[],'nvidia':False})
self.patch(installer.access,'ask_addresses',side_effect=lambda ui,bridge,names,*a:({name:'dhcp' for name in names},None))
self.patch(installer.access,'ask_ipv4',return_value=('dhcp',None))
for module,name in [(installer,'run_remote_install'),(cli,'run_remote_install'),
(cli.images,'offer_removal'),(subprocess,'run'),(subprocess,'Popen'),(subprocess,'check_output'),
(Path,'write_text'),(Path,'write_bytes'),(Path,'unlink')]:
self.forbidden.append(self.patch(module,name,side_effect=AssertionError('Forbidden effect: '+name)))
self.addCleanup(self.no_effects)
def patch(self,module,name,*args,**kwargs):
item=patch.object(module,name,*args,**kwargs);value=item.start();self.addCleanup(item.stop);return value
def no_effects(self):
for effect in self.forbidden:effect.assert_not_called()
def recorder(self,answers=None):
return Recorder({i18n.translate(key):value for key,value in (answers or {}).items()})
def build(self,app,answers=None,mode=installer.ADVANCED_MODE):
ui=self.recorder(answers)
plan=installer.build_deployment(self.catalog.compose(app),ui,mode)
return ui,plan
class StorageWordingTests(StorageFixture):
def test_tandoor_files_question_and_minimum_error_keep_the_limit(self):
ui,plan=self.build('tandoor',{'Files volume size in GB':'9'})
self.assertIn('Veľkosť dátového zväzku pre súbory v GB',ui.asked)
self.assertEqual(plan['media']['size_gb'],9)
with self.assertRaisesRegex(installer.InstallError,'Dátový zväzok súborov Tandoor musí mať najmenej 2 GB'):
self.build('tandoor',{'Files volume size in GB':'1'})
def test_paperless_storage_documents_and_consume_export_names(self):
ui,plan=self.build('paperless-ngx',{'Documents volume size in GB':'30',
'Where to store the consume and export folders':'managed-volume','Size of each consume/export volume in GB':'3'})
for text in ['Úložisko pre údaje a dokumenty aplikácie Paperless',
'Kam uložiť priečinky consume a export',
'Veľkosť dátového zväzku pre dokumenty v GB',
'Veľkosť každého dátového zväzku pre consume a export v GB']:
self.assertIn(text,ui.asked)
self.assertEqual((plan['media_size_gb'],plan['transfer']['size_gb'],plan['transfer']['backup']),(30,3,True))
def test_paperless_volume_limit_and_transfer_limit_errors(self):
with self.assertRaisesRegex(installer.InstallError,'Trvalé dátové zväzky aplikácie Paperless musia mať najmenej 8 GB'):
self.build('paperless-ngx',{'Documents volume size in GB':'7'})
with self.assertRaisesRegex(installer.InstallError,'Dátové zväzky pre consume a export musia mať najmenej 1 GB'):
self.build('paperless-ngx',{'Where to store the consume and export folders':'managed-volume',
'Size of each consume/export volume in GB':'0'})
def test_postgresql_question_and_error_are_shared_by_immich_and_nextcloud(self):
for app in ('immich','nextcloud-stack'):
with self.subTest(app=app):
ui,plan=self.build(app,{'PostgreSQL volume size in GB':'12'})
self.assertIn('Veľkosť dátového zväzku PostgreSQL v GB',ui.asked)
self.assertEqual(plan['database_size_gb'],12)
with self.assertRaisesRegex(installer.InstallError,'Dátový zväzok PostgreSQL musí mať najmenej 8 GB'):
self.build(app,{'PostgreSQL volume size in GB':'7'})
def test_host_directory_remains_outside_the_container_backup(self):
ui,plan=self.build('nextcloud-stack',{'Where to store the Nextcloud files, configuration and data':'host-bind',
'Shared host directory':'/fiction-only/existing-data'})
self.assertIn('Zdieľaný adresár hostiteľa',ui.asked)
self.assertEqual((plan['application']['mode'],plan['application']['host_path'],plan['application']['backup']),
('host-bind','/fiction-only/existing-data',False))
def test_default_and_advanced_builders_preserve_language_independent_plans(self):
for app in ('jellyfin','immich','nextcloud-stack','paperless-ngx','tandoor'):
for mode in (installer.DEFAULT_MODE,installer.ADVANCED_MODE):
with self.subTest(app=app,mode=mode):
_ui,slovak=self.build(app,mode=mode)
with patch.object(i18n,'_language','en'):
_ui,english=self.build(app,mode=mode)
self.assertEqual(slovak,english)
def test_multi_path_cancel_and_invalid_selection_preserve_all_members(self):
services=[{'name':'main','main':True,'deployment':{'mounts':[]}},
{'name':'database','deployment':{'mounts':[]}}]
original=copy.deepcopy(services)
ui=self.recorder({'Add extra paths to this stack':True,'Containers that will receive this path':[]})
with self.assertRaisesRegex(ValueError,'^Vyberte aspoň jeden kontajner aplikácie$'):
custom_mounts.ask_stack_custom_mounts(ui,services,'fiction-storage')
self.assertIn('Pridať ďalšie cesty do aplikácie s viacerými kontajnermi',ui.asked)
self.assertEqual(services,original)
ui=self.recorder({'Add extra paths to this stack':True,'Containers that will receive this path':['main'],
'Data location':None})
with self.assertRaises(native_ui.UserCancelled):custom_mounts.ask_stack_custom_mounts(ui,services,'fiction-storage')
self.assertEqual(services,original)
def test_shared_selection_error_in_device_flow_also_preserves_members(self):
services=[{'name':'main','main':True,'deployment':{}}];original=copy.deepcopy(services)
ui=self.recorder()
with patch.object(extra_devices,'ask_extra_devices',return_value=[{'kind':'usb','host_path':'/fiction-only/device'}]), \
patch.object(ui,'checklist',return_value=[]):
with self.assertRaisesRegex(ValueError,'^Vyberte aspoň jeden kontajner aplikácie$'):
extra_devices.ask_stack_extra_devices(ui,services)
self.assertEqual(services,original)
def test_selected_path_is_added_only_to_selected_member_with_backup(self):
services=[{'name':'main','main':True,'deployment':{'mounts':[]}},
{'name':'database','deployment':{'mounts':[]}}]
ui=self.recorder({'Add extra paths to this stack':True,'Containers that will receive this path':['main']})
custom_mounts.ask_stack_custom_mounts(ui,services,'fiction-storage')
self.assertEqual(services[1]['deployment']['mounts'],[])
mount=services[0]['deployment']['mounts'][0]
self.assertEqual((mount['source'],mount['container_path'],mount['size_gb'],mount['backup']),
('fiction-storage','/media-extra',8,True))
self.assertIn('Pridať ďalšiu cestu do aplikácie s viacerými kontajnermi',ui.asked)
def test_rclone_shared_directory_question_is_localized_without_creating_it(self):
ui=self.recorder({'Shared directory for copy/sync operations':'/fiction-only/rclone'})
with patch.object(ui,'confirm',return_value=True):
plan=installer.build_deployment(self.catalog.compose('rclone'),ui,installer.ADVANCED_MODE)
self.assertIn('Zdieľaný adresár na kopírovanie a synchronizáciu',ui.asked)
self.assertTrue(any(m.get('source')=='/fiction-only/rclone' and not m['backup'] for m in plan['mounts']))
class InputAndCancellationTests(StorageFixture):
def test_dialog_required_value_and_password_use_slovak_then_retry(self):
dialog=native_ui.DialogUI()
empty=subprocess.CompletedProcess('dialog',0,stderr='')
value=subprocess.CompletedProcess('dialog',0,stderr='fiction-value')
with patch.object(dialog,'_run',side_effect=[empty,value]),patch.object(dialog,'message') as message:
self.assertEqual(dialog.ask('fiction-field'),'fiction-value')
message.assert_called_once_with('Táto hodnota je povinná.')
with patch.object(dialog,'_run',side_effect=[empty,value,value]),patch.object(dialog,'message') as message:
self.assertEqual(dialog.password('fiction-field'),'fiction-value')
message.assert_called_once_with('Táto hodnota je povinná.')
def test_terminal_required_value_and_password_are_also_localized(self):
terminal=native_ui.TerminalUI()
with patch('builtins.input',side_effect=['','fiction-value']),redirect_stdout(io.StringIO()) as output:
self.assertEqual(terminal.ask('fiction-field'),'fiction-value')
self.assertIn('Táto hodnota je povinná.',output.getvalue())
with patch.object(native_ui.getpass,'getpass',side_effect=['','fiction-value','fiction-value']),redirect_stdout(io.StringIO()) as output:
self.assertEqual(terminal.password('fiction-field'),'fiction-value')
self.assertIn('Táto hodnota je povinná.',output.getvalue())
def test_real_install_flow_storage_cancel_and_final_no_never_dispatch(self):
template=self.catalog.compose('tandoor');original=copy.deepcopy(template)
dialog=native_ui.DialogUI()
for code in (1,255):
replies=[subprocess.CompletedProcess('dialog',0,stderr=value) for value in ('','fiction-stack','2','2048')]
replies.append(subprocess.CompletedProcess('dialog',code,stderr=''))
with self.subTest(code=code),patch.object(dialog,'_run',side_effect=replies) as run:
with self.assertRaises(native_ui.UserCancelled):cli.install_template(dialog,template,'tandoor',installer.ADVANCED_MODE)
self.assertEqual(run.call_count,5)
self.assertEqual(template,original)
self.assertIsNone(cli.install_template(self.recorder(),template,'tandoor',installer.ADVANCED_MODE))
self.assertEqual(template,original)
def test_numeric_retry_keeps_previous_answers_and_already_correct_slovak(self):
ui=Recorder();wizard=native_ui.BacktrackUI(ui)
wizard.answers=[('ask','previous'),('ask','abc')];wizard.cursor=2
self.assertTrue(wizard.retry_last(ValueError("invalid literal for int() with base 10: 'abc'")))
self.assertEqual(ui.messages,["Hodnota musí byť číslo: 'abc'\n\nZadajte hodnotu znova."])
self.assertEqual((wizard.answers,wizard.cursor),([('ask','previous')],0));wizard.close()
def test_real_menu_error_handler_has_slovak_heading_and_preserves_detail(self):
ui=Mock();ui.choose.side_effect=['custom',None]
with patch.object(cli,'interactive_ui',return_value=ui),patch.object(custom,'explore',side_effect=ValueError('fiction-detail')), \
patch.object(updates,'UpdateCheck') as checker:
check=checker.return_value.__enter__.return_value
check.active=False;check.notice.return_value='';check.status.return_value={'available':[]}
self.assertEqual(cli.interactive(self.catalog),0)
ui.message.assert_called_once_with('Operáciu sa nepodarilo dokončiť:\n\nfiction-detail')
@@ -0,0 +1,168 @@
"""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()
@@ -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),
+1 -1
View File
@@ -121,7 +121,7 @@ class OciWordingTests(TestCase):
def test_failed_registry_inspection_points_to_registry_accounts(self):
ui = SimpleNamespace(message=Mock())
fn = function(CUSTOM, 'explore', read_definition=lambda _: ('definition', 'source'),
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'),
@@ -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)
@@ -0,0 +1,250 @@
"""PBS CLI language and no-effect boundaries using real functions, isolated files."""
import contextlib
import copy
import importlib.util
import io
import json
import os
from pathlib import Path
import subprocess
import tempfile
import unittest
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[3]
def function(path, name):
text = (ROOT/path).read_text()
start = text.index(name+'() {')
return text[start:text.index('\n}\n', start)+3]
def repository(components, suite='trixie'):
return dict(Types=['deb'], URIs=['https://credential.invalid/private'], Suites=[suite],
Components=components.split(), Enabled=True)
class RepositoryCliTests(unittest.TestCase):
def run_policy(self, case, language='sk', mode='plan', malformed=None, product='pbs'):
spec = importlib.util.spec_from_file_location('policy_fixture', ROOT/'scripts/global/repository_policy.py')
module = importlib.util.module_from_spec(spec); spec.loader.exec_module(module)
state = dict(files=[dict(path='/etc/apt/sources.list.d/pbs.sources', repositories=[repository('pbs-enterprise')]),
dict(path='/etc/apt/sources.list.d/debian.sources', repositories=[repository('main')])],
errors=[], digest='fixture', **{'standard-repos':[{'handle':'no-subscription'}]})
status='notfound'; calls=[]
if case=='inventory': state['errors']=['PRIVATE RAW DIAGNOSTIC']
if case=='status': status='unknown'
if case=='suite': state['files'][0]['repositories'][0]['Suites']=['bookworm']
if case=='active': status='active'; state['files']=state['files'][1:]
if case=='debian': state['files']=state['files'][:1]
if case=='mixed': state['files'][0]['repositories'][0]['Components'].append('main')
if case=='handle': state['standard-repos']=[]
if case=='preserve': state['files'][0]['repositories'][0]['Components']=['pbs-no-subscription']
original=copy.deepcopy(state)
def run(args):
calls.append(args)
if args[0]=='proxmox-backup-manager': return json.dumps(dict(status=status, serverid='PRIVATE SERVER ID'))
if 'create' in args:
state['files'][0]['repositories'][0]['Enabled']=False
if case=='partial': state['errors']=['PRIVATE RAW DIAGNOSTIC']
return ''
if 'set' in args:
state['files'].append(dict(path='/etc/apt/sources.list.d/public.sources',repositories=[repository('pbs-no-subscription')]))
return ''
return json.dumps(state)
with tempfile.TemporaryDirectory(prefix='pbs-policy-i18n-') as folder:
module.BASE=Path(folder)
(module.BASE/'lang').mkdir()
(module.BASE/'config.json').write_text(json.dumps({'language':language}))
catalog=(ROOT/'lang/sk.json').read_text() if malformed is None else malformed
(module.BASE/'lang/sk.json').write_text(catalog)
stdout,stderr=io.StringIO(),io.StringIO()
suite='sid' if case=='unsupported' else 'trixie'
with patch.object(module,'run',side_effect=run), patch.object(module.sys,'argv',['repository_policy.py',mode,suite,product]), contextlib.redirect_stdout(stdout),contextlib.redirect_stderr(stderr):
result=module.main()
return result,stdout.getvalue(),stderr.getvalue(),calls,state,original
def test_pbs_errors_are_localized_without_changing_decisions_or_exposing_raw_data(self):
expectations={'inventory':'Zoznam repozitárov APT', 'status':'Stav predplatného',
'suite':'iné vydanie', 'active':'aktívne predplatné', 'debian':'základný repozitár Debianu',
'mixed':'spoločný záznam APT', 'handle':'no-subscription', 'unsupported':'Nepodporované vydanie',
'partial':'stav repozitárov je neznámy'}
for case,expected in expectations.items():
with self.subTest(case=case):
result,out,error,calls,state,original=self.run_policy(case,mode='apply' if case=='partial' else 'plan')
self.assertEqual(result,1); self.assertEqual(out,''); self.assertIn(expected,error)
self.assertNotIn('PRIVATE',error); self.assertNotIn('credential.invalid',error)
writes=[c for c in calls if 'create' in c or 'set' in c]
if case=='partial':
self.assertEqual(len(writes),1)
self.assertFalse(state['files'][0]['repositories'][0]['Enabled'])
self.assertIn('Automatický návrat',error)
else:
self.assertEqual(writes,[]); self.assertEqual(state,original)
if case=='unsupported': self.assertEqual(calls,[])
def test_english_error_and_stdout_tokens_remain_compatible(self):
en=self.run_policy('unsupported',language='en')
self.assertEqual(en[2],'Unsupported Proxmox suite; no repository changed.\n')
pve=self.run_policy('unsupported',product='pve')
self.assertEqual(pve[2],en[2])
for mode,expected in [('plan','offer\n'),('apply','changed\n')]:
result,out,error,*_=self.run_policy('offer',mode=mode)
self.assertEqual((result,out,error),(0,expected,''))
result,out,error,calls,*_=self.run_policy('preserve',mode='apply')
self.assertEqual((result,out,error),(0,'preserve\n',''))
self.assertFalse(any(c[0]=='proxmox-backup-manager' or 'set' in c or 'create' in c for c in calls))
def test_bad_or_missing_translation_falls_back_to_original_english(self):
for catalog in ('{}','not json','[]',json.dumps({'Unsupported Proxmox suite; no repository changed.':42}),
json.dumps({'Unsupported Proxmox suite; no repository changed.':'{unknown}'})):
with self.subTest(catalog=catalog):
result,out,error,calls,*_=self.run_policy('unsupported',malformed=catalog)
self.assertEqual(result,1);self.assertEqual(out,'');self.assertEqual(calls,[])
self.assertEqual(error,'Unsupported Proxmox suite; no repository changed.\n')
def test_translation_cannot_drop_actual_repository_details(self):
spec=importlib.util.spec_from_file_location('policy_format_fixture',ROOT/'scripts/global/repository_policy.py')
module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
with tempfile.TemporaryDirectory(prefix='pbs-policy-format-') as folder:
module.BASE=Path(folder);(module.BASE/'lang').mkdir()
(module.BASE/'config.json').write_text('{"language":"sk"}')
error=module.PolicyError('Inspect {ui}; no repository changed.',ui='Administration > Repositories')
for invalid in ('Skontrolujte repozitáre.','{ui.__class__}','{other}','{ui:.0}','{ui!r}'):
(module.BASE/'lang/sk.json').write_text(json.dumps({error.template:invalid}))
self.assertEqual(module.cli_error(error,'pbs'),str(error))
for invalid in ('not json','[]','{"language":"../../private"}'):
(module.BASE/'config.json').write_text(invalid)
self.assertEqual(module.cli_error(error,'pbs'),str(error))
UPDATE_HARNESS=r'''
BASE_DIR="$FIXTURE"
LANG_DIR="$REPO/lang"
SCRIPT_TITLE='fixture'
TAB='' BGN='' CL='' GN='' BL=''
mkdir -p "$FIXTURE/log" "$FIXTURE/tmp" "$FIXTURE/boot" "$FIXTURE/run"
: > "$FIXTURE/calls"
record() { printf '%s\n' "$*" >> "$FIXTURE/calls"; }
pmx_journal_context() { :; }
pmx_is_pbs() { [[ "$CASE" != pve ]]; }
pmx_platform_field() { [[ "$1" != supported ]] && printf '%s' "$SUITE" || printf true; }
ensure_pbs_repositories() { record repository-plan; return 0; }
dpkg-query() { [[ "$SUITE" == bookworm ]] && printf 3.4.0 || printf 4.0.1; }
apt-cache() { record apt-cache "$@"; printf ' Candidate: 4.0.2\n'; }
df() { printf 'Filesystem 1K-blocks Used Available Use%% Mounted\nfixture 10000 1000 %s 10%% /\n' "$([[ "$CASE" == disk ]] && echo 512000 || echo 4096000)"; }
apt-get() { record apt-get "$@"; [[ "$CASE" != refresh ]]; }
apt() {
record apt "$@"
if [[ "$*" == 'list --upgradable' ]]; then
printf 'Listing...\n'
[[ "$CASE" == current ]] || printf 'fixture/trixie 1.2 amd64 [upgradable from: 1.1]\n'
return 0
fi
[[ "$CASE" != failure ]]
}
fixture_env() { "$@"; }
APT_ENV='fixture_env'
pmx_format_upgradable() { cat; }
pmx_record_execution() { record journal "$1"; }
pmx_dpkg_changes_since() { :; }
msg_error() { printf '%s\n' "$*"; }
msg_ok() { printf '%s\n' "$*"; }
msg_info2() { printf '%s\n' "$*"; }
msg_warn() { printf '%s\n' "$*"; }
msg_title() { :; }
clear() { :; }
show_proxmenux_logo() { :; }
whiptail() { printf '%s\n' "$*" >> "$FIXTURE/dialog"; [[ "$CASE" != cancel ]]; }
uname() { printf old-kernel; }
touch "$FIXTURE/boot/vmlinuz-new-kernel"
'''
class UpdateCliTests(unittest.TestCase):
def run_update(self,case,language='sk',suite='trixie'):
with tempfile.TemporaryDirectory(prefix='pbs-update-i18n-') as folder:
body=function('scripts/global/update-pbs.sh','update_pbs')
for old,new in [('/var/log/',folder+'/log/'),('/tmp/proxmenux_screen_capture_',folder+'/tmp/proxmenux_screen_capture_'),('/var/run/',folder+'/run/'),('/boot/',folder+'/boot/')]: body=body.replace(old,new)
script=UPDATE_HARNESS+function('scripts/utils.sh','translate')+body+'\nupdate_pbs\n'
result=subprocess.run(['bash','-c',script],capture_output=True,text=True,timeout=15,
env={**os.environ,'REPO':str(ROOT),'FIXTURE':folder,'CASE':case,'LANGUAGE':language,'SUITE':suite})
files={str(p.relative_to(folder)):p.read_text() for p in Path(folder).rglob('*') if p.is_file()}
return result,files
def test_pve_guard_cannot_reach_repository_or_apt_calls(self):
result,files=self.run_update('pve')
self.assertEqual(result.returncode,1,result.stderr)
self.assertEqual(files['calls'],'')
self.assertFalse(any(name.startswith('log/') for name in files))
def test_disk_and_upgrade_errors_keep_their_actual_parameters(self):
result,files=self.run_update('disk')
self.assertEqual(result.returncode,1,result.stderr)
self.assertIn('Nedostatok miesta na disku. Dostupné:',result.stdout)
self.assertIn('500MB',result.stdout)
self.assertNotIn('apt',files['calls'])
result,files=self.run_update('failure')
self.assertEqual(result.returncode,1,result.stderr)
self.assertIn('Aktualizácia systému zlyhala. Skontrolujte denník:',result.stdout)
self.assertIn('/log/proxmox-update-',result.stdout)
self.assertNotIn('autoremove',files['calls'])
def test_update_status_cancel_refresh_error_and_reboot_notice(self):
for suite in ('bookworm','trixie'):
for case in ('current','cancel','refresh','success'):
with self.subTest(suite=suite,case=case):
result,files=self.run_update(case,suite=suite)
self.assertEqual(result.returncode,1 if case=='refresh' else 0,result.stderr)
if case!='success': self.assertNotIn('full-upgrade',files['calls'])
if case=='success': self.assertIn('reštartujte server',result.stdout)
if case in ('current','cancel'): self.assertIn('Aktuálna verzia PBS',files['dialog'])
class DialogTests(unittest.TestCase):
def test_automated_confirmation_and_banner_cancel_do_not_run_actions(self):
cases=[('scripts/menus/menu_post_install.sh','confirm_automated_script',
['repozitár no-subscription','upozornenie na predplatné','pamäť','jadro','sieť']),
('scripts/post_install/auto_post_install.sh','remove_subscription_banner',
['Proxmox Backup Server – odstránenie upozornenia na predplatné'])]
for path,name,expected in cases:
with tempfile.TemporaryDirectory(prefix='pbs-dialog-') as folder:
script=function('scripts/utils.sh','translate')+function(path,name)+r'''
PMX_ON_PBS=true
LOCAL_SCRIPTS='/fixture/scripts'
clear() { :; }; sleep() { :; }; stop_spinner() { :; }
msg_warn() { printf '%s\n' "$*"; }
bash() { printf forbidden >> "$CAPTURE/actions"; }
register_tool() { printf forbidden >> "$CAPTURE/actions"; }
dialog() { printf '%s\n' "$*" >> "$CAPTURE/dialog"; return 1; }
whiptail() { dialog "$@"; }
'''+name+'\n'
result=subprocess.run(['bash','-c',script],capture_output=True,text=True,timeout=15,
env={**os.environ,'LANGUAGE':'sk','LANG_DIR':str(ROOT/'lang'),'CAPTURE':folder})
self.assertEqual(result.returncode,1 if name=='remove_subscription_banner' else 0,result.stderr)
dialog=(Path(folder)/'dialog').read_text()
for text in expected:self.assertIn(text,dialog)
self.assertFalse((Path(folder)/'actions').exists())
if name=='remove_subscription_banner':self.assertIn('bolo zrušené používateľom',result.stdout)
def test_community_labels_preserve_dispatch_names_and_translate_display(self):
script=function('scripts/utils.sh','translate')+function('scripts/menus/menu_post_install.sh','post_install_label')+'\npost_install_label "$1"\n'
for text,expected in [('Proxmox Backup Server Post Install','nastavenie po inštalácii'),
('Proxmox Backup Server Microcode','mikrokód procesora')]:
result=subprocess.run(['bash','-c',script,'bash',text],capture_output=True,text=True,timeout=15,
env={**os.environ,'LANGUAGE':'sk','LANG_DIR':str(ROOT/'lang')})
self.assertEqual(result.returncode,0,result.stderr);self.assertIn(expected,result.stdout)
def test_failed_apt_hook_check_removes_only_fixture_hook_with_localized_warning(self):
with tempfile.TemporaryDirectory(prefix='pbs-hook-') as folder:
script=function('scripts/utils.sh','translate')+function('scripts/global/remove-banner-pve-v3.sh','create_apt_hook')+r'''
apt-config() { return 1; }
msg_warn() { printf '%s\n' "$*"; }
create_apt_hook
'''
result=subprocess.run(['bash','-c',script],capture_output=True,text=True,timeout=15,
env={**os.environ,'LANGUAGE':'sk','LANG_DIR':str(ROOT/'lang'),'APT_HOOK':folder+'/hook'})
self.assertEqual(result.returncode,0,result.stderr)
self.assertIn('Nesprávna syntax hooku APT',result.stdout)
self.assertFalse((Path(folder)/'hook').exists())
@@ -69,13 +69,43 @@ class TerminalCatalogsTests(unittest.TestCase):
def test_shipped_active_catalogs_keep_literal_contracts(self):
result = subprocess.run(
[sys.executable, '-B', str(SCRIPTS / 'check_terminal_catalogs.py'),
'--root', str(ROOT), '--require-complete'],
'--root', str(ROOT)],
capture_output=True, text=True, timeout=60,
)
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
for locale in MODULE.DEFAULT_LANGUAGES:
self.assertIn(f'[{locale}] active=', result.stdout)
self.assertNotIn('missing=1', result.stdout)
# New hand-reviewed Slovak keys ship with their source changes;
# other locales may use the runtime English fallback until translated.
result = subprocess.run(
[sys.executable, '-B', str(SCRIPTS / 'check_terminal_catalogs.py'),
'--root', str(ROOT), '--languages', 'sk', '--require-complete'],
capture_output=True, text=True, timeout=60,
)
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
self.assertIn('missing=0', result.stdout)
def test_optional_completeness_still_rejects_invalid_present_translations(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
(root / 'scripts').mkdir()
(root / 'lang').mkdir()
(root / 'scripts/check.sh').write_text('translate "Check /etc/pve/storage.cfg"')
(root / 'lang/sk.json').write_text(json.dumps({'Check /etc/pve/storage.cfg': 'Skontrolujte /etc/pve/storage.cfg'}))
(root / 'lang/es.json').write_text('{}')
def check(languages, *flags):
return subprocess.run(
[sys.executable, '-B', str(SCRIPTS / 'check_terminal_catalogs.py'),
'--root', str(root), '--languages', languages, *flags],
capture_output=True, text=True, timeout=60,
)
self.assertEqual(check('es,sk').returncode, 0)
self.assertEqual(check('sk', '--require-complete').returncode, 0)
self.assertNotEqual(check('es', '--require-complete').returncode, 0)
(root / 'lang/es.json').write_text(json.dumps({'Check /etc/pve/storage.cfg': 'Compruebe /etc/PVE/storage.cfg'}))
self.assertNotEqual(check('es,sk').returncode, 0)
if __name__ == '__main__':
@@ -0,0 +1,168 @@
"""Render the real ZRAM CLI paths in isolated files with host effects stubbed."""
import json
import os
from pathlib import Path
import subprocess
import tempfile
import unittest
ROOT = Path(__file__).resolve().parents[3]
DETAILS = {
"version": ("Cannot determine installed systemd-zram-generator version", "Nepodarilo sa zistiť verziu nainštalovaného balíka systemd-zram-generator"),
"mismatch": ("Existing ZRAM profile is active but does not match its configuration", "Existujúci profil ZRAM je aktívny, ale nezodpovedá svojej konfigurácii"),
"stop": ("Could not stop existing dev-zram0.swap", "Nepodarilo sa zastaviť existujúcu jednotku dev-zram0.swap"),
"directory": ("Could not create ZRAM generator directory", "Nepodarilo sa vytvoriť adresár pre konfiguráciu generátora ZRAM"),
"config": ("Could not write ZRAM generator configuration", "Nepodarilo sa zapísať konfiguráciu generátora ZRAM"),
"marker": ("Could not write ZRAM package ownership state", "Nepodarilo sa zapísať údaj o tom, či balík ZRAM nainštaloval ProxMenux"),
"start": ("Could not activate dev-zram0.swap", "Nepodarilo sa aktivovať jednotku dev-zram0.swap"),
"verify": ("Active ZRAM algorithm, capacity or swap sources did not match the profile", "Algoritmus kompresie, veľkosť alebo zdroje swapu aktívnej ZRAM nezodpovedajú profilu"),
}
REFUSAL = "Another ZRAM generator configuration exists; refusing to overwrite it:"
REFUSAL_SK = "Existuje iná konfigurácia generátora ZRAM; neprepíše sa:"
def function(path, name):
text = path.read_text()
start = text.index(name + "() {")
return text[start:text.index("\n}\n", start) + 3]
HARNESS = r'''
set -u
BASE_DIR="$FIXTURE/state"
LANG_DIR="$REPO/lang"
mkdir -p "$BASE_DIR" "$FIXTURE/etc/systemd/zram-generator.conf.d" "$FIXTURE/proc"
CONFIG="$FIXTURE/etc/systemd/zram-generator.conf.d/90-proxmenux.conf"
MARKER="$BASE_DIR/zram-swap.state"
printf 'Filename\tType\tSize\tUsed\tPriority\n' > "$FIXTURE/proc/swaps"
printf '/swapfile none swap sw 0 0\n' > "$FIXTURE/etc/fstab"
: > "$FIXTURE/calls"
record() { printf '%s\n' "$*" >> "$FIXTURE/calls"; }
pmx_journal_context() { :; }
msg_error() { printf '%s\n' "$*"; }
msg_success() { printf '%s\n' "$*"; }
register_tool() { record register "$@"; }
pmx_install_pkg() { record package-install; return 1; }
apt-get() { record apt "$@"; return 1; }
dpkg-query() {
record dpkg-query
case "$*" in
*'${Status}'*) printf 'install ok installed\n';;
*'${Version}'*) [[ "$CASE" == version ]] || printf '1.2.0-1\n';;
esac
}
dpkg() { return 0; }
install() { record directory; [[ "$CASE" != directory ]]; }
pmx_write_file() {
record write "$1"
if [[ "$CASE" == config && "$1" == "$CONFIG" ]] ||
[[ "$CASE" == marker && "$1" == "$MARKER" ]]; then
cat >/dev/null
return 1
fi
cat > "$1"
}
pmx_remove_file() { record remove "$1"; rm -f -- "$1"; }
pmx_record_uninstall() { record record-uninstall; }
swapoff() { record swapoff; return 0; }
systemctl() {
record systemctl "$@"
[[ "$CASE" != start || "$1" != start ]] &&
[[ "$CASE" != stop || "$1" != stop ]]
}
source "$REPO/scripts/post_install/zram_profile_helpers.sh"
zram_profile_verify_active() { [[ "$CASE" != verify && "$CASE" != mismatch ]]; }
if [[ "$CASE" == mismatch || "$CASE" == stop ]]; then
printf '# ProxMenux ZRAM profile\n[zram0]\nzram-size = ram / 10\n' > "$CONFIG"
[[ "$CASE" != mismatch ]] || zram_render_generator_config true > "$CONFIG"
printf 'installed_by_proxmenux=false\n' > "$MARKER"
printf '/dev/zram0\tpartition\t3355443\t8\t32767\n' >> "$FIXTURE/proc/swaps"
fi
if [[ "$CASE" == foreign ]]; then
printf '[zram1]\nzram-size = ram / 5\n' > "$FIXTURE/etc/systemd/zram-generator.conf"
fi
if [[ "$CASE" == unreadable ]]; then
zram_foreign_generator_config() { return 2; }
fi
'''
class ZramLocalizationTests(unittest.TestCase):
def test_missing_optional_translation_uses_original_english(self):
with tempfile.TemporaryDirectory(prefix="zram-fallback-") as folder:
(Path(folder) / 'xx.json').write_text('{}')
script = function(ROOT / "scripts/utils.sh", "translate") + '\ntranslate "$1"\n'
result = subprocess.run(["bash", "-c", script, "bash", DETAILS['version'][0]],
text=True, capture_output=True, timeout=15,
env={**os.environ, 'LANGUAGE': 'xx', 'LANG_DIR': folder})
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(result.stdout.strip(), DETAILS['version'][0])
def run_case(self, case, language="sk"):
with tempfile.TemporaryDirectory(prefix="zram-i18n-") as folder:
directory = Path(folder)
installer = function(ROOT / "scripts/post_install/customizable_post_install.sh", "configure_zram_swap")
installer = installer.replace("/etc/systemd/zram-generator.conf.d/90-proxmenux.conf", folder + "/etc/systemd/zram-generator.conf.d/90-proxmenux.conf")
installer = installer.replace("/proc/swaps", folder + "/proc/swaps").replace("/etc/fstab", folder + "/etc/fstab")
installer = installer.replace('zram_foreign_generator_config "$config"', 'zram_foreign_generator_config "$config" "$FIXTURE"')
installer = installer.replace('zram_list_foreign_systemd_zram_units)', 'zram_list_foreign_systemd_zram_units "$FIXTURE")')
script = HARNESS + function(ROOT / "scripts/utils.sh", "translate") + installer + "\nconfigure_zram_swap\n"
result = subprocess.run(["bash", "-c", script], text=True, capture_output=True,
env={**os.environ, "REPO": str(ROOT), "FIXTURE": folder, "CASE": case, "LANGUAGE": language}, timeout=15)
calls = (directory / "calls").read_text().splitlines()
files = {str(p.relative_to(directory)): p.read_text() for p in directory.rglob("*") if p.is_file()}
return result, calls, files
def test_all_activation_details_are_localized_and_english_is_preserved(self):
for case, (english, slovak) in DETAILS.items():
for language, expected in (("sk", slovak), ("en", english)):
with self.subTest(case=case, language=language):
result, calls, files = self.run_case(case, language)
self.assertEqual(result.returncode, 1, result.stderr)
self.assertIn(expected, result.stdout)
self.assertEqual(files["etc/fstab"], "/swapfile none swap sw 0 0\n")
if case in ("version", "mismatch"):
self.assertTrue(all(call == "dpkg-query" for call in calls), calls)
elif case == "stop":
self.assertEqual(files["state/zram-swap.state"], "installed_by_proxmenux=false\n")
self.assertIn("# ProxMenux ZRAM profile", files["etc/systemd/zram-generator.conf.d/90-proxmenux.conf"])
else:
self.assertNotIn("state/zram-swap.state", files)
self.assertNotIn("etc/systemd/zram-generator.conf.d/90-proxmenux.conf", files)
self.assertFalse(any(call.startswith(("apt ", "package-install", "register ")) for call in calls), calls)
def test_foreign_generator_is_refused_in_both_languages_without_any_effect(self):
for language, expected in (("sk", REFUSAL_SK), ("en", REFUSAL)):
result, calls, files = self.run_case("foreign", language)
self.assertEqual(result.returncode, 1, result.stderr)
self.assertIn(expected, result.stdout)
self.assertIn("/etc/systemd/zram-generator.conf", result.stdout)
self.assertEqual(calls, [])
self.assertEqual(files["etc/systemd/zram-generator.conf"], "[zram1]\nzram-size = ram / 5\n")
self.assertNotIn("state/zram-swap.state", files)
self.assertNotIn("etc/systemd/zram-generator.conf.d/90-proxmenux.conf", files)
self.assertEqual(files["etc/fstab"], "/swapfile none swap sw 0 0\n")
self.assertEqual(files["proc/swaps"], "Filename\tType\tSize\tUsed\tPriority\n")
def test_failed_inspection_is_not_mistaken_for_no_conflict(self):
result, calls, files = self.run_case("unreadable")
self.assertEqual(result.returncode, 1, result.stderr)
self.assertIn("Nie je možné skontrolovať existujúcu konfiguráciu generátora ZRAM", result.stdout)
self.assertEqual(calls, [])
self.assertNotIn("state/zram-swap.state", files)
self.assertNotIn("etc/systemd/zram-generator.conf.d/90-proxmenux.conf", files)
def test_uninstall_label_and_cancel_do_not_call_uninstall(self):
for language, label in (("sk", "Profil swapu ZRAM"), ("en", "ZRAM swap profile")):
with tempfile.TemporaryDirectory(prefix="zram-menu-") as folder:
tools = Path(folder) / "tools.json"
tools.write_text('{"zram_swap":true}')
script = function(ROOT / "scripts/utils.sh", "translate")
script += function(ROOT / "scripts/post_install/uninstall-tools.sh", "show_uninstall_menu")
script += '\nensure_tools_json() { :; }; migrate_installed_tools() { :; }\nuninstall_zram_swap() { printf forbidden; return 1; }\ndialog() { printf "%s\\n" "$@" > "$MENU_CAPTURE"; return 1; }\nshow_uninstall_menu\n'
result = subprocess.run(["bash", "-c", script], text=True, capture_output=True,
env={**os.environ, "LANGUAGE": language, "LANG_DIR": str(ROOT / "lang"), "TOOLS_JSON": str(tools), "MENU_CAPTURE": folder + "/dialog.txt"}, timeout=15)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn(label, (Path(folder) / "dialog.txt").read_text())
self.assertNotIn("forbidden", result.stdout + result.stderr)
self.assertEqual(json.loads(tools.read_text()), {"zram_swap": True})