n2vc/tests/unit/test_kubectl.py

   1 # Copyright 2020 Canonical Ltd.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #     Unless required by applicable law or agreed to in writing, software
  10 #     distributed under the License is distributed on an "AS IS" BASIS,
  11 #     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 #     See the License for the specific language governing permissions and
  13 #     limitations under the License.
  14
  15 from unittest import TestCase, mock
  16 from n2vc.kubectl import Kubectl
  17 from n2vc.utils import Dict
  18 from kubernetes.client.rest import ApiException
  19
  20 fake_list_services = Dict(
  21     {
  22         "items": [
  23             Dict(
  24                 {
  25                     "metadata": Dict(
  26                         {
  27                             "name": "squid",
  28                             "namespace": "test",
  29                             "labels": {"juju-app": "squid"},
  30                         }
  31                     ),
  32                     "spec": Dict(
  33                         {
  34                             "cluster_ip": "10.152.183.79",
  35                             "type": "LoadBalancer",
  36                             "ports": [
  37                                 Dict(
  38                                     {
  39                                         "name": None,
  40                                         "node_port": None,
  41                                         "port": 30666,
  42                                         "protocol": "TCP",
  43                                         "target_port": 30666,
  44                                     }
  45                                 )
  46                             ],
  47                         }
  48                     ),
  49                     "status": Dict(
  50                         {
  51                             "load_balancer": Dict(
  52                                 {
  53                                     "ingress": [
  54                                         Dict({"hostname": None, "ip": "192.168.0.201"})
  55                                     ]
  56                                 }
  57                             )
  58                         }
  59                     ),
  60                 }
  61             )
  62         ]
  63     }
  64 )
  65
  66
  67 class FakeCoreV1Api:
  68     def list_service_for_all_namespaces(self, **kwargs):
  69         return fake_list_services
  70
  71
  72 class ProvisionerTest(TestCase):
  73     @mock.patch("n2vc.kubectl.config.load_kube_config")
  74     @mock.patch("n2vc.kubectl.client.CoreV1Api")
  75     def setUp(self, mock_core, mock_config):
  76         mock_core.return_value = mock.MagicMock()
  77         mock_config.return_value = mock.MagicMock()
  78         self.kubectl = Kubectl()
  79
  80     @mock.patch("n2vc.kubectl.client.CoreV1Api")
  81     def test_get_service(self, mock_corev1api):
  82         mock_corev1api.return_value = FakeCoreV1Api()
  83         services = self.kubectl.get_services(
  84             field_selector="metadata.namespace", label_selector="juju-operator=squid"
  85         )
  86         keys = ["name", "cluster_ip", "type", "ports", "external_ip"]
  87         self.assertTrue(k in service for service in services for k in keys)
  88
  89     @mock.patch("n2vc.kubectl.client.CoreV1Api.list_service_for_all_namespaces")
  90     def test_get_service_exception(self, list_services):
  91         list_services.side_effect = ApiException()
  92         with self.assertRaises(ApiException):
  93             self.kubectl.get_services()