import requests
+from cloudify import ctx
+
# it is safe to assume that consul agent is at localhost:8500 along with cloudify manager
CONSUL_SERVICE_URL = "http://localhost:8500/v1/catalog/service/{0}"
def discover_service_url(service_name):
"""find the service record in consul"""
- response = requests.get(CONSUL_SERVICE_URL.format(service_name))
- response.raise_for_status()
- resp_json = response.json()
- if resp_json:
- service = resp_json[0]
- return "http://{0}:{1}".format(service["ServiceAddress"], service["ServicePort"])
+ service_url_url = CONSUL_SERVICE_URL.format(service_name)
+ ctx.logger.info("getting service_url at {0}".format(service_url_url))
+
+ response = requests.get(service_url_url)
+
+ ctx.logger.info("got service_url at {0} status({1}) response: {2}"
+ .format(service_url_url, response.status_code, response.text))
+
+ if response.status_code == requests.codes.ok:
+ resp_json = response.json()
+ if resp_json:
+ service = resp_json[0]
+ return "http://{0}:{1}".format(service["ServiceAddress"], service["ServicePort"])
SERVICE_NAME_POLICY_HANDLER = "policy_handler"
X_ECOMP_REQUESTID = 'X-ECOMP-RequestID'
STATUS_CODE_POLICIES_NOT_FOUND = 404
+ DEFAULT_URL = "http://policy-handler"
_url = None
@staticmethod
return
PolicyHandler._url = discover_service_url(PolicyHandler.SERVICE_NAME_POLICY_HANDLER)
+ if not PolicyHandler._url:
+ PolicyHandler._url = PolicyHandler.DEFAULT_URL
@staticmethod
def get_latest_policy(policy_id):
ph_path = "{0}/policy_latest/{1}".format(PolicyHandler._url, policy_id)
headers = {PolicyHandler.X_ECOMP_REQUESTID: str(uuid.uuid4())}
- ctx.logger.info("getting latest policy from {0} headers={1}".format( \
+ ctx.logger.info("getting latest policy from {0} headers={1}".format(
ph_path, json.dumps(headers)))
res = requests.get(ph_path, headers=headers)
ctx.logger.info("latest policy for policy_id({0}) status({1}) response: {2}"
def test_discovery_failure(monkeypatch):
"""test finding policy-handler in consul"""
monkeypatch.setattr('requests.get', monkeyed_discovery_get_failure)
- expected = None
- tasks.PolicyHandler._lazy_init()
- assert expected == tasks.PolicyHandler._url
+
+ 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 tasks.PolicyHandler.DEFAULT_URL == 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"""
def test_discovery(monkeypatch):
"""test finding policy-handler in consul"""
monkeypatch.setattr('requests.get', monkeyed_discovery_get)
- expected = "http://monkey-policy-handler-address:9999"
- tasks.PolicyHandler._lazy_init()
- assert expected == tasks.PolicyHandler._url
-def monkeyed_policy_handler_get(full_path, headers):
+ 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()
+
+
+def monkeyed_policy_handler_get(full_path, headers=None):
"""monkeypatch for the GET to policy-engine"""
return MonkeyedResponse(full_path, headers, \
MonkeyedPolicyBody.create_policy(MONKEYED_POLICY_ID))
def test_policy_get(monkeypatch):
"""test policy_get operation on dcae.nodes.policy node"""
+ tasks.PolicyHandler._url = tasks.PolicyHandler.DEFAULT_URL
monkeypatch.setattr('requests.get', monkeyed_policy_handler_get)
node_policy = MonkeyedNode(
def test_policies_find(monkeypatch):
"""test policy_get operation on dcae.nodes.policies node"""
+ tasks.PolicyHandler._url = tasks.PolicyHandler.DEFAULT_URL
monkeypatch.setattr('requests.post', monkeyed_policy_handler_find)
node_policies = MonkeyedNode(