ProxMenux/oci/tests/test_modify_review.py

"""Maintainer review: fallback, complete mount edits and retained unused disks."""
import contextlib
import copy
from pathlib import Path
import re
import sys
import tempfile
import unittest
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parents[1]/'remote'))
import oci_instances as instances
import oci_instance_transaction as transaction
import oci_native_modify as native
import oci_update_current as update
import oci_operation_notice
from test_modify_image_reuse import record, CONFIG


def installation():
    result=record()
    result['deployment'].update(rootfs={'storage':'local-zfs','size_gb':8}, onboot=True,
                               security={'unprivileged':True})
    result['observed']['image']['defaults']={}
    config=CONFIG+b'unprivileged: 1\nonboot: 1\nnet0: name=eth0,bridge=vmbr0,ip=dhcp,hwaddr=AA:BB:CC:DD:EE:FF\n'
    result['observed'].update(config=config.decode(), config_sha256=transaction.sha(config))
    return result,config


class SnapshotSectionGuardTests(unittest.TestCase):
    def test_ipv6_listen_address_in_oci_entrypoint_is_not_a_snapshot(self):
        saved, config = installation()
        config += b"entrypoint: entrypoint.sh php -S '[::]:8080' -t /var/www/html\n"
        saved['observed'].update(config=config.decode(), config_sha256=transaction.sha(config))
        with patch.object(transaction, 'external_changes', return_value={}):
            transaction.preflight(saved, copy.deepcopy(saved), config)

    def test_actual_proxmox_snapshot_section_is_still_refused(self):
        saved, config = installation()
        config += b"\n[before-upgrade]\nmemory: 512\n"
        saved['observed'].update(config=config.decode(), config_sha256=transaction.sha(config))
        with patch.object(transaction, 'external_changes', return_value={}), \
                self.assertRaisesRegex(ValueError, 'advanced Proxmox settings'):
            transaction.preflight(saved, copy.deepcopy(saved), config)


