--- /dev/null
+# ============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.
+
+from dockerplugin import decorators as dec
+
+
+def test_wrapper_merge_inputs():
+ properties = { "app_config": {"nested": { "a": 123, "b": 456 }, "foo": "duh"},
+ "image": "some-docker-image" }
+ kwargs = { "app_config": {"nested": {"a": 789, "c": "zyx"}} }
+
+ def task_func(**inputs):
+ return inputs
+
+ expected = { "app_config": {"nested": { "a": 789, "b": 456, "c": "zyx" },
+ "foo": "duh"}, "image": "some-docker-image" }
+
+ assert expected == dec._wrapper_merge_inputs(task_func, properties, **kwargs)
+
# ============LICENSE_START=======================================================
# org.onap.dcae
# ================================================================================
-# Copyright (c) 2017 AT&T Intellectual Property. All rights reserved.
+# 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.
wrapped_foo("a", "b", "c")
+def test_generate_service_component_name():
+ component_type = "some-component-type"
+ name = dis.generate_service_component_name(component_type)
+ assert name.split("_")[1] == component_type
+
+
def test_find_matching_services():
services = { "component_dockerhost_1": ["foo", "bar"],
"platform_dockerhost": [], "component_dockerhost_2": ["baz"] }
"component_dockerhost", ["foo"])
assert [] == dis._find_matching_services(services, "unknown", ["foo"])
+
+
+def test_is_healthy_pure():
+ def fake_is_healthy(name):
+ return 0, [{ "Checks": [{"Status": "passing"}] }]
+
+ assert True == dis._is_healthy_pure(fake_is_healthy, "some-component")
+
# ============LICENSE_START=======================================================
# org.onap.dcae
# ================================================================================
-# Copyright (c) 2017 AT&T Intellectual Property. All rights reserved.
+# 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.
import copy
import pytest
-from cloudify.exceptions import NonRecoverableError
+from cloudify.exceptions import NonRecoverableError, RecoverableError
import dockerplugin
from dockerplugin import tasks
+from dockerplugin.exceptions import DockerPluginDeploymentError
def test_generate_component_name():
assert expected == tasks._parse_streams(**test_input)
+def test_setup_for_discovery(monkeypatch):
+ test_input = { "name": "some-name",
+ "application_config": { "one": "a", "two": "b" } }
+
+ def fake_push_config(conn, name, application_config):
+ return
+
+ monkeypatch.setattr(dockerplugin.discovery, "push_service_component_config",
+ fake_push_config)
+
+ assert test_input == tasks._setup_for_discovery(**test_input)
+
+ def fake_push_config_connection_error(conn, name, application_config):
+ raise dockerplugin.discovery.DiscoveryConnectionError("Boom")
+
+ monkeypatch.setattr(dockerplugin.discovery, "push_service_component_config",
+ fake_push_config_connection_error)
+
+ with pytest.raises(RecoverableError):
+ tasks._setup_for_discovery(**test_input)
+
+
def test_setup_for_discovery_streams(monkeypatch):
test_input = {'feed01': {'type': 'data_router', 'name': 'feed01',
'username': 'hero', 'password': '123456', 'location': 'Bedminster'},
tasks._setup_for_discovery_streams(**test_input)
+def test_lookup_service(monkeypatch):
+ def fake_lookup(conn, scn):
+ return [{"ServiceAddress": "192.168.1.1", "ServicePort": "80"}]
+
+ monkeypatch.setattr(dockerplugin.discovery, "lookup_service",
+ fake_lookup)
+
+ assert "192.168.1.1" == tasks._lookup_service("some-component")
+ assert "192.168.1.1:80" == tasks._lookup_service("some-component",
+ with_port=True)
+
+
+def test_verify_container(monkeypatch):
+ def fake_is_healthy_success(ch, scn):
+ return True
+
+ monkeypatch.setattr(dockerplugin.discovery, "is_healthy",
+ fake_is_healthy_success)
+
+ assert tasks._verify_container("some-name", 3)
+
+ def fake_is_healthy_never_good(ch, scn):
+ return False
+
+ monkeypatch.setattr(dockerplugin.discovery, "is_healthy",
+ fake_is_healthy_never_good)
+
+ with pytest.raises(DockerPluginDeploymentError):
+ tasks._verify_container("some-name", 2)
+
+
def test_update_delivery_url(monkeypatch):
test_input = {'feed01': {'type': 'data_router', 'name': 'feed01',
'username': 'hero', 'password': '123456', 'location': 'Bedminster',
actual = tasks._enhance_docker_params(**test_kwargs)
assert actual["envs"] == {"SERVICE_TAGS": "abc,zed"}
+
+
+def test_notify_container():
+ test_input = { "docker_config": { "trigger_type": "unknown" } }
+ assert test_input == tasks._notify_container(**test_input)
--- /dev/null
+# ============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.
+
+from dockerplugin import utils
+
+
+def test_random_string():
+ target_length = 10
+ assert len(utils.random_string(target_length)) == target_length
+
+
+def test_update_dict():
+ d = { "a": 1, "b": 2 }
+ u = { "a": 2, "b": 3 }
+ assert utils.update_dict(d, u) == u
pytest
coverage
pytest-cov
-commands=pytest --junitxml xunit-results.xml --cov {envsitepackagesdir}/dockerplugin --cov-report=xml
+commands=pytest --junitxml xunit-results.xml --cov {envsitepackagesdir}/dockerplugin
Code Review / dcaegen2 / platform / plugins.git / commitdiff