k8s/tests/test_discovery.py

   1 # ============LICENSE_START=======================================================
   2 # org.onap.dcae
   3 # ================================================================================
   4 # Copyright (c) 2017-2020 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
  20 import pytest
  21 from functools import partial
  22 import requests
  23
  24 def test_wrap_consul_call(mockconfig):
  25     from k8splugin import discovery as dis
  26
  27     def foo(a, b, c="default"):
  28         return " ".join([a, b, c])
  29
  30     wrapped_foo = partial(dis._wrap_consul_call, foo)
  31     assert wrapped_foo("hello", "world") == "hello world default"
  32     assert wrapped_foo("hello", "world", c="new masters") == "hello world new masters"
  33
  34     def foo_connection_error(a, b, c):
  35         raise requests.exceptions.ConnectionError("simulate failed connection")
  36
  37     wrapped_foo = partial(dis._wrap_consul_call, foo_connection_error)
  38     with pytest.raises(dis.DiscoveryConnectionError):
  39         wrapped_foo("a", "b", "c")
  40
  41 def test_generate_service_component_name(mockconfig):
  42     from k8splugin import discovery as dis
  43
  44     component_type = "some-component-type"
  45     name = dis.generate_service_component_name(component_type)
  46     assert name.split("-", 1)[1] == component_type
  47
  48 def test_find_matching_services(mockconfig):
  49     from k8splugin import discovery as dis
  50
  51     services = { "component_dockerhost_1": ["foo", "bar"],
  52             "platform_dockerhost": [], "component_dockerhost_2": ["baz"] }
  53     assert sorted(["component_dockerhost_1", "component_dockerhost_2"]) \
  54         == sorted(dis._find_matching_services(services, "component_dockerhost", []))
  55
  56     assert ["component_dockerhost_1"] == dis._find_matching_services(services, \
  57             "component_dockerhost", ["foo", "bar"])
  58
  59     assert ["component_dockerhost_1"] == dis._find_matching_services(services, \
  60             "component_dockerhost", ["foo"])
  61
  62     assert [] == dis._find_matching_services(services, "unknown", ["foo"])
  63
  64 def test_is_healthy_pure(mockconfig):
  65     from k8splugin import discovery as dis
  66
  67     def fake_is_healthy(name):
  68         return 0, [{ "Checks": [{"Status": "passing"}] }]
  69
  70     assert True == dis._is_healthy_pure(fake_is_healthy, "some-component")