class FallbackTests(unittest.TestCase):
    def test_all_installer_adaptations_fall_back_before_any_native_write(self):
        saved,config=installation()
        def nvidia(c):c['deployment']['devices']=[{'kind':'nvidia-runtime','host_path':'/dev/nvidia0'}]
        def groups(c):c['deployment']['devices']=[{'host_path':'/dev/dri/renderD128','append_host_device_gid_to_environment':'GROUPS'}]
        def permissions(c):
            c['deployment']['devices']=[{'host_path':'/dev/dri/renderD128'}]
            c['deployment']['device_permissions']={'uid':1000}
        mutations=[nvidia,groups,permissions,
            lambda c:c['deployment'].update(completion_notes=['Installer note']),
            lambda c:c['deployment'].update(post_start_configurations=[{'id':'adapter'}]),
            lambda c:c['template']['catalog_ui'].update(title='Refreshed catalog template'),
            lambda c:c['deployment'].update(mounts=[{'type':'host-bind','source':'/host/file','container_path':'/config/file','backup':False,'size_gb':None}])]
        for mutation in mutations:
            with self.subTest(mutation=mutation),tempfile.TemporaryDirectory() as folder:
                candidate=copy.deepcopy(saved);mutation(candidate)
                root=Path(folder);instances.write(instances.location(root,115),saved)
                with patch.object(instances,'ROOT',root),patch.object(instances,'locked',return_value=contextlib.nullcontext()), \
                        patch.object(instances,'command',return_value=config), \
                        patch.object(transaction,'external_changes',return_value={}),patch.object(transaction,'preflight'), \
                        patch.object(transaction,'is_file_bind',side_effect=lambda m:m.get('source')=='/host/file'), \
                        patch.object(native,'apply',side_effect=AssertionError('partial native apply')), \
                        patch.object(update,'resolve_archive',return_value=(Path('/cached.tar'),'sha256:resolved')) as resolve, \
                        patch.object(transaction,'apply') as recreate, \
                        patch.object(oci_operation_notice,'operation',return_value=contextlib.nullcontext()):
                    update.update(115,proposal={'operation':'recreate','candidate':candidate})
                recreate.assert_called_once();resolve.assert_called_once()
                self.assertEqual(resolve.call_args.kwargs['local_archives'],[saved['observed']['archive_path']])

    def test_native_validation_failure_never_falls_back_to_recreation(self):
        saved,config=installation();candidate=copy.deepcopy(saved)
        candidate['deployment']['environment']=[{'name':'INVALID NAME','value':'value'}]
        with tempfile.TemporaryDirectory() as folder:
            root=Path(folder);instances.write(instances.location(root,115),saved)
            with patch.object(instances,'ROOT',root),patch.object(instances,'locked',return_value=contextlib.nullcontext()), \
                    patch.object(instances,'command',return_value=config), \
                    patch.object(transaction,'preflight'),patch.object(transaction,'external_changes',return_value={}), \
                    patch.object(transaction,'apply') as recreate,patch.object(update,'resolve_archive') as resolve, \
                    patch.object(oci_operation_notice,'operation',return_value=contextlib.nullcontext()):
                with self.assertRaises(ValueError):update.update(115,proposal={'operation':'recreate','candidate':candidate})
            recreate.assert_not_called();resolve.assert_not_called()

    def test_native_api_failure_never_falls_back_after_starting_the_edit(self):
        saved,config=installation();candidate=copy.deepcopy(saved)
        candidate['deployment']['environment']=[{'name':'VALID','value':'value'}]
        with tempfile.TemporaryDirectory() as folder:
            root=Path(folder);instances.write(instances.location(root,115),saved)
            with patch.object(instances,'ROOT',root),patch.object(instances,'locked',return_value=contextlib.nullcontext()), \
                    patch.object(instances,'command',return_value=config), \
                    patch.object(transaction,'preflight'),patch.object(transaction,'external_changes',return_value={}), \
                    patch.object(transaction,'apply') as recreate,patch.object(update,'resolve_archive') as resolve, \
                    patch.object(native,'apply',side_effect=RuntimeError('native API failed')), \
                    patch.object(oci_operation_notice,'operation',return_value=contextlib.nullcontext()):
                with self.assertRaisesRegex(RuntimeError,'native API failed'):
                    update.update(115,proposal={'operation':'recreate','candidate':candidate})
            recreate.assert_not_called();resolve.assert_not_called()


