diff options
Diffstat (limited to 'dmaap/tests')
| -rw-r--r-- | dmaap/tests/conftest.py | 82 | ||||
| -rw-r--r-- | dmaap/tests/test_consulif.py | 120 | ||||
| -rw-r--r-- | dmaap/tests/test_dmaapcontrollerif.py | 126 | ||||
| -rw-r--r-- | dmaap/tests/test_dr_lifecycle.py | 72 | ||||
| -rw-r--r-- | dmaap/tests/test_mr_lifecycle.py | 64 | ||||
| -rw-r--r-- | dmaap/tests/test_utils.py | 27 |
6 files changed, 491 insertions, 0 deletions
diff --git a/dmaap/tests/conftest.py b/dmaap/tests/conftest.py new file mode 100644 index 0000000..240a1af --- /dev/null +++ b/dmaap/tests/conftest.py @@ -0,0 +1,82 @@ +# ============LICENSE_START======================================================= +# org.onap.dcae +# ================================================================================ +# 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. + +import pytest + +import requests + +@pytest.fixture() +def mockconsul(monkeypatch): + """ Override the regular Consul interface""" + def fake_get_config(self, key): + config={'dmaap': {'username': 'm06814@controller.dcae.ecomp.att.com', + 'url': 'https://dcae-dbcl.d54852.dyh1b.tci.att.com:8443/webapi', + 'password' : 'Dca3wr3x', + 'owner': 'dcaeorch' }} + return config + + def fake_get_service(self, service_name): + service_address = "myAddress" + service_port= "8443" + return service_address, service_port + + def fake_add_to_entry(self, key, add_name, add_value): + return True + + def fake_delete_entry(self, entry_name): + return True + + def fake_init(self, api_url, user, password, logger): + pass + + from consulif.consulif import ConsulHandle + monkeypatch.setattr(ConsulHandle, 'get_config', fake_get_config) + monkeypatch.setattr(ConsulHandle, 'get_service', fake_get_service) + monkeypatch.setattr(ConsulHandle, 'add_to_entry', fake_add_to_entry) + monkeypatch.setattr(ConsulHandle, 'delete_entry', fake_delete_entry) + monkeypatch.setattr(ConsulHandle, '__init__', fake_init) + +@pytest.fixture() +def mockdmaapbc(monkeypatch): + + def fake_get(url, auth): + # print "fake_get: {0}, {1}".format(url, auth) + r = requests.Response() + r.status_code = 200 + return r + def fake_post(url, auth, json): + # print "fake_post: {0}, {1}, {2}".format(url, auth, json) + r = requests.Response() + r.status_code = 200 + return r + def fake_delete(url, auth): + # print "fake_delete: {0}, {1}".format(url, auth) + r = requests.Response() + r.status_code = 200 + return r + def fake_json(self): + return {"fqtn":"test_fqtn"} + + import requests + monkeypatch.setattr(requests.Response, "json", fake_json) + monkeypatch.setattr(requests, "get", fake_get) + monkeypatch.setattr(requests, "post", fake_post) + monkeypatch.setattr(requests, "delete", fake_delete) + diff --git a/dmaap/tests/test_consulif.py b/dmaap/tests/test_consulif.py new file mode 100644 index 0000000..2d26576 --- /dev/null +++ b/dmaap/tests/test_consulif.py @@ -0,0 +1,120 @@ +# ============LICENSE_START======================================================= +# org.onap.dcae +# ================================================================================ +# Copyright (c) 2017-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. + + +import pytest +from cloudify.exceptions import NonRecoverableError +import os +from consulif.consulif import ConsulHandle + + +#When run unit test, please update the consul interface parameters based on local consul configuration. +CONSUL_HOST = "consul" # Should always be a local consul agent on Cloudify Manager +#CONSUL_PORT = '8510' +CONSUL_PORT = '8500' +DBCL_KEY_NAME = "dmaap_dbcl_info" # Consul key containing DMaaP data bus credentials +DBC_SERVICE_NAME= "dmaap_bus_controller" # Name under which the DMaaP bus controller is registered + + +def _fake_get_config(self, key): + fix ={'dmaap': {'username': 'm06814@controller.dcae.ecomp.att.com', + 'url': 'https://dcae-dbcl.d54852.dyh1b.tci.att.com:8443/webapi', + 'password' : 'Dca3wr3x', + 'owner': 'dcaeorch' }} + return fix + +def _fake_service(self, key): + service_address = "myAddress" + service_port= "8443" + return service_address, service_port + +def test_get_config_service(monkeypatch): + monkeypatch.setattr(ConsulHandle, 'get_config', _fake_get_config) + monkeypatch.setattr(ConsulHandle, 'get_service', _fake_service) + try: + err_msg = "Error getting ConsulHandle when configuring dmaap plugin: {0}" + _ch = ConsulHandle("http://{0}:{1}".format(CONSUL_HOST, CONSUL_PORT), None, None, None) + assert None != _ch + + err_msg = "Error getting config for '{0}' from ConsulHandle when configuring dmaap plugin: ".format(DBCL_KEY_NAME) + "{0}" + config = _ch.get_config(DBCL_KEY_NAME) + + err_msg = "Error setting DMAAP_USER while configuring dmaap plugin: {0}" + DMAAP_USER = config['dmaap']['username'] + + err_msg = "Error setting DMAAP_PASS while configuring dmaap plugin: {0}" + DMAAP_PASS = config['dmaap']['password'] + + err_msg = "Error setting DMAAP_OWNER while configuring dmaap plugin: {0}" + DMAAP_OWNER = config['dmaap']['owner'] + + err_msg = "Error setting DMAAP_PROTOCOL while configuring dmaap plugin: {0}" + if 'protocol' in config['dmaap']: + DMAAP_PROTOCOL = config['dmaap']['protocol'] + else: + DMAAP_PROTOCOL = 'https' # Default to https (service discovery should give us this but doesn't + + err_msg = "Error setting DMAAP_PATH while configuring dmaap plugin: {0}" + if 'path' in config['dmaap']: + DMAAP_PATH = config['dmaap']['path'] + else: + DMAAP_PATH = 'webapi' # SHould come from service discovery but Consul doesn't support it + + err_msg = "Error getting service_address and service_port for '{0}' from ConsulHandle when configuring dmaap plugin: ".format(DBC_SERVICE_NAME) + "{0}" + service_address, service_port = _ch.get_service(DBC_SERVICE_NAME) + + err_msg = "Error setting DMAAP_API_URL while configuring dmaap plugin: {0}" + DMAAP_API_URL = '{0}://{1}:{2}/{3}'.format(DMAAP_PROTOCOL, service_address, service_port, DMAAP_PATH) + assert DMAAP_API_URL != None + dmaap_config = {'DMAAP_USER':DMAAP_USER, 'DMAAP_API_URL':DMAAP_API_URL, 'DMAAP_PASS':DMAAP_PASS, 'DMAAP_OWNER':DMAAP_OWNER} + print "return dmaap config info from consul: {0}".format(dmaap_config) + return dmaap_config + + except Exception as e: + raise NonRecoverableError(err_msg.format(e)) + +def _fake_add_to_entry(self, key, add_name, add_value): + return True + +def _fake_delete_entry(self, entry_name): + return True + +def test_add_entry(monkeypatch): + monkeypatch.setattr(ConsulHandle, 'get_config', _fake_get_config) + monkeypatch.setattr(ConsulHandle, 'add_to_entry', _fake_add_to_entry) + monkeypatch.setattr(ConsulHandle, 'delete_entry', _fake_delete_entry) + try: + _ch = ConsulHandle("http://{0}:{1}".format(CONSUL_HOST, CONSUL_PORT), None, None, None) + assert None != _ch + + key = 'DMAAP_TEST' + name = 'dmaap_test_name' + value = 'dmaap_test_value' + _ch.add_to_entry(key, name, value) + + name = "dmaap_test_name_2" + value = 'dmaap_test_value_2' + _ch.add_to_entry(key, name, value) + + _ch.delete_entry(key) + + except Exception as e: + raise NonRecoverableError("Error in test_add_entry: {0}".format(e)) + diff --git a/dmaap/tests/test_dmaapcontrollerif.py b/dmaap/tests/test_dmaapcontrollerif.py new file mode 100644 index 0000000..f647530 --- /dev/null +++ b/dmaap/tests/test_dmaapcontrollerif.py @@ -0,0 +1,126 @@ +# ============LICENSE_START======================================================= +# org.onap.dcae +# ================================================================================ +# Copyright (c) 2017-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. + +import pytest +import requests +from cloudify.mocks import MockCloudifyContext +from cloudify.state import current_ctx +from cloudify import ctx +from cloudify.decorators import operation +from cloudify.exceptions import NonRecoverableError + + +import test_consulif +from dmaapcontrollerif.dmaap_requests import DMaaPControllerHandle + +import logging +logger = logging.getLogger("test_mr_lifecycle") + +_goodosv2 = { + 'auth_url': 'https://example.com/identity/v2.0', + 'password': 'pw', + 'region': 'r', + 'tenant_name': 'tn', + 'username': 'un' +} + + +def test_dmaapc (monkeypatch, mockconsul, mockdmaapbc): + from dmaapplugin.dmaaputils import random_string + + config = test_consulif.test_get_config_service(monkeypatch) + DMAAP_API_URL = config['DMAAP_API_URL'] + DMAAP_USER = config['DMAAP_USER'] + DMAAP_PASS = config['DMAAP_PASS'] + DMAAP_OWNER = config['DMAAP_OWNER'] + assert DMAAP_API_URL != None + + properties = {'fqdn': 'a.x.example.com', 'openstack': _goodosv2 } + mock_ctx = MockCloudifyContext(node_id='test_node_id', node_name='test_node_name', properties=properties, + runtime_properties = { + "admin": { "user": "admin_user" }, + "user": { "user": "user_user" }, + "viewer": { "user": "viewer_user" } + } + ) + + try: + current_ctx.set(mock_ctx) + except Exception as e: + raise NonRecoverableError(e) +# finally: +# current_ctx.clear() + + kwargs = { "topic_name": "ONAP_test", + "topic_description": "onap dmaap plugin unit test topic"} + + try: + # Make sure there's a topic_name + if "topic_name" in ctx.node.properties: + topic_name = ctx.node.properties["topic_name"] + if topic_name == '' or topic_name.isspace(): + topic_name = random_string(12) + else: + topic_name = random_string(12) + + # Make sure there's a topic description + if "topic_description" in ctx.node.properties: + topic_description = ctx.node.properties["topic_description"] + else: + topic_description = "No description provided" + + # ..and the truly optional setting + if "txenable" in ctx.node.properties: + txenable = ctx.node.properties["txenable"] + else: + txenable= False + + if "replication_case" in ctx.node.properties: + replication_case = ctx.node.properties["replication_case"] + else: + replication_case = None + + if "global_mr_url" in ctx.node.properties: + global_mr_url = ctx.node.properties["global_mr_url"] + else: + global_mr_url = None + + dmc = DMaaPControllerHandle(DMAAP_API_URL, DMAAP_USER, DMAAP_PASS, ctx.logger) + ctx.logger.info("Attempting to create topic name {0}".format(topic_name)) + t = dmc.create_topic(topic_name, topic_description, txenable, DMAAP_OWNER, replication_case, global_mr_url) + + # Capture important properties from the result + topic = t.json() + ctx.instance.runtime_properties["fqtn"] = topic["fqtn"] + + except Exception as e: + raise NonRecoverableError(e) + + # test DMaaPControllerHandle functions + try: + path = "myPath" + url = dmc._make_url(path) + rc = dmc._get_resource(path) + rc = dmc._create_resource(path, None) + rc = dmc._delete_resource(path) + + except Exception as e: + raise NonRecoverableError(e) + diff --git a/dmaap/tests/test_dr_lifecycle.py b/dmaap/tests/test_dr_lifecycle.py new file mode 100644 index 0000000..c6b6d4b --- /dev/null +++ b/dmaap/tests/test_dr_lifecycle.py @@ -0,0 +1,72 @@ +# ============LICENSE_START======================================================= +# org.onap.dcae +# ================================================================================ +# Copyright (c) 2017-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. + +import pytest +import requests +from cloudify.mocks import MockCloudifyContext +from cloudify.state import current_ctx +from cloudify import ctx +from cloudify.decorators import operation +from cloudify.exceptions import NonRecoverableError +from cloudify.exceptions import RecoverableError + +_goodosv2 = { + 'auth_url': 'https://example.com/identity/v2.0', + 'password': 'pw', + 'region': 'r', + 'tenant_name': 'tn', + 'username': 'un' +} + + +def test_create_feed(monkeypatch, mockdmaapbc): + import dmaapplugin + from dmaapplugin import dr_lifecycle + + properties = {'fqdn': 'a.x.example.com', 'openstack': _goodosv2, 'feed_id': 'test_feed_id' } + mock_ctx = MockCloudifyContext(node_id='test_node_id', node_name='test_node_name', properties=properties, + runtime_properties = { + "admin": { "user": "admin_user" }, + "user": { "user": "user_user" }, + "viewer": { "user": "viewer_user" } + } + ) + + try: + current_ctx.set(mock_ctx) + except Exception as e: + raise NonRecoverableError(e) +# finally: +# current_ctx.clear() + + kwargs = { "feed_name": "ONAP_test", + "feed_description": "onap dmaap plugin unit test feed"} + + def fake_feed(self): + return {"feedId":"test_feedId", "publishURL":"test_publishURL", "logURL":"test_logURL" } + monkeypatch.setattr(requests.Response, "json", fake_feed) + + try: + dr_lifecycle.create_feed(**kwargs) + dr_lifecycle.get_existing_feed(**kwargs) + dr_lifecycle.delete_feed(**kwargs) + + except Exception as e: + raise NonRecoverableError(e) diff --git a/dmaap/tests/test_mr_lifecycle.py b/dmaap/tests/test_mr_lifecycle.py new file mode 100644 index 0000000..98a9ed5 --- /dev/null +++ b/dmaap/tests/test_mr_lifecycle.py @@ -0,0 +1,64 @@ +# ============LICENSE_START======================================================= +# org.onap.dcae +# ================================================================================ +# Copyright (c) 2017-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. + +import pytest +import requests +from cloudify.mocks import MockCloudifyContext +from cloudify.state import current_ctx +from cloudify import ctx +from cloudify.decorators import operation +from cloudify.exceptions import NonRecoverableError +from cloudify.exceptions import RecoverableError + +_goodosv2 = { + 'auth_url': 'https://example.com/identity/v2.0', + 'password': 'pw', + 'region': 'r', + 'tenant_name': 'tn', + 'username': 'un' +} + +def test_create_topic(monkeypatch, mockdmaapbc): + import dmaapplugin + from dmaapplugin import mr_lifecycle + properties = {'fqdn': 'a.x.example.com', 'openstack': _goodosv2, 'fqtn': 'test_fqtn' } + mock_ctx = MockCloudifyContext(node_id='test_node_id', node_name='test_node_name', properties=properties, + runtime_properties = { + "admin": { "user": "admin_user" }, + "user": { "user": "user_user" }, + "viewer": { "user": "viewer_user" } + } + ) + + try: + current_ctx.set(mock_ctx) + except Exception as e: + raise NonRecoverableError(e) + + kwargs = { "topic_name": "ONAP_test", + "topic_description": "onap dmaap plugin unit test topic"} + + try: + mr_lifecycle.create_topic(**kwargs) + mr_lifecycle.get_existing_topic(**kwargs) + mr_lifecycle.delete_topic(**kwargs) + + except Exception as e: + raise NonRecoverableError(e) diff --git a/dmaap/tests/test_utils.py b/dmaap/tests/test_utils.py new file mode 100644 index 0000000..6ec6ce8 --- /dev/null +++ b/dmaap/tests/test_utils.py @@ -0,0 +1,27 @@ +# ============LICENSE_START======================================================= +# org.onap.dcae +# ================================================================================ +# Copyright (c) 2017-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. + +import pytest + + +def test_random_string(monkeypatch): + from dmaapplugin import dmaaputils + target_length = 10 + assert len(dmaaputils.random_string(target_length)) == target_length |