#!/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()