class MountTests(unittest.TestCase):
    def setUp(self):
        self.record,self.config=installation()
        self.mount={'type':'host-bind','source':'/mnt/old','container_path':'/media',
                    'size_gb':None,'backup':False,'read_only':False,'create_if_missing':False}
        self.record['deployment']['mounts']=[self.mount]
        self.config+=b'mp3: /mnt/old,mp=/media,backup=0,ro=0,replicate=0\n'
        self.record['observed'].update(config=self.config.decode(),config_sha256=transaction.sha(self.config))
        self.candidate=copy.deepcopy(self.record)

    def test_existing_bind_source_and_flags_are_applied_with_other_options_preserved(self):
        self.candidate['deployment']['mounts'][0].update(source='/mnt/new',read_only=True)
        with patch.object(Path,'is_dir',return_value=True):
            plan=native.plan(self.record,self.candidate,self.config)
        self.assertEqual(plan['values'],{'mp3':'/mnt/new,mp=/media,backup=0,ro=1,replicate=0'})
        self.assertTrue(plan['restart']);self.assertEqual(plan['delete'],[])

    def test_missing_replacement_directory_is_rejected_unless_creation_was_requested(self):
        self.candidate['deployment']['mounts'][0]['source']='/mnt/new'
        with patch.object(Path,'is_dir',return_value=False):
            with self.assertRaisesRegex(ValueError,'directory does not exist'):
                native.plan(self.record,self.candidate,self.config)
            self.candidate['deployment']['mounts'][0]['create_if_missing']=True
            self.assertIn('/mnt/new,',native.plan(self.record,self.candidate,self.config)['values']['mp3'])

    def test_verification_detects_a_source_that_was_not_applied(self):
        operation={'values':{'mp3':'/mnt/new,mp=/media,backup=0,ro=0'},'delete':[]}
        with self.assertRaisesRegex(ValueError,'could not apply'):native.verify_values(self.config,operation)
        native.verify_values(self.config.replace(b'/mnt/old',b'/mnt/new'),operation)

    def test_unknown_mount_options_and_permission_grants_use_installer(self):
        for fields in ({'grant_access':True},{'mount_options':['noexec']},{'type':'managed-volume','source':'local-zfs','size_gb':2,'backup':True}):
            candidate=copy.deepcopy(self.candidate);candidate['deployment']['mounts'][0].update(fields)
            self.assertIsNone(native.changes(self.record,candidate,self.config))

    def test_recursive_bind_adaptations_use_installer(self):
        self.candidate['deployment']['mounts'][0]['read_only']=True
        derived=b'lxc.mount.entry: /mnt/old media none rbind,rslave,create=dir 0 0\n'
        self.assertIsNone(native.changes(self.record,self.candidate,self.config+derived))

    def test_storage_and_disk_size_changes_keep_existing_migration_guard(self):
        self.record['deployment']['mounts'][0].update(type='managed-volume',source='local-zfs',size_gb=1,backup=True)
        self.config=self.config.replace(b'/mnt/old,mp=/media,backup=0',b'local-zfs:subvol-115-disk-1,mp=/media,backup=1')
        self.record['observed']['config']=self.config.decode()
        for fields in ({'size_gb':2},{'source':'other-storage'}):
            candidate=copy.deepcopy(self.record);candidate['deployment']['mounts'][0].update(fields)
            self.assertIsNone(native.changes(self.record,candidate,self.config))
            with self.assertRaisesRegex(ValueError,'migration'):transaction.preflight(self.record,candidate,self.config)


class DetachedPreflightTests(unittest.TestCase):
    def test_removing_a_mount_then_updating_accepts_the_recorded_unused_disk(self):
        saved,config=installation()
        saved['deployment']['mounts']=[{'type':'managed-volume','source':'local-zfs','size_gb':1,
            'container_path':'/extra','backup':True,'read_only':False}]
        config+=b'mp4: local-zfs:subvol-115-disk-1,mp=/extra,backup=1,size=1G\n'
        saved['observed']['config']=config.decode()
        candidate=copy.deepcopy(saved);candidate['deployment']['mounts']=[]
        self.assertEqual(native.plan(saved,candidate,config)['delete'],['mp4'])
        detached=config.replace(b'mp4: local-zfs:subvol-115-disk-1,mp=/extra,backup=1,size=1G\n',b'unused0: local-zfs:subvol-115-disk-1\n')
        candidate['observed'].update(config=detached.decode(),config_sha256=transaction.sha(detached))
        cfg,_,_=transaction.preflight(candidate,copy.deepcopy(candidate),detached)
        self.assertIn('unused0',cfg)

    def test_unrecorded_detached_disks_are_not_adopted_silently(self):
        saved,config=installation()
        with self.assertRaisesRegex(ValueError,'disks do not match|discard those changes'):
            transaction.preflight(saved,copy.deepcopy(saved),config+b'unused0: local-zfs:subvol-115-disk-1\n')

    def test_invalid_duplicate_or_mounted_unused_disks_are_rejected(self):
        for extra in (b'unused0: /etc/passwd\n',b'unused0: local-zfs:subvol-115-disk-0\n',
                      b'unused0: local-zfs:subvol-115-disk-1\nunused1: local-zfs:subvol-115-disk-1\n'):
            saved,config=installation();config+=extra;saved['observed']['config']=config.decode()
            with self.assertRaisesRegex(ValueError,'disks do not match'):transaction.preflight(saved,copy.deepcopy(saved),config)


