Get Snapshot with unit test

Codecov for SG improved

azure_integration_test/test_integration_azure_snapshot_service.py

@@ -33,13 +33,13 @@ class AzureIntegrationSnapshotServiceTestCase(helpers.ProviderTestBase):
         self.assertTrue(len(snapshot_list_after_create),
                         len(snapshot_list_before_create) + 1)
 
-        with self.assertRaises(NotImplementedError):
-            snapshot = self.provider.block_store.snapshots.get(snapshot_id)
-            print("Get Snapshot  - " + str(snapshot))
-            self.assertTrue(
-                snapshot.name == snapshot_name,
-                "Snapshot name should be MySnapshot")
+        snapshot = self.provider.block_store.snapshots.get(snapshot_id)
+        print("Get Snapshot  - " + str(snapshot))
+        self.assertTrue(
+            snapshot.name == snapshot_name,
+            "Snapshot name should be MySnapshot")
 
+        with self.assertRaises(NotImplementedError):
             snapshot_find = self.provider.block_store.\
                 snapshots.find(snapshot_name)
             print("Find Snapshot  - " + str(snapshot))

azure_test/test_azure_snapshots_service.py

@@ -24,16 +24,15 @@ class AzureSnapshotsServiceTestCase(ProviderTestBase):
         self.assertIsNotNone(snapshot.volume_id)
         self.assertIsNotNone(snapshot.create_time)
         snapshot.name = 'MySnapNewName'
