dcae-policy/tests/test_discovery.py

   1 # ================================================================================
   2 # Copyright (c) 2019-2020 AT&T Intellectual Property. All rights reserved.
   3 # Copyright (c) 2019 Pantheon.tech. All rights reserved.
   4 # ================================================================================
   5 # Licensed under the Apache License, Version 2.0 (the "License");
   6 # you may not use this file except in compliance with the License.
   7 # You may obtain a copy of the License at
   8 #
   9 #      http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 # Unless required by applicable law or agreed to in writing, software
  12 # distributed under the License is distributed on an "AS IS" BASIS,
  13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16 # ============LICENSE_END=========================================================
  17 #
  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, **kwargs):
  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, **kwargs):
  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         ).encode()).decode()
  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, **kwargs):
 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()