test_cloud_improvements.py #1

#!/usr/bin/env python3
"""Offline cloud backup tests. Every provider call is a mocked CLI response."""
from copy import deepcopy
from datetime import datetime, timezone
import importlib.util
import json
import os
from pathlib import Path
import subprocess
import unittest
from unittest.mock import patch

CLOUD = Path(__file__).resolve().parents[1] / 'Server/Unix/p4/common/cloud'
spec = importlib.util.spec_from_file_location('sdp_snapshot', CLOUD / 'bin/sdp_snapshot.py')
cloud = importlib.util.module_from_spec(spec)
spec.loader.exec_module(cloud)
NOW = datetime(2026, 10, 16, tzinfo=timezone.utc)
RUN = '20261016T000000Z-012345abcdef'
OLD_RUN = '20260101T000000Z-abcdef012345'
ACCOUNT = '123456789012'
SUB = '11111111-1111-1111-1111-111111111111'


class AWS:
    def __init__(self):
        self.calls, self.created, self.old = [], [], []
        self.fail, self.change_tags, self.incomplete = None, False, False
        self.names = {'host-root': 'vol-1', 'host-hxdepots': 'vol-2', 'host-hxlogs': 'vol-3'}
        self.volumes = {name: {'VolumeId': vid, 'Tags': [{'Key': 'Name', 'Value': name}],
                              'Attachments': [{'InstanceId': 'i-abc', 'State': 'attached'}]}
                        for name, vid in self.names.items()}
        self.vm = {'InstanceId': 'i-abc', 'RootDeviceName': '/dev/root', 'BlockDeviceMappings':
                   [{'DeviceName': '/dev/root' if n == 1 else '/dev/data' + str(n), 'Ebs': {'VolumeId': 'vol-' + str(n)}} for n in range(1, 5)]}

    def __call__(self, argv):
        self.calls.append(argv)
        cmd = argv[2]
        if cmd == self.fail:
            raise cloud.SnapshotError('AWS mock failure: ' + cmd)
        if cmd == 'get-caller-identity':
            return {'Account': ACCOUNT}
        if cmd == 'describe-volumes':
            name = json.loads(argv[argv.index('--filters') + 1])[0]['Values'][0]
            return {'Volumes': deepcopy(self.volumes.get(name, [])) if isinstance(self.volumes.get(name), list) else [deepcopy(self.volumes[name])]}
        if cmd == 'describe-instances':
            return {'Reservations': [{'Instances': [deepcopy(self.vm)]}]}
        if cmd == 'create-snapshots':
            tags = json.loads(argv[argv.index('--tag-specifications') + 1])[0]['Tags']
            self.created = [{'SnapshotId': 'snap-a' + str(i), 'VolumeId': vid, 'OwnerId': ACCOUNT,
                             'Tags': deepcopy(tags), 'State': 'pending', 'StartTime': NOW.isoformat()}
                            for i, vid in enumerate(self.names.values())]
            if self.change_tags:
                self.created[0]['Tags'] = []
            return {'Snapshots': deepcopy(self.created[:-1] if self.incomplete else self.created)}
        if cmd == 'wait':
            for snap in self.created:
                snap['State'] = 'completed'
            return None
        if cmd == 'describe-snapshots':
            return {'Snapshots': deepcopy(self.created if '--snapshot-ids' in argv else self.old + self.created)}
        if cmd == 'delete-snapshot':
            return None
        raise AssertionError('Unexpected AWS mock command ' + repr(argv))

    def old_set(self, backup, count=3):
        tags = dict(backup.scope(ACCOUNT), SDPSourceInstance='i-abc', SDPRun=OLD_RUN,
                    SDPSetSize=str(count), SDPJournal='7')
        return [{'SnapshotId': 'snap-b' + str(i), 'VolumeId': 'vol-' + str(i+1), 'OwnerId': ACCOUNT,
                 'Tags': [{'Key': k, 'Value': v} for k, v in tags.items()], 'State': 'completed',
                 'StartTime': '2026-01-01T00:00:00Z'} for i in range(count)]


