framebridge/source/tests/test_package_upgrade.py

import hashlib
from contextlib import nullcontext
import json
import os
from pathlib import Path
from types import SimpleNamespace
import tempfile
import unittest
from unittest import mock

from framebridge_expanded import package_upgrade as upgrade


class PackageUpgradeTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        self.original_read = upgrade._read
        self.state = self.root / 'profile'
        self.state.mkdir(mode=0o700)
        self.config = {'role': 'node', 'node_id': 'test-node'}
        self.write(self.state / 'config.json', self.config)
        self.proc = self.root / '1234'
        self.proc.mkdir()
        self.identity = {'pid': 1234, 'birth': '99', 'uid': os.getuid()}
        self.write(self.state / 'manager.json', self.identity)
        self.args = ['/usr/bin/python3', '-m', 'framebridge_expanded.runtime',
                     '_manager', '--state', str(self.state)]
        (self.proc / 'cmdline').write_bytes(b'\0'.join(x.encode() for x in self.args) + b'\0')
        self.environ()
        (self.proc / 'stat').write_text('1234 (python3) S ' + ' '.join(['0'] * 18 + ['99']))
        (self.proc / 'cgroup').write_text('0::/user.slice/user-1000.slice/user@1000.service/app.slice/test.scope\n')

    def write(self, path, data):
        path.write_text(json.dumps(data))
        path.chmod(0o600)

    def environ(self, prefix=None):
        prefix = upgrade.PREFIX if prefix is None else prefix
        (self.proc / 'environ').write_bytes(b'\0'.join(x.encode() for x in [
            'FRAMEBRIDGE_EXPANDED_PREFIX=' + prefix,
            'FRAMEBRIDGE_EXPANDED_STATE=' + str(self.state),
            'PYTHONPATH=' + prefix + '/expanded:' + prefix + '/portable']) + b'\0')

    def identify(self):
        if os.getuid() == 0:
            self.skipTest('Identity fixtures run as the ordinary build owner, not root')
        return upgrade.identify(self.proc)

    def test_identifies_only_installed_manager_and_exact_owned_record(self):
        row = self.identify()
        self.assertEqual(row['pid'], 1234)
        self.assertEqual(row['state'], str(self.state))
        self.assertEqual(row['birth'], '99')
        self.assertIsNone(row['unit'])
        self.assertNotIn('config', row)

    def test_development_clone_is_not_an_installed_service(self):
        self.environ('/home/example/source')
        self.assertIsNone(self.identify())

    def test_arbitrary_python_command_is_ignored(self):
        (self.proc / 'cmdline').write_bytes(b'/usr/bin/python3\0-c\0print(1)\0')
        self.assertIsNone(upgrade.identify(self.proc))

    def test_pid_reuse_is_rejected(self):
        self.write(self.state / 'manager.json', dict(self.identity, birth='98'))
        if os.getuid() == 0:
            self.skipTest('Ordinary-owner fixture required')
        with self.assertRaisesRegex(RuntimeError, 'RECORD_MISMATCH'):
            upgrade.identify(self.proc)

    def test_recognizes_qa_unit_without_selecting_unrelated_services(self):
        (self.proc / 'cgroup').write_text('0::/user.slice/user@1000.service/app.slice/framebridge-qa-mint223-private2-c3418304.service\n')
        self.assertEqual(self.identify()['unit'], 'framebridge-qa-mint223-private2-c3418304.service')

    def test_unrelated_unit_is_not_restarted(self):
        (self.proc / 'cgroup').write_text('0::/user.slice/user@1000.service/app.slice/owner-important.service\n')
        self.assertIsNone(self.identify()['unit'])

    def test_world_readable_metadata_is_refused(self):
        path = self.state / 'config.json'
        path.chmod(0o644)
        with self.assertRaisesRegex(RuntimeError, 'FILE_UNTRUSTED'):
            upgrade._read(path, os.getuid())

    def test_metadata_symlink_is_refused(self):
        path = self.state / 'alias.json'
        path.symlink_to(self.state / 'config.json')
        with self.assertRaises(OSError):
            upgrade._read(path, os.getuid())

    def test_shared_profile_directory_is_refused(self):
        self.state.chmod(0o755)
        with self.assertRaisesRegex(RuntimeError, 'DIRECTORY_UNTRUSTED'):
            upgrade._private(self.state, os.getuid())

    def root_read(self, path, uid, maximum=upgrade.MAXIMUM):
        # Root-owned production intent is represented by the test owner's file.
        return self.original_read(path, os.getuid(), maximum)

    def pending(self, profiles=None):
        value = {'schema': 1, 'transaction': '1' * 16, 'phase': 'CAPTURED', 'profiles': profiles or []}
        self.write(self.root / 'pending.json', value)
        return value

    def test_fresh_install_does_not_schedule_or_start_profiles(self):
        with mock.patch.object(upgrade, 'ROOT', self.root), \
             mock.patch.object(upgrade, '_root'), \
             mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
             mock.patch.object(upgrade.subprocess, 'run') as run:
            self.assertEqual(upgrade.schedule()['phase'], 'NO_UPGRADE_SNAPSHOT')
        run.assert_not_called()

    def test_empty_upgrade_stays_stopped(self):
        self.pending()
        with mock.patch.object(upgrade, 'ROOT', self.root), \
             mock.patch.object(upgrade, '_root'), \
             mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
             mock.patch.object(upgrade.subprocess, 'run') as run:
            self.assertEqual(upgrade.schedule()['phase'], 'COMPLETE')
        run.assert_not_called()

    def test_restart_is_detached_and_delayed_past_package_transaction(self):
        self.pending([{'state': str(self.state)}])
        with mock.patch.object(upgrade, 'ROOT', self.root), \
             mock.patch.object(upgrade, '_root'), \
             mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
             mock.patch.object(upgrade.subprocess, 'run', return_value=SimpleNamespace(returncode=0)) as run:
            self.assertEqual(upgrade.schedule()['phase'], 'SCHEDULED')
        args = run.call_args.args[0]
        self.assertEqual(args[0], '/usr/bin/systemd-run')
        self.assertIn('--on-active=2s', args)
        self.assertEqual(args[-2:], [upgrade.MODULE, 'resume'])

    def test_scheduling_failure_keeps_captured_intent(self):
        self.pending([{'state': str(self.state)}])
        with mock.patch.object(upgrade, 'ROOT', self.root), \
             mock.patch.object(upgrade, '_root'), \
             mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
             mock.patch.object(upgrade.subprocess, 'run', return_value=SimpleNamespace(returncode=1)):
            with self.assertRaisesRegex(RuntimeError, 'SCHEDULE_FAILED'):
                upgrade.schedule()
        self.assertEqual(json.loads((self.root / 'pending.json').read_text())['phase'], 'CAPTURED')

    def test_repeated_configure_does_not_schedule_second_restart(self):
        self.write(self.root / 'pending.json', {'phase': 'SCHEDULED'})
        with mock.patch.object(upgrade, 'ROOT', self.root), \
             mock.patch.object(upgrade, '_root'), \
             mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
             mock.patch.object(upgrade.subprocess, 'run') as run:
            self.assertEqual(upgrade.schedule()['phase'], 'SCHEDULED')
        run.assert_not_called()

    def test_changed_manager_is_not_stopped_and_failure_is_recorded(self):
        row = {'uid': 1000, 'user': 'owner', 'home': '/home/owner',
               'state': str(self.state), 'pid': 1234, 'birth': '99',
               'config_sha256': 'a' * 64, 'unit': None, 'phase': 'CAPTURED'}
        self.write(self.root / 'pending.json', {'schema': 1, 'phase': 'SCHEDULED', 'profiles': [row]})
        with mock.patch.object(upgrade, 'ROOT', self.root), \
             mock.patch.object(upgrade, '_root'), \
             mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
             mock.patch.object(upgrade, '_wait_dpkg', return_value=nullcontext()), \
             mock.patch.object(upgrade, '_resume_lock', return_value=nullcontext()), \
             mock.patch.object(upgrade, 'identify', return_value=dict(row, birth='100')), \
             mock.patch.object(upgrade, '_owner') as owner:
            with self.assertRaisesRegex(RuntimeError, 'PROFILE_RESTART_FAILED'):
                upgrade.resume()
        owner.assert_not_called()
        value = json.loads((self.root / 'pending.json').read_text())
        self.assertEqual(value['phase'], 'FAILED')
        self.assertIn('IDENTITY_CHANGED', value['profiles'][0]['error'])

    def test_manual_profile_restart_never_enables_service(self):
        row = {'uid': 1000, 'user': 'owner', 'home': '/home/owner',
               'state': str(self.state), 'pid': 1234, 'birth': '99',
               'config_sha256': 'a' * 64, 'unit': None, 'phase': 'CAPTURED'}
        self.write(self.root / 'pending.json', {'schema': 1, 'phase': 'SCHEDULED', 'profiles': [row]})
        with mock.patch.object(upgrade, 'ROOT', self.root), \
             mock.patch.object(upgrade, '_root'), \
             mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
             mock.patch.object(upgrade, '_wait_dpkg', return_value=nullcontext()), \
             mock.patch.object(upgrade, '_resume_lock', return_value=nullcontext()), \
             mock.patch.object(upgrade, 'identify', return_value=row), \
             mock.patch.object(upgrade, '_owner', return_value={'ok': True, 'ready': True}) as owner, \
             mock.patch.object(upgrade, '_start_manual', return_value='framebridge-profile-' + 'a' * 20 + '.service'), \
             mock.patch.object(upgrade.subprocess, 'run') as run:
            self.assertEqual(upgrade.resume()['phase'], 'COMPLETE')
        self.assertEqual([c.args[1] for c in owner.call_args_list], ['user-desktop', 'user-stop', 'user-ready'])
        run.assert_not_called()


    def test_snapshot_does_not_stop_the_process_running_apt(self):
        row = {'uid': os.getuid(), 'state': str(self.state), 'pid': 1234}
        with mock.patch.object(upgrade, 'ROOT', self.root), \
             mock.patch.object(upgrade, '_root'), \
             mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
             mock.patch.object(upgrade, 'identify', return_value=row), \
             mock.patch.object(upgrade.subprocess, 'run') as run:
            value = upgrade.snapshot(self.root)
        self.assertEqual(value['phase'], 'CAPTURED')
        run.assert_not_called()

    def test_reinstall_after_failure_preserves_receipt_and_keeps_stopped_profiles_stopped(self):
        failed = {'schema': 1, 'transaction': 'a' * 16, 'phase': 'FAILED',
                  'profiles': [{'state': str(self.state), 'phase': 'FAILED',
                                'error': 'UPGRADE_OWNER_OPERATION_FAILED:user-ready'}]}
        self.write(self.root / 'pending.json', failed)
        processes = self.root / 'no-running-managers'
        processes.mkdir()
        marker = self.state / 'manual-stop.json'
        marker.write_bytes(b'{"manual_stop": true}\n')
        before_marker = marker.read_bytes()
        with mock.patch.object(upgrade, 'ROOT', self.root), \
             mock.patch.object(upgrade, '_root'), \
             mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
             mock.patch.object(upgrade.subprocess, 'run') as run:
            value = upgrade.snapshot(processes)
            self.assertEqual(value['profiles'], [])
            self.assertEqual(upgrade.schedule()['phase'], 'COMPLETE')
        self.assertEqual(json.loads((self.root / 'previous.json').read_text()), failed)
        self.assertEqual(marker.read_bytes(), before_marker)
        run.assert_not_called()

    def test_reinstall_after_failure_captures_only_current_managers_without_replaying_old_intent(self):
        failed = {'schema': 1, 'phase': 'FAILED',
                  'profiles': [{'state': '/home/owner/previous-profile', 'phase': 'FAILED'}]}
        self.write(self.root / 'pending.json', failed)
        row = {'uid': 1000, 'state': str(self.state), 'pid': 1234, 'phase': 'CAPTURED'}
        with mock.patch.object(upgrade, 'ROOT', self.root), \
             mock.patch.object(upgrade, '_root'), \
             mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
             mock.patch.object(upgrade, 'identify', return_value=row), \
             mock.patch.object(upgrade.subprocess, 'run', return_value=SimpleNamespace(returncode=0)) as run:
            value = upgrade.snapshot(self.root)
            self.assertEqual(value['profiles'], [row])
            self.assertEqual(upgrade.schedule()['phase'], 'SCHEDULED')
        self.assertEqual(json.loads((self.root / 'previous.json').read_text()), failed)
        self.assertEqual(run.call_count, 1)
        self.assertEqual(run.call_args.args[0][-2:], [upgrade.MODULE, 'resume'])

    def test_reinstall_does_not_replace_a_snapshot_with_a_live_upgrade_worker(self):
        for phase in ('SCHEDULED', 'RUNNING'):
            with self.subTest(phase=phase):
                self.write(self.root / 'pending.json', {'schema': 1, 'phase': phase})
                before = (self.root / 'pending.json').read_bytes()
                with mock.patch.object(upgrade, 'ROOT', self.root), \
                     mock.patch.object(upgrade, '_root'), \
                     mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
                     mock.patch.object(upgrade, 'identify', return_value=None):
                    with self.assertRaisesRegex(RuntimeError, 'PREVIOUS_WORKER_PENDING'):
                        upgrade.snapshot(self.root)
                self.assertEqual((self.root / 'pending.json').read_bytes(), before)
                self.assertFalse((self.root / 'previous.json').exists())

    def test_optional_desktop_migration_failure_does_not_stop_runtime_recovery(self):
        row = {'uid': 1000, 'user': 'owner', 'home': '/home/owner',
               'state': str(self.state), 'pid': 1234, 'birth': '99',
               'config_sha256': 'a' * 64, 'unit': None, 'phase': 'CAPTURED'}
        self.write(self.root / 'pending.json', {'schema': 1, 'phase': 'SCHEDULED', 'profiles': [row]})
        def owner(profile, operation):
            if operation == 'user-desktop':
                raise RuntimeError('Legacy console is closed')
            return {'ok': True, 'ready': True}
        with mock.patch.object(upgrade, 'ROOT', self.root), \
             mock.patch.object(upgrade, '_root'), \
             mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
             mock.patch.object(upgrade, '_wait_dpkg', return_value=nullcontext()), \
             mock.patch.object(upgrade, '_resume_lock', return_value=nullcontext()), \
             mock.patch.object(upgrade, 'identify', return_value=row), \
             mock.patch.object(upgrade, '_owner', side_effect=owner), \
             mock.patch.object(upgrade, '_start_manual', return_value='framebridge-profile-' + 'a' * 20 + '.service'):
            value = upgrade.resume()
        self.assertEqual(value['phase'], 'COMPLETE')
        self.assertEqual(value['profiles'][0]['phase'], 'READY')
        self.assertFalse(value['profiles'][0]['desktop']['ok'])

    def check_stop_failure_retains_marker(self, before):
        if os.getuid() == 0:
            self.skipTest('Ordinary-owner fixture required')
        from framebridge_expanded import runtime, autostart
        directory = self.state / 'autostart'
        directory.mkdir(mode=0o700)
        marker = directory / 'manual-stop.json'
        if before is not None:
            marker.write_bytes(before)
            marker.chmod(0o600)
        def fail(state, **kwargs):
            marker.write_text('{"temporary":true}')
            marker.chmod(0o600)
            raise RuntimeError('simulated bounded stop failure')
        with mock.patch.object(runtime, 'stop', side_effect=fail), \
             mock.patch.object(upgrade, '_birth', return_value='99'), \
             mock.patch.object(upgrade, '_background', return_value=None), \
             mock.patch.object(runtime, 'start') as start, \
             mock.patch.object(autostart, '_configuration_action', side_effect=lambda state, action: action(state)):
            with self.assertRaisesRegex(RuntimeError, 'bounded stop failure'):
                upgrade.user_operation('user-stop', str(self.state),
                    hashlib.sha256((self.state / 'config.json').read_bytes()).hexdigest(),
                    manager_pid=1234, manager_birth='99')
        start.assert_not_called()
        if before is None:
            self.assertFalse(marker.exists())
        else:
            self.assertEqual(marker.read_bytes(), before)

    def test_stop_failure_retains_absent_manual_stop_policy(self):
        self.check_stop_failure_retains_marker(None)

    def test_stop_failure_retains_existing_manual_stop_bytes(self):
        self.check_stop_failure_retains_marker(b'{"manual_stop": true}\n')



    def test_user_unit_restart_rejects_account_reassignment(self):
        row = {'uid': 1000, 'user': 'owner', 'home': '/home/owner'}
        account = SimpleNamespace(pw_name='different', pw_dir='/home/different')
        with mock.patch.object(upgrade.pwd, 'getpwuid', return_value=account):
            with self.assertRaisesRegex(RuntimeError, 'ACCOUNT_CHANGED'):
                upgrade._unit(row)



    def test_manual_start_uses_an_independent_user_service_cgroup(self):
        row = {'uid': os.getuid(), 'user': 'owner', 'home': '/home/owner', 'state': str(self.state)}
        account = SimpleNamespace(pw_name='owner', pw_dir='/home/owner')
        with mock.patch.object(upgrade.pwd, 'getpwuid', return_value=account), \
             mock.patch.object(upgrade.subprocess, 'run', return_value=SimpleNamespace(returncode=0)) as run:
            if os.getuid() == 0:
                self.skipTest('Ordinary-owner fixture required')
            unit = upgrade._start_manual(row)
        args = run.call_args.args[0]
        self.assertIn('/usr/bin/systemd-run', args)
        self.assertIn('--property=ExitType=cgroup', args)
        self.assertIn('--property=KillMode=control-group', args)
        self.assertIn('--unit=' + unit, args)
        self.assertEqual(args[-5:], ['/usr/bin/fb', '--state-dir', str(self.state), '--json', 'start'])
        self.assertNotIn('enable', args)

    def test_background_identity_failure_is_refused_without_signaling(self):
        from framebridge_expanded import connection_background as background
        with mock.patch.object(background, '_directory', return_value=self.state), \
             mock.patch.object(background, '_read', return_value={'pid': os.getpid()}), \
             mock.patch.object(background, '_valid_identity', return_value=False), \
             mock.patch.object(background, 'disconnect') as disconnect:
            with self.assertRaisesRegex(RuntimeError, 'BACKGROUND_IDENTITY_UNVERIFIED'):
                upgrade._background(self.state)
        disconnect.assert_not_called()

    def test_verified_background_connection_is_retained_without_reading_credentials(self):
        from framebridge_expanded import connection_background as background
        record = {'pid': 1234, 'start_ticks': 88}
        with mock.patch.object(background, '_directory', return_value=self.state), \
             mock.patch.object(background, '_read', return_value=record), \
             mock.patch.object(background, '_valid_identity', return_value=True), \
             mock.patch.object(background, 'status', return_value={'verified': True}):
            value = upgrade._background(self.state)
        self.assertTrue(value['preserved_in_memory'])
        self.assertFalse(value['process_reloaded'])
        self.assertEqual(value['start_ticks'], 88)

    def check_runtime_stop_connection(self, preserve):
        from framebridge_expanded import runtime, connection_background, tty, autostart
        with mock.patch.object(runtime, 'load_config', return_value=self.config), \
             mock.patch.object(runtime, '_read', return_value={}), \
             mock.patch.object(runtime, '_terminate'), \
             mock.patch.object(runtime, '_atomic'), \
             mock.patch.object(runtime, 'status', return_value={'running': False}), \
             mock.patch.object(tty, 'stop', return_value={'ok': True, 'cleanup_verified': True}), \
             mock.patch.object(autostart, 'mark_manual_stop'), \
             mock.patch.object(connection_background, 'disconnect', return_value={'ok': True}) as disconnect, \
             mock.patch.object(connection_background, 'status', return_value={'ok': True, 'alive': True}) as status:
            if preserve:
                runtime.stop(self.state, preserve_connection=True)
                disconnect.assert_not_called()
                status.assert_called_once_with(self.state)
            else:
                runtime.stop(self.state)
                disconnect.assert_called_once_with(self.state)
                status.assert_not_called()

    def test_normal_runtime_stop_still_disconnects(self):
        self.check_runtime_stop_connection(False)

    def test_upgrade_runtime_stop_retains_existing_connection(self):
        self.check_runtime_stop_connection(True)


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