k8s/tests/test_k8sclient.py

   1 # ============LICENSE_START=======================================================
   2 # org.onap.dcae
   3 # ================================================================================
   4 # Copyright (c) 2018-2019 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 import pytest
  20
  21 def test_parse_interval():
  22     from k8sclient.k8sclient import _parse_interval
  23
  24     good_intervals = [{"in": input, "ex": expected}
  25         for (input, expected) in [
  26             (30, 30),
  27             ("30", 30),
  28             ("30s", 30),
  29             ("2m", 2 * 60),
  30             ("2h", 2 * 60 * 60),
  31             ("24h", 24 * 60 * 60),
  32             (354123, 354123),
  33             ("354123", 354123),
  34             ("354123s", 354123),
  35             (1234567890123456789012345678901234567890L,1234567890123456789012345678901234567890L),
  36             ("1234567890123456789012345678901234567890",1234567890123456789012345678901234567890L),
  37             ("1234567890123456789012345678901234567890s",1234567890123456789012345678901234567890L),
  38             ("05s", 5),
  39             ("00000000000000000000000000000000005m", 5 * 60)
  40         ]
  41     ]
  42
  43     bad_intervals = [
  44         -99,
  45         "-99",
  46         "-99s",
  47         "-99m",
  48         "-99h",
  49         "30d",
  50         "30w",
  51         "30y",
  52         "3 0s",
  53         "3 5m",
  54         30.0,
  55         "30.0s",
  56         "30.0m",
  57         "30.0h",
  58         "a 30s",
  59         "30s a",
  60         "a 30s a",
  61         "a 30",
  62         "30 a",
  63         "a 30 a",
  64         "i want an interval of 30s",
  65         "thirty seconds",
  66         "30 s",
  67         "30 m",
  68         "30 h",
  69         10E0,
  70         "10E0",
  71         3.14159,
  72         "3.14159s"
  73         "3:05",
  74         "3m05s",
  75         "3seconds",
  76         "3S",
  77         "1minute",
  78         "1stanbul"
  79     ]
  80
  81     for test_case in good_intervals:
  82         assert _parse_interval(test_case["in"]) == test_case["ex"]
  83
  84     for interval in bad_intervals:
  85         with pytest.raises(ValueError):
  86             _parse_interval(interval)
  87
  88 def test_parse_ports():
  89     from k8sclient.k8sclient import _parse_ports
  90
  91     good_ports = [{"in": input, "ex": expected}
  92         for (input, expected) in [
  93             ("9101:0", (9101, 0, "TCP")),
  94             ("9101/TCP:0", (9101, 0, "TCP")),
  95             ("9101/tcp:0", (9101, 0, "TCP")),
  96             ("9101/UDP:0", (9101, 0, "UDP")),
  97             ("9101/udp:0", (9101, 0, "UDP")),
  98             ("9101:31043", (9101, 31043, "TCP")),
  99             ("9101/TCP:31043", (9101, 31043, "TCP")),
 100             ("9101/tcp:31043", (9101, 31043, "TCP")),
 101             ("9101/UDP:31043", (9101, 31043, "UDP")),
 102             ("9101/udp:31043", (9101, 31043, "UDP"))
 103         ]
 104     ]
 105
 106     bad_ports = [
 107         "9101",
 108         "9101:",
 109         "9101:0x453",
 110         "9101:0/udp",
 111         "9101/:0",
 112         "9101/u:0",
 113         "9101/http:404",
 114         "9101:-1"
 115     ]
 116
 117     port_list = [
 118         "9101:0",
 119         "5053/tcp:5053",
 120         "5053/udp:5053",
 121         "9661:19661",
 122         "9661/udp:19661",
 123         "8080/tcp:8080"
 124     ]
 125
 126     expected_port_map = {
 127         (9101,"TCP") : 0,
 128         (5053,"TCP") : 5053,
 129         (5053,"UDP") : 5053,
 130         (9661,"TCP") : 19661,
 131         (9661,"UDP") : 19661,
 132         (8080,"TCP") : 8080
 133     }
 134
 135     for test_case in good_ports:
 136         container_ports, port_map = _parse_ports([test_case["in"]])
 137         (cport, hport, proto) = test_case["ex"]
 138         assert container_ports == [(cport, proto)]
 139         assert port_map == {(cport, proto) : hport}
 140
 141     for port in bad_ports:
 142         with pytest.raises(ValueError):
 143             _parse_ports([port])
 144
 145     container_ports, port_map = _parse_ports(port_list)
 146     assert port_map == expected_port_map
 147
 148 def test_create_container():
 149     from k8sclient.k8sclient import _create_container_object
 150     from kubernetes import client
 151
 152     container = _create_container_object("c1","nginx",False, container_ports=[(80, "TCP"), (53, "UDP")])
 153
 154     assert container.ports[0].container_port == 80 and container.ports[0].protocol == "TCP"
 155     assert container.ports[1].container_port == 53 and container.ports[1].protocol == "UDP"
 156
 157 def test_create_probe():
 158     from k8sclient.k8sclient import _create_probe
 159     from kubernetes import client
 160
 161     http_checks = [
 162         {"type" : "http", "endpoint" : "/example/health"}
 163     ]
 164
 165     script_checks = [
 166         {"type" : "docker", "script": "/opt/app/health_check.sh"}
 167     ]
 168
 169     for hc in http_checks:
 170         probe = _create_probe(hc, 13131)
 171         assert probe.http_get.path == hc["endpoint"]
 172         assert probe.http_get.scheme == hc["type"].upper()
 173
 174     for hc in script_checks:
 175         probe = _create_probe(hc, 13131)
 176         assert probe._exec.command[0] == hc["script"]
 177
 178 def test_deploy(monkeypatch):
 179     import k8sclient.k8sclient
 180     from kubernetes import client
 181
 182     # We need to patch the kubernetes 'client' module
 183     # Awkward because of the way it requires a function call
 184     # to get an API object
 185     core = client.CoreV1Api()
 186     ext = client.ExtensionsV1beta1Api()
 187
 188     def pseudo_deploy(namespace, dep):
 189         return dep
 190
 191     def pseudo_service(namespace, svc):
 192         return svc
 193
 194     # patched_core returns a CoreV1Api object with the
 195     # create_namespaced_service method stubbed out so that there
 196     # is no attempt to call the k8s API server
 197     def patched_core():
 198         monkeypatch.setattr(core, "create_namespaced_service", pseudo_service)
 199         return core
 200
 201     # patched_ext returns an ExtensionsV1beta1Api object with the
 202     # create_namespaced_deployment method stubbed out so that there
 203     # is no attempt to call the k8s API server
 204     def patched_ext():
 205         monkeypatch.setattr(ext,"create_namespaced_deployment", pseudo_deploy)
 206         return ext
 207
 208     def pseudo_configure(loc):
 209         pass
 210
 211     monkeypatch.setattr(k8sclient.k8sclient,"_configure_api", pseudo_configure)
 212     monkeypatch.setattr(client, "CoreV1Api", patched_core)
 213     monkeypatch.setattr(client,"ExtensionsV1beta1Api", patched_ext)
 214
 215     k8s_test_config = {
 216         "image_pull_secrets" : ["secret0", "secret1"],
 217         "filebeat" : {
 218             "log_path": "/var/log/onap",
 219             "data_path": "/usr/share/filebeat/data",
 220             "config_path": "/usr/share/filebeat/filebeat.yml",
 221             "config_subpath": "filebeat.yml",
 222             "image" : "filebeat-repo/filebeat:latest",
 223             "config_map" : "dcae-filebeat-configmap"
 224         },
 225         "tls" : {
 226             "cert_path": "/opt/certs",
 227             "image": "tlsrepo/tls-init-container:1.2.3"
 228         }
 229     }
 230
 231     resources = {
 232         "limits": {
 233             "cpu" : 0.5,
 234             "memory" : "2Gi"
 235         },
 236         "requests": {
 237             "cpu" : 0.5,
 238             "memory" : "2Gi"
 239         }
 240     }
 241
 242     kwargs = {
 243         "volumes": [
 244             {"host":{"path": "/path/on/host"}, "container":{"bind":"/path/on/container","mode":"rw"}}
 245         ],
 246         "ports": ["80:0", "443:0"],
 247         "env": {"name0": "value0", "name1": "value1"},
 248         "log_info": {"log_directory": "/path/to/container/log/directory"},
 249         "tls_info": {"use_tls": True, "cert_directory": "/path/to/container/cert/directory" },
 250         "readiness": {"type": "http", "endpoint" : "/ready"}
 251     }
 252     dep, deployment_description = k8sclient.k8sclient.deploy("k8stest","testcomponent","example.com/testcomponent:1.4.3",1,False, k8s_test_config, resources, **kwargs)
 253
 254     assert deployment_description["deployment"] == "dep-testcomponent"
 255     assert deployment_description["namespace"] == "k8stest"
 256     assert deployment_description["services"][0] == "testcomponent"
 257
 258     # For unit test purposes, we want to make sure that the deployment object
 259     # we're passing to the k8s API is correct
 260     app_container = dep.spec.template.spec.containers[0]
 261     assert app_container.image == "example.com/testcomponent:1.4.3"
 262     assert app_container.image_pull_policy == "IfNotPresent"
 263     assert len(app_container.ports) == 2
 264     assert app_container.ports[0].container_port == 80
 265     assert app_container.ports[1].container_port == 443
 266     assert app_container.readiness_probe.http_get.path == "/ready"
 267     assert app_container.readiness_probe.http_get.scheme == "HTTP"
 268     assert len(app_container.volume_mounts) == 3
 269     assert app_container.volume_mounts[0].mount_path == "/path/on/container"
 270     assert app_container.volume_mounts[1].mount_path == "/path/to/container/log/directory"
 271     assert app_container.volume_mounts[2].mount_path == "/path/to/container/cert/directory"
 272
 273     # Needs to be correctly labeled so that the Service can find it
 274     assert dep.spec.template.metadata.labels["app"] == "testcomponent"