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                                 {
  38                                     "name": None,
  39                                     "node_port": None,
  40                                     "port": 30666,
  41                                     "protocol": "TCP",
  42                                     "target_port": 30666,
  43                                 }
  44                             ],
  45                         }
  46                     ),
  47                     "status": Dict(
  48                         {
  49                             "load_balancer": Dict(
  50                                 {
  51                                     "ingress": [
  52                                         Dict({"hostname": None, "ip": "192.168.0.201"})
  53                                     ]
  54                                 }
  55                             )
  56                         }
  57                     ),
  58                 }
  59             )
  60         ]
  61     }
  62 )
  63
  64
  65 class FakeCoreV1Api:
  66     def list_service_for_all_namespaces(self, **kwargs):
  67         return fake_list_services
  68
  69
  70 class ProvisionerTest(TestCase):
  71     @mock.patch("n2vc.kubectl.config.load_kube_config")
  72     @mock.patch("n2vc.kubectl.client.CoreV1Api")
  73     def setUp(self, mock_core, mock_config):
  74         mock_core.return_value = mock.MagicMock()
  75         mock_config.return_value = mock.MagicMock()
  76         self.kubectl = Kubectl()
  77
  78     @mock.patch("n2vc.kubectl.client.CoreV1Api")
  79     def test_get_service(self, mock_corev1api):
  80         mock_corev1api.return_value = FakeCoreV1Api()
  81         services = self.kubectl.get_services(
  82             field_selector="metadata.namespace", label_selector="juju-operator=squid"
  83         )
  84         keys = ["name", "cluster_ip", "type", "ports", "external_ip"]
  85         self.assertTrue(k in service for service in services for k in keys)
  86
  87     @mock.patch("n2vc.kubectl.client.CoreV1Api.list_service_for_all_namespaces")
  88     def test_get_service_exception(self, list_services):
  89         list_services.side_effect = ApiException()
  90         with self.assertRaises(ApiException):
  91             self.kubectl.get_services()