Fix bug 1263
[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
21 class FakeK8sResourceMetadata:
22 def __init__(
23 self,
24 name: str = None,
25 namespace: str = None,
26 annotations: dict = {},
27 labels: dict = {},
28 ):
29 self._annotations = annotations
30 self._name = name or "name"
31 self._namespace = namespace or "namespace"
32 self._labels = labels or {"juju-app": "squid"}
33
34 @property
35 def name(self):
36 return self._name
37
38 @property
39 def namespace(self):
40 return self._namespace
41
42 @property
43 def labels(self):
44 return self._labels
45
46 @property
47 def annotations(self):
48 return self._annotations
49
50
51 class FakeK8sStorageClass:
52 def __init__(self, metadata=None):
53 self._metadata = metadata or FakeK8sResourceMetadata()
54
55 @property
56 def metadata(self):
57 return self._metadata
58
59
60 class FakeK8sStorageClassesList:
61 def __init__(self, items=[]):
62 self._items = items
63
64 @property
65 def items(self):
66 return self._items
67
68
69 fake_list_services = Dict(
70 {
71 "items": [
72 Dict(
73 {
74 "metadata": Dict(
75 {
76 "name": "squid",
77 "namespace": "test",
78 "labels": {"juju-app": "squid"},
79 }
80 ),
81 "spec": Dict(
82 {
83 "cluster_ip": "10.152.183.79",
84 "type": "LoadBalancer",
85 "ports": [
86 Dict(
87 {
88 "name": None,
89 "node_port": None,
90 "port": 30666,
91 "protocol": "TCP",
92 "target_port": 30666,
93 }
94 )
95 ],
96 }
97 ),
98 "status": Dict(
99 {
100 "load_balancer": Dict(
101 {
102 "ingress": [
103 Dict({"hostname": None, "ip": "192.168.0.201"})
104 ]
105 }
106 )
107 }
108 ),
109 }
110 )
111 ]
112 }
113 )
114
115
116 class KubectlTestCase(TestCase):
117 def setUp(self,):
118 pass
119
120
121 class FakeCoreV1Api:
122 def list_service_for_all_namespaces(self, **kwargs):
123 return fake_list_services
124
125
126 class ProvisionerTest(TestCase):
127 @mock.patch("n2vc.kubectl.config.load_kube_config")
128 @mock.patch("n2vc.kubectl.client.CoreV1Api")
129 def setUp(self, mock_core, mock_config):
130 mock_core.return_value = mock.MagicMock()
131 mock_config.return_value = mock.MagicMock()
132 self.kubectl = Kubectl()
133
134 @mock.patch("n2vc.kubectl.client.CoreV1Api")
135 def test_get_service(self, mock_corev1api):
136 mock_corev1api.return_value = FakeCoreV1Api()
137 services = self.kubectl.get_services(
138 field_selector="metadata.namespace", label_selector="juju-operator=squid"
139 )
140 keys = ["name", "cluster_ip", "type", "ports", "external_ip"]
141 self.assertTrue(k in service for service in services for k in keys)
142
143 @mock.patch("n2vc.kubectl.client.CoreV1Api.list_service_for_all_namespaces")
144 def test_get_service_exception(self, list_services):
145 list_services.side_effect = ApiException()
146 with self.assertRaises(ApiException):
147 self.kubectl.get_services()
148
149
150 @mock.patch("kubernetes.config.kube_config.Configuration")
151 @mock.patch("n2vc.kubectl.config.load_kube_config")
152 class GetConfiguration(KubectlTestCase):
153 def setUp(self):
154 super(GetConfiguration, self).setUp()
155
156 def test_get_configuration(self, mock_load_kube_config, mock_configuration):
157 kubectl = Kubectl()
158 kubectl.get_configuration()
159 mock_configuration.assert_called_once()
160
161
162 @mock.patch("kubernetes.client.StorageV1Api.list_storage_class")
163 @mock.patch("kubernetes.config.load_kube_config")
164 class GetDefaultStorageClass(KubectlTestCase):
165 def setUp(self):
166 super(GetDefaultStorageClass, self).setUp()
167
168 # Default Storage Class
169 self.default_sc_name = "default-sc"
170 default_sc_metadata = FakeK8sResourceMetadata(
171 name=self.default_sc_name,
172 annotations={"storageclass.kubernetes.io/is-default-class": "true"},
173 )
174 self.default_sc = FakeK8sStorageClass(metadata=default_sc_metadata)
175
176 # Default Storage Class with old annotation
177 self.default_sc_old_name = "default-sc-old"
178 default_sc_old_metadata = FakeK8sResourceMetadata(
179 name=self.default_sc_old_name,
180 annotations={"storageclass.beta.kubernetes.io/is-default-class": "true"},
181 )
182 self.default_sc_old = FakeK8sStorageClass(metadata=default_sc_old_metadata)
183
184 # Storage class - not default
185 self.sc_name = "default-sc-old"
186 self.sc = FakeK8sStorageClass(
187 metadata=FakeK8sResourceMetadata(name=self.sc_name)
188 )
189
190 def test_get_default_storage_class_exists_default(
191 self, mock_load_kube_config, mock_list_storage_class
192 ):
193 kubectl = Kubectl()
194 items = [self.default_sc]
195 mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
196 sc_name = kubectl.get_default_storage_class()
197 self.assertEqual(sc_name, self.default_sc_name)
198 mock_list_storage_class.assert_called_once()
199
200 def test_get_default_storage_class_exists_default_old(
201 self, mock_load_kube_config, mock_list_storage_class
202 ):
203 kubectl = Kubectl()
204 items = [self.default_sc_old]
205 mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
206 sc_name = kubectl.get_default_storage_class()
207 self.assertEqual(sc_name, self.default_sc_old_name)
208 mock_list_storage_class.assert_called_once()
209
210 def test_get_default_storage_class_none(
211 self, mock_load_kube_config, mock_list_storage_class
212 ):
213 kubectl = Kubectl()
214 mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=[])
215 sc_name = kubectl.get_default_storage_class()
216 self.assertEqual(sc_name, None)
217 mock_list_storage_class.assert_called_once()
218
219 def test_get_default_storage_class_exists_not_default(
220 self, mock_load_kube_config, mock_list_storage_class
221 ):
222 kubectl = Kubectl()
223 items = [self.sc]
224 mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
225 sc_name = kubectl.get_default_storage_class()
226 self.assertEqual(sc_name, self.sc_name)
227 mock_list_storage_class.assert_called_once()
228
229 def test_get_default_storage_class_choose(
230 self, mock_load_kube_config, mock_list_storage_class
231 ):
232 kubectl = Kubectl()
233 items = [self.sc, self.default_sc]
234 mock_list_storage_class.return_value = FakeK8sStorageClassesList(items=items)
235 sc_name = kubectl.get_default_storage_class()
236 self.assertEqual(sc_name, self.default_sc_name)
237 mock_list_storage_class.assert_called_once()