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("kubernetes.config.kube_config.Configuration")
154 @mock.patch("n2vc.kubectl.config.load_kube_config")
155 class GetConfiguration(KubectlTestCase
):
157 super(GetConfiguration
, self
).setUp()
159 def test_get_configuration(self
, mock_load_kube_config
, mock_configuration
):
161 kubectl
.configuration
162 mock_configuration
.assert_called_once()
165 @mock.patch("kubernetes.client.StorageV1Api.list_storage_class")
166 @mock.patch("kubernetes.config.load_kube_config")
167 class GetDefaultStorageClass(KubectlTestCase
):
169 super(GetDefaultStorageClass
, self
).setUp()
171 # Default Storage Class
172 self
.default_sc_name
= "default-sc"
173 default_sc_metadata
= FakeK8sResourceMetadata(
174 name
=self
.default_sc_name
,
175 annotations
={"storageclass.kubernetes.io/is-default-class": "true"},
177 self
.default_sc
= FakeK8sStorageClass(metadata
=default_sc_metadata
)
179 # Default Storage Class with old annotation
180 self
.default_sc_old_name
= "default-sc-old"
181 default_sc_old_metadata
= FakeK8sResourceMetadata(
182 name
=self
.default_sc_old_name
,
183 annotations
={"storageclass.beta.kubernetes.io/is-default-class": "true"},
185 self
.default_sc_old
= FakeK8sStorageClass(metadata
=default_sc_old_metadata
)
187 # Storage class - not default
188 self
.sc_name
= "default-sc-old"
189 self
.sc
= FakeK8sStorageClass(
190 metadata
=FakeK8sResourceMetadata(name
=self
.sc_name
)
193 def test_get_default_storage_class_exists_default(
194 self
, mock_load_kube_config
, mock_list_storage_class
197 items
= [self
.default_sc
]
198 mock_list_storage_class
.return_value
= FakeK8sStorageClassesList(items
=items
)
199 sc_name
= kubectl
.get_default_storage_class()
200 self
.assertEqual(sc_name
, self
.default_sc_name
)
201 mock_list_storage_class
.assert_called_once()
203 def test_get_default_storage_class_exists_default_old(
204 self
, mock_load_kube_config
, mock_list_storage_class
207 items
= [self
.default_sc_old
]
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_old_name
)
211 mock_list_storage_class
.assert_called_once()
213 def test_get_default_storage_class_none(
214 self
, mock_load_kube_config
, mock_list_storage_class
217 mock_list_storage_class
.return_value
= FakeK8sStorageClassesList(items
=[])
218 sc_name
= kubectl
.get_default_storage_class()
219 self
.assertEqual(sc_name
, None)
220 mock_list_storage_class
.assert_called_once()
222 def test_get_default_storage_class_exists_not_default(
223 self
, mock_load_kube_config
, mock_list_storage_class
227 mock_list_storage_class
.return_value
= FakeK8sStorageClassesList(items
=items
)
228 sc_name
= kubectl
.get_default_storage_class()
229 self
.assertEqual(sc_name
, self
.sc_name
)
230 mock_list_storage_class
.assert_called_once()
232 def test_get_default_storage_class_choose(
233 self
, mock_load_kube_config
, mock_list_storage_class
236 items
= [self
.sc
, self
.default_sc
]
237 mock_list_storage_class
.return_value
= FakeK8sStorageClassesList(items
=items
)
238 sc_name
= kubectl
.get_default_storage_class()
239 self
.assertEqual(sc_name
, self
.default_sc_name
)
240 mock_list_storage_class
.assert_called_once()