Add get_metrics command to n2vc_juju_conn and libjuju
[osm/N2VC.git] / 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()