ProxMenux/oci/tests/test_update_menu.py

"""Explicit update selection and live status redraw without lifecycle shortcuts."""
from pathlib import Path
import ast
import json
import os
import re
import signal
import subprocess
import sys
from types import SimpleNamespace
import unittest
from unittest.mock import Mock, patch

ROOT = Path(__file__).resolve().parents[1]
sys.path[:0] = [str(ROOT / 'src'), str(ROOT / 'remote')]
from proxmenux_oci import cli, management, updates
from proxmenux_oci.ui import DialogUI, MenuRefresh, ReadableTags, UserCancelled

AVAILABLE = [{'vmid': 110, 'title': 'First app'}, {'vmid': 111, 'title': 'Second app'}]


class SelectionTests(unittest.TestCase):
    def select(self, picked, available=AVAILABLE, review=True, fresh=None, outcomes=None):
        ui = Mock()
        ui.checklist.side_effect = picked if isinstance(picked, Exception) else [picked]
        ui.review.return_value = review
        current = [{**row, 'reason': 'matched'} for row in available]
        with patch.object(updates, 'UpdateCheck') as checker, \
                patch.object(management, 'saved_inventory', return_value=current), \
                patch.object(management, 'manage_instance', side_effect=outcomes or [True] * len(available)) as run:
            checker.return_value.status.return_value = {'available': available if fresh is None else fresh}
            result = updates.select_updates(ROOT, ui, available)
        return result, ui, run

    def test_one_update_is_not_checked_by_default(self):
        _, ui, run = self.select([], available=AVAILABLE[:1])
        self.assertEqual(ui.checklist.call_args.args[2], [])
        self.assertEqual(ui.checklist.call_args.args[1][0], ('all', 'Select all'))
        run.assert_not_called()

    def test_only_selected_app_uses_existing_update_flow(self):
        _, ui, run = self.select(['111'])
        self.assertEqual(run.call_count, 1)
        self.assertEqual(run.call_args.args[2]['vmid'], 111)
        self.assertEqual(run.call_args.kwargs, {'action': 'update', 'heading': 'Update OCI (1/1) · Second app'})
        self.assertIn('Second app', ui.review.call_args.args[0])
        self.assertNotIn('First app', ui.review.call_args.args[0])

    def test_select_all_includes_every_app_once(self):
        _, ui, run = self.select(['all', '111'])
        self.assertEqual([c.args[2]['vmid'] for c in run.call_args_list], [110, 111])
        # Each update shows its place in the batch while its container is checked.
        self.assertEqual([c.kwargs['heading'] for c in run.call_args_list],
                         ['Update OCI (1/2) · First app', 'Update OCI (2/2) · Second app'])
        self.assertFalse(ui.review.call_args.kwargs['default'])

    def test_cancel_does_nothing(self):
        _, ui, run = self.select(UserCancelled())
        ui.review.assert_not_called(); run.assert_not_called()

    def test_selection_has_an_update_button_and_no_extra_action(self):
        result, ui, run = self.select([])
        self.assertEqual(ui.checklist.call_args.kwargs['ok_label'], 'Update')
        # The same height and width as the OCI main menu.
        self.assertEqual(ui.checklist.call_args.kwargs['size'], updates.menu_size())
        self.assertNotIn('extra_button', ui.checklist.call_args.kwargs)
        self.assertFalse(result); ui.review.assert_not_called(); run.assert_not_called()

    def test_changed_app_blocks_batch_before_any_lifecycle(self):
        _, ui, run = self.select(['all'], fresh=AVAILABLE[1:])
        run.assert_not_called()
        self.assertIn('changed since', ui.message.call_args.args[0])

    def test_failure_or_per_app_cancel_stops_later_selections(self):
        _, _, run = self.select(['all'], outcomes=[False])
        self.assertEqual(run.call_count, 1)

    def test_declining_review_returns_to_selection_instead_of_updating(self):
        ui = Mock(); ui.checklist.side_effect = [['all'], []]; ui.review.return_value = False
        with patch.object(management, 'manage_instance') as run:
            self.assertFalse(updates.select_updates(ROOT, ui, AVAILABLE))
        run.assert_not_called()
        self.assertEqual(ui.checklist.call_args.args[2], ['110', '111'])


