File: test_mgmt_recoveryservices_backup.py

package info (click to toggle)
python-azure 20181112%2Bgit-2
  • links: PTS, VCS
  • area: main
  • in suites: buster
  • size: 407,300 kB
  • sloc: python: 717,190; makefile: 201; sh: 76
file content (100 lines) | stat: -rw-r--r-- 4,786 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
# coding: utf-8

# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
import time
from contextlib import contextmanager

import azure.mgmt.recoveryservicesbackup
from devtools_testutils import AzureMgmtTestCase, ResourceGroupPreparer
from recoveryservicesbackup_testcase import MgmtRecoveryServicesBackupTestDefinition, MgmtRecoveryServicesBackupTestHelper


class MgmtRecoveryServicesBackupTests(AzureMgmtTestCase):

    def setUp(self):
        super(MgmtRecoveryServicesBackupTests, self).setUp()

        self.vault_client = self.create_mgmt_client(
            azure.mgmt.recoveryservices.RecoveryServicesClient
        )
        self.backup_client = self.create_mgmt_client(
            azure.mgmt.recoveryservicesbackup.RecoveryServicesBackupClient
        )

        # Using pre-existing vault until vault client is available
        self.resource_group_name = "PythonSDKBackupTestRg"
        self.group_name = self.resource_group_name
        self.vault_name = "PySDKBackupTestRsVault"
        self.test_definition = MgmtRecoveryServicesBackupTestDefinition(self.settings.SUBSCRIPTION_ID, self.vault_name, self.resource_group_name)
        self.test_helper = MgmtRecoveryServicesBackupTestHelper(self)
        self.region = "southeastasia"

    def sleep(self, duration):
        if self.is_live:
            time.sleep(duration)

    @contextmanager
    def vault(self):
        self.test_helper.create_vault()
        yield
        self.test_helper.delete_vault()

    @ResourceGroupPreparer()
    def test_iaasvm_e2e(self, resource_group, location):
        with self.vault():
            self.test_helper.enable_protection(self.test_definition.container_name, self.test_definition.vm_name, "DefaultPolicy")

            protected_items = self.test_helper.list_protected_items()
            self.assertIsNotNone(protected_items)
            protected_items = list(protected_items)
            protected_item = next(protected_item for protected_item in protected_items
                                  if protected_item.name.lower() == self.test_definition.vm_name.lower()).properties

            # Trigger Backup
            backup_job_id = self.test_helper.trigger_backup(self.test_definition.container_name, self.test_definition.vm_name)

            backup_jobs = self.test_helper.list_backup_jobs()
            self.assertIsNotNone(backup_jobs)
            backup_jobs = list(backup_jobs)
            self.assertTrue(any(backup_job_id == backup_job.name for backup_job in backup_jobs))

            self.test_helper.wait_for_job_completion(backup_job_id)

            recovery_points = self.test_helper.list_recovery_points(self.test_definition.container_name, self.test_definition.vm_name)
            self.assertIsNotNone(recovery_points)
            recovery_points = list(recovery_points)
            self.assertTrue(any(recovery_points))

            recovery_point = recovery_points[0]
            self.assertIsNotNone(recovery_point)

            # Trigger Restore
            restore_job_id = self.test_helper.trigger_restore(self.test_definition.container_name, self.test_definition.vm_name, recovery_point.name,
                                                              protected_item.virtual_machine_id, self.test_definition.storage_account_id)
            backup_jobs = self.test_helper.list_backup_jobs()
            self.assertIsNotNone(backup_jobs)
            backup_jobs = list(backup_jobs)
            self.assertTrue(any(restore_job_id == backup_job.name for backup_job in backup_jobs))

            self.test_helper.wait_for_job_completion(restore_job_id)

            # Provision ILR
            self.test_helper.provision_item_level_recovery(self.test_definition.container_name, self.test_definition.vm_name,
                                                                               recovery_point.name, protected_item.virtual_machine_id)

            # Revoke ILR
            self.test_helper.revoke_item_level_recovery(self.test_definition.container_name, self.test_definition.vm_name, recovery_point.name)

            self.test_helper.disable_protection_with_retain_data(self.test_definition.container_name, self.test_definition.vm_name, protected_item)

            # Disable Protection
            self.test_helper.delete_protection(self.test_definition.container_name, self.test_definition.vm_name)

    def test_operations_api(self):
        operations = self.test_helper.list_operations()
        self.assertIsNotNone(operations)
        self.assertTrue(any(operations))