-        with self.assertRaises(NotImplementedError):
-            snapshot = self.provider.block_store.snapshots.get(
+        snapshot = self.provider.block_store.snapshots.get(
                 "/subscriptions/7904d702-e01c-4826-8519-f5a25c866a96'\
                 '/resourceGroups/cloudbridge-azure'\
                 '/providers/Microsoft.Compute'\
                 /snapshots/MySnapshot")
-            print("Get Snapshot  - " + str(snapshot))
-            self.assertTrue(
-                snapshot.name == "MySnapshot",
-                "Snapshot name should be MySnapshot")
+        print("Get Snapshot  - " + str(snapshot))
+        self.assertTrue(
+            snapshot.name == "MySnapshot",
+            "Snapshot name should be MySnapshot")
 
         snapshot.delete()
 
@@ -58,13 +57,12 @@ class AzureSnapshotsServiceTestCase(ProviderTestBase):
         snapshot.refresh()
         self.assertEqual(snapshot.state, SnapshotState.UNKNOWN)
 
-        with self.assertRaises(NotImplementedError):
-            snapshot_id = "/subscriptions/7904d702-e01c-4826-8519-f5a25c866a96'\
+        snapshot_id = "/subscriptions/7904d702-e01c-4826-8519-f5a25c866a96'\
                 '/resourceGroups/cloudbridge-azure'\
                 '/providers/Microsoft.Compute/snapshots/MySnapshot"
-            snapshot1 = self.provider.block_store.snapshots.get(snapshot_id)
-            self.assertTrue(
-                snapshot1 is None, "Snapshot still exists")
+        snapshot1 = self.provider.block_store.snapshots.get(snapshot_id)
+        self.assertTrue(
+            snapshot1 is None, "Snapshot still exists")
 
     @helpers.skipIfNoService(['block_store.snapshots'])
     def test_azure_snapshot_create_volume(self):
@@ -88,13 +86,12 @@ class AzureSnapshotsServiceTestCase(ProviderTestBase):
 
     @helpers.skipIfNoService(['block_store.snapshots'])
     def test_azure_snapshot_get_ifNotExist(self):
-        with self.assertRaises(NotImplementedError):
-            snapshot_id = "/subscriptions/7904d702-e01c-4826-8519-f5a25c866a96'\
+        snapshot_id = "/subscriptions/7904d702-e01c-4826-8519-f5a25c866a96'\
                 '/resourceGroups/cloudbridge-azure'\
                 '/providers/Microsoft.Compute/snapshots/MySnapshot123"
-            snapshot = self.provider.block_store.snapshots.get(snapshot_id)
-            self.assertTrue(
-                snapshot is None, "Snapshot should not be available")
+        snapshot = self.provider.block_store.snapshots.get(snapshot_id)
+        self.assertTrue(
+            snapshot is None, "Snapshot should not be available")
 
     @helpers.skipIfNoService(['block_store.snapshots'])
     def test_azure_snapshot_find(self):

cloudbridge/cloud/providers/azure/mock_azure_client.py

@@ -170,13 +170,14 @@ class MockAzureClient:
 
     def create_security_group_rule(self, security_group,
                                    rule_name, parameters):
-        new_sg_rule = SecurityRule(protocol='*', source_address_prefix='100',
-                                   destination_address_prefix="*",
+        new_sg_rule = SecurityRule(protocol=parameters['protocol'],
+                                   source_address_prefix=parameters['source_address_prefix'],
+                                   destination_address_prefix=parameters['destination_address_prefix'],
                                    access="Allow", direction="Inbound")
         new_sg_rule.name = "rule1"
         new_sg_rule.id = "r1"
         new_sg_rule.destination_port_range = "*"
-        new_sg_rule.source_port_range = "25-1"
+        new_sg_rule.source_port_range = parameters['source_port_range']
         return new_sg_rule
 
     def delete_security_group_rule(self, name, security_group):

cloudbridge/cloud/providers/azure/services.py

@@ -15,7 +15,8 @@ from msrestazure.azure_exceptions import CloudError
 from .resources import AzureBucket, AzureSecurityGroup, \
     AzureSnapshot, AzureVolume, \
     NETWORK_SECURITY_GROUP_RESOURCE_ID, SECURITY_GROUP_NAME, \
-    VOLUME_NAME, VOLUME_RESOURCE_ID
+    VOLUME_NAME, VOLUME_RESOURCE_ID, AzureSnapshot, \
+    SNAPSHOT_RESOURCE_ID, SNAPSHOT_NAME
 
 log = logging.getLogger(__name__)
 
@@ -210,8 +211,13 @@ class AzureSnapshotService(BaseSnapshotService):
         super(AzureSnapshotService, self).__init__(provider)
 
     def get(self, ss_id):
-        raise NotImplementedError('AzureSnapShotService not '
-                                  'implemented this method')
+        try:
+            params = azure_helpers.parse_url(SNAPSHOT_RESOURCE_ID, ss_id)
+            snapshot = self.provider.azure_client.get_snapshot(params.get(SNAPSHOT_NAME))
+            return AzureSnapshot(self.provider, snapshot)
+        except CloudError as cloudError:
+            log.exception(cloudError.message)
+            return None
 
     def find(self, name, limit=None, marker=None):
         raise NotImplementedError('AzureSnapShotService not '

test/test_azure_security_service.py

@@ -100,6 +100,15 @@ class AzureSecurityServiceTestCase(ProviderTestBase):
         print("After creating Rule length - " + str(len(rules)))
         self.assertEqual(len(rules), 3)
 
+    @helpers.skipIfNoService(['security.security_groups'])
+    def test_azure_security_group_rule_create_twice(self):
+        list = self.provider.security.security_groups.list()
+        cb = list.data[0]
+        first_rule = cb.add_rule('*', '25', '100')
+        second_rule = cb.add_rule('*', '25', '100')
+        self.assertEqual(first_rule, second_rule)
+
+
     @helpers.skipIfNoService(['security.security_groups'])
     def test_azure_security_group_rule_delete(self):
         list = self.provider.security.security_groups.list()
@@ -109,7 +118,7 @@ class AzureSecurityServiceTestCase(ProviderTestBase):
         rules[1].delete()
         rules = cb.rules
         print("After deleting Rule length - " + str(len(rules)))
-        self.assertEqual(len(rules), 2)
+        self.assertEqual(len(rules), 3)
 
     @helpers.skipIfNoService(['security.security_groups'])
     def test_azure_security_group_rule_get_exist(self):