k8s/tests/test_k8sclient.py

   1 # ============LICENSE_START=======================================================
   2 # org.onap.dcae
   3 # ================================================================================
   4 # Copyright (c) 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 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"