1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
|
# ================================================================================
# Copyright (c) 2018 AT&T Intellectual Property. All rights reserved.
# ================================================================================
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ============LICENSE_END=========================================================
#
# ECOMP is a trademark and service mark of AT&T Intellectual Property.
"""unit tests for discovery in dcaepolicyplugin"""
import base64
import json
import pytest
import requests
from cloudify.exceptions import NonRecoverableError
from cloudify.state import current_ctx
from dcaepolicyplugin import discovery, tasks
from tests.log_ctx import CtxLogger
from tests.mock_cloudify_ctx import MockCloudifyContextFull
from tests.mock_setup import (MONKEYED_POLICY_ID, POLICY_ID, MonkeyedNode,
MonkeyedResponse)
POLICY_HANDLER_FROM_KV = "http://policy_handler_from_kv:25577"
def monkeyed_discovery_get_failure(full_path):
"""monkeypatch for the GET to consul"""
raise requests.ConnectionError("monkey-boom")
def test_discovery_failure(monkeypatch):
"""test finding policy-handler in consul"""
monkeypatch.setattr('requests.get', monkeyed_discovery_get_failure)
node_policy = MonkeyedNode(
'test_dcae_policy_node_id',
'test_dcae_policy_node_name',
tasks.DCAE_POLICY_TYPE,
{POLICY_ID: MONKEYED_POLICY_ID}
)
try:
current_ctx.set(node_policy.ctx)
with pytest.raises(NonRecoverableError) as excinfo:
tasks.PolicyHandler._lazy_init()
CtxLogger.log_ctx_info("test_discovery_failure: {0}".format(str(excinfo.value)))
assert str(excinfo.value).startswith("ConnectionError")
finally:
tasks.PolicyHandler._url = None
MockCloudifyContextFull.clear()
current_ctx.clear()
def monkeyed_discovery_get_kv(full_path):
"""monkeypatch for the GET to consul"""
if full_path.startswith(discovery.CONSUL_SERVICE_URL.format("")):
return MonkeyedResponse(full_path)
if full_path.startswith(discovery.CONSUL_KV_MASK.format("")):
value = base64.b64encode(json.dumps(
{tasks.DCAE_POLICY_PLUGIN: {
tasks.PolicyHandler.SERVICE_NAME_POLICY_HANDLER: {
"url": POLICY_HANDLER_FROM_KV}}}
))
return MonkeyedResponse(full_path, {}, [{"Value": value}])
return MonkeyedResponse(full_path)
def test_discovery_kv(monkeypatch):
"""test finding policy-handler in consul"""
monkeypatch.setattr('requests.get', monkeyed_discovery_get_kv)
node_policy = MonkeyedNode(
'test_dcae_policy_node_id',
'test_dcae_policy_node_name',
tasks.DCAE_POLICY_TYPE,
{POLICY_ID: MONKEYED_POLICY_ID}
)
try:
current_ctx.set(node_policy.ctx)
tasks.PolicyHandler._lazy_init()
assert POLICY_HANDLER_FROM_KV == tasks.PolicyHandler._url
finally:
tasks.PolicyHandler._url = None
MockCloudifyContextFull.clear()
current_ctx.clear()
def monkeyed_discovery_get(full_path):
"""monkeypatch for the GET to consul"""
return MonkeyedResponse(full_path, {},
[{"ServiceAddress": "monkey-policy-handler-address", "ServicePort": "9999"}])
def test_discovery(monkeypatch):
"""test finding policy-handler in consul"""
monkeypatch.setattr('requests.get', monkeyed_discovery_get)
node_policy = MonkeyedNode(
'test_dcae_policy_node_id',
'test_dcae_policy_node_name',
tasks.DCAE_POLICY_TYPE,
{POLICY_ID: MONKEYED_POLICY_ID}
)
try:
current_ctx.set(node_policy.ctx)
expected = "http://monkey-policy-handler-address:9999"
CtxLogger.log_ctx_info("before PolicyHandler._lazy_init")
tasks.PolicyHandler._lazy_init()
CtxLogger.log_ctx_info("after PolicyHandler._lazy_init")
assert expected == tasks.PolicyHandler._url
finally:
tasks.PolicyHandler._url = None
MockCloudifyContextFull.clear()
current_ctx.clear()
|
