k8s/tests/test_k8sclient.py

   1 # ============LICENSE_START=======================================================
   2 # org.onap.dcae
   3 # ================================================================================
   4 # Copyright (c) 2018-2020 AT&T Intellectual Property. All rights reserved.
   5 # Copyright (c) 2019 Pantheon.tech. All rights reserved.
   6 # ================================================================================
   7 # Licensed under the Apache License, Version 2.0 (the "License");
   8 # you may not use this file except in compliance with the License.
   9 # You may obtain a copy of the License at
  10 #
  11 #      http://www.apache.org/licenses/LICENSE-2.0
  12 #
  13 # Unless required by applicable law or agreed to in writing, software
  14 # distributed under the License is distributed on an "AS IS" BASIS,
  15 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16 # See the License for the specific language governing permissions and
  17 # limitations under the License.
  18 # ============LICENSE_END=========================================================
  19
  20 # Basic sanity tests for  k8sclient functions
  21
  22 import pytest
  23
  24 def test_parse_interval():
  25     from k8sclient.k8sclient import _parse_interval
  26
  27     good_intervals = [{"in": input, "ex": expected}
  28         for (input, expected) in [
  29             (30, 30),
  30             ("30", 30),
  31             ("30s", 30),
  32             ("2m", 2 * 60),
  33             ("2h", 2 * 60 * 60),
  34             ("24h", 24 * 60 * 60),
  35             (354123, 354123),
  36             ("354123", 354123),
  37             ("354123s", 354123),
  38             (1234567890123456789012345678901234567890,1234567890123456789012345678901234567890),
  39             ("1234567890123456789012345678901234567890",1234567890123456789012345678901234567890),
  40             ("1234567890123456789012345678901234567890s",1234567890123456789012345678901234567890),
  41             ("05s", 5),
  42             ("00000000000000000000000000000000005m", 5 * 60)
  43         ]
  44     ]
  45
  46     bad_intervals = [
  47         -99,
  48         "-99",
  49         "-99s",
  50         "-99m",
  51         "-99h",
  52         "30d",
  53         "30w",
  54         "30y",
  55         "3 0s",
  56         "3 5m",
  57         30.0,
  58         "30.0s",
  59         "30.0m",
  60         "30.0h",
  61         "a 30s",
  62         "30s a",
  63         "a 30s a",
  64         "a 30",
  65         "30 a",
  66         "a 30 a",
  67         "i want an interval of 30s",
  68         "thirty seconds",
  69         "30 s",
  70         "30 m",
  71         "30 h",
  72         10E0,
  73         "10E0",
  74         3.14159,
  75         "3.14159s"
  76         "3:05",
  77         "3m05s",
  78         "3seconds",
  79         "3S",
  80         "1minute",
  81         "1stanbul"
  82     ]
  83
  84     for test_case in good_intervals:
  85         assert _parse_interval(test_case["in"]) == test_case["ex"]
  86
  87     for interval in bad_intervals:
  88         with pytest.raises(ValueError):
  89             _parse_interval(interval)
  90
  91 def test_parse_ports():
  92     from k8sclient.k8sclient import parse_ports
  93
  94     good_ports = [{"in": input, "ex": expected}
  95         for (input, expected) in [
  96             ("9101:0", (9101, 0, "TCP")),
  97             ("9101/TCP:0", (9101, 0, "TCP")),
  98             ("9101/tcp:0", (9101, 0, "TCP")),
  99             ("9101/UDP:0", (9101, 0, "UDP")),
 100             ("9101/udp:0", (9101, 0, "UDP")),
 101             ("9101:31043", (9101, 31043, "TCP")),
 102             ("9101/TCP:31043", (9101, 31043, "TCP")),
 103             ("9101/tcp:31043", (9101, 31043, "TCP")),
 104             ("9101/UDP:31043", (9101, 31043, "UDP")),
 105             ("9101/udp:31043", (9101, 31043, "UDP"))
 106         ]
 107     ]
 108
 109     bad_ports = [
 110         "9101",
 111         "9101:",
 112         "9101:0x453",
 113         "9101:0/udp",
 114         "9101/:0",
 115         "9101/u:0",
 116         "9101/http:404",
 117         "9101:-1"
 118     ]
 119
 120     port_list = [
 121         "9101:0",
 122         "5053/tcp:5053",
 123         "5053/udp:5053",
 124         "9661:19661",
 125         "9661/udp:19661",
 126         "8080/tcp:8080"
 127     ]
 128
 129     expected_port_map = {
 130         (9101,"TCP") : 0,
 131         (5053,"TCP") : 5053,
 132         (5053,"UDP") : 5053,
 133         (9661,"TCP") : 19661,
 134         (9661,"UDP") : 19661,
 135         (8080,"TCP") : 8080
 136     }
 137
 138     for test_case in good_ports:
 139         container_ports, port_map = parse_ports([test_case["in"]])
 140         (cport, hport, proto) = test_case["ex"]
 141         assert container_ports == [(cport, proto)]
 142         assert port_map == {(cport, proto) : hport}
 143
 144     for port in bad_ports:
 145         with pytest.raises(ValueError):
 146             parse_ports([port])
 147
 148     container_ports, port_map = parse_ports(port_list)
 149     assert port_map == expected_port_map
 150
 151 def test_create_container():
 152     from k8sclient.k8sclient import _create_container_object
 153     from kubernetes import client
 154
 155     container = _create_container_object("c1","nginx",False, container_ports=[(80, "TCP"), (53, "UDP")])
 156
 157     assert container.ports[0].container_port == 80 and container.ports[0].protocol == "TCP"
 158     assert container.ports[1].container_port == 53 and container.ports[1].protocol == "UDP"
 159
 160 def test_create_probe():
 161     from k8sclient.k8sclient import _create_probe
 162     from kubernetes import client
 163
 164     http_checks = [
 165         {"type" : "http", "endpoint" : "/example/health"}
 166     ]
 167
 168     script_checks = [
 169         {"type" : "docker", "script": "/opt/app/health_check.sh"}
 170     ]
 171
 172     for hc in http_checks:
 173         probe = _create_probe(hc, 13131)
 174         assert probe.http_get.path == hc["endpoint"]
 175         assert probe.http_get.scheme == hc["type"].upper()
 176
 177     for hc in script_checks:
 178         probe = _create_probe(hc, 13131)
 179         assert probe._exec.command[0] == hc["script"]