class StatusTests(unittest.TestCase):
    def checker(self):
        return updates.UpdateCheck(ROOT)

    def test_labels_show_cached_count_or_manual_check(self):
        checker = self.checker()
        self.assertEqual(checker.label({'available': []}), 'Check for updates')
        self.assertEqual(checker.label({'available': AVAILABLE}), 'Updates available (2)')
        checker.process = Mock(stdout=None); checker.process.poll.return_value = None
        self.assertEqual(checker.label({'available': AVAILABLE}), 'Checking for updates (0/0)')

    def test_partial_check_keeps_the_entry_clean_without_a_header_warning(self):
        checker = self.checker()
        state = {'checked': True, 'available': [], 'unknown': 1, 'applications': 2}
        self.assertEqual(checker.label(state), 'Check for updates')
        self.assertEqual(checker.notice(state), '')

    def test_completed_checks_show_only_the_update_count(self):
        state = {'checked': True, 'available': AVAILABLE, 'unknown': 0, 'applications': 2}
        self.assertEqual(self.checker().label(state), 'Updates available (2)')

    def progress_checker(self):
        reader, writer = os.pipe()
        stream = os.fdopen(reader, 'rb', buffering=0)
        self.addCleanup(stream.close); self.addCleanup(os.close, writer)
        os.set_blocking(reader, False)
        checker = self.checker(); checker.total = 24
        checker.process = Mock(stdout=stream); checker.process.poll.return_value = None
        return checker, writer

    def test_numeric_progress_redraws_only_when_the_count_changes(self):
        checker, writer = self.progress_checker()
        self.assertEqual(checker.label({}), 'Checking for updates (0/24)')
        self.assertFalse(checker.changed())
        os.write(writer, b'{"completed":12,"total":24}\n')
        self.assertTrue(checker.changed())
        self.assertEqual(checker.label({}), 'Checking for updates (12/24)')
        self.assertFalse(checker.changed())
        os.write(writer, b'{"completed":12,"total":24}\n')
        self.assertFalse(checker.changed())

    def test_split_and_invalid_worker_output_never_appears_in_the_menu(self):
        checker, writer = self.progress_checker()
        os.write(writer, b'password=secret\n{"completed":-1,"total":24}\n{"completed":')
        self.assertEqual(checker.label({}), 'Checking for updates (0/24)')
        os.write(writer, b'7,"total":24}\n{"completed":25,"total":24}\n')
        self.assertEqual(checker.label({}), 'Checking for updates (7/24)')

    def test_worker_failure_does_not_reuse_an_old_successful_result(self):
        checker = self.checker(); checker.process = Mock(stdout=None)
        checker.process.poll.return_value = checker.process.wait.return_value = 1
        old = {'checked': True, 'available': AVAILABLE, 'unknown': 0, 'applications': 2}
        with patch.object(checker.backend, 'cached', return_value=old):
            state = checker.status()
        self.assertEqual(checker.label(state), 'Check for updates')
        self.assertEqual(state['available'], [])
        self.assertEqual(checker.notice(state), 'The update check could not be completed. Try again later.')

    def test_finished_worker_keeps_the_final_redraw_pending_until_status_is_consumed(self):
        checker = self.checker(); checker.completed = checker.total = 2
        checker.process = Mock(stdout=None)
        checker.process.poll.return_value = checker.process.wait.return_value = 0
        self.assertFalse(checker.running)
        self.assertTrue(checker.active)
        self.assertEqual(checker.label({}), 'Checking for updates (2/2)')
        self.assertTrue(checker.changed())
        state = {'checked': True, 'available': AVAILABLE, 'unknown': 0, 'applications': 2}
        with patch.object(checker.backend, 'cached', return_value=state):
            final = checker.status()
        self.assertFalse(checker.active)
        self.assertEqual(checker.label(final), 'Updates available (2)')

    def test_check_process_is_isolated_and_displays_no_registry_errors(self):
        checker = self.checker()
        with patch.object(subprocess, 'Popen') as popen, patch.object(updates.os, 'set_blocking') as blocking, \
                patch.object(checker.backend, 'monitor_available', return_value=True):
            checker.start()
        args = popen.call_args
        self.assertTrue(args.kwargs['start_new_session'])
        self.assertEqual(args.kwargs['stderr'], subprocess.DEVNULL)
        self.assertEqual(args.args[0][1], str(ROOT / 'remote/oci_update_check.py'))
        self.assertIn('--progress', args.args[0])
        self.assertEqual(args.kwargs['stdout'], subprocess.PIPE)
        blocking.assert_called_once_with(checker.process.stdout.fileno(), False)

    def test_failed_applications_are_reported_and_cleared_by_a_new_check(self):
        checker = self.checker(); read, write = os.pipe()
        os.write(write, b'{"completed": 2, "total": 2}\n{"failed": [110]}\n'); os.close(write)
        os.set_blocking(read, False)
        checker.process = Mock(stdout=Mock(fileno=Mock(return_value=read)))
        checker.process.poll.return_value = checker.process.wait.return_value = 0
        state = {'checked': True, 'available': [], 'unknown': 1, 'applications': 2}
        with patch.object(checker.backend, 'cached', return_value=state) as cached:
            final = checker.status()
        os.close(read)
        self.assertEqual(cached.call_args.kwargs['failed'], {110})
        self.assertEqual(checker.notice(final),
                         'Some applications could not be checked (1). Try again later.')
        with patch.object(subprocess, 'Popen'), patch.object(updates.os, 'set_blocking'), \
                patch.object(checker.backend, 'monitor_available', return_value=True), \
                patch.object(checker.backend, 'cached', return_value=state):
            checker.start()
        self.assertEqual(checker.failed_applications, set())

    def test_a_failed_check_is_reported_in_the_menu(self):
        checker = self.checker(); checker.failed = True
        self.assertEqual(checker.notice({}), 'The update check could not be completed. Try again later.')

    def test_missing_monitor_never_starts_a_fallback_checker(self):
        checker = self.checker()
        with patch.object(checker.backend, 'monitor_available', return_value=False), \
                patch.object(subprocess, 'Popen') as popen:
            with self.assertRaisesRegex(ValueError, 'ProxMenux Monitor'):
                checker.start()
        popen.assert_not_called()

    def test_cancelling_stops_the_worker_process_group(self):
        checker = self.checker(); process = Mock(); checker.process = process
        process.pid = 4321; process.poll.return_value = None
        with patch.object(updates.os, 'killpg') as kill:
            checker.close()
        kill.assert_called_once_with(4321, signal.SIGTERM)
        process.wait.assert_called_once_with(timeout=2)

    def test_finished_worker_is_not_signalled_after_its_pid_could_be_reused(self):
        checker = self.checker(); checker.process = Mock(); checker.process.poll.return_value = 0
        with patch.object(updates.os, 'killpg') as kill:
            checker.close()
        kill.assert_not_called()


