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
, FakeApplication
23 from n2vc
.exceptions
import (
27 from n2vc
.vca
.connection_data
import ConnectionData
30 class K8sJujuConnTestCase(asynctest
.TestCase
):
31 @asynctest.mock
.patch("n2vc.k8s_juju_conn.Libjuju")
32 @asynctest.mock
.patch("n2vc.k8s_juju_conn.MotorStore")
33 @asynctest.mock
.patch("n2vc.k8s_juju_conn.get_connection")
34 @asynctest.mock
.patch("n2vc.vca.connection_data.base64_to_cacert")
37 mock_base64_to_cacert
=None,
38 mock_get_connection
=None,
42 self
.loop
= asyncio
.get_event_loop()
44 mock_base64_to_cacert
.return_value
= """
45 -----BEGIN CERTIFICATE-----
47 -----END CERTIFICATE-----"""
48 mock_libjuju
.return_value
= Mock()
49 mock_store
.return_value
= AsyncMock()
50 mock_vca_connection
= Mock()
51 mock_get_connection
.return_value
= mock_vca_connection
52 mock_vca_connection
.data
.return_value
= ConnectionData(
54 "endpoints": ["1.2.3.4:17070"],
60 "lxd-credentials": "credentials",
61 "k8s-cloud": "k8s_cloud",
62 "k8s-credentials": "k8s_credentials",
64 "api-proxy": "api_proxy",
67 logging
.disable(logging
.CRITICAL
)
69 self
.k8s_juju_conn
= K8sJujuConnector(
76 self
.k8s_juju_conn
._store
.get_vca_id
.return_value
= None
77 self
.k8s_juju_conn
.libjuju
= Mock()
80 @asynctest.mock
.patch("n2vc.kubectl.Kubectl.get_default_storage_class")
81 class InitEnvTest(K8sJujuConnTestCase
):
83 super(InitEnvTest
, self
).setUp()
84 self
.k8s_juju_conn
.libjuju
.add_k8s
= AsyncMock()
85 self
.k8s_juju_conn
._create
_cluster
_role
= Mock()
86 self
.k8s_juju_conn
._create
_service
_account
= Mock()
87 self
.k8s_juju_conn
._create
_cluster
_role
_binding
= Mock()
88 self
.k8s_juju_conn
._delete
_cluster
_role
= Mock()
89 self
.k8s_juju_conn
._delete
_service
_account
= Mock()
90 self
.k8s_juju_conn
._delete
_cluster
_role
_binding
= Mock()
91 self
.k8s_juju_conn
._get
_secret
_data
= AsyncMock()
92 self
.k8s_juju_conn
._get
_secret
_data
.return_value
= ("token", "cacert")
94 def test_with_cluster_uuid(
96 mock_get_default_storage_class
,
98 reuse_cluster_uuid
= "uuid"
99 uuid
, created
= self
.loop
.run_until_complete(
100 self
.k8s_juju_conn
.init_env(
101 k8s_creds
=kubeconfig
, reuse_cluster_uuid
=reuse_cluster_uuid
105 self
.assertTrue(created
)
106 self
.assertEqual(uuid
, reuse_cluster_uuid
)
107 mock_get_default_storage_class
.assert_called_once()
108 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
110 def test_with_no_cluster_uuid(
112 mock_get_default_storage_class
,
114 uuid
, created
= self
.loop
.run_until_complete(
115 self
.k8s_juju_conn
.init_env(k8s_creds
=kubeconfig
)
118 self
.assertTrue(created
)
119 self
.assertTrue(isinstance(uuid
, str))
120 mock_get_default_storage_class
.assert_called_once()
121 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
123 def test_init_env_exception(
125 mock_get_default_storage_class
,
127 self
.k8s_juju_conn
.libjuju
.add_k8s
.side_effect
= Exception()
130 with self
.assertRaises(Exception):
131 uuid
, created
= self
.loop
.run_until_complete(
132 self
.k8s_juju_conn
.init_env(k8s_creds
=kubeconfig
)
135 self
.assertIsNone(created
)
136 self
.assertIsNone(uuid
)
137 mock_get_default_storage_class
.assert_called_once()
138 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
141 class NotImplementedTest(K8sJujuConnTestCase
):
143 super(NotImplementedTest
, self
).setUp()
145 def test_repo_add(self
):
146 with self
.assertRaises(MethodNotImplemented
):
147 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_add("", ""))
149 def test_repo_list(self
):
150 with self
.assertRaises(MethodNotImplemented
):
151 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_list())
153 def test_repo_remove(self
):
154 with self
.assertRaises(MethodNotImplemented
):
155 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_remove(""))
157 def test_synchronize_repos(self
):
159 self
.loop
.run_until_complete(self
.k8s_juju_conn
.synchronize_repos("", ""))
162 def test_upgrade(self
):
163 with self
.assertRaises(MethodNotImplemented
):
164 self
.loop
.run_until_complete(self
.k8s_juju_conn
.upgrade("", ""))
166 def test_rollback(self
):
167 with self
.assertRaises(MethodNotImplemented
):
168 self
.loop
.run_until_complete(self
.k8s_juju_conn
.rollback("", ""))
170 def test_get_namespace(self
):
171 self
.assertIsNone(self
.k8s_juju_conn
.get_namespace(""))
173 def test_instances_list(self
):
174 res
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.instances_list(""))
175 self
.assertEqual(res
, [])
178 class ResetTest(K8sJujuConnTestCase
):
180 super(ResetTest
, self
).setUp()
181 self
.k8s_juju_conn
.libjuju
.remove_cloud
= AsyncMock()
182 self
.k8s_juju_conn
.libjuju
.get_cloud_credentials
= AsyncMock()
184 cloud_creds
.result
= {"attrs": {RBAC_LABEL_KEY_NAME
: "asd"}}
185 self
.k8s_juju_conn
.libjuju
.get_cloud_credentials
.return_value
= [cloud_creds
]
186 self
.k8s_juju_conn
._delete
_cluster
_role
_binding
= Mock()
187 self
.k8s_juju_conn
._delete
_service
_account
= Mock()
188 self
.k8s_juju_conn
._delete
_cluster
_role
= Mock()
189 self
.k8s_juju_conn
.get_credentials
= Mock()
190 self
.k8s_juju_conn
.get_credentials
.return_value
= kubeconfig
192 def test_success(self
):
193 removed
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.reset("uuid"))
194 self
.assertTrue(removed
)
195 self
.k8s_juju_conn
.libjuju
.remove_cloud
.assert_called_once()
197 def test_exception(self
):
199 self
.k8s_juju_conn
.libjuju
.remove_cloud
.side_effect
= Exception()
200 with self
.assertRaises(Exception):
201 removed
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.reset("uuid"))
202 self
.assertIsNone(removed
)
203 self
.k8s_juju_conn
.libjuju
.remove_cloud
.assert_called_once()
206 @asynctest.mock
.patch("os.chdir")
207 class InstallTest(K8sJujuConnTestCase
):
209 super(InstallTest
, self
).setUp()
210 self
.db_dict
= {"filter": {"_id": "id"}}
211 self
.local_bundle
= "bundle"
212 self
.cs_bundle
= "cs:bundle"
213 self
.http_bundle
= "https://example.com/bundle.yaml"
214 self
.kdu_name
= "kdu_name"
215 self
.cluster_uuid
= "cluster"
216 self
.kdu_instance
= "{}-{}".format(self
.kdu_name
, "id")
217 self
.k8s_juju_conn
.libjuju
.add_model
= AsyncMock()
218 self
.k8s_juju_conn
.libjuju
.deploy
= AsyncMock()
220 def test_success_local(self
, mock_chdir
):
221 self
.loop
.run_until_complete(
222 self
.k8s_juju_conn
.install(
227 kdu_name
=self
.kdu_name
,
228 db_dict
=self
.db_dict
,
232 self
.assertEqual(mock_chdir
.call_count
, 2)
233 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
234 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
235 "local:{}".format(self
.local_bundle
),
236 model_name
=self
.kdu_instance
,
241 def test_success_cs(self
, mock_chdir
):
242 self
.loop
.run_until_complete(
243 self
.k8s_juju_conn
.install(
248 kdu_name
=self
.kdu_name
,
249 db_dict
=self
.db_dict
,
253 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
254 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
256 model_name
=self
.kdu_instance
,
261 def test_success_http(self
, mock_chdir
):
262 self
.loop
.run_until_complete(
263 self
.k8s_juju_conn
.install(
268 kdu_name
=self
.kdu_name
,
269 db_dict
=self
.db_dict
,
273 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
274 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
276 model_name
=self
.kdu_instance
,
281 def test_success_not_kdu_name(self
, mock_chdir
):
282 self
.loop
.run_until_complete(
283 self
.k8s_juju_conn
.install(
288 db_dict
=self
.db_dict
,
292 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
293 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
295 model_name
=self
.kdu_instance
,
300 def test_missing_db_dict(self
, mock_chdir
):
302 with self
.assertRaises(K8sException
):
303 self
.loop
.run_until_complete(
304 self
.k8s_juju_conn
.install(
309 kdu_name
=self
.kdu_name
,
313 self
.assertIsNone(kdu_instance
)
314 self
.k8s_juju_conn
.libjuju
.add_model
.assert_not_called()
315 self
.k8s_juju_conn
.libjuju
.deploy
.assert_not_called()
317 @asynctest.mock
.patch("os.getcwd")
318 def test_getcwd_exception(self
, mock_getcwd
, mock_chdir
):
319 mock_getcwd
.side_effect
= FileNotFoundError()
320 self
.loop
.run_until_complete(
321 self
.k8s_juju_conn
.install(
326 kdu_name
=self
.kdu_name
,
327 db_dict
=self
.db_dict
,
331 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
332 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
334 model_name
=self
.kdu_instance
,
339 def test_missing_bundle(self
, mock_chdir
):
340 with self
.assertRaises(K8sException
):
341 self
.loop
.run_until_complete(
342 self
.k8s_juju_conn
.install(
347 kdu_name
=self
.kdu_name
,
349 db_dict
=self
.db_dict
,
352 self
.k8s_juju_conn
.libjuju
.add_model
.assert_not_called()
353 self
.k8s_juju_conn
.libjuju
.deploy
.assert_not_called()
355 def test_missing_exception(self
, mock_chdir
):
356 self
.k8s_juju_conn
.libjuju
.deploy
.side_effect
= Exception()
357 with self
.assertRaises(Exception):
358 self
.loop
.run_until_complete(
359 self
.k8s_juju_conn
.install(
364 kdu_name
=self
.kdu_name
,
365 db_dict
=self
.db_dict
,
369 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
370 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
371 "local:{}".format(self
.local_bundle
),
372 model_name
=self
.kdu_instance
,
378 class UninstallTest(K8sJujuConnTestCase
):
380 super(UninstallTest
, self
).setUp()
381 self
.k8s_juju_conn
.libjuju
.destroy_model
= AsyncMock()
383 def test_success(self
):
384 destroyed
= self
.loop
.run_until_complete(
385 self
.k8s_juju_conn
.uninstall("cluster_uuid", "model_name")
387 self
.assertTrue(destroyed
)
388 self
.k8s_juju_conn
.libjuju
.destroy_model
.assert_called_once()
390 def test_exception(self
):
392 self
.k8s_juju_conn
.libjuju
.destroy_model
.side_effect
= Exception()
393 with self
.assertRaises(Exception):
394 destroyed
= self
.loop
.run_until_complete(
395 self
.k8s_juju_conn
.uninstall("cluster_uuid", "model_name")
397 self
.assertIsNone(destroyed
)
398 self
.k8s_juju_conn
.libjuju
.destroy_model
.assert_called_once()
401 class ExecPrimitivesTest(K8sJujuConnTestCase
):
403 super(ExecPrimitivesTest
, self
).setUp()
404 self
.action_name
= "touch"
405 self
.application_name
= "myapp"
406 self
.model_name
= "model"
407 self
.k8s_juju_conn
.libjuju
.get_actions
= AsyncMock()
408 self
.k8s_juju_conn
.libjuju
.execute_action
= AsyncMock()
410 def test_success(self
):
411 params
= {"application-name": self
.application_name
}
412 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
413 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (
418 output
= self
.loop
.run_until_complete(
419 self
.k8s_juju_conn
.exec_primitive(
420 "cluster", self
.model_name
, self
.action_name
, params
=params
424 self
.assertEqual(output
, "success")
425 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
426 self
.application_name
, self
.model_name
428 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
429 self
.application_name
, self
.model_name
, self
.action_name
, **params
432 def test_exception(self
):
433 params
= {"application-name": self
.application_name
}
434 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
435 self
.k8s_juju_conn
.libjuju
.execute_action
.side_effect
= Exception()
438 with self
.assertRaises(Exception):
439 output
= self
.loop
.run_until_complete(
440 self
.k8s_juju_conn
.exec_primitive(
441 "cluster", self
.model_name
, self
.action_name
, params
=params
445 self
.assertIsNone(output
)
446 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
447 self
.application_name
, self
.model_name
449 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
450 self
.application_name
, self
.model_name
, self
.action_name
, **params
453 def test_missing_application_name_in_params(self
):
457 with self
.assertRaises(K8sException
):
458 output
= self
.loop
.run_until_complete(
459 self
.k8s_juju_conn
.exec_primitive(
460 "cluster", self
.model_name
, self
.action_name
, params
=params
464 self
.assertIsNone(output
)
465 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called()
466 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
468 def test_missing_params(self
):
470 with self
.assertRaises(K8sException
):
471 output
= self
.loop
.run_until_complete(
472 self
.k8s_juju_conn
.exec_primitive(
473 "cluster", self
.model_name
, self
.action_name
477 self
.assertIsNone(output
)
478 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called()
479 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
481 def test_missing_action(self
):
483 params
= {"application-name": self
.application_name
}
484 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
485 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (
489 with self
.assertRaises(K8sException
):
490 output
= self
.loop
.run_until_complete(
491 self
.k8s_juju_conn
.exec_primitive(
492 "cluster", self
.model_name
, "non-existing-action", params
=params
496 self
.assertIsNone(output
)
497 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
498 self
.application_name
, self
.model_name
500 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
502 def test_missing_not_completed(self
):
504 params
= {"application-name": self
.application_name
}
505 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
506 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (None, "failed")
507 with self
.assertRaises(K8sException
):
508 output
= self
.loop
.run_until_complete(
509 self
.k8s_juju_conn
.exec_primitive(
510 "cluster", self
.model_name
, self
.action_name
, params
=params
514 self
.assertIsNone(output
)
515 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
516 self
.application_name
, self
.model_name
518 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
519 self
.application_name
, self
.model_name
, self
.action_name
, **params
523 class InspectKduTest(K8sJujuConnTestCase
):
525 super(InspectKduTest
, self
).setUp()
527 @asynctest.mock
.patch("builtins.open")
528 @asynctest.mock
.patch("os.path.exists")
529 def test_existing_file(self
, mock_exists
, mock_open
):
530 mock_exists
.return_value
= True
532 'description': 'test bundle',
533 'bundle': 'kubernetes',
534 'applications': {'app':{ }, 'app2': { }}
536 mock_open
.return_value
= FakeFileWrapper(content
=content
)
537 kdu
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.inspect_kdu("model"))
538 self
.assertEqual(kdu
, {"app": {}, "app2": {}})
539 mock_exists
.assert_called_once()
540 mock_open
.assert_called_once()
542 @asynctest.mock
.patch("builtins.open")
543 @asynctest.mock
.patch("os.path.exists")
544 def test_not_existing_file(self
, mock_exists
, mock_open
):
546 mock_exists
.return_value
= False
547 with self
.assertRaises(K8sException
):
548 kdu
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.inspect_kdu("model"))
549 self
.assertEqual(kdu
, None)
550 mock_exists
.assert_called_once_with("model")
551 mock_open
.assert_not_called()
554 class HelpKduTest(K8sJujuConnTestCase
):
556 super(HelpKduTest
, self
).setUp()
558 @asynctest.mock
.patch("builtins.open")
559 @asynctest.mock
.patch("os.listdir")
560 def test_existing_file(self
, mock_listdir
, mock_open
):
561 content
= "Readme file content"
562 mock_open
.return_value
= FakeFileWrapper(content
=content
)
563 for file in ["README.md", "README.txt", "README"]:
564 mock_listdir
.return_value
= [file]
565 help = self
.loop
.run_until_complete(
566 self
.k8s_juju_conn
.help_kdu("kdu_instance")
568 self
.assertEqual(help, content
)
570 self
.assertEqual(mock_listdir
.call_count
, 3)
571 self
.assertEqual(mock_open
.call_count
, 3)
573 @asynctest.mock
.patch("builtins.open")
574 @asynctest.mock
.patch("os.listdir")
575 def test_not_existing_file(self
, mock_listdir
, mock_open
):
576 for file in ["src/charm.py", "tox.ini", "requirements.txt"]:
577 mock_listdir
.return_value
= [file]
578 help = self
.loop
.run_until_complete(
579 self
.k8s_juju_conn
.help_kdu("kdu_instance")
581 self
.assertEqual(help, None)
583 self
.assertEqual(mock_listdir
.call_count
, 3)
584 self
.assertEqual(mock_open
.call_count
, 0)
587 class StatusKduTest(K8sJujuConnTestCase
):
589 super(StatusKduTest
, self
).setUp()
590 self
.k8s_juju_conn
.libjuju
.get_model_status
= AsyncMock()
592 def test_success(self
):
593 applications
= {"app": {"status": {"status": "active"}}}
594 model
= FakeModel(applications
=applications
)
595 self
.k8s_juju_conn
.libjuju
.get_model_status
.return_value
= model
596 status
= self
.loop
.run_until_complete(
597 self
.k8s_juju_conn
.status_kdu("cluster", "kdu_instance")
599 self
.assertEqual(status
, {"app": {"status": "active"}})
600 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
602 def test_exception(self
):
603 self
.k8s_juju_conn
.libjuju
.get_model_status
.side_effect
= Exception()
605 with self
.assertRaises(Exception):
606 status
= self
.loop
.run_until_complete(
607 self
.k8s_juju_conn
.status_kdu("cluster", "kdu_instance")
609 self
.assertIsNone(status
)
610 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
613 class GetServicesTest(K8sJujuConnTestCase
):
615 super(GetServicesTest
, self
).setUp()
617 @asynctest.mock
.patch("n2vc.kubectl.Kubectl.get_services")
618 @asynctest.mock
.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
619 def test_success(self
, mock_get_credentials
, mock_get_services
):
620 mock_get_credentials
.return_value
= kubeconfig
621 self
.loop
.run_until_complete(self
.k8s_juju_conn
.get_services("", "", ""))
622 mock_get_credentials
.assert_called_once()
623 mock_get_services
.assert_called_once()
626 class GetServiceTest(K8sJujuConnTestCase
):
628 super(GetServiceTest
, self
).setUp()
630 @asynctest.mock
.patch("n2vc.kubectl.Kubectl.get_services")
631 @asynctest.mock
.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
632 def test_success(self
, mock_get_credentials
, mock_get_services
):
633 mock_get_credentials
.return_value
= kubeconfig
634 self
.loop
.run_until_complete(self
.k8s_juju_conn
.get_service("", "", ""))
635 mock_get_credentials
.assert_called_once()
636 mock_get_services
.assert_called_once()
639 class GetCredentialsTest(K8sJujuConnTestCase
):
641 super(GetCredentialsTest
, self
).setUp()
643 @asynctest.mock
.patch("yaml.safe_dump")
644 def test_success(self
, mock_safe_dump
):
645 self
.k8s_juju_conn
.db
.get_one
.return_value
= {
647 "credentials": "credentials",
648 "schema_version": "2",
650 self
.k8s_juju_conn
.get_credentials("cluster_uuid")
651 self
.k8s_juju_conn
.db
.get_one
.assert_called_once()
652 self
.k8s_juju_conn
.db
.encrypt_decrypt_fields
.assert_called_once()
653 mock_safe_dump
.assert_called_once()
656 class UpdateVcaStatusTest(K8sJujuConnTestCase
):
658 super(UpdateVcaStatusTest
, self
).setUp()
659 self
.vcaStatus
= {"model": {"applications": {"app": {"actions": {}}}}}
660 self
.kdu_name
= "kdu_name"
661 self
.kdu_instance
= "{}-{}".format(self
.kdu_name
, "id")
662 self
.k8s_juju_conn
.libjuju
.get_executed_actions
= AsyncMock()
663 self
.k8s_juju_conn
.libjuju
.get_actions
= AsyncMock()
664 self
.k8s_juju_conn
.libjuju
.get_application_configs
= AsyncMock()
666 def test_success(self
):
667 self
.loop
.run_until_complete(
668 self
.k8s_juju_conn
.update_vca_status(self
.vcaStatus
, self
.kdu_instance
)
670 self
.k8s_juju_conn
.libjuju
.get_executed_actions
.assert_called_once()
671 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once()
672 self
.k8s_juju_conn
.libjuju
.get_application_configs
.assert_called_once()
674 def test_exception(self
):
675 self
.k8s_juju_conn
.libjuju
.get_model
.return_value
= None
676 self
.k8s_juju_conn
.libjuju
.get_executed_actions
.side_effect
= Exception()
677 with self
.assertRaises(Exception):
678 self
.loop
.run_until_complete(
679 self
.k8s_juju_conn
.update_vca_status(self
.vcaStatus
, self
.kdu_instance
)
681 self
.k8s_juju_conn
.libjuju
.get_executed_actions
.assert_not_called()
682 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called_once()
683 self
.k8s_juju_conn
.libjuju
.get_application_configs
.assert_not_called_once()
686 class ScaleTest(K8sJujuConnTestCase
):
688 super(ScaleTest
, self
).setUp()
689 self
.application_name
= "app"
690 self
.kdu_name
= "kdu-instance"
692 self
.k8s_juju_conn
.libjuju
.scale_application
= AsyncMock()
694 def test_success(self
):
695 self
.loop
.run_until_complete(
696 self
.k8s_juju_conn
.scale(self
.kdu_name
, self
._scale
, self
.application_name
)
698 self
.k8s_juju_conn
.libjuju
.scale_application
.assert_called_once()
700 def test_exception(self
):
701 self
.k8s_juju_conn
.libjuju
.scale_application
.side_effect
= Exception()
702 with self
.assertRaises(Exception):
703 self
.loop
.run_until_complete(
704 self
.k8s_juju_conn
.scale(
705 self
.kdu_name
, self
._scale
, self
.application_name
708 self
.k8s_juju_conn
.libjuju
.scale_application
.assert_called_once()
711 class GetScaleCount(K8sJujuConnTestCase
):
713 super(GetScaleCount
, self
).setUp()
714 self
.k8s_juju_conn
.libjuju
.get_model_status
= AsyncMock()
716 def test_success(self
):
717 applications
= {"app": FakeApplication()}
718 model
= FakeModel(applications
=applications
)
719 self
.k8s_juju_conn
.libjuju
.get_model_status
.return_value
= model
720 status
= self
.loop
.run_until_complete(
721 self
.k8s_juju_conn
.get_scale_count("app", "kdu_instance")
723 self
.assertEqual(status
, 2)
724 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
726 def test_exception(self
):
727 self
.k8s_juju_conn
.libjuju
.get_model_status
.side_effect
= Exception()
729 with self
.assertRaises(Exception):
730 status
= self
.loop
.run_until_complete(
731 self
.k8s_juju_conn
.status_kdu("app", "kdu_instance")
733 self
.assertIsNone(status
)
734 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()