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 self
.kubectl
.get_secret_data
= AsyncMock()
81 self
.kubectl
.get_secret_data
.return_value
= ("token", "cacert")
82 self
.kubectl
.get_services
.return_value
= [{}]
83 self
.k8s_juju_conn
._get
_kubectl
= Mock()
84 self
.k8s_juju_conn
._get
_kubectl
.return_value
= self
.kubectl
87 class InitEnvTest(K8sJujuConnTestCase
):
89 super(InitEnvTest
, self
).setUp()
90 self
.k8s_juju_conn
.libjuju
.add_k8s
= AsyncMock()
92 def test_with_cluster_uuid(
95 reuse_cluster_uuid
= "uuid"
96 uuid
, created
= self
.loop
.run_until_complete(
97 self
.k8s_juju_conn
.init_env(
98 k8s_creds
=kubeconfig
, reuse_cluster_uuid
=reuse_cluster_uuid
102 self
.assertTrue(created
)
103 self
.assertEqual(uuid
, reuse_cluster_uuid
)
104 self
.kubectl
.get_default_storage_class
.assert_called_once()
105 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
107 def test_with_no_cluster_uuid(
110 uuid
, created
= self
.loop
.run_until_complete(
111 self
.k8s_juju_conn
.init_env(k8s_creds
=kubeconfig
)
114 self
.assertTrue(created
)
115 self
.assertTrue(isinstance(uuid
, str))
116 self
.kubectl
.get_default_storage_class
.assert_called_once()
117 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
119 def test_init_env_exception(
122 self
.k8s_juju_conn
.libjuju
.add_k8s
.side_effect
= Exception()
125 with self
.assertRaises(Exception):
126 uuid
, created
= self
.loop
.run_until_complete(
127 self
.k8s_juju_conn
.init_env(k8s_creds
=kubeconfig
)
130 self
.assertIsNone(created
)
131 self
.assertIsNone(uuid
)
132 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
135 class NotImplementedTest(K8sJujuConnTestCase
):
137 super(NotImplementedTest
, self
).setUp()
139 def test_repo_add(self
):
140 with self
.assertRaises(MethodNotImplemented
):
141 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_add("", ""))
143 def test_repo_list(self
):
144 with self
.assertRaises(MethodNotImplemented
):
145 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_list())
147 def test_repo_remove(self
):
148 with self
.assertRaises(MethodNotImplemented
):
149 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_remove(""))
151 def test_synchronize_repos(self
):
153 self
.loop
.run_until_complete(self
.k8s_juju_conn
.synchronize_repos("", ""))
156 def test_upgrade(self
):
157 with self
.assertRaises(MethodNotImplemented
):
158 self
.loop
.run_until_complete(self
.k8s_juju_conn
.upgrade("", ""))
160 def test_rollback(self
):
161 with self
.assertRaises(MethodNotImplemented
):
162 self
.loop
.run_until_complete(self
.k8s_juju_conn
.rollback("", ""))
164 def test_get_namespace(self
):
165 self
.assertIsNone(self
.k8s_juju_conn
.get_namespace(""))
167 def test_instances_list(self
):
168 res
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.instances_list(""))
169 self
.assertEqual(res
, [])
172 class ResetTest(K8sJujuConnTestCase
):
174 super(ResetTest
, self
).setUp()
175 self
.k8s_juju_conn
.libjuju
.remove_cloud
= AsyncMock()
176 self
.k8s_juju_conn
.libjuju
.get_cloud_credentials
= AsyncMock()
178 cloud_creds
.result
= {"attrs": {RBAC_LABEL_KEY_NAME
: "asd"}}
179 self
.k8s_juju_conn
.libjuju
.get_cloud_credentials
.return_value
= [cloud_creds
]
180 self
.k8s_juju_conn
.get_credentials
= Mock()
181 self
.k8s_juju_conn
.get_credentials
.return_value
= kubeconfig
183 def test_success(self
):
184 removed
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.reset("uuid"))
185 self
.assertTrue(removed
)
186 self
.k8s_juju_conn
.libjuju
.remove_cloud
.assert_called_once()
188 def test_exception(self
):
190 self
.k8s_juju_conn
.libjuju
.remove_cloud
.side_effect
= Exception()
191 with self
.assertRaises(Exception):
192 removed
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.reset("uuid"))
193 self
.assertIsNone(removed
)
194 self
.k8s_juju_conn
.libjuju
.remove_cloud
.assert_called_once()
197 @asynctest.mock
.patch("os.chdir")
198 class InstallTest(K8sJujuConnTestCase
):
200 super(InstallTest
, self
).setUp()
201 self
.db_dict
= {"filter": {"_id": "id"}}
202 self
.local_bundle
= "bundle"
203 self
.cs_bundle
= "cs:bundle"
204 self
.http_bundle
= "https://example.com/bundle.yaml"
205 self
.kdu_name
= "kdu_name"
206 self
.cluster_uuid
= "cluster"
207 self
.kdu_instance
= "{}-{}".format(self
.kdu_name
, "id")
208 self
.k8s_juju_conn
.libjuju
.add_model
= AsyncMock()
209 self
.k8s_juju_conn
.libjuju
.deploy
= AsyncMock()
211 def test_success_local(self
, mock_chdir
):
212 self
.loop
.run_until_complete(
213 self
.k8s_juju_conn
.install(
218 kdu_name
=self
.kdu_name
,
219 db_dict
=self
.db_dict
,
223 self
.assertEqual(mock_chdir
.call_count
, 2)
224 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
225 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
226 "local:{}".format(self
.local_bundle
),
227 model_name
=self
.kdu_instance
,
232 def test_success_cs(self
, mock_chdir
):
233 self
.loop
.run_until_complete(
234 self
.k8s_juju_conn
.install(
239 kdu_name
=self
.kdu_name
,
240 db_dict
=self
.db_dict
,
244 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
245 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
247 model_name
=self
.kdu_instance
,
252 def test_success_http(self
, mock_chdir
):
253 self
.loop
.run_until_complete(
254 self
.k8s_juju_conn
.install(
259 kdu_name
=self
.kdu_name
,
260 db_dict
=self
.db_dict
,
264 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
265 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
267 model_name
=self
.kdu_instance
,
272 def test_success_not_kdu_name(self
, mock_chdir
):
273 self
.loop
.run_until_complete(
274 self
.k8s_juju_conn
.install(
279 db_dict
=self
.db_dict
,
283 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
284 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
286 model_name
=self
.kdu_instance
,
291 def test_missing_db_dict(self
, mock_chdir
):
293 with self
.assertRaises(K8sException
):
294 self
.loop
.run_until_complete(
295 self
.k8s_juju_conn
.install(
300 kdu_name
=self
.kdu_name
,
304 self
.assertIsNone(kdu_instance
)
305 self
.k8s_juju_conn
.libjuju
.add_model
.assert_not_called()
306 self
.k8s_juju_conn
.libjuju
.deploy
.assert_not_called()
308 @asynctest.mock
.patch("os.getcwd")
309 def test_getcwd_exception(self
, mock_getcwd
, mock_chdir
):
310 mock_getcwd
.side_effect
= FileNotFoundError()
311 self
.loop
.run_until_complete(
312 self
.k8s_juju_conn
.install(
317 kdu_name
=self
.kdu_name
,
318 db_dict
=self
.db_dict
,
322 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
323 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
325 model_name
=self
.kdu_instance
,
330 def test_missing_bundle(self
, mock_chdir
):
331 with self
.assertRaises(K8sException
):
332 self
.loop
.run_until_complete(
333 self
.k8s_juju_conn
.install(
338 kdu_name
=self
.kdu_name
,
340 db_dict
=self
.db_dict
,
343 self
.k8s_juju_conn
.libjuju
.add_model
.assert_not_called()
344 self
.k8s_juju_conn
.libjuju
.deploy
.assert_not_called()
346 def test_missing_exception(self
, mock_chdir
):
347 self
.k8s_juju_conn
.libjuju
.deploy
.side_effect
= Exception()
348 with self
.assertRaises(Exception):
349 self
.loop
.run_until_complete(
350 self
.k8s_juju_conn
.install(
355 kdu_name
=self
.kdu_name
,
356 db_dict
=self
.db_dict
,
360 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
361 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
362 "local:{}".format(self
.local_bundle
),
363 model_name
=self
.kdu_instance
,
369 class UninstallTest(K8sJujuConnTestCase
):
371 super(UninstallTest
, self
).setUp()
372 self
.k8s_juju_conn
.libjuju
.destroy_model
= AsyncMock()
374 def test_success(self
):
375 destroyed
= self
.loop
.run_until_complete(
376 self
.k8s_juju_conn
.uninstall("cluster_uuid", "model_name")
378 self
.assertTrue(destroyed
)
379 self
.k8s_juju_conn
.libjuju
.destroy_model
.assert_called_once()
381 def test_exception(self
):
383 self
.k8s_juju_conn
.libjuju
.destroy_model
.side_effect
= Exception()
384 with self
.assertRaises(Exception):
385 destroyed
= self
.loop
.run_until_complete(
386 self
.k8s_juju_conn
.uninstall("cluster_uuid", "model_name")
388 self
.assertIsNone(destroyed
)
389 self
.k8s_juju_conn
.libjuju
.destroy_model
.assert_called_once()
392 class ExecPrimitivesTest(K8sJujuConnTestCase
):
394 super(ExecPrimitivesTest
, self
).setUp()
395 self
.action_name
= "touch"
396 self
.application_name
= "myapp"
397 self
.model_name
= "model"
398 self
.k8s_juju_conn
.libjuju
.get_actions
= AsyncMock()
399 self
.k8s_juju_conn
.libjuju
.execute_action
= AsyncMock()
401 def test_success(self
):
402 params
= {"application-name": self
.application_name
}
403 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
404 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (
409 output
= self
.loop
.run_until_complete(
410 self
.k8s_juju_conn
.exec_primitive(
411 "cluster", self
.model_name
, self
.action_name
, params
=params
415 self
.assertEqual(output
, "success")
416 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
417 self
.application_name
, self
.model_name
419 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
420 self
.application_name
, self
.model_name
, self
.action_name
, **params
423 def test_exception(self
):
424 params
= {"application-name": self
.application_name
}
425 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
426 self
.k8s_juju_conn
.libjuju
.execute_action
.side_effect
= Exception()
429 with self
.assertRaises(Exception):
430 output
= self
.loop
.run_until_complete(
431 self
.k8s_juju_conn
.exec_primitive(
432 "cluster", self
.model_name
, self
.action_name
, params
=params
436 self
.assertIsNone(output
)
437 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
438 self
.application_name
, self
.model_name
440 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
441 self
.application_name
, self
.model_name
, self
.action_name
, **params
444 def test_missing_application_name_in_params(self
):
448 with self
.assertRaises(K8sException
):
449 output
= self
.loop
.run_until_complete(
450 self
.k8s_juju_conn
.exec_primitive(
451 "cluster", self
.model_name
, self
.action_name
, params
=params
455 self
.assertIsNone(output
)
456 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called()
457 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
459 def test_missing_params(self
):
461 with self
.assertRaises(K8sException
):
462 output
= self
.loop
.run_until_complete(
463 self
.k8s_juju_conn
.exec_primitive(
464 "cluster", self
.model_name
, self
.action_name
468 self
.assertIsNone(output
)
469 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called()
470 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
472 def test_missing_action(self
):
474 params
= {"application-name": self
.application_name
}
475 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
476 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (
480 with self
.assertRaises(K8sException
):
481 output
= self
.loop
.run_until_complete(
482 self
.k8s_juju_conn
.exec_primitive(
483 "cluster", self
.model_name
, "non-existing-action", params
=params
487 self
.assertIsNone(output
)
488 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
489 self
.application_name
, self
.model_name
491 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
493 def test_missing_not_completed(self
):
495 params
= {"application-name": self
.application_name
}
496 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
497 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (None, "failed")
498 with self
.assertRaises(K8sException
):
499 output
= self
.loop
.run_until_complete(
500 self
.k8s_juju_conn
.exec_primitive(
501 "cluster", self
.model_name
, self
.action_name
, params
=params
505 self
.assertIsNone(output
)
506 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
507 self
.application_name
, self
.model_name
509 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
510 self
.application_name
, self
.model_name
, self
.action_name
, **params
514 class InspectKduTest(K8sJujuConnTestCase
):
516 super(InspectKduTest
, self
).setUp()
518 @asynctest.mock
.patch("builtins.open")
519 @asynctest.mock
.patch("os.path.exists")
520 def test_existing_file(self
, mock_exists
, mock_open
):
521 mock_exists
.return_value
= True
523 'description': 'test bundle',
524 'bundle': 'kubernetes',
525 'applications': {'app':{ }, 'app2': { }}
527 mock_open
.return_value
= FakeFileWrapper(content
=content
)
528 kdu
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.inspect_kdu("model"))
529 self
.assertEqual(kdu
, {"app": {}, "app2": {}})
530 mock_exists
.assert_called_once()
531 mock_open
.assert_called_once()
533 @asynctest.mock
.patch("builtins.open")
534 @asynctest.mock
.patch("os.path.exists")
535 def test_not_existing_file(self
, mock_exists
, mock_open
):
537 mock_exists
.return_value
= False
538 with self
.assertRaises(K8sException
):
539 kdu
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.inspect_kdu("model"))
540 self
.assertEqual(kdu
, None)
541 mock_exists
.assert_called_once_with("model")
542 mock_open
.assert_not_called()
545 class HelpKduTest(K8sJujuConnTestCase
):
547 super(HelpKduTest
, self
).setUp()
549 @asynctest.mock
.patch("builtins.open")
550 @asynctest.mock
.patch("os.listdir")
551 def test_existing_file(self
, mock_listdir
, mock_open
):
552 content
= "Readme file content"
553 mock_open
.return_value
= FakeFileWrapper(content
=content
)
554 for file in ["README.md", "README.txt", "README"]:
555 mock_listdir
.return_value
= [file]
556 help = self
.loop
.run_until_complete(
557 self
.k8s_juju_conn
.help_kdu("kdu_instance")
559 self
.assertEqual(help, content
)
561 self
.assertEqual(mock_listdir
.call_count
, 3)
562 self
.assertEqual(mock_open
.call_count
, 3)
564 @asynctest.mock
.patch("builtins.open")
565 @asynctest.mock
.patch("os.listdir")
566 def test_not_existing_file(self
, mock_listdir
, mock_open
):
567 for file in ["src/charm.py", "tox.ini", "requirements.txt"]:
568 mock_listdir
.return_value
= [file]
569 help = self
.loop
.run_until_complete(
570 self
.k8s_juju_conn
.help_kdu("kdu_instance")
572 self
.assertEqual(help, None)
574 self
.assertEqual(mock_listdir
.call_count
, 3)
575 self
.assertEqual(mock_open
.call_count
, 0)
578 class StatusKduTest(K8sJujuConnTestCase
):
580 super(StatusKduTest
, self
).setUp()
581 self
.k8s_juju_conn
.libjuju
.get_model_status
= AsyncMock()
583 def test_success(self
):
584 applications
= {"app": {"status": {"status": "active"}}}
585 model
= FakeModel(applications
=applications
)
586 self
.k8s_juju_conn
.libjuju
.get_model_status
.return_value
= model
587 status
= self
.loop
.run_until_complete(
588 self
.k8s_juju_conn
.status_kdu("cluster", "kdu_instance")
590 self
.assertEqual(status
, {"app": {"status": "active"}})
591 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
593 def test_exception(self
):
594 self
.k8s_juju_conn
.libjuju
.get_model_status
.side_effect
= Exception()
596 with self
.assertRaises(Exception):
597 status
= self
.loop
.run_until_complete(
598 self
.k8s_juju_conn
.status_kdu("cluster", "kdu_instance")
600 self
.assertIsNone(status
)
601 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
604 class GetServicesTest(K8sJujuConnTestCase
):
606 super(GetServicesTest
, self
).setUp()
608 @asynctest.mock
.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
609 def test_success(self
, mock_get_credentials
):
610 mock_get_credentials
.return_value
= kubeconfig
611 self
.loop
.run_until_complete(self
.k8s_juju_conn
.get_services("", "", ""))
612 mock_get_credentials
.assert_called_once()
613 self
.kubectl
.get_services
.assert_called_once()
616 class GetServiceTest(K8sJujuConnTestCase
):
618 super(GetServiceTest
, self
).setUp()
620 @asynctest.mock
.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
621 def test_success(self
, mock_get_credentials
):
622 mock_get_credentials
.return_value
= kubeconfig
623 self
.loop
.run_until_complete(self
.k8s_juju_conn
.get_service("", "", ""))
624 mock_get_credentials
.assert_called_once()
625 self
.kubectl
.get_services
.assert_called_once()
628 class GetCredentialsTest(K8sJujuConnTestCase
):
630 super(GetCredentialsTest
, self
).setUp()
632 @asynctest.mock
.patch("yaml.safe_dump")
633 def test_success(self
, mock_safe_dump
):
634 self
.k8s_juju_conn
.db
.get_one
.return_value
= {
636 "credentials": "credentials",
637 "schema_version": "2",
639 self
.k8s_juju_conn
.get_credentials("cluster_uuid")
640 self
.k8s_juju_conn
.db
.get_one
.assert_called_once()
641 self
.k8s_juju_conn
.db
.encrypt_decrypt_fields
.assert_called_once()
642 mock_safe_dump
.assert_called_once()
645 class UpdateVcaStatusTest(K8sJujuConnTestCase
):
647 super(UpdateVcaStatusTest
, self
).setUp()
648 self
.vcaStatus
= {"model": {"applications": {"app": {"actions": {}}}}}
649 self
.kdu_name
= "kdu_name"
650 self
.kdu_instance
= "{}-{}".format(self
.kdu_name
, "id")
651 self
.k8s_juju_conn
.libjuju
.get_executed_actions
= AsyncMock()
652 self
.k8s_juju_conn
.libjuju
.get_actions
= AsyncMock()
653 self
.k8s_juju_conn
.libjuju
.get_application_configs
= AsyncMock()
655 def test_success(self
):
656 self
.loop
.run_until_complete(
657 self
.k8s_juju_conn
.update_vca_status(self
.vcaStatus
, self
.kdu_instance
)
659 self
.k8s_juju_conn
.libjuju
.get_executed_actions
.assert_called_once()
660 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once()
661 self
.k8s_juju_conn
.libjuju
.get_application_configs
.assert_called_once()
663 def test_exception(self
):
664 self
.k8s_juju_conn
.libjuju
.get_model
.return_value
= None
665 self
.k8s_juju_conn
.libjuju
.get_executed_actions
.side_effect
= Exception()
666 with self
.assertRaises(Exception):
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_not_called()
671 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called_once()
672 self
.k8s_juju_conn
.libjuju
.get_application_configs
.assert_not_called_once()
675 class ScaleTest(K8sJujuConnTestCase
):
677 super(ScaleTest
, self
).setUp()
678 self
.application_name
= "app"
679 self
.kdu_name
= "kdu-instance"
681 self
.k8s_juju_conn
.libjuju
.scale_application
= AsyncMock()
683 def test_success(self
):
684 self
.loop
.run_until_complete(
685 self
.k8s_juju_conn
.scale(self
.kdu_name
, self
._scale
, self
.application_name
)
687 self
.k8s_juju_conn
.libjuju
.scale_application
.assert_called_once()
689 def test_exception(self
):
690 self
.k8s_juju_conn
.libjuju
.scale_application
.side_effect
= Exception()
691 with self
.assertRaises(Exception):
692 self
.loop
.run_until_complete(
693 self
.k8s_juju_conn
.scale(
694 self
.kdu_name
, self
._scale
, self
.application_name
697 self
.k8s_juju_conn
.libjuju
.scale_application
.assert_called_once()
700 class GetScaleCount(K8sJujuConnTestCase
):
702 super(GetScaleCount
, self
).setUp()
703 self
.k8s_juju_conn
.libjuju
.get_model_status
= AsyncMock()
705 def test_success(self
):
706 applications
= {"app": FakeApplication()}
707 model
= FakeModel(applications
=applications
)
708 self
.k8s_juju_conn
.libjuju
.get_model_status
.return_value
= model
709 status
= self
.loop
.run_until_complete(
710 self
.k8s_juju_conn
.get_scale_count("app", "kdu_instance")
712 self
.assertEqual(status
, 2)
713 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
715 def test_exception(self
):
716 self
.k8s_juju_conn
.libjuju
.get_model_status
.side_effect
= Exception()
718 with self
.assertRaises(Exception):
719 status
= self
.loop
.run_until_complete(
720 self
.k8s_juju_conn
.status_kdu("app", "kdu_instance")
722 self
.assertIsNone(status
)
723 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()