dcae-policy/tests/test_tasks.py

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