dcae-policy/tests/test_discovery.py

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