AzStorageAccountAllowTrustedServicesEvent tests

Added test cases for the `AzStorageAccountAllowTrustedServicesEvent` plugin. Also, in this commit a few changes have been made to the `AzStorageAccountAllowTrustedServicesEvent` plugin to take care of the corner cases.
cloudmarker · Mar 5, 2020 · 5ecdb4c · 5ecdb4c
1 parent e4d0ac6
commit 5ecdb4c
Show file tree

Hide file tree

Showing 2 changed files with 85 additions and 0 deletions.
diff --git a/cloudmarker/events/azstorageaccountallowtrustedservicesevent.py b/cloudmarker/events/azstorageaccountallowtrustedservicesevent.py
@@ -33,7 +33,14 @@ def eval(self, record):
 
         """
         com = record.get('com')
+        com = record.get('com', {})
+        if com is None:
+            return
+        if com.get('cloud_type') != 'azure':
+            return
         ext = record.get('ext')
+        if ext is None:
+            return
         if ext.get('record_type') != 'storage_account_properties':
             return
 

diff --git a/cloudmarker/test/test_azstorageaccountallowtrustedservicesevent.py b/cloudmarker/test/test_azstorageaccountallowtrustedservicesevent.py
@@ -0,0 +1,78 @@
+"""Tests for AzStorageAccountAllowTrustedServicesEvent plugin."""
+
+
+import copy
+import unittest
+
+from cloudmarker.events import azstorageaccountallowtrustedservicesevent
+
+base_record = {
+    'com':  {
+        'cloud_type': 'azure',
+        'record_type': 'storage_account_properties',
+    },
+    'ext': {
+        'record_type': 'storage_account_properties',
+        'trusted_services_allowed': False
+    }
+}
+
+
+class AzStorageAccountAllowTrustedServicesEventTest(unittest.TestCase):
+    """Tests for AzStorageAccountAllowTrustedServicesEvent plugin."""
+
+    def test_com_bucket_missing(self):
+        record = copy.deepcopy(base_record)
+        record['com'] = None
+        plugin = azstorageaccountallowtrustedservicesevent. \
+            AzStorageAccountAllowTrustedServicesEvent()
+        events = list(plugin.eval(record))
+        self.assertEqual(events, [])
+
+    def test_cloud_non_azure(self):
+        record = copy.deepcopy(base_record)
+        record['com']['cloud_type'] = 'non_azure'
+        plugin = azstorageaccountallowtrustedservicesevent. \
+            AzStorageAccountAllowTrustedServicesEvent()
+        events = list(plugin.eval(record))
+        self.assertEqual(events, [])
+
+    def test_record_type_non_storage_account_properties(self):
+        record = copy.deepcopy(base_record)
+        record['ext']['record_type'] = 'non_storage_account_properties'
+        plugin = azstorageaccountallowtrustedservicesevent. \
+            AzStorageAccountAllowTrustedServicesEvent()
+        events = list(plugin.eval(record))
+        self.assertEqual(events, [])
+
+    def test_ext_bucket_missing(self):
+        record = copy.deepcopy(base_record)
+        record['ext'] = None
+        plugin = azstorageaccountallowtrustedservicesevent. \
+            AzStorageAccountAllowTrustedServicesEvent()
+        events = list(plugin.eval(record))
+        self.assertEqual(events, [])
+
+    def test_trusted_services_allowed(self):
+        record = copy.deepcopy(base_record)
+        record['ext']['trusted_services_allowed'] = True
+        plugin = azstorageaccountallowtrustedservicesevent. \
+            AzStorageAccountAllowTrustedServicesEvent()
+        events = list(plugin.eval(record))
+        self.assertEqual(events, [])
+
+    def test_trusted_services_not_allowed(self):
+        record = copy.deepcopy(base_record)
+        plugin = azstorageaccountallowtrustedservicesevent. \
+            AzStorageAccountAllowTrustedServicesEvent()
+        events = list(plugin.eval(record))
+        self.assertEqual(len(events), 1)
+        self.assertEqual(events[0]['ext']['record_type'],
+                         'storage_account_allow_trusted_services_event')
+        self.assertEqual(events[0]['com']['cloud_type'],
+                         'azure')
+        self.assertEqual(events[0]['com']['record_type'],
+                         'storage_account_allow_trusted_services_event')
+        self.assertTrue('reference' in events[0]['com'])
+        self.assertIsNotNone(events[0]['com']['description'])
+        self.assertIsNotNone(events[0]['com']['recommendation'])