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.
19 from unittest
.mock
import Mock
20 from n2vc
.k8s_juju_conn
import K8sJujuConnector
, RBAC_LABEL_KEY_NAME
21 from osm_common
import fslocal
22 from .utils
import kubeconfig
, FakeModel
, FakeFileWrapper
, AsyncMock
23 from n2vc
.exceptions
import (
26 N2VCBadArgumentsException
,
30 class K8sJujuConnTestCase(asynctest
.TestCase
):
31 @asynctest.mock
.patch("juju.controller.Controller.update_endpoints")
32 @asynctest.mock
.patch("juju.client.connector.Connector.connect")
33 @asynctest.mock
.patch("juju.controller.Controller.connection")
34 @asynctest.mock
.patch("n2vc.k8s_juju_conn.base64_to_cacert")
35 @asynctest.mock
.patch("n2vc.k8s_juju_conn.Libjuju")
39 mock_base64_to_cacert
=None,
42 mock_update_endpoints
=None,
44 self
.loop
= asyncio
.get_event_loop()
45 mock_libjuju
.return_value
= AsyncMock()
49 "api_proxy": "api_proxy",
51 "k8s_cloud": "k8s_cloud",
58 logging
.disable(logging
.CRITICAL
)
60 self
.k8s_juju_conn
= K8sJujuConnector(
65 vca_config
=vca_config
,
70 class K8sJujuConnInitSuccessTestCase(asynctest
.TestCase
):
74 logging
.disable(logging
.CRITICAL
)
76 @asynctest.mock
.patch("juju.controller.Controller.update_endpoints")
77 @asynctest.mock
.patch("juju.client.connector.Connector.connect")
78 @asynctest.mock
.patch("juju.controller.Controller.connection")
79 @asynctest.mock
.patch("n2vc.k8s_juju_conn.base64_to_cacert")
80 @asynctest.mock
.patch("n2vc.libjuju.Libjuju.__init__")
84 mock_base64_to_cacert
=None,
87 mock_update_endpoints
=None,
89 mock_libjuju
.return_value
= None
90 loop
= asyncio
.get_event_loop()
91 log
= logging
.getLogger()
96 "k8s_cloud": "k8s_cloud",
103 fs
=fslocal
.FsLocal(),
107 vca_config
=vca_config
,
111 mock_libjuju
.assert_called_once_with(
112 endpoint
="1.1.1.1:17070",
113 api_proxy
=None, # Not needed for k8s charms
114 enable_os_upgrade
=True,
118 cacert
=mock_base64_to_cacert
.return_value
,
125 class K8sJujuConnectorInitFailureTestCase(asynctest
.TestCase
):
129 self
.loop
= asyncio
.get_event_loop()
130 logging
.disable(logging
.CRITICAL
)
133 "api_proxy": "api_proxy",
135 "k8s_cloud": "k8s_cloud",
142 def test_missing_vca_config_host(self
):
144 self
.vca_config
.pop("host")
145 with self
.assertRaises(N2VCBadArgumentsException
):
146 self
.k8s_juju_conn
= K8sJujuConnector(
147 fs
=fslocal
.FsLocal(),
151 vca_config
=self
.vca_config
,
155 def test_missing_vca_config_user(self
):
157 self
.vca_config
.pop("user")
158 with self
.assertRaises(N2VCBadArgumentsException
):
159 self
.k8s_juju_conn
= K8sJujuConnector(
160 fs
=fslocal
.FsLocal(),
164 vca_config
=self
.vca_config
,
168 def test_missing_vca_config_secret(self
):
170 self
.vca_config
.pop("secret")
171 with self
.assertRaises(N2VCBadArgumentsException
):
172 self
.k8s_juju_conn
= K8sJujuConnector(
173 fs
=fslocal
.FsLocal(),
177 vca_config
=self
.vca_config
,
181 def test_missing_vca_config_ca_cert(self
):
183 self
.vca_config
.pop("ca_cert")
184 with self
.assertRaises(N2VCBadArgumentsException
):
185 self
.k8s_juju_conn
= K8sJujuConnector(
186 fs
=fslocal
.FsLocal(),
190 vca_config
=self
.vca_config
,
195 @asynctest.mock
.patch("n2vc.kubectl.Kubectl.get_default_storage_class")
196 class InitEnvTest(K8sJujuConnTestCase
):
198 super(InitEnvTest
, self
).setUp()
199 self
.k8s_juju_conn
.libjuju
.add_k8s
= AsyncMock()
200 self
.k8s_juju_conn
._create
_cluster
_role
= Mock()
201 self
.k8s_juju_conn
._create
_service
_account
= Mock()
202 self
.k8s_juju_conn
._create
_cluster
_role
_binding
= Mock()
203 self
.k8s_juju_conn
._get
_secret
_data
= AsyncMock()
204 self
.k8s_juju_conn
._get
_secret
_data
.return_value
= ("token", "cacert")
206 def test_with_cluster_uuid(
208 mock_get_default_storage_class
,
210 reuse_cluster_uuid
= "uuid"
211 uuid
, created
= self
.loop
.run_until_complete(
212 self
.k8s_juju_conn
.init_env(
213 k8s_creds
=kubeconfig
, reuse_cluster_uuid
=reuse_cluster_uuid
217 self
.assertTrue(created
)
218 self
.assertEqual(uuid
, reuse_cluster_uuid
)
219 mock_get_default_storage_class
.assert_called_once()
220 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
222 def test_with_no_cluster_uuid(
224 mock_get_default_storage_class
,
226 uuid
, created
= self
.loop
.run_until_complete(
227 self
.k8s_juju_conn
.init_env(k8s_creds
=kubeconfig
)
230 self
.assertTrue(created
)
231 self
.assertTrue(isinstance(uuid
, str))
232 mock_get_default_storage_class
.assert_called_once()
233 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
235 def test_init_env_exception(
237 mock_get_default_storage_class
,
239 self
.k8s_juju_conn
.libjuju
.add_k8s
.side_effect
= Exception()
242 with self
.assertRaises(Exception):
243 uuid
, created
= self
.loop
.run_until_complete(
244 self
.k8s_juju_conn
.init_env(k8s_creds
=kubeconfig
)
247 self
.assertIsNone(created
)
248 self
.assertIsNone(uuid
)
249 mock_get_default_storage_class
.assert_called_once()
250 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
253 class NotImplementedTest(K8sJujuConnTestCase
):
255 super(NotImplementedTest
, self
).setUp()
257 def test_repo_add(self
):
258 with self
.assertRaises(MethodNotImplemented
):
259 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_add("", ""))
261 def test_repo_list(self
):
262 with self
.assertRaises(MethodNotImplemented
):
263 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_list())
265 def test_repo_remove(self
):
266 with self
.assertRaises(MethodNotImplemented
):
267 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_remove(""))
269 def test_synchronize_repos(self
):
271 self
.loop
.run_until_complete(self
.k8s_juju_conn
.synchronize_repos("", ""))
274 def test_upgrade(self
):
275 with self
.assertRaises(MethodNotImplemented
):
276 self
.loop
.run_until_complete(self
.k8s_juju_conn
.upgrade("", ""))
278 def test_rollback(self
):
279 with self
.assertRaises(MethodNotImplemented
):
280 self
.loop
.run_until_complete(self
.k8s_juju_conn
.rollback("", ""))
282 def test_get_namespace(self
):
283 self
.assertIsNone(self
.k8s_juju_conn
.get_namespace(""))
285 def test_instances_list(self
):
286 res
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.instances_list(""))
287 self
.assertEqual(res
, [])
290 class ResetTest(K8sJujuConnTestCase
):
292 super(ResetTest
, self
).setUp()
293 self
.k8s_juju_conn
.libjuju
.remove_cloud
= AsyncMock()
294 self
.k8s_juju_conn
.libjuju
.get_cloud_credentials
= AsyncMock()
296 cloud_creds
.result
= {"attrs": {RBAC_LABEL_KEY_NAME
: "asd"}}
297 self
.k8s_juju_conn
.libjuju
.get_cloud_credentials
.return_value
= [cloud_creds
]
298 self
.k8s_juju_conn
._delete
_cluster
_role
_binding
= Mock()
299 self
.k8s_juju_conn
._delete
_service
_account
= Mock()
300 self
.k8s_juju_conn
._delete
_cluster
_role
= Mock()
301 self
.k8s_juju_conn
.get_credentials
= Mock()
302 self
.k8s_juju_conn
.get_credentials
.return_value
= kubeconfig
304 def test_success(self
):
305 removed
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.reset("uuid"))
306 self
.assertTrue(removed
)
307 self
.k8s_juju_conn
.libjuju
.remove_cloud
.assert_called_once()
309 def test_exception(self
):
311 self
.k8s_juju_conn
.libjuju
.remove_cloud
.side_effect
= Exception()
312 with self
.assertRaises(Exception):
313 removed
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.reset("uuid"))
314 self
.assertIsNone(removed
)
315 self
.k8s_juju_conn
.libjuju
.remove_cloud
.assert_called_once()
318 @asynctest.mock
.patch("os.chdir")
319 class InstallTest(K8sJujuConnTestCase
):
321 super(InstallTest
, self
).setUp()
322 self
.db_dict
= {"filter": {"_id": "id"}}
323 self
.local_bundle
= "bundle"
324 self
.cs_bundle
= "cs:bundle"
325 self
.http_bundle
= "https://example.com/bundle.yaml"
326 self
.kdu_name
= "kdu_name"
327 self
.cluster_uuid
= "cluster"
328 self
.k8s_juju_conn
.libjuju
.add_model
= AsyncMock()
329 self
.k8s_juju_conn
.libjuju
.deploy
= AsyncMock()
331 def test_success_local(self
, mock_chdir
):
332 expected_kdu_instance
= "{}-{}".format(self
.kdu_name
, "id")
333 kdu_instance
= self
.loop
.run_until_complete(
334 self
.k8s_juju_conn
.install(
338 kdu_name
=self
.kdu_name
,
339 db_dict
=self
.db_dict
,
343 self
.assertEqual(kdu_instance
, expected_kdu_instance
)
344 self
.assertEqual(mock_chdir
.call_count
, 2)
345 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once_with(
346 model_name
=expected_kdu_instance
,
347 cloud_name
=self
.cluster_uuid
,
348 credential_name
="cred-{}".format(self
.cluster_uuid
),
350 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
351 "local:{}".format(self
.local_bundle
),
352 model_name
=expected_kdu_instance
,
357 def test_success_cs(self
, mock_chdir
):
358 expected_kdu_instance
= "{}-{}".format(self
.kdu_name
, "id")
359 kdu_instance
= self
.loop
.run_until_complete(
360 self
.k8s_juju_conn
.install(
364 kdu_name
=self
.kdu_name
,
365 db_dict
=self
.db_dict
,
369 self
.assertEqual(kdu_instance
, expected_kdu_instance
)
370 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once_with(
371 model_name
=expected_kdu_instance
,
372 cloud_name
=self
.cluster_uuid
,
373 credential_name
="cred-{}".format(self
.cluster_uuid
),
375 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
377 model_name
=expected_kdu_instance
,
382 def test_success_http(self
, mock_chdir
):
383 expected_kdu_instance
= "{}-{}".format(self
.kdu_name
, "id")
384 kdu_instance
= self
.loop
.run_until_complete(
385 self
.k8s_juju_conn
.install(
389 kdu_name
=self
.kdu_name
,
390 db_dict
=self
.db_dict
,
394 self
.assertEqual(kdu_instance
, expected_kdu_instance
)
395 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once_with(
396 model_name
=expected_kdu_instance
,
397 cloud_name
=self
.cluster_uuid
,
398 credential_name
="cred-{}".format(self
.cluster_uuid
),
400 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
402 model_name
=expected_kdu_instance
,
407 def test_success_not_kdu_name(self
, mock_chdir
):
408 expected_kdu_instance
= "id"
409 kdu_instance
= self
.loop
.run_until_complete(
410 self
.k8s_juju_conn
.install(
414 db_dict
=self
.db_dict
,
418 self
.assertEqual(kdu_instance
, expected_kdu_instance
)
419 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once_with(
420 model_name
=expected_kdu_instance
,
421 cloud_name
=self
.cluster_uuid
,
422 credential_name
="cred-{}".format(self
.cluster_uuid
),
424 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
426 model_name
=expected_kdu_instance
,
431 def test_missing_db_dict(self
, mock_chdir
):
433 with self
.assertRaises(K8sException
):
434 kdu_instance
= self
.loop
.run_until_complete(
435 self
.k8s_juju_conn
.install(
439 kdu_name
=self
.kdu_name
,
443 self
.assertIsNone(kdu_instance
)
444 self
.k8s_juju_conn
.libjuju
.add_model
.assert_not_called()
445 self
.k8s_juju_conn
.libjuju
.deploy
.assert_not_called()
447 @asynctest.mock
.patch("os.getcwd")
448 def test_getcwd_exception(self
, mock_getcwd
, mock_chdir
):
449 mock_getcwd
.side_effect
= FileNotFoundError()
450 expected_kdu_instance
= "{}-{}".format(self
.kdu_name
, "id")
451 kdu_instance
= self
.loop
.run_until_complete(
452 self
.k8s_juju_conn
.install(
456 kdu_name
=self
.kdu_name
,
457 db_dict
=self
.db_dict
,
461 self
.assertEqual(kdu_instance
, expected_kdu_instance
)
462 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once_with(
463 model_name
=expected_kdu_instance
,
464 cloud_name
=self
.cluster_uuid
,
465 credential_name
="cred-{}".format(self
.cluster_uuid
),
467 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
469 model_name
=expected_kdu_instance
,
474 def test_missing_bundle(self
, mock_chdir
):
476 with self
.assertRaises(K8sException
):
477 kdu_instance
= self
.loop
.run_until_complete(
478 self
.k8s_juju_conn
.install(
482 kdu_name
=self
.kdu_name
,
484 db_dict
=self
.db_dict
,
487 self
.assertIsNone(kdu_instance
)
488 self
.k8s_juju_conn
.libjuju
.add_model
.assert_not_called()
489 self
.k8s_juju_conn
.libjuju
.deploy
.assert_not_called()
491 def test_missing_exception(self
, mock_chdir
):
492 expected_kdu_instance
= "{}-{}".format(self
.kdu_name
, "id")
494 self
.k8s_juju_conn
.libjuju
.deploy
.side_effect
= Exception()
495 with self
.assertRaises(Exception):
496 kdu_instance
= self
.loop
.run_until_complete(
497 self
.k8s_juju_conn
.install(
501 kdu_name
=self
.kdu_name
,
502 db_dict
=self
.db_dict
,
506 self
.assertIsNone(kdu_instance
)
507 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once_with(
508 model_name
=expected_kdu_instance
,
509 cloud_name
=self
.cluster_uuid
,
510 credential_name
="cred-{}".format(self
.cluster_uuid
),
512 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
513 "local:{}".format(self
.local_bundle
),
514 model_name
=expected_kdu_instance
,
520 class UninstallTest(K8sJujuConnTestCase
):
522 super(UninstallTest
, self
).setUp()
523 self
.k8s_juju_conn
.libjuju
.destroy_model
= AsyncMock()
525 def test_success(self
):
526 destroyed
= self
.loop
.run_until_complete(
527 self
.k8s_juju_conn
.uninstall("cluster_uuid", "model_name")
529 self
.assertTrue(destroyed
)
530 self
.k8s_juju_conn
.libjuju
.destroy_model
.assert_called_once()
532 def test_exception(self
):
534 self
.k8s_juju_conn
.libjuju
.destroy_model
.side_effect
= Exception()
535 with self
.assertRaises(Exception):
536 destroyed
= self
.loop
.run_until_complete(
537 self
.k8s_juju_conn
.uninstall("cluster_uuid", "model_name")
539 self
.assertIsNone(destroyed
)
540 self
.k8s_juju_conn
.libjuju
.destroy_model
.assert_called_once()
543 class ExecPrimitivesTest(K8sJujuConnTestCase
):
545 super(ExecPrimitivesTest
, self
).setUp()
546 self
.action_name
= "touch"
547 self
.application_name
= "myapp"
548 self
.model_name
= "model"
549 self
.k8s_juju_conn
.libjuju
.get_actions
= AsyncMock()
550 self
.k8s_juju_conn
.libjuju
.execute_action
= AsyncMock()
552 def test_success(self
):
553 params
= {"application-name": self
.application_name
}
554 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
555 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (
560 output
= self
.loop
.run_until_complete(
561 self
.k8s_juju_conn
.exec_primitive(
562 "cluster", self
.model_name
, self
.action_name
, params
=params
566 self
.assertEqual(output
, "success")
567 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
568 self
.application_name
, self
.model_name
570 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
571 self
.application_name
, self
.model_name
, self
.action_name
, **params
574 def test_exception(self
):
575 params
= {"application-name": self
.application_name
}
576 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
577 self
.k8s_juju_conn
.libjuju
.execute_action
.side_effect
= Exception()
580 with self
.assertRaises(Exception):
581 output
= self
.loop
.run_until_complete(
582 self
.k8s_juju_conn
.exec_primitive(
583 "cluster", self
.model_name
, self
.action_name
, params
=params
587 self
.assertIsNone(output
)
588 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
589 self
.application_name
, self
.model_name
591 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
592 self
.application_name
, self
.model_name
, self
.action_name
, **params
595 def test_missing_application_name_in_params(self
):
599 with self
.assertRaises(K8sException
):
600 output
= self
.loop
.run_until_complete(
601 self
.k8s_juju_conn
.exec_primitive(
602 "cluster", self
.model_name
, self
.action_name
, params
=params
606 self
.assertIsNone(output
)
607 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called()
608 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
610 def test_missing_params(self
):
612 with self
.assertRaises(K8sException
):
613 output
= self
.loop
.run_until_complete(
614 self
.k8s_juju_conn
.exec_primitive(
615 "cluster", self
.model_name
, self
.action_name
619 self
.assertIsNone(output
)
620 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called()
621 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
623 def test_missing_action(self
):
625 params
= {"application-name": self
.application_name
}
626 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
627 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (
631 with self
.assertRaises(K8sException
):
632 output
= self
.loop
.run_until_complete(
633 self
.k8s_juju_conn
.exec_primitive(
634 "cluster", self
.model_name
, "non-existing-action", params
=params
638 self
.assertIsNone(output
)
639 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
640 self
.application_name
, self
.model_name
642 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
644 def test_missing_not_completed(self
):
646 params
= {"application-name": self
.application_name
}
647 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
648 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (None, "failed")
649 with self
.assertRaises(K8sException
):
650 output
= self
.loop
.run_until_complete(
651 self
.k8s_juju_conn
.exec_primitive(
652 "cluster", self
.model_name
, self
.action_name
, params
=params
656 self
.assertIsNone(output
)
657 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
658 self
.application_name
, self
.model_name
660 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
661 self
.application_name
, self
.model_name
, self
.action_name
, **params
665 class InspectKduTest(K8sJujuConnTestCase
):
667 super(InspectKduTest
, self
).setUp()
669 @asynctest.mock
.patch("builtins.open")
670 @asynctest.mock
.patch("os.path.exists")
671 def test_existing_file(self
, mock_exists
, mock_open
):
672 mock_exists
.return_value
= True
674 'description': 'test bundle',
675 'bundle': 'kubernetes',
676 'applications': {'app':{ }, 'app2': { }}
678 mock_open
.return_value
= FakeFileWrapper(content
=content
)
679 kdu
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.inspect_kdu("model"))
680 self
.assertEqual(kdu
, {"app": {}, "app2": {}})
681 mock_exists
.assert_called_once()
682 mock_open
.assert_called_once()
684 @asynctest.mock
.patch("builtins.open")
685 @asynctest.mock
.patch("os.path.exists")
686 def test_not_existing_file(self
, mock_exists
, mock_open
):
688 mock_exists
.return_value
= False
689 with self
.assertRaises(K8sException
):
690 kdu
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.inspect_kdu("model"))
691 self
.assertEqual(kdu
, None)
692 mock_exists
.assert_called_once_with("model")
693 mock_open
.assert_not_called()
696 class HelpKduTest(K8sJujuConnTestCase
):
698 super(HelpKduTest
, self
).setUp()
700 @asynctest.mock
.patch("builtins.open")
701 @asynctest.mock
.patch("os.listdir")
702 def test_existing_file(self
, mock_listdir
, mock_open
):
703 content
= "Readme file content"
704 mock_open
.return_value
= FakeFileWrapper(content
=content
)
705 for file in ["README.md", "README.txt", "README"]:
706 mock_listdir
.return_value
= [file]
707 help = self
.loop
.run_until_complete(
708 self
.k8s_juju_conn
.help_kdu("kdu_instance")
710 self
.assertEqual(help, content
)
712 self
.assertEqual(mock_listdir
.call_count
, 3)
713 self
.assertEqual(mock_open
.call_count
, 3)
715 @asynctest.mock
.patch("builtins.open")
716 @asynctest.mock
.patch("os.listdir")
717 def test_not_existing_file(self
, mock_listdir
, mock_open
):
718 for file in ["src/charm.py", "tox.ini", "requirements.txt"]:
719 mock_listdir
.return_value
= [file]
720 help = self
.loop
.run_until_complete(
721 self
.k8s_juju_conn
.help_kdu("kdu_instance")
723 self
.assertEqual(help, None)
725 self
.assertEqual(mock_listdir
.call_count
, 3)
726 self
.assertEqual(mock_open
.call_count
, 0)
729 class StatusKduTest(K8sJujuConnTestCase
):
731 super(StatusKduTest
, self
).setUp()
732 self
.k8s_juju_conn
.libjuju
.get_model_status
= AsyncMock()
734 def test_success(self
):
735 applications
= {"app": {"status": {"status": "active"}}}
736 model
= FakeModel(applications
=applications
)
737 self
.k8s_juju_conn
.libjuju
.get_model_status
.return_value
= model
738 status
= self
.loop
.run_until_complete(
739 self
.k8s_juju_conn
.status_kdu("cluster", "kdu_instance")
741 self
.assertEqual(status
, {"app": {"status": "active"}})
742 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
744 def test_exception(self
):
745 self
.k8s_juju_conn
.libjuju
.get_model_status
.side_effect
= Exception()
747 with self
.assertRaises(Exception):
748 status
= self
.loop
.run_until_complete(
749 self
.k8s_juju_conn
.status_kdu("cluster", "kdu_instance")
751 self
.assertIsNone(status
)
752 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
755 class GetServicesTest(K8sJujuConnTestCase
):
757 super(GetServicesTest
, self
).setUp()
759 @asynctest.mock
.patch("n2vc.kubectl.Kubectl.get_services")
760 @asynctest.mock
.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
761 def test_success(self
, mock_get_credentials
, mock_get_services
):
762 mock_get_credentials
.return_value
= kubeconfig
763 self
.loop
.run_until_complete(self
.k8s_juju_conn
.get_services("", "", ""))
764 mock_get_credentials
.assert_called_once()
765 mock_get_services
.assert_called_once()
768 class GetServiceTest(K8sJujuConnTestCase
):
770 super(GetServiceTest
, self
).setUp()
772 @asynctest.mock
.patch("n2vc.kubectl.Kubectl.get_services")
773 @asynctest.mock
.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
774 def test_success(self
, mock_get_credentials
, mock_get_services
):
775 mock_get_credentials
.return_value
= kubeconfig
776 self
.loop
.run_until_complete(self
.k8s_juju_conn
.get_service("", "", ""))
777 mock_get_credentials
.assert_called_once()
778 mock_get_services
.assert_called_once()
781 class GetCredentialsTest(K8sJujuConnTestCase
):
783 super(GetCredentialsTest
, self
).setUp()
785 @asynctest.mock
.patch("yaml.safe_dump")
786 def test_success(self
, mock_safe_dump
):
787 self
.k8s_juju_conn
.db
.get_one
.return_value
= {
789 "credentials": "credentials",
790 "schema_version": "2",
792 self
.k8s_juju_conn
.get_credentials("cluster_uuid")
793 self
.k8s_juju_conn
.db
.get_one
.assert_called_once()
794 self
.k8s_juju_conn
.db
.encrypt_decrypt_fields
.assert_called_once()
795 mock_safe_dump
.assert_called_once()