k8s/tests/test_discovery.py

   1 # ============LICENSE_START=======================================================
   2 # org.onap.dcae
   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
   9 #
  10 #      http://www.apache.org/licenses/LICENSE-2.0
  11 #
  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=========================================================
  18 #
  19 # ECOMP is a trademark and service mark of AT&T Intellectual Property.
  20
  21 import pytest
  22 from functools import partial
  23 import requests
  24
  25 def test_wrap_consul_call(mockconfig):
  26     from k8splugin import discovery as dis
  27
  28     def foo(a, b, c="default"):
  29         return " ".join([a, b, c])
  30
  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"
  34
  35     def foo_connection_error(a, b, c):
  36         raise requests.exceptions.ConnectionError("simulate failed connection")
  37
  38     wrapped_foo = partial(dis._wrap_consul_call, foo_connection_error)
  39     with pytest.raises(dis.DiscoveryConnectionError):
  40         wrapped_foo("a", "b", "c")
  41
  42 def test_generate_service_component_name(mockconfig):
  43     from k8splugin import discovery as dis
  44
  45     component_type = "some-component-type"
  46     name = dis.generate_service_component_name(component_type)
  47     assert name.split("-", 1)[1] == component_type
  48
  49 def test_find_matching_services(mockconfig):
  50     from k8splugin import discovery as dis
  51
  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", []))
  56
  57     assert ["component_dockerhost_1"] == dis._find_matching_services(services, \
  58             "component_dockerhost", ["foo", "bar"])
  59
  60     assert ["component_dockerhost_1"] == dis._find_matching_services(services, \
  61             "component_dockerhost", ["foo"])
  62
  63     assert [] == dis._find_matching_services(services, "unknown", ["foo"])
  64
  65 def test_is_healthy_pure(mockconfig):
  66     from k8splugin import discovery as dis
  67
  68     def fake_is_healthy(name):
  69         return 0, [{ "Checks": [{"Status": "passing"}] }]
  70
  71     assert True == dis._is_healthy_pure(fake_is_healthy, "some-component")