class DialogTests(unittest.TestCase):
    def test_status_change_redraws_menu_and_reaps_dialog(self):
        process = Mock(); process.poll.return_value = None
        process.communicate.side_effect = [subprocess.TimeoutExpired('dialog', .1), (None, '')]
        popen = Mock(); popen.__enter__ = Mock(return_value=process); popen.__exit__ = Mock(return_value=False)
        with patch.object(subprocess, 'Popen', return_value=popen):
            with self.assertRaises(MenuRefresh):
                DialogUI().choose('OCI', ReadableTags([('updates', 'Checking for updates...')]),
                                  'updates', size=(20, 75, 12), refresh=lambda: True)
        process.terminate.assert_called_once()
        self.assertEqual(process.communicate.call_count, 2)

    def test_escape_from_live_menu_remains_cancel(self):
        process = Mock(); process.poll.return_value = 255; process.returncode = 255
        process.communicate.return_value = (None, '')
        popen = Mock(); popen.__enter__ = Mock(return_value=process); popen.__exit__ = Mock(return_value=False)
        with patch.object(subprocess, 'Popen', return_value=popen):
            self.assertIsNone(DialogUI().choose('OCI', [('1', 'Checking')], refresh=lambda: False))
        process.terminate.assert_not_called()

    def test_checklist_starts_unselected_with_update_and_cancel_only(self):
        ui = DialogUI()
        with patch.object(ui, '_run', return_value=SimpleNamespace(returncode=0, stderr='111\n')) as run:
            picked = ui.checklist('Updates', [('all', 'Select all'), ('111', 'App')], [],
                                  ok_label='Update')
        self.assertEqual(picked, ['111'])
        widget = run.call_args.args[0]
        self.assertNotIn('--extra-button', widget)
        self.assertEqual(widget[widget.index('--ok-label') + 1], 'Update')
        self.assertNotIn('on', widget)
        with patch.object(ui, '_run', return_value=SimpleNamespace(returncode=0, stderr='')) as run:
            ui.checklist('Updates', [('111', 'App')], [], size=(24, 75, 16))
        widget = run.call_args.args[0]
        start = widget.index('--checklist')
        self.assertEqual(widget[start + 2:start + 5], ['24', '75', '16'])


