dcae-policy/tests/test_tasks.py

   1 # ================================================================================
   2 # Copyright (c) 2017-2020 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
  18 """unit tests for tasks in dcaepolicyplugin"""
  19
  20 import json
  21
  22 import pytest
  23 from cloudify.exceptions import NonRecoverableError
  24 from cloudify.state import current_ctx
  25
  26 from dcaepolicyplugin import tasks
  27 from tests.log_ctx import CtxLogger
  28 from tests.mock_cloudify_ctx import (TARGET_NODE_ID, TARGET_NODE_NAME,
  29                                      MockCloudifyContextFull)
  30 from tests.mock_setup import (CONFIG_NAME, MONKEYED_POLICY_ID, POLICY_BODY,
  31                               POLICY_ID, POLICY_NAME, MonkeyedNode,
  32                               MonkeyedPolicyBody, MonkeyedResponse)
  33
  34
  35 LATEST_POLICIES = "latest_policies"
  36
  37
  38 def monkeyed_policy_handler_get(full_path, headers=None, **kwargs):
  39     """monkeypatch for the GET to policy-engine"""
  40     return MonkeyedResponse(full_path, headers,
  41         MonkeyedPolicyBody.create_policy(MONKEYED_POLICY_ID))
  42
  43
  44 def test_policy_get(monkeypatch):
  45     """test policy_get operation on dcae.nodes.policy node"""
  46     tasks.PolicyHandler._url = tasks.PolicyHandler.DEFAULT_URL
  47     monkeypatch.setattr('requests.get', monkeyed_policy_handler_get)
  48
  49     node_policy = MonkeyedNode(
  50         'test_dcae_policy_node_id',
  51         'test_dcae_policy_node_name',
  52         tasks.DCAE_POLICY_TYPE,
  53         {POLICY_ID: MONKEYED_POLICY_ID}
  54     )
  55
  56     try:
  57         current_ctx.set(node_policy.ctx)
  58         CtxLogger.log_ctx_info("before policy_get")
  59         tasks.policy_get()
  60         CtxLogger.log_ctx_info("after policy_get")
  61
  62         expected = {
  63             POLICY_BODY: MonkeyedPolicyBody.create_policy_body(MONKEYED_POLICY_ID)
  64         }
  65         result = node_policy.ctx.instance.runtime_properties
  66         node_policy.ctx.logger.info("expected runtime_properties: {0}".format(
  67             json.dumps(expected)))
  68         node_policy.ctx.logger.info("runtime_properties: {0}".format(json.dumps(result)))
  69         assert MonkeyedPolicyBody.is_the_same_dict(result, expected)
  70         assert MonkeyedPolicyBody.is_the_same_dict(expected, result)
  71
  72     finally:
  73         MockCloudifyContextFull.clear()
  74         current_ctx.clear()
  75
  76
  77 def test_policy_get_fail(monkeypatch):
  78     """test policy_get operation on non dcae.nodes.policy node"""
  79     tasks.PolicyHandler._url = tasks.PolicyHandler.DEFAULT_URL
  80     monkeypatch.setattr('requests.get', monkeyed_policy_handler_get)
  81
  82     node_policy = MonkeyedNode(
  83         'test_dcae_policy_node_id',
  84         'test_dcae_policy_node_name',
  85         tasks.DCAE_POLICY_TYPE,
  86         {POLICY_ID: MONKEYED_POLICY_ID}
  87     )
  88
  89     node_ms = MonkeyedNode(
  90         'test_ms_id', 'test_ms_name', "ms.nodes.type", None,
  91         [{TARGET_NODE_ID: node_policy.node_id, TARGET_NODE_NAME: node_policy.node_name}]
  92     )
  93
  94     try:
  95         current_ctx.set(node_ms.ctx)
  96         CtxLogger.log_ctx_info("ctx of node_ms not policy type")
  97         with pytest.raises(NonRecoverableError) as excinfo:
  98             tasks.policy_get()
  99         CtxLogger.log_ctx_info("node_ms not policy type boom: {0}".format(str(excinfo.value)))
 100         assert "unexpected node type " in str(excinfo.value)
 101
 102     finally:
 103         MockCloudifyContextFull.clear()
 104         current_ctx.clear()
 105
 106
 107 def monkeyed_policy_handler_find(full_path, json, headers, **kwargs):
 108     """monkeypatch for the GET to policy-engine"""
 109     return MonkeyedResponse(
 110         full_path, headers,
 111         {LATEST_POLICIES: {
 112             MONKEYED_POLICY_ID: MonkeyedPolicyBody.create_policy(MONKEYED_POLICY_ID)}}
 113     )
 114
 115
 116 def test_policies_find(monkeypatch):
 117     """test policy_get operation on dcae.nodes.policies node"""
 118     tasks.PolicyHandler._url = tasks.PolicyHandler.DEFAULT_URL
 119     monkeypatch.setattr('requests.post', monkeyed_policy_handler_find)
 120
 121     node_policies = MonkeyedNode(
 122         'test_dcae_policies_node_id',
 123         'test_dcae_policies_node_name',
 124         tasks.DCAE_POLICIES_TYPE,
 125         {
 126             tasks.POLICY_FILTER: {
 127                 POLICY_NAME: MONKEYED_POLICY_ID,
 128                 tasks.CONFIG_ATTRIBUTES: json.dumps({
 129                     CONFIG_NAME: "alex_config_name"
 130                 })
 131             }
 132         }
 133     )
 134
 135     try:
 136         current_ctx.set(node_policies.ctx)
 137         CtxLogger.log_ctx_info("before policy_get")
 138         tasks.policy_get()
 139         CtxLogger.log_ctx_info("after policy_get")
 140
 141         expected = {
 142             tasks.POLICIES_FILTERED: {
 143                 MONKEYED_POLICY_ID: MonkeyedPolicyBody.create_policy(MONKEYED_POLICY_ID)}}
 144
 145         result = node_policies.ctx.instance.runtime_properties
 146         node_policies.ctx.logger.info("expected runtime_properties: {0}".format(
 147             json.dumps(expected)))
 148         node_policies.ctx.logger.info("runtime_properties: {0}".format(json.dumps(result)))
 149         assert MonkeyedPolicyBody.is_the_same_dict(result, expected)
 150         assert MonkeyedPolicyBody.is_the_same_dict(expected, result)
 151
 152     finally:
 153         MockCloudifyContextFull.clear()
 154         current_ctx.clear()
 155
 156
 157 def test_policies_find_fail(monkeypatch):
 158     """test policy_get operation on non dcae.nodes.policies node"""
 159     tasks.PolicyHandler._url = tasks.PolicyHandler.DEFAULT_URL
 160     monkeypatch.setattr('requests.post', monkeyed_policy_handler_find)
 161
 162     node_policies = MonkeyedNode(
 163         'test_dcae_policies_node_id',
 164         'test_dcae_policies_node_name',
 165         tasks.DCAE_POLICIES_TYPE,
 166         {
 167             tasks.POLICY_FILTER: {
 168                 POLICY_NAME: MONKEYED_POLICY_ID,
 169                 tasks.CONFIG_ATTRIBUTES: json.dumps({
 170                     CONFIG_NAME: "alex_config_name"
 171                 })
 172             }
 173         }
 174     )
 175     node_ms_multi = MonkeyedNode(
 176         'test_ms_multi_id', 'test_ms_multi_name', "ms.nodes.type",
 177         None,
 178         [{TARGET_NODE_ID: node_policies.node_id, TARGET_NODE_NAME: node_policies.node_name}]
 179     )
 180
 181     try:
 182         current_ctx.set(node_ms_multi.ctx)
 183         CtxLogger.log_ctx_info("ctx of node_ms_multi not policy type")
 184         with pytest.raises(NonRecoverableError) as excinfo:
 185             tasks.policy_get()
 186         CtxLogger.log_ctx_info("node_ms_multi not policy type boom: {0}".format(str(excinfo.value)))
 187         assert "unexpected node type " in str(excinfo.value)
 188
 189     finally:
 190         MockCloudifyContextFull.clear()
 191         current_ctx.clear()