class Azure:
    def __init__(self):
        self.calls, self.old, self.created = [], [], None
        self.fail, self.existing = None, True
        self.group, self.name, self.collection_name = 'rg', 'vm', 'owned'
        self.vm_id = '/subscriptions/' + SUB + '/resourceGroups/rg/providers/Microsoft.Compute/virtualMachines/vm'
        self.collection_id = '/subscriptions/' + SUB + '/resourceGroups/rg/providers/Microsoft.Compute/restorePointCollections/owned'
        self.disk_ids = ['/subscriptions/' + SUB + '/resourceGroups/rg/providers/Microsoft.Compute/disks/' + disk for disk in ('os', 'data')]
        self.vm = {'id': self.vm_id, 'location': 'eastus', 'vmId': 'vm-uuid',
                   'storageProfile': {'osDisk': {'osType': 'Linux', 'managedDisk': {'id': self.disk_ids[0]}},
                                      'dataDisks': [{'managedDisk': {'id': self.disk_ids[1]}}]}}
        self.disks = {disk: {'id': disk, 'sku': {'name': 'Standard_LRS'}, 'maxShares': 1} for disk in self.disk_ids}
        self.collection = {'id': self.collection_id, 'name': 'owned', 'source': {'id': self.vm_id}, 'provisioningState': 'Succeeded',
                           'tags': {'SDPManagedBy': cloud.MANAGER, 'SDPSchema': '1', 'SDPAccount': SUB,
                                    'SDPHost': 'host', 'SDPInstance': '1', 'SDPSourceVM': 'vm-uuid'}}
        self.point_change = lambda value: value

    def point(self, name, created='2026-01-01T00:00:00Z'):
        point_id = self.collection_id + '/restorePoints/' + name
        disks = [{'managedDisk': {'id': disk}, 'diskRestorePoint': {'id': point_id + '/diskRestorePoints/' + str(i)}}
                 for i, disk in enumerate(self.disk_ids)]
        return {'id': point_id, 'name': name, 'provisioningState': 'Succeeded', 'consistencyMode': 'CrashConsistent', 'timeCreated': created,
                'sourceMetadata': {'vmId': 'vm-uuid', 'storageProfile': {'osDisk': disks[0], 'dataDisks': disks[1:]}},
                'instanceView': {'diskRestorePoints': [{'id': disk['diskRestorePoint']['id'],
                 'replicationStatus': {'completionPercent': 100, 'status': {'code': 'ReplicationState/succeeded'}}} for disk in disks]}}

    def __call__(self, argv):
        self.calls.append(argv)
        args = argv[1:]
        cmd = ' '.join(args[:3] if args[:2] == ['restore-point', 'collection'] else args[:2])
        if self.fail == cmd:
            raise cloud.SnapshotError('Azure mock failure: ' + cmd)
        if cmd == 'account show':
            return {'id': SUB}
        if cmd == 'vm show':
            return deepcopy(self.vm)
        if cmd == 'disk show':
            return deepcopy(self.disks[args[args.index('--ids')+1]])
        if cmd == 'restore-point collection list':
            return [deepcopy(self.collection)] if self.existing else []
        if cmd == 'restore-point collection create':
            self.existing = True
            return deepcopy(self.collection)
        if cmd == 'restore-point collection show':
            result = deepcopy(self.collection)
            if '--restore-points' in args:
                result['restorePoints'] = deepcopy(self.old + ([self.created] if self.created else []))
            return result
        if cmd == 'restore-point create':
            name = args[args.index('--name')+1]
            self.created = self.point_change(self.point(name, NOW.isoformat()))
            return deepcopy(self.created)
        if cmd == 'restore-point wait':
            return None
        if cmd == 'restore-point show':
            name = args[args.index('--name')+1]
            return deepcopy(next(point for point in self.old + [self.created] if point and point['name'] == name))
        if cmd == 'restore-point delete':
            return None
        raise AssertionError('Unexpected Azure mock command ' + repr(argv))


