dcae-services-policy-sync/tests/test_inventory.py

   1 # ============LICENSE_START=======================================================
   2 # Copyright (c) 2021 AT&T Intellectual Property. All rights reserved.
   3 # ================================================================================
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #      http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15 # ============LICENSE_END=========================================================
  16
  17 import pytest, json, aiohttp, asyncio
  18 from policysync.inventory import (
  19     Inventory,
  20     ACTION_GATHERED,
  21     ACTION_UPDATED,
  22 )
  23 from tests.mocks import MockClient
  24
  25
  26 class MockMessage:
  27     def __init__(self, type, data):
  28         self.type = type
  29         self.data = data
  30
  31
  32 @pytest.fixture()
  33 def inventory(request, tmpdir):
  34     f1 = tmpdir.mkdir("sub").join("myfile")
  35     print(f1)
  36     return Inventory(["DCAE.Config_MS_AGING_UVERSE_PROD_.*"], [], f1, MockClient())
  37
  38
  39 class TestInventory:
  40     @pytest.mark.asyncio
  41     async def test_close(self, inventory):
  42         await inventory.close()
  43         assert inventory.client.closed
  44
  45     @pytest.mark.asyncio
  46     async def test_get_policy_content(self, inventory):
  47         await inventory.get_policy_content()
  48         with open(inventory.file) as f:
  49             data = json.load(f)
  50
  51         assert data["policies"] == {
  52             "items": [
  53                 {
  54                     "policyConfigMessage": "Config Retrieved!",
  55                     "policyConfigStatus": "CONFIG_RETRIEVED",
  56                     "type": "JSON",
  57                     "config": {
  58                         "service": "DCAE_HighlandPark_AgingConfig",
  59                         "location": " Edge",
  60                         "uuid": "TestUUID",
  61                         "policyName": "DCAE.AGING_UVERS_PROD_Tosca_HP_GOC_Model_cl55973_IT64_testAging",
  62                         "configName": "DCAE_HighlandPark_AgingConfig",
  63                         "templateVersion": "1607",
  64                         "priority": "4",
  65                         "version": 11.0,
  66                         "policyScope": "resource=Test1,service=vSCP,type=configuration,closedLoopControlName=vSCP_F5_Firewall_d925ed73_7831_4d02_9545_db4e101f88f8",
  67                         "riskType": "test",
  68                         "riskLevel": "2",
  69                         "guard": "False",
  70                         "content": {
  71                             "signature": {
  72                                 "filter_clause": "event.faultFields.alarmCondition LIKE('%chrisluckenbaugh%')"
  73                             },
  74                             "name": "testAging",
  75                             "context": ["PROD"],
  76                             "priority": 1,
  77                             "prePublishAging": 40,
  78                             "preCorrelationAging": 20,
  79                         },
  80                         "policyNameWithPrefix": "DCAE.AGING_UVERSE_PSL_Tosca_HP_GOC_Model_cl55973_IT64_testAging",
  81                     },
  82                     "policyName": "DCAE.Config_MS_AGING_UVERSE_PROD_Tosca_HP_AGING_Model_cl55973_IT64_testAging.78.xml",
  83                     "policyType": "MicroService",
  84                     "policyVersion": "78",
  85                     "matchingConditions": {
  86                         "ECOMPName": "DCAE",
  87                         "ONAPName": "DCAE",
  88                         "ConfigName": "DCAE_HighlandPark_AgingConfig",
  89                         "service": "DCAE_HighlandPark_AgingConfig",
  90                         "uuid": "TestUUID",
  91                         "Location": " Edge",
  92                     },
  93                     "responseAttributes": {},
  94                     "property": None,
  95                 },
  96                 {
  97                     "policyConfigMessage": "Config Retrieved! ",
  98                     "policyConfigStatus": "CONFIG_RETRIEVED",
  99                     "type": "JSON",
 100                     "config": "adlskjfadslkjf",
 101                     "policyName": "DCAE.Config_MS_AGING_UVERSE_PROD_Tosca_HP_AGING_Model_cl55973_IT64_testAging.78.xml",
 102                     "policyType": "MicroService",
 103                     "policyVersion": "78",
 104                     "matchingConditions": {
 105                         "ECOMPName": "DCAE",
 106                         "ONAPName": "DCAE",
 107                         "ConfigName": "DCAE_HighlandPark_AgingConfig",
 108                         "service": "DCAE_HighlandPark_AgingConfig",
 109                         "uuid": "TestUUID",
 110                         "Location": " Edge",
 111                     },
 112                     "responseAttributes": {},
 113                     "property": None,
 114                 },
 115             ]
 116         }
 117
 118         assert data["event"]["action"] == ACTION_UPDATED
 119
 120     @pytest.mark.asyncio
 121     async def test_update(self, inventory):
 122         await inventory.update()
 123         assert len(inventory.hp_active_inventory) == 1
 124
 125         assert not await inventory.update()
 126
 127     @pytest.mark.asyncio
 128     async def test_update_listpolicies_exception(self, inventory):
 129         inventory.client.raise_on_listpolicies = True
 130         assert not await inventory.update()
 131
 132     @pytest.mark.asyncio
 133     async def test_update_getconfig_exception(self, inventory):
 134         inventory.client.raise_on_getconfig = True
 135         await inventory.get_policy_content()
 136
 137     @pytest.mark.asyncio
 138     async def test_gather(self, inventory):
 139         await inventory.gather()
 140
 141         # We should gather one policy
 142         assert len(inventory.hp_active_inventory) == 1
 143
 144         # type in event should be gather
 145         with open(inventory.file) as f:
 146             data = json.load(f)
 147
 148         assert data["event"]["action"] == ACTION_GATHERED
 149
 150     @pytest.mark.asyncio
 151     async def test_ws_text(self, inventory):
 152         result = await inventory.check_and_update()
 153         assert result == True