1 # Copyright 2020 Canonical Ltd.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
15 from unittest
import TestCase
, mock
16 from n2vc
.kubectl
import Kubectl
, CORE_CLIENT
17 from n2vc
.utils
import Dict
18 from kubernetes
.client
.rest
import ApiException
21 class FakeK8sResourceMetadata
:
25 namespace
: str = None,
26 annotations
: dict = {},
29 self
._annotations
= annotations
30 self
._name
= name
or "name"
31 self
._namespace
= namespace
or "namespace"
32 self
._labels
= labels
or {"juju-app": "squid"}
40 return self
._namespace
47 def annotations(self
):
48 return self
._annotations
51 class FakeK8sStorageClass
:
52 def __init__(self
, metadata
=None):
53 self
._metadata
= metadata
or FakeK8sResourceMetadata()
60 class FakeK8sStorageClassesList
:
61 def __init__(self
, items
=[]):
69 fake_list_services
= Dict(
78 "labels": {"juju-app": "squid"},
83 "cluster_ip": "10.152.183.79",
84 "type": "LoadBalancer",
100 "load_balancer": Dict(
103 Dict({"hostname": None, "ip": "192.168.0.201"})
116 class KubectlTestCase(TestCase
):
124 def list_service_for_all_namespaces(self
, **kwargs
):
125 return fake_list_services
128 class GetServices(TestCase
):
129 @mock.patch("n2vc.kubectl.config.load_kube_config")
130 @mock.patch("n2vc.kubectl.client.CoreV1Api")
131 def setUp(self
, mock_core
, mock_config
):
132 mock_core
.return_value
= mock
.MagicMock()
133 mock_config
.return_value
= mock
.MagicMock()
134 self
.kubectl
= Kubectl()
136 @mock.patch("n2vc.kubectl.client.CoreV1Api")
137 def test_get_service(self
, mock_corev1api
):
138 mock_corev1api
.return_value
= FakeCoreV1Api()
139 services
= self
.kubectl
.get_services(
140 field_selector
="metadata.namespace", label_selector
="juju-operator=squid"
142 keys
= ["name", "cluster_ip", "type", "ports", "external_ip"]
143 self
.assertTrue(k
in service
for service
in services
for k
in keys
)
145 def test_get_service_exception(self
):
146 self
.kubectl
.clients
[
148 ].list_service_for_all_namespaces
.side_effect
= ApiException()
149 with self
.assertRaises(ApiException
):
150 self
.kubectl
.get_services()
153 @mock.patch("n2vc.kubectl.client")
154 @mock.patch("n2vc.kubectl.config.kube_config.Configuration.get_default_copy")
155 @mock.patch("n2vc.kubectl.config.load_kube_config")
156 class GetConfiguration(KubectlTestCase
):
158 super(GetConfiguration
, self
).setUp()
160 def test_get_configuration(
162 mock_load_kube_config
,
167 kubectl
.configuration
168 mock_configuration
.assert_called_once()
169 mock_load_kube_config
.assert_called_once()
170 mock_client
.CoreV1Api
.assert_called_once()
171 mock_client
.RbacAuthorizationV1Api
.assert_called_once()
172 mock_client
.StorageV1Api
.assert_called_once()
175 @mock.patch("kubernetes.client.StorageV1Api.list_storage_class")
176 @mock.patch("kubernetes.config.load_kube_config")
177 class GetDefaultStorageClass(KubectlTestCase
):
179 super(GetDefaultStorageClass
, self
).setUp()
181 # Default Storage Class
182 self
.default_sc_name
= "default-sc"
183 default_sc_metadata
= FakeK8sResourceMetadata(
184 name
=self
.default_sc_name
,
185 annotations
={"storageclass.kubernetes.io/is-default-class": "true"},
187 self
.default_sc
= FakeK8sStorageClass(metadata
=default_sc_metadata
)
189 # Default Storage Class with old annotation
190 self
.default_sc_old_name
= "default-sc-old"
191 default_sc_old_metadata
= FakeK8sResourceMetadata(
192 name
=self
.default_sc_old_name
,
193 annotations
={"storageclass.beta.kubernetes.io/is-default-class": "true"},
195 self
.default_sc_old
= FakeK8sStorageClass(metadata
=default_sc_old_metadata
)
197 # Storage class - not default
198 self
.sc_name
= "default-sc-old"
199 self
.sc
= FakeK8sStorageClass(
200 metadata
=FakeK8sResourceMetadata(name
=self
.sc_name
)
203 def test_get_default_storage_class_exists_default(
204 self
, mock_load_kube_config
, mock_list_storage_class
207 items
= [self
.default_sc
]
208 mock_list_storage_class
.return_value
= FakeK8sStorageClassesList(items
=items
)
209 sc_name
= kubectl
.get_default_storage_class()
210 self
.assertEqual(sc_name
, self
.default_sc_name
)
211 mock_list_storage_class
.assert_called_once()
213 def test_get_default_storage_class_exists_default_old(
214 self
, mock_load_kube_config
, mock_list_storage_class
217 items
= [self
.default_sc_old
]
218 mock_list_storage_class
.return_value
= FakeK8sStorageClassesList(items
=items
)
219 sc_name
= kubectl
.get_default_storage_class()
220 self
.assertEqual(sc_name
, self
.default_sc_old_name
)
221 mock_list_storage_class
.assert_called_once()
223 def test_get_default_storage_class_none(
224 self
, mock_load_kube_config
, mock_list_storage_class
227 mock_list_storage_class
.return_value
= FakeK8sStorageClassesList(items
=[])
228 sc_name
= kubectl
.get_default_storage_class()
229 self
.assertEqual(sc_name
, None)
230 mock_list_storage_class
.assert_called_once()
232 def test_get_default_storage_class_exists_not_default(
233 self
, mock_load_kube_config
, mock_list_storage_class
237 mock_list_storage_class
.return_value
= FakeK8sStorageClassesList(items
=items
)
238 sc_name
= kubectl
.get_default_storage_class()
239 self
.assertEqual(sc_name
, self
.sc_name
)
240 mock_list_storage_class
.assert_called_once()
242 def test_get_default_storage_class_choose(
243 self
, mock_load_kube_config
, mock_list_storage_class
246 items
= [self
.sc
, self
.default_sc
]
247 mock_list_storage_class
.return_value
= FakeK8sStorageClassesList(items
=items
)
248 sc_name
= kubectl
.get_default_storage_class()
249 self
.assertEqual(sc_name
, self
.default_sc_name
)
250 mock_list_storage_class
.assert_called_once()