class AWSTests(unittest.TestCase):
    def setUp(self):
        self.cli = AWS()
        self.backup = cloud.Backup('1', 'host', '42', now=NOW, run_id=RUN, runner=self.cli, environ={})

    def deletions(self):
        return [a for a in self.cli.calls if a[2] == 'delete-snapshot']

    def test_single_coordinated_request_tags_at_creation_excludes_unselected_disks_waits_before_pruning(self):
        self.cli.old = self.cli.old_set(self.backup)
        self.backup.run_aws()
        create, = [a for a in self.cli.calls if a[2] == 'create-snapshots']
        self.assertEqual(json.loads(create[create.index('--instance-specification')+1]),
                         {'InstanceId': 'i-abc', 'ExcludeBootVolume': False, 'ExcludeDataVolumeIds': ['vol-4']})
        tags = cloud.tags_dict(json.loads(create[create.index('--tag-specifications')+1])[0]['Tags'])
        self.assertEqual(tags['SDPRun'], RUN)
        self.assertEqual(tags['SDPJournal'], '42')
        self.assertEqual(tags['SDPAccount'], ACCOUNT)
        self.assertEqual(len(self.deletions()), 3)
        self.assertLess(next(i for i,a in enumerate(self.cli.calls) if a[2] == 'wait'),
                        next(i for i,a in enumerate(self.cli.calls) if a[2] == 'delete-snapshot'))
        self.assertTrue(all(a[a.index('--owner-ids')+1] == 'self' for a in self.cli.calls if a[2] == 'describe-snapshots'))

    def test_excluded_boot_volume_never_appears_in_excluded_data_ids(self):
        self.cli.vm['RootDeviceName'] = '/dev/data4'
        self.backup.run_aws()
        create = next(a for a in self.cli.calls if a[2] == 'create-snapshots')
        spec = json.loads(create[create.index('--instance-specification')+1])
        self.assertTrue(spec['ExcludeBootVolume'])
        self.assertNotIn('ExcludeDataVolumeIds', spec)

    def test_duplicate_volume_names_fail_before_creation(self):
        self.cli.volumes['host-root'] = [self.cli.volumes['host-root']] * 2
        with self.assertRaises(cloud.SnapshotError): self.backup.run_aws()
        self.assertFalse(any(a[2] == 'create-snapshots' for a in self.cli.calls))

    def test_cross_instance_and_multiattach_volume_fail_before_creation(self):
        for attachments in [[{'InstanceId': 'i-def', 'State': 'attached'}],
                            [{'InstanceId': 'i-abc', 'State': 'attached'}] * 2]:
            with self.subTest(attachments=attachments):
                self.cli.volumes['host-hxlogs']['Attachments'] = attachments
                with self.assertRaises(cloud.SnapshotError): self.backup.run_aws()
        self.assertFalse(any(a[2] == 'create-snapshots' for a in self.cli.calls))

    def test_create_wait_or_tag_failure_never_expires_previous_backups(self):
        for failure in ['create-snapshots', 'wait', 'missing-tags', 'incomplete-set']:
            with self.subTest(failure=failure):
                cli = AWS()
                cli.fail = failure
                cli.change_tags = failure == 'missing-tags'
                cli.incomplete = failure == 'incomplete-set'
                backup = cloud.Backup('1', 'host', '42', now=NOW, run_id=RUN, runner=cli, environ={})
                cli.old = cli.old_set(backup)
                with self.assertRaises(cloud.SnapshotError): backup.run_aws()
                self.assertFalse(any('--filters' in a and a[2] == 'describe-snapshots' for a in cli.calls))
                self.assertFalse(any(a[2] == 'delete-snapshot' for a in cli.calls))

    def test_unrelated_account_host_instance_legacy_tags_are_retained(self):
        records = self.cli.old_set(self.backup)
        for key, value in [('SDPHost','other'),('SDPInstance','2'),('SDPManagedBy','other'),('SDPAccount','987654321012')]:
            self.cli.old = deepcopy(records)
            for snap in self.cli.old:
                for tag in snap['Tags']:
                    if tag['Key'] == key: tag['Value'] = value
            self.backup.run_aws()
            self.assertEqual(self.deletions(), [])
        self.cli.old = deepcopy(records)
        for snap in self.cli.old: snap['Tags'] = [{'Key': 'Contents', 'Value': 'Perforce Checkpoints and Archives'}]
        self.backup.run_aws()
        self.assertEqual(self.deletions(), [])

    def test_incomplete_or_pending_old_set_retained(self):
        self.cli.old = self.cli.old_set(self.backup)[:-1]
        self.backup.run_aws()
        self.assertEqual(self.deletions(), [])
        self.cli.old = self.cli.old_set(self.backup)
        self.cli.old[0]['State'] = 'pending'
        self.backup.run_aws()
        self.assertEqual(self.deletions(), [])

    def test_invalid_owned_date_blocks_all_expiration(self):
        self.cli.old = self.cli.old_set(self.backup)
        self.cli.old[-1]['StartTime'] = 'invalid'
        with self.assertRaises(cloud.SnapshotError): self.backup.run_aws()
        self.assertEqual(self.deletions(), [])

    def test_delete_failure_is_reported(self):
        self.cli.old = self.cli.old_set(self.backup)
        self.cli.fail = 'delete-snapshot'
        with self.assertRaisesRegex(cloud.SnapshotError, 'Could not delete expired snapshot'): self.backup.run_aws()

    def test_recent_sets_and_current_set_are_retained(self):
        self.cli.old = self.cli.old_set(self.backup)
        for snap in self.cli.old: snap['StartTime'] = NOW.isoformat()
        self.backup.run_aws()
        self.assertEqual(self.deletions(), [])

    def test_instance_pin_and_exclusion_limit_checked_before_mutation(self):
        self.backup.env['SDP_AWS_INSTANCE_ID'] = 'i-def'
        with self.assertRaises(cloud.SnapshotError): self.backup.run_aws()
        self.backup.env.clear()
        for n in range(5, 46): self.cli.vm['BlockDeviceMappings'].append({'DeviceName': '/dev/x'+str(n), 'Ebs': {'VolumeId': 'vol-'+format(n,'x')}})
        with self.assertRaises(cloud.SnapshotError): self.backup.run_aws()
        self.assertFalse(any(a[2] == 'create-snapshots' for a in self.cli.calls))


