1 # ============LICENSE_START=======================================================
3 # ================================================================================
4 # Copyright (c) 2017-2018 AT&T Intellectual Property. All rights reserved.
5 # ================================================================================
6 # Licensed under the Apache License, Version 2.0 (the "License");
7 # you may not use this file except in compliance with the License.
8 # You may obtain a copy of the License at
10 # http://www.apache.org/licenses/LICENSE-2.0
12 # Unless required by applicable law or agreed to in writing, software
13 # distributed under the License is distributed on an "AS IS" BASIS,
14 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15 # See the License for the specific language governing permissions and
16 # limitations under the License.
17 # ============LICENSE_END=========================================================
19 # ECOMP is a trademark and service mark of AT&T Intellectual Property.
22 from functools import partial
25 def test_wrap_consul_call(mockconfig):
26 from k8splugin import discovery as dis
28 def foo(a, b, c="default"):
29 return " ".join([a, b, c])
31 wrapped_foo = partial(dis._wrap_consul_call, foo)
32 assert wrapped_foo("hello", "world") == "hello world default"
33 assert wrapped_foo("hello", "world", c="new masters") == "hello world new masters"
35 def foo_connection_error(a, b, c):
36 raise requests.exceptions.ConnectionError("simulate failed connection")
38 wrapped_foo = partial(dis._wrap_consul_call, foo_connection_error)
39 with pytest.raises(dis.DiscoveryConnectionError):
40 wrapped_foo("a", "b", "c")
42 def test_generate_service_component_name(mockconfig):
43 from k8splugin import discovery as dis
45 component_type = "some-component-type"
46 name = dis.generate_service_component_name(component_type)
47 assert name.split("-", 1)[1] == component_type
49 def test_find_matching_services(mockconfig):
50 from k8splugin import discovery as dis
52 services = { "component_dockerhost_1": ["foo", "bar"],
53 "platform_dockerhost": [], "component_dockerhost_2": ["baz"] }
54 assert sorted(["component_dockerhost_1", "component_dockerhost_2"]) \
55 == sorted(dis._find_matching_services(services, "component_dockerhost", []))
57 assert ["component_dockerhost_1"] == dis._find_matching_services(services, \
58 "component_dockerhost", ["foo", "bar"])
60 assert ["component_dockerhost_1"] == dis._find_matching_services(services, \
61 "component_dockerhost", ["foo"])
63 assert [] == dis._find_matching_services(services, "unknown", ["foo"])
65 def test_is_healthy_pure(mockconfig):
66 from k8splugin import discovery as dis
68 def fake_is_healthy(name):
69 return 0, [{ "Checks": [{"Status": "passing"}] }]
71 assert True == dis._is_healthy_pure(fake_is_healthy, "some-component")