class DetachedTransferTests(unittest.TestCase):
    def setUp(self):
        saved,config=installation()
        self.configs={115:config+b'unused0: local-zfs:subvol-115-disk-1\nunused2: local-zfs:subvol-115-disk-2\n',
                      900:b'description: proxmenux-transaction=test-tx\nrootfs: local-zfs:subvol-900-disk-0\n'}
        self.state={'vmid':115,'stage':900,'id':'test-tx','record':saved,
                    'retained_unused':transaction.unused_volumes(self.configs[115])}
        self.calls=[]
        def owned(vmid,marker):
            self.assertIn(marker.encode(),self.configs[vmid]);return self.configs[vmid]
        def run(*args):
            self.calls.append(args);self.assertEqual(args[:2],('pct','move-volume'))
            source,key,target=int(args[2]),args[3],int(args[5]);self.assertEqual(key,args[7])
            line=next(line for line in self.configs[source].splitlines(keepends=True) if line.startswith((key+': ').encode()))
            self.configs[source]=self.configs[source].replace(line,b'')
            self.configs[target]+=line.replace(('subvol-'+str(source)+'-').encode(),('subvol-'+str(target)+'-').encode())
        for p in (patch.object(transaction,'owned',side_effect=owned),patch.object(transaction,'run',side_effect=run)):
            p.start();self.addCleanup(p.stop)

    def test_unused_disks_are_parked_and_returned_without_freeing_any_volume(self):
        transaction.move_retained_unused(self.state,True)
        self.assertEqual(transaction.unused_volumes(self.configs[115]),{})
        self.assertEqual(set(transaction.unused_volumes(self.configs[900])),{'unused0','unused2'})
        transaction.move_retained_unused(self.state,False)
        self.assertEqual(transaction.unused_volumes(self.configs[115]),self.state['retained_unused'])
        self.assertEqual(transaction.unused_volumes(self.configs[900]),{})
        self.assertEqual(len(self.calls),4)

    def test_an_interrupted_transfer_can_be_retried_in_both_directions(self):
        original=transaction.run.side_effect
        def interrupt(*args):
            original(*args);raise RuntimeError('interrupted after reassignment')
        with patch.object(transaction,'run',side_effect=interrupt):
            with self.assertRaises(RuntimeError):transaction.move_retained_unused(self.state,True)
        transaction.move_retained_unused(self.state,True)
        with patch.object(transaction,'run',side_effect=interrupt):
            with self.assertRaises(RuntimeError):transaction.move_retained_unused(self.state,False)
        transaction.move_retained_unused(self.state,False)
        self.assertEqual(transaction.unused_volumes(self.configs[115]),self.state['retained_unused'])

    def test_unrelated_unused_disks_are_not_moved(self):
        self.configs[115]+=b'unused3: local-zfs:subvol-115-disk-3\n'
        with self.assertRaises(ValueError):transaction.move_retained_unused(self.state,True)
        self.assertEqual(self.calls,[])

    def test_a_missing_disk_or_target_collision_blocks_the_transfer(self):
        del self.state['retained_unused']['unused2']
        self.configs[115]=self.configs[115].replace(b'unused2: local-zfs:subvol-115-disk-2\n',b'')
        self.configs[900]+=b'unused0: local-zfs:subvol-900-disk-9\n'
        with self.assertRaises(ValueError):transaction.move_retained_unused(self.state,True)
        self.configs[115]=self.configs[115].replace(b'unused0: local-zfs:subvol-115-disk-1\n',b'')
        self.configs[900]=self.configs[900].replace(b'unused0: local-zfs:subvol-900-disk-9\n',b'')
        with self.assertRaises(ValueError):transaction.move_retained_unused(self.state,False)


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