class AzureTests(unittest.TestCase):
    def setUp(self):
        self.cli = Azure()
        self.env = {'SDP_AZURE_SUBSCRIPTION': SUB, 'SDP_AZURE_RESOURCE_GROUP': 'rg', 'SDP_AZURE_VM': 'vm', 'SDP_AZURE_COLLECTION': 'owned'}
        self.backup = cloud.Backup('1', 'host', '42', now=NOW, run_id=RUN, runner=self.cli, environ=self.env)

    def deletions(self):
        return [a for a in self.cli.calls if a[1:3] == ['restore-point', 'delete']]

    def test_crash_consistent_set_waits_for_disk_copy_before_scoped_expiration(self):
        self.cli.old = [self.cli.point('sdp-v1-j7-' + OLD_RUN), self.cli.point('unmanaged')]
        self.backup.run_azure()
        create, = [a for a in self.cli.calls if a[1:3] == ['restore-point', 'create']]
        self.assertEqual(create[create.index('--consistency-mode')+1], 'CrashConsistent')
        self.assertEqual(len(self.deletions()), 1)
        self.assertEqual(self.deletions()[0][self.deletions()[0].index('--name')+1], 'sdp-v1-j7-' + OLD_RUN)
        self.assertTrue(any('--custom' in a and '--expand' in a for a in self.cli.calls))
        self.assertFalse(any(a[1:4] == ['restore-point', 'collection', 'delete'] for a in self.cli.calls))
        self.assertTrue(all(a[a.index('--subscription')+1] == SUB for a in self.cli.calls))

    def test_new_collection_is_owned_at_creation(self):
        self.cli.existing = False
        self.backup.run_azure()
        create, = [a for a in self.cli.calls if a[1:4] == ['restore-point', 'collection', 'create']]
        self.assertIn('SDPManagedBy=' + cloud.MANAGER, create)
        self.assertEqual(create[create.index('--source-id')+1], self.cli.vm_id)

    def test_wrong_owner_or_source_collection_is_never_modified(self):
        for mode in ['owner', 'source']:
            self.cli = Azure()
            self.backup.runner = self.cli
            if mode == 'owner': self.cli.collection['tags']['SDPHost'] = 'other'
            else: self.cli.collection['source']['id'] += '-other'
            with self.assertRaises(cloud.SnapshotError): self.backup.run_azure()
            self.assertFalse(any('create' in a or 'delete' in a for a in self.cli.calls))

    def test_unmanaged_or_unsupported_disks_stop_before_collection_writes(self):
        mutations = [lambda: self.cli.vm['storageProfile']['osDisk'].update(diffDiskSettings={'option':'Local'}),
                     lambda: self.cli.vm['storageProfile']['dataDisks'][0].update(writeAcceleratorEnabled=True),
                     lambda: self.cli.disks[self.cli.disk_ids[1]].update(maxShares=2),
                     lambda: self.cli.disks[self.cli.disk_ids[1]]['sku'].update(name='PremiumV2_LRS'),
                     lambda: self.cli.disks[self.cli.disk_ids[1]]['sku'].update(name='UltraSSD_LRS')]
        for mutate in mutations:
            self.cli = Azure()
            self.backup.runner = self.cli
            mutate()
            with self.assertRaises(cloud.SnapshotError): self.backup.run_azure()
            self.assertFalse(any('create' in a or 'delete' in a for a in self.cli.calls))

    def test_create_or_wait_failure_preserves_previous_restore_points(self):
        for stage in ['restore-point create', 'restore-point wait']:
            self.cli = Azure()
            self.backup.runner = self.cli
            self.cli.fail = stage
            self.cli.old = [self.cli.point('sdp-v1-j7-' + OLD_RUN)]
            with self.assertRaises(cloud.SnapshotError): self.backup.run_azure()
            self.assertEqual(self.deletions(), [])

    def test_incomplete_source_set_or_disk_copy_blocks_expiration(self):
        for mode in ['copy', 'set', 'vm']:
            self.cli = Azure()
            self.backup.runner = self.cli
            def change(point):
                if mode == 'copy': point['instanceView']['diskRestorePoints'][0]['replicationStatus']['completionPercent'] = 99
                if mode == 'set': point['sourceMetadata']['storageProfile']['dataDisks'] = []
                if mode == 'vm': point['sourceMetadata']['vmId'] = 'other'
                return point
            self.cli.point_change = change
            self.cli.old = [self.cli.point('sdp-v1-j7-' + OLD_RUN)]
            with self.assertRaises(cloud.SnapshotError): self.backup.run_azure()
            self.assertEqual(self.deletions(), [])

    def test_other_vm_pending_recent_or_legacy_points_retained(self):
        old = self.cli.point('sdp-v1-j7-' + OLD_RUN)
        for mode in ['vm', 'pending', 'recent', 'legacy']:
            point = deepcopy(old)
            if mode == 'vm': point['sourceMetadata']['vmId'] = 'other'
            if mode == 'pending': point['provisioningState'] = 'Creating'
            if mode == 'recent': point['timeCreated'] = NOW.isoformat()
            if mode == 'legacy': point['name'] = 'legacy'
            self.cli.old = [point]
            self.backup.run_azure()
            self.assertEqual(self.deletions(), [])

    def test_wrong_collection_child_or_invalid_time_blocks_deletion(self):
        for mode in ['scope','date']:
            self.cli.old = [self.cli.point('sdp-v1-j7-' + OLD_RUN)]
            if mode == 'scope': self.cli.old[0]['id'] = '/unrelated/restorePoints/old'
            else: self.cli.old[0]['timeCreated'] = 'bad'
            with self.assertRaises(cloud.SnapshotError): self.backup.run_azure()
            self.assertEqual(self.deletions(), [])

    def test_delete_failure_is_reported(self):
        self.cli.old = [self.cli.point('sdp-v1-j7-' + OLD_RUN)]
        self.cli.fail = 'restore-point delete'
        with self.assertRaisesRegex(cloud.SnapshotError, 'Azure mock failure'): self.backup.run_azure()

    def test_missing_configuration_stops_before_any_provider_call(self):
        self.env.pop('SDP_AZURE_SUBSCRIPTION')
        with self.assertRaises(cloud.SnapshotError): self.backup.run_azure()
        self.assertEqual(self.cli.calls, [])


class Commands(unittest.TestCase):
    def test_cli_receives_argument_vector_and_rejects_failed_or_non_json_output(self):
        for result in [subprocess.CompletedProcess([], 7, '', 'specific failure'), subprocess.CompletedProcess([],0,'bad','')]:
            with patch.object(cloud.subprocess, 'run', return_value=result) as run, self.assertRaises(cloud.SnapshotError):
                cloud.command(['aws', 'safe argument with spaces'])
            self.assertNotIn('shell', run.call_args.kwargs)
            self.assertEqual(run.call_args.args[0], ['aws', 'safe argument with spaces'])

    def test_retention_and_journal_validation(self):
        for retention in ['0','-1','invalid']:
            with self.assertRaises(cloud.SnapshotError): cloud.Backup('1','host','42', retention=retention)
        with self.assertRaises(cloud.SnapshotError): cloud.Backup('1','host','invalid')

    def test_arm_properties_layout_supported(self):
        source = {'id': 'resource', 'properties': {'provisioningState':'Succeeded'}}
        self.assertEqual(cloud.properties(source)['provisioningState'], 'Succeeded')


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