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
117 cacert
=mock_base64_to_cacert
.return_value
,
124 class K8sJujuConnectorInitFailureTestCase(asynctest
.TestCase
):
128 self
.loop
= asyncio
.get_event_loop()
129 logging
.disable(logging
.CRITICAL
)
132 "api_proxy": "api_proxy",
134 "k8s_cloud": "k8s_cloud",
141 def test_missing_vca_config_host(self
):
143 self
.vca_config
.pop("host")
144 with self
.assertRaises(N2VCBadArgumentsException
):
145 self
.k8s_juju_conn
= K8sJujuConnector(
146 fs
=fslocal
.FsLocal(),
150 vca_config
=self
.vca_config
,
154 def test_missing_vca_config_user(self
):
156 self
.vca_config
.pop("user")
157 with self
.assertRaises(N2VCBadArgumentsException
):
158 self
.k8s_juju_conn
= K8sJujuConnector(
159 fs
=fslocal
.FsLocal(),
163 vca_config
=self
.vca_config
,
167 def test_missing_vca_config_secret(self
):
169 self
.vca_config
.pop("secret")
170 with self
.assertRaises(N2VCBadArgumentsException
):
171 self
.k8s_juju_conn
= K8sJujuConnector(
172 fs
=fslocal
.FsLocal(),
176 vca_config
=self
.vca_config
,
180 def test_missing_vca_config_ca_cert(self
):
182 self
.vca_config
.pop("ca_cert")
183 with self
.assertRaises(N2VCBadArgumentsException
):
184 self
.k8s_juju_conn
= K8sJujuConnector(
185 fs
=fslocal
.FsLocal(),
189 vca_config
=self
.vca_config
,
194 @asynctest.mock
.patch("n2vc.kubectl.Kubectl.get_default_storage_class")
195 class InitEnvTest(K8sJujuConnTestCase
):
197 super(InitEnvTest
, self
).setUp()
198 self
.k8s_juju_conn
.libjuju
.add_k8s
= AsyncMock()
199 self
.k8s_juju_conn
._create
_cluster
_role
= Mock()
200 self
.k8s_juju_conn
._create
_service
_account
= Mock()
201 self
.k8s_juju_conn
._create
_cluster
_role
_binding
= Mock()
202 self
.k8s_juju_conn
._delete
_cluster
_role
= Mock()
203 self
.k8s_juju_conn
._delete
_service
_account
= Mock()
204 self
.k8s_juju_conn
._delete
_cluster
_role
_binding
= Mock()
205 self
.k8s_juju_conn
._get
_secret
_data
= AsyncMock()
206 self
.k8s_juju_conn
._get
_secret
_data
.return_value
= ("token", "cacert")
208 def test_with_cluster_uuid(
210 mock_get_default_storage_class
,
212 reuse_cluster_uuid
= "uuid"
213 uuid
, created
= self
.loop
.run_until_complete(
214 self
.k8s_juju_conn
.init_env(
215 k8s_creds
=kubeconfig
, reuse_cluster_uuid
=reuse_cluster_uuid
219 self
.assertTrue(created
)
220 self
.assertEqual(uuid
, reuse_cluster_uuid
)
221 mock_get_default_storage_class
.assert_called_once()
222 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
224 def test_with_no_cluster_uuid(
226 mock_get_default_storage_class
,
228 uuid
, created
= self
.loop
.run_until_complete(
229 self
.k8s_juju_conn
.init_env(k8s_creds
=kubeconfig
)
232 self
.assertTrue(created
)
233 self
.assertTrue(isinstance(uuid
, str))
234 mock_get_default_storage_class
.assert_called_once()
235 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
237 def test_init_env_exception(
239 mock_get_default_storage_class
,
241 self
.k8s_juju_conn
.libjuju
.add_k8s
.side_effect
= Exception()
244 with self
.assertRaises(Exception):
245 uuid
, created
= self
.loop
.run_until_complete(
246 self
.k8s_juju_conn
.init_env(k8s_creds
=kubeconfig
)
249 self
.assertIsNone(created
)
250 self
.assertIsNone(uuid
)
251 mock_get_default_storage_class
.assert_called_once()
252 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
255 class NotImplementedTest(K8sJujuConnTestCase
):
257 super(NotImplementedTest
, self
).setUp()
259 def test_repo_add(self
):
260 with self
.assertRaises(MethodNotImplemented
):
261 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_add("", ""))
263 def test_repo_list(self
):
264 with self
.assertRaises(MethodNotImplemented
):
265 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_list())
267 def test_repo_remove(self
):
268 with self
.assertRaises(MethodNotImplemented
):
269 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_remove(""))
271 def test_synchronize_repos(self
):
273 self
.loop
.run_until_complete(self
.k8s_juju_conn
.synchronize_repos("", ""))
276 def test_upgrade(self
):
277 with self
.assertRaises(MethodNotImplemented
):
278 self
.loop
.run_until_complete(self
.k8s_juju_conn
.upgrade("", ""))
280 def test_rollback(self
):
281 with self
.assertRaises(MethodNotImplemented
):
282 self
.loop
.run_until_complete(self
.k8s_juju_conn
.rollback("", ""))
284 def test_get_namespace(self
):
285 self
.assertIsNone(self
.k8s_juju_conn
.get_namespace(""))
287 def test_instances_list(self
):
288 res
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.instances_list(""))
289 self
.assertEqual(res
, [])
292 class ResetTest(K8sJujuConnTestCase
):
294 super(ResetTest
, self
).setUp()
295 self
.k8s_juju_conn
.libjuju
.remove_cloud
= AsyncMock()
296 self
.k8s_juju_conn
.libjuju
.get_cloud_credentials
= AsyncMock()
298 cloud_creds
.result
= {"attrs": {RBAC_LABEL_KEY_NAME
: "asd"}}
299 self
.k8s_juju_conn
.libjuju
.get_cloud_credentials
.return_value
= [cloud_creds
]
300 self
.k8s_juju_conn
._delete
_cluster
_role
_binding
= Mock()
301 self
.k8s_juju_conn
._delete
_service
_account
= Mock()
302 self
.k8s_juju_conn
._delete
_cluster
_role
= Mock()
303 self
.k8s_juju_conn
.get_credentials
= Mock()
304 self
.k8s_juju_conn
.get_credentials
.return_value
= kubeconfig
306 def test_success(self
):
307 removed
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.reset("uuid"))
308 self
.assertTrue(removed
)
309 self
.k8s_juju_conn
.libjuju
.remove_cloud
.assert_called_once()
311 def test_exception(self
):
313 self
.k8s_juju_conn
.libjuju
.remove_cloud
.side_effect
= Exception()
314 with self
.assertRaises(Exception):
315 removed
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.reset("uuid"))
316 self
.assertIsNone(removed
)
317 self
.k8s_juju_conn
.libjuju
.remove_cloud
.assert_called_once()
320 @asynctest.mock
.patch("os.chdir")
321 class InstallTest(K8sJujuConnTestCase
):
323 super(InstallTest
, self
).setUp()
324 self
.db_dict
= {"filter": {"_id": "id"}}
325 self
.local_bundle
= "bundle"
326 self
.cs_bundle
= "cs:bundle"
327 self
.http_bundle
= "https://example.com/bundle.yaml"
328 self
.kdu_name
= "kdu_name"
329 self
.cluster_uuid
= "cluster"
330 self
.kdu_instance
= "{}-{}".format(self
.kdu_name
, "id")
331 self
.k8s_juju_conn
.libjuju
.add_model
= AsyncMock()
332 self
.k8s_juju_conn
.libjuju
.deploy
= AsyncMock()
334 def test_success_local(self
, mock_chdir
):
335 self
.loop
.run_until_complete(
336 self
.k8s_juju_conn
.install(
341 kdu_name
=self
.kdu_name
,
342 db_dict
=self
.db_dict
,
346 self
.assertEqual(mock_chdir
.call_count
, 2)
347 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once_with(
348 model_name
=self
.kdu_instance
,
349 cloud_name
=self
.cluster_uuid
,
350 credential_name
="cred-{}".format(self
.cluster_uuid
),
352 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
353 "local:{}".format(self
.local_bundle
),
354 model_name
=self
.kdu_instance
,
359 def test_success_cs(self
, mock_chdir
):
360 self
.loop
.run_until_complete(
361 self
.k8s_juju_conn
.install(
366 kdu_name
=self
.kdu_name
,
367 db_dict
=self
.db_dict
,
371 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once_with(
372 model_name
=self
.kdu_instance
,
373 cloud_name
=self
.cluster_uuid
,
374 credential_name
="cred-{}".format(self
.cluster_uuid
),
376 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
378 model_name
=self
.kdu_instance
,
383 def test_success_http(self
, mock_chdir
):
384 self
.loop
.run_until_complete(
385 self
.k8s_juju_conn
.install(
390 kdu_name
=self
.kdu_name
,
391 db_dict
=self
.db_dict
,
395 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once_with(
396 model_name
=self
.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
=self
.kdu_instance
,
407 def test_success_not_kdu_name(self
, mock_chdir
):
408 self
.loop
.run_until_complete(
409 self
.k8s_juju_conn
.install(
414 db_dict
=self
.db_dict
,
418 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once_with(
419 model_name
=self
.kdu_instance
,
420 cloud_name
=self
.cluster_uuid
,
421 credential_name
="cred-{}".format(self
.cluster_uuid
),
423 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
425 model_name
=self
.kdu_instance
,
430 def test_missing_db_dict(self
, mock_chdir
):
432 with self
.assertRaises(K8sException
):
433 self
.loop
.run_until_complete(
434 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 self
.loop
.run_until_complete(
451 self
.k8s_juju_conn
.install(
456 kdu_name
=self
.kdu_name
,
457 db_dict
=self
.db_dict
,
461 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once_with(
462 model_name
=self
.kdu_instance
,
463 cloud_name
=self
.cluster_uuid
,
464 credential_name
="cred-{}".format(self
.cluster_uuid
),
466 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
468 model_name
=self
.kdu_instance
,
473 def test_missing_bundle(self
, mock_chdir
):
474 with self
.assertRaises(K8sException
):
475 self
.loop
.run_until_complete(
476 self
.k8s_juju_conn
.install(
481 kdu_name
=self
.kdu_name
,
483 db_dict
=self
.db_dict
,
486 self
.k8s_juju_conn
.libjuju
.add_model
.assert_not_called()
487 self
.k8s_juju_conn
.libjuju
.deploy
.assert_not_called()
489 def test_missing_exception(self
, mock_chdir
):
490 self
.k8s_juju_conn
.libjuju
.deploy
.side_effect
= Exception()
491 with self
.assertRaises(Exception):
492 self
.loop
.run_until_complete(
493 self
.k8s_juju_conn
.install(
498 kdu_name
=self
.kdu_name
,
499 db_dict
=self
.db_dict
,
503 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once_with(
504 model_name
=self
.kdu_instance
,
505 cloud_name
=self
.cluster_uuid
,
506 credential_name
="cred-{}".format(self
.cluster_uuid
),
508 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
509 "local:{}".format(self
.local_bundle
),
510 model_name
=self
.kdu_instance
,
516 class UninstallTest(K8sJujuConnTestCase
):
518 super(UninstallTest
, self
).setUp()
519 self
.k8s_juju_conn
.libjuju
.destroy_model
= AsyncMock()
521 def test_success(self
):
522 destroyed
= self
.loop
.run_until_complete(
523 self
.k8s_juju_conn
.uninstall("cluster_uuid", "model_name")
525 self
.assertTrue(destroyed
)
526 self
.k8s_juju_conn
.libjuju
.destroy_model
.assert_called_once()
528 def test_exception(self
):
530 self
.k8s_juju_conn
.libjuju
.destroy_model
.side_effect
= Exception()
531 with self
.assertRaises(Exception):
532 destroyed
= self
.loop
.run_until_complete(
533 self
.k8s_juju_conn
.uninstall("cluster_uuid", "model_name")
535 self
.assertIsNone(destroyed
)
536 self
.k8s_juju_conn
.libjuju
.destroy_model
.assert_called_once()
539 class ExecPrimitivesTest(K8sJujuConnTestCase
):
541 super(ExecPrimitivesTest
, self
).setUp()
542 self
.action_name
= "touch"
543 self
.application_name
= "myapp"
544 self
.model_name
= "model"
545 self
.k8s_juju_conn
.libjuju
.get_actions
= AsyncMock()
546 self
.k8s_juju_conn
.libjuju
.execute_action
= AsyncMock()
548 def test_success(self
):
549 params
= {"application-name": self
.application_name
}
550 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
551 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (
556 output
= self
.loop
.run_until_complete(
557 self
.k8s_juju_conn
.exec_primitive(
558 "cluster", self
.model_name
, self
.action_name
, params
=params
562 self
.assertEqual(output
, "success")
563 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
564 self
.application_name
, self
.model_name
566 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
567 self
.application_name
, self
.model_name
, self
.action_name
, **params
570 def test_exception(self
):
571 params
= {"application-name": self
.application_name
}
572 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
573 self
.k8s_juju_conn
.libjuju
.execute_action
.side_effect
= Exception()
576 with self
.assertRaises(Exception):
577 output
= self
.loop
.run_until_complete(
578 self
.k8s_juju_conn
.exec_primitive(
579 "cluster", self
.model_name
, self
.action_name
, params
=params
583 self
.assertIsNone(output
)
584 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
585 self
.application_name
, self
.model_name
587 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
588 self
.application_name
, self
.model_name
, self
.action_name
, **params
591 def test_missing_application_name_in_params(self
):
595 with self
.assertRaises(K8sException
):
596 output
= self
.loop
.run_until_complete(
597 self
.k8s_juju_conn
.exec_primitive(
598 "cluster", self
.model_name
, self
.action_name
, params
=params
602 self
.assertIsNone(output
)
603 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called()
604 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
606 def test_missing_params(self
):
608 with self
.assertRaises(K8sException
):
609 output
= self
.loop
.run_until_complete(
610 self
.k8s_juju_conn
.exec_primitive(
611 "cluster", self
.model_name
, self
.action_name
615 self
.assertIsNone(output
)
616 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called()
617 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
619 def test_missing_action(self
):
621 params
= {"application-name": self
.application_name
}
622 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
623 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (
627 with self
.assertRaises(K8sException
):
628 output
= self
.loop
.run_until_complete(
629 self
.k8s_juju_conn
.exec_primitive(
630 "cluster", self
.model_name
, "non-existing-action", params
=params
634 self
.assertIsNone(output
)
635 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
636 self
.application_name
, self
.model_name
638 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
640 def test_missing_not_completed(self
):
642 params
= {"application-name": self
.application_name
}
643 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
644 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (None, "failed")
645 with self
.assertRaises(K8sException
):
646 output
= self
.loop
.run_until_complete(
647 self
.k8s_juju_conn
.exec_primitive(
648 "cluster", self
.model_name
, self
.action_name
, params
=params
652 self
.assertIsNone(output
)
653 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
654 self
.application_name
, self
.model_name
656 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
657 self
.application_name
, self
.model_name
, self
.action_name
, **params
661 class InspectKduTest(K8sJujuConnTestCase
):
663 super(InspectKduTest
, self
).setUp()
665 @asynctest.mock
.patch("builtins.open")
666 @asynctest.mock
.patch("os.path.exists")
667 def test_existing_file(self
, mock_exists
, mock_open
):
668 mock_exists
.return_value
= True
670 'description': 'test bundle',
671 'bundle': 'kubernetes',
672 'applications': {'app':{ }, 'app2': { }}
674 mock_open
.return_value
= FakeFileWrapper(content
=content
)
675 kdu
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.inspect_kdu("model"))
676 self
.assertEqual(kdu
, {"app": {}, "app2": {}})
677 mock_exists
.assert_called_once()
678 mock_open
.assert_called_once()
680 @asynctest.mock
.patch("builtins.open")
681 @asynctest.mock
.patch("os.path.exists")
682 def test_not_existing_file(self
, mock_exists
, mock_open
):
684 mock_exists
.return_value
= False
685 with self
.assertRaises(K8sException
):
686 kdu
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.inspect_kdu("model"))
687 self
.assertEqual(kdu
, None)
688 mock_exists
.assert_called_once_with("model")
689 mock_open
.assert_not_called()
692 class HelpKduTest(K8sJujuConnTestCase
):
694 super(HelpKduTest
, self
).setUp()
696 @asynctest.mock
.patch("builtins.open")
697 @asynctest.mock
.patch("os.listdir")
698 def test_existing_file(self
, mock_listdir
, mock_open
):
699 content
= "Readme file content"
700 mock_open
.return_value
= FakeFileWrapper(content
=content
)
701 for file in ["README.md", "README.txt", "README"]:
702 mock_listdir
.return_value
= [file]
703 help = self
.loop
.run_until_complete(
704 self
.k8s_juju_conn
.help_kdu("kdu_instance")
706 self
.assertEqual(help, content
)
708 self
.assertEqual(mock_listdir
.call_count
, 3)
709 self
.assertEqual(mock_open
.call_count
, 3)
711 @asynctest.mock
.patch("builtins.open")
712 @asynctest.mock
.patch("os.listdir")
713 def test_not_existing_file(self
, mock_listdir
, mock_open
):
714 for file in ["src/charm.py", "tox.ini", "requirements.txt"]:
715 mock_listdir
.return_value
= [file]
716 help = self
.loop
.run_until_complete(
717 self
.k8s_juju_conn
.help_kdu("kdu_instance")
719 self
.assertEqual(help, None)
721 self
.assertEqual(mock_listdir
.call_count
, 3)
722 self
.assertEqual(mock_open
.call_count
, 0)
725 class StatusKduTest(K8sJujuConnTestCase
):
727 super(StatusKduTest
, self
).setUp()
728 self
.k8s_juju_conn
.libjuju
.get_model_status
= AsyncMock()
730 def test_success(self
):
731 applications
= {"app": {"status": {"status": "active"}}}
732 model
= FakeModel(applications
=applications
)
733 self
.k8s_juju_conn
.libjuju
.get_model_status
.return_value
= model
734 status
= self
.loop
.run_until_complete(
735 self
.k8s_juju_conn
.status_kdu("cluster", "kdu_instance")
737 self
.assertEqual(status
, {"app": {"status": "active"}})
738 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
740 def test_exception(self
):
741 self
.k8s_juju_conn
.libjuju
.get_model_status
.side_effect
= Exception()
743 with self
.assertRaises(Exception):
744 status
= self
.loop
.run_until_complete(
745 self
.k8s_juju_conn
.status_kdu("cluster", "kdu_instance")
747 self
.assertIsNone(status
)
748 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
751 class GetServicesTest(K8sJujuConnTestCase
):
753 super(GetServicesTest
, self
).setUp()
755 @asynctest.mock
.patch("n2vc.kubectl.Kubectl.get_services")
756 @asynctest.mock
.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
757 def test_success(self
, mock_get_credentials
, mock_get_services
):
758 mock_get_credentials
.return_value
= kubeconfig
759 self
.loop
.run_until_complete(self
.k8s_juju_conn
.get_services("", "", ""))
760 mock_get_credentials
.assert_called_once()
761 mock_get_services
.assert_called_once()
764 class GetServiceTest(K8sJujuConnTestCase
):
766 super(GetServiceTest
, self
).setUp()
768 @asynctest.mock
.patch("n2vc.kubectl.Kubectl.get_services")
769 @asynctest.mock
.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
770 def test_success(self
, mock_get_credentials
, mock_get_services
):
771 mock_get_credentials
.return_value
= kubeconfig
772 self
.loop
.run_until_complete(self
.k8s_juju_conn
.get_service("", "", ""))
773 mock_get_credentials
.assert_called_once()
774 mock_get_services
.assert_called_once()
777 class GetCredentialsTest(K8sJujuConnTestCase
):
779 super(GetCredentialsTest
, self
).setUp()
781 @asynctest.mock
.patch("yaml.safe_dump")
782 def test_success(self
, mock_safe_dump
):
783 self
.k8s_juju_conn
.db
.get_one
.return_value
= {
785 "credentials": "credentials",
786 "schema_version": "2",
788 self
.k8s_juju_conn
.get_credentials("cluster_uuid")
789 self
.k8s_juju_conn
.db
.get_one
.assert_called_once()
790 self
.k8s_juju_conn
.db
.encrypt_decrypt_fields
.assert_called_once()
791 mock_safe_dump
.assert_called_once()