1 # ================================================================================
2 # Copyright (c) 2019 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
8 # http://www.apache.org/licenses/LICENSE-2.0
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=========================================================
17 # ECOMP is a trademark and service mark of AT&T Intellectual Property.
19 """unit tests for discovery in dcaepolicyplugin"""
26 from cloudify.exceptions import NonRecoverableError
27 from cloudify.state import current_ctx
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,
35 POLICY_HANDLER_FROM_KV = "http://policy_handler_from_kv:25577"
38 def monkeyed_discovery_get_failure(full_path, **kwargs):
39 """monkeypatch for the GET to consul"""
40 raise requests.ConnectionError("monkey-boom")
43 def test_discovery_failure(monkeypatch):
44 """test finding policy-handler in consul"""
45 monkeypatch.setattr('requests.get', monkeyed_discovery_get_failure)
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}
54 current_ctx.set(node_policy.ctx)
55 with pytest.raises(NonRecoverableError) as excinfo:
56 tasks.PolicyHandler._lazy_init()
58 CtxLogger.log_ctx_info("test_discovery_failure: {0}".format(str(excinfo.value)))
59 assert str(excinfo.value).startswith("ConnectionError")
63 tasks.PolicyHandler._url = None
64 MockCloudifyContextFull.clear()
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)
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}}}
79 return MonkeyedResponse(full_path, {}, [{"Value": value}])
81 return MonkeyedResponse(full_path)
84 def test_discovery_kv(monkeypatch):
85 """test finding policy-handler in consul"""
86 monkeypatch.setattr('requests.get', monkeyed_discovery_get_kv)
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}
95 current_ctx.set(node_policy.ctx)
96 tasks.PolicyHandler._lazy_init()
97 assert POLICY_HANDLER_FROM_KV == tasks.PolicyHandler._url
100 tasks.PolicyHandler._url = None
101 MockCloudifyContextFull.clear()
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"}])
111 def test_discovery(monkeypatch):
112 """test finding policy-handler in consul"""
113 monkeypatch.setattr('requests.get', monkeyed_discovery_get)
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}
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
131 tasks.PolicyHandler._url = None
132 MockCloudifyContextFull.clear()