class MainMenuTests(unittest.TestCase):
    def test_status_changes_in_the_same_main_menu_entry(self):
        class Checker:
            running = False
            completed = 0
            @property
            def active(self): return self.running
            def __enter__(self): return self
            def __exit__(self, *exc): pass
            def start(self): self.running = True
            def status(self): return {'available': AVAILABLE if getattr(self, 'done', False) else []}
            def changed(self): return True
            def label(self, state):
                return (f'Checking for updates ({self.completed}/2)' if self.running else
                        'Updates available (2)' if state['available'] else 'Check for updates')
            def notice(self, state): return ''
        checker = Checker(); seen = []
        class UI:
            live_refresh = True
            def choose(self, text, entries, default, **kw):
                seen.append(dict(entries)['updates'])
                if len(seen) == 1: return 'updates'
                if len(seen) == 2:
                    checker.completed = 1
                    raise MenuRefresh()
                if len(seen) == 3:
                    checker.running = False; checker.done = True
                    raise MenuRefresh()
                return None
        with patch.object(cli, 'interactive_ui', return_value=UI()), \
                patch.object(updates, 'UpdateCheck', return_value=checker):
            self.assertEqual(cli.interactive(cli.Catalog(ROOT)), 0)
        self.assertEqual(seen, ['Check for updates', 'Checking for updates (0/2)',
                               'Checking for updates (1/2)', 'Updates available (2)'])


class TranslationTests(unittest.TestCase):
    def test_update_menu_literals_are_translated_in_every_shipped_language(self):
        source = ast.parse((ROOT / 'src/proxmenux_oci/updates.py').read_text())
        texts = {node.args[0].value for node in ast.walk(source)
                 if isinstance(node, ast.Call) and isinstance(node.func, ast.Name)
                 and node.func.id == 'translate' and node.args
                 and isinstance(node.args[0], ast.Constant)}
        for path in sorted((ROOT.parent / 'lang').glob('*.json')):
            dictionary = json.loads(path.read_text())
            for text in texts:
                with self.subTest(locale=path.stem, text=text):
                    self.assertTrue(dictionary.get(text))
                    self.assertEqual(set(re.findall(r'\{\w+\}', text)),
                                     set(re.findall(r'\{\w+\}', dictionary[text])))


if __name__ == '__main__':
    unittest.main()