diff options
Diffstat (limited to 'cdap/cdapplugin/tests/test_cdap_plugin.py')
-rw-r--r-- | cdap/cdapplugin/tests/test_cdap_plugin.py | 103 |
1 files changed, 0 insertions, 103 deletions
diff --git a/cdap/cdapplugin/tests/test_cdap_plugin.py b/cdap/cdapplugin/tests/test_cdap_plugin.py deleted file mode 100644 index f28485d..0000000 --- a/cdap/cdapplugin/tests/test_cdap_plugin.py +++ /dev/null @@ -1,103 +0,0 @@ -# org.onap.dcae -# ================================================================================ -# Copyright (c) 2017 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. - -from cdapcloudify.cdap_plugin import _validate_conns, BadConnections, try_raise_nonr -import pytest -from cloudify.exceptions import NonRecoverableError - -#todo.. add more tests.. #shame - -def _get_good_connection(): - connections = {} - connections["streams_publishes"] = [ - {"name" : "test_n", - "location" : "test_l", - "client_role" : "test_cr", - "type" : "message_router", - "config_key" : "test_c", - "aaf_username": "test_u", - "aaf_password": "test_p" - }, - {"name" : "test_n2", - "location" : "test_l", - "client_role" : "test_cr", - "type" : "message_router", - "config_key" : "test_c", - "aaf_username": "test_u", - "aaf_password": "test_p" - }, - {"name" : "test_feed00", - "location" : "test_l", - "type" : "data_router", - "config_key" : "mydrconfigkey" - } - ] - connections["streams_subscribes"] = [ - {"name" : "test_n", - "location" : "test_l", - "client_role" : "test_cr", - "type" : "message_router", - "config_key" : "test_c", - "aaf_username": "test_u", - "aaf_password": "test_p" - }, - {"name" : "test_n2", - "location" : "test_l", - "client_role" : "test_cr", - "type" : "message_router", - "config_key" : "test_c", - "aaf_username": "test_u", - "aaf_password": "test_p" - } - ] - return connections - -def test_validate_cons(): - #test good streams - good_conn = _get_good_connection() - _validate_conns(good_conn) - - #mutate - nosub = _get_good_connection().pop("streams_subscribes") - with pytest.raises(BadConnections) as excinfo: - _validate_conns(nosub) - - nopub = _get_good_connection().pop("streams_publishes") - with pytest.raises(BadConnections) as excinfo: - _validate_conns(nopub) - - noloc = _get_good_connection()["streams_publishes"][0].pop("location") - with pytest.raises(BadConnections) as excinfo: - _validate_conns(noloc) - -def test_nonr_dec(): - def blow(): - d = {} - d["emptyinside"] += 1 - return d - #apply decorator - blow = try_raise_nonr(blow) - with pytest.raises(NonRecoverableError): - blow() - - def work(): - return 666 - work = try_raise_nonr(work) - assert work() == 666 - |