dcae-policy/tests/test_discovery.py

   1 # ================================================================================
   2 # Copyright (c) 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 discovery in dcaepolicyplugin"""
  20
  21 import base64
  22 import json
  23
  24 import pytest
  25 import requests
  26 from cloudify.exceptions import NonRecoverableError
  27 from cloudify.state import current_ctx
  28
  29 from dcaepolicyplugin import discovery, tasks
  30 from tests.log_ctx import CtxLogger
  31 from tests.mock_cloudify_ctx import MockCloudifyContextFull
  32 from tests.mock_setup import (MONKEYED_POLICY_ID, POLICY_ID, MonkeyedNode,
  33                               MonkeyedResponse)
  34
  35 POLICY_HANDLER_FROM_KV = "http://policy_handler_from_kv:25577"
  36
  37
  38 def monkeyed_discovery_get_failure(full_path):
  39     """monkeypatch for the GET to consul"""
  40     raise requests.ConnectionError("monkey-boom")
  41
  42
  43 def test_discovery_failure(monkeypatch):
  44     """test finding policy-handler in consul"""
  45     monkeypatch.setattr('requests.get', monkeyed_discovery_get_failure)
  46
  47     node_policy = MonkeyedNode(
  48         'test_dcae_policy_node_id',
  49         'test_dcae_policy_node_name',
  50         tasks.DCAE_POLICY_TYPE,
  51         {POLICY_ID: MONKEYED_POLICY_ID}
  52     )
  53     try:
  54         current_ctx.set(node_policy.ctx)
  55         with pytest.raises(NonRecoverableError) as excinfo:
  56             tasks.PolicyHandler._lazy_init()
  57
  58         CtxLogger.log_ctx_info("test_discovery_failure: {0}".format(str(excinfo.value)))
  59         assert str(excinfo.value).startswith("ConnectionError")
  60
  61
  62     finally:
  63         tasks.PolicyHandler._url = None
  64         MockCloudifyContextFull.clear()
  65         current_ctx.clear()
  66
  67
  68 def monkeyed_discovery_get_kv(full_path):
  69     """monkeypatch for the GET to consul"""
  70     if full_path.startswith(discovery.CONSUL_SERVICE_URL.format("")):
  71         return MonkeyedResponse(full_path)
  72
  73     if full_path.startswith(discovery.CONSUL_KV_MASK.format("")):
  74         value = base64.b64encode(json.dumps(
  75             {tasks.DCAE_POLICY_PLUGIN: {
  76                 tasks.PolicyHandler.SERVICE_NAME_POLICY_HANDLER: {
  77                     "url": POLICY_HANDLER_FROM_KV}}}
  78         ))
  79         return MonkeyedResponse(full_path, {}, [{"Value": value}])
  80
  81     return MonkeyedResponse(full_path)
  82
  83
  84 def test_discovery_kv(monkeypatch):
  85     """test finding policy-handler in consul"""
  86     monkeypatch.setattr('requests.get', monkeyed_discovery_get_kv)
  87
  88     node_policy = MonkeyedNode(
  89         'test_dcae_policy_node_id',
  90         'test_dcae_policy_node_name',
  91         tasks.DCAE_POLICY_TYPE,
  92         {POLICY_ID: MONKEYED_POLICY_ID}
  93     )
  94     try:
  95         current_ctx.set(node_policy.ctx)
  96         tasks.PolicyHandler._lazy_init()
  97         assert POLICY_HANDLER_FROM_KV == tasks.PolicyHandler._url
  98
  99     finally:
 100         tasks.PolicyHandler._url = None
 101         MockCloudifyContextFull.clear()
 102         current_ctx.clear()
 103
 104
 105 def monkeyed_discovery_get(full_path):
 106     """monkeypatch for the GET to consul"""
 107     return MonkeyedResponse(full_path, {},
 108         [{"ServiceAddress": "monkey-policy-handler-address", "ServicePort": "9999"}])
 109
 110
 111 def test_discovery(monkeypatch):
 112     """test finding policy-handler in consul"""
 113     monkeypatch.setattr('requests.get', monkeyed_discovery_get)
 114
 115     node_policy = MonkeyedNode(
 116         'test_dcae_policy_node_id',
 117         'test_dcae_policy_node_name',
 118         tasks.DCAE_POLICY_TYPE,
 119         {POLICY_ID: MONKEYED_POLICY_ID}
 120     )
 121
 122     try:
 123         current_ctx.set(node_policy.ctx)
 124         expected = "http://monkey-policy-handler-address:9999"
 125         CtxLogger.log_ctx_info("before PolicyHandler._lazy_init")
 126         tasks.PolicyHandler._lazy_init()
 127         CtxLogger.log_ctx_info("after PolicyHandler._lazy_init")
 128         assert expected == tasks.PolicyHandler._url
 129
 130     finally:
 131         tasks.PolicyHandler._url = None
 132         MockCloudifyContextFull.clear()
 133         current_ctx.clear()