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
.definitions
import Offer
, RelationEndpoint
21 from n2vc
.k8s_juju_conn
import K8sJujuConnector
, RBAC_LABEL_KEY_NAME
22 from osm_common
import fslocal
23 from .utils
import kubeconfig
, FakeModel
, FakeFileWrapper
, AsyncMock
, FakeApplication
24 from n2vc
.exceptions
import (
28 from n2vc
.vca
.connection_data
import ConnectionData
31 class K8sJujuConnTestCase(asynctest
.TestCase
):
32 @asynctest.mock
.patch("n2vc.k8s_juju_conn.Libjuju")
33 @asynctest.mock
.patch("n2vc.k8s_juju_conn.MotorStore")
34 @asynctest.mock
.patch("n2vc.k8s_juju_conn.get_connection")
35 @asynctest.mock
.patch("n2vc.vca.connection_data.base64_to_cacert")
38 mock_base64_to_cacert
=None,
39 mock_get_connection
=None,
43 self
.loop
= asyncio
.get_event_loop()
45 mock_base64_to_cacert
.return_value
= """
46 -----BEGIN CERTIFICATE-----
48 -----END CERTIFICATE-----"""
49 mock_libjuju
.return_value
= Mock()
50 mock_store
.return_value
= AsyncMock()
51 mock_vca_connection
= Mock()
52 mock_get_connection
.return_value
= mock_vca_connection
53 mock_vca_connection
.data
.return_value
= ConnectionData(
55 "endpoints": ["1.2.3.4:17070"],
61 "lxd-credentials": "credentials",
62 "k8s-cloud": "k8s_cloud",
63 "k8s-credentials": "k8s_credentials",
65 "api-proxy": "api_proxy",
68 logging
.disable(logging
.CRITICAL
)
70 self
.k8s_juju_conn
= K8sJujuConnector(
77 self
.k8s_juju_conn
._store
.get_vca_id
.return_value
= None
78 self
.k8s_juju_conn
.libjuju
= Mock()
81 self
.kubectl
.get_secret_data
= AsyncMock()
82 self
.kubectl
.get_secret_data
.return_value
= ("token", "cacert")
83 self
.kubectl
.get_services
.return_value
= [{}]
84 self
.k8s_juju_conn
._get
_kubectl
= Mock()
85 self
.k8s_juju_conn
._get
_kubectl
.return_value
= self
.kubectl
88 class InitEnvTest(K8sJujuConnTestCase
):
90 super(InitEnvTest
, self
).setUp()
91 self
.k8s_juju_conn
.libjuju
.add_k8s
= AsyncMock()
93 def test_with_cluster_uuid(
96 reuse_cluster_uuid
= "uuid"
97 uuid
, created
= self
.loop
.run_until_complete(
98 self
.k8s_juju_conn
.init_env(
99 k8s_creds
=kubeconfig
, reuse_cluster_uuid
=reuse_cluster_uuid
103 self
.assertTrue(created
)
104 self
.assertEqual(uuid
, reuse_cluster_uuid
)
105 self
.kubectl
.get_default_storage_class
.assert_called_once()
106 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
108 def test_with_no_cluster_uuid(
111 uuid
, created
= self
.loop
.run_until_complete(
112 self
.k8s_juju_conn
.init_env(k8s_creds
=kubeconfig
)
115 self
.assertTrue(created
)
116 self
.assertTrue(isinstance(uuid
, str))
117 self
.kubectl
.get_default_storage_class
.assert_called_once()
118 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
120 def test_init_env_exception(
123 self
.k8s_juju_conn
.libjuju
.add_k8s
.side_effect
= Exception()
126 with self
.assertRaises(Exception):
127 uuid
, created
= self
.loop
.run_until_complete(
128 self
.k8s_juju_conn
.init_env(k8s_creds
=kubeconfig
)
131 self
.assertIsNone(created
)
132 self
.assertIsNone(uuid
)
133 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
136 class NotImplementedTest(K8sJujuConnTestCase
):
138 super(NotImplementedTest
, self
).setUp()
140 def test_repo_add(self
):
141 with self
.assertRaises(MethodNotImplemented
):
142 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_add("", ""))
144 def test_repo_list(self
):
145 with self
.assertRaises(MethodNotImplemented
):
146 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_list())
148 def test_repo_remove(self
):
149 with self
.assertRaises(MethodNotImplemented
):
150 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_remove(""))
152 def test_synchronize_repos(self
):
154 self
.loop
.run_until_complete(self
.k8s_juju_conn
.synchronize_repos("", ""))
157 def test_upgrade(self
):
158 with self
.assertRaises(MethodNotImplemented
):
159 self
.loop
.run_until_complete(self
.k8s_juju_conn
.upgrade("", ""))
161 def test_rollback(self
):
162 with self
.assertRaises(MethodNotImplemented
):
163 self
.loop
.run_until_complete(self
.k8s_juju_conn
.rollback("", ""))
165 def test_get_namespace(self
):
166 self
.assertIsNone(self
.k8s_juju_conn
.get_namespace(""))
168 def test_instances_list(self
):
169 res
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.instances_list(""))
170 self
.assertEqual(res
, [])
173 class ResetTest(K8sJujuConnTestCase
):
175 super(ResetTest
, self
).setUp()
176 self
.k8s_juju_conn
.libjuju
.remove_cloud
= AsyncMock()
177 self
.k8s_juju_conn
.libjuju
.get_cloud_credentials
= AsyncMock()
179 cloud_creds
.result
= {"attrs": {RBAC_LABEL_KEY_NAME
: "asd"}}
180 self
.k8s_juju_conn
.libjuju
.get_cloud_credentials
.return_value
= [cloud_creds
]
181 self
.k8s_juju_conn
.get_credentials
= Mock()
182 self
.k8s_juju_conn
.get_credentials
.return_value
= kubeconfig
184 def test_success(self
):
185 removed
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.reset("uuid"))
186 self
.assertTrue(removed
)
187 self
.k8s_juju_conn
.libjuju
.remove_cloud
.assert_called_once()
189 def test_exception(self
):
191 self
.k8s_juju_conn
.libjuju
.remove_cloud
.side_effect
= Exception()
192 with self
.assertRaises(Exception):
193 removed
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.reset("uuid"))
194 self
.assertIsNone(removed
)
195 self
.k8s_juju_conn
.libjuju
.remove_cloud
.assert_called_once()
198 @asynctest.mock
.patch("os.chdir")
199 class InstallTest(K8sJujuConnTestCase
):
201 super(InstallTest
, self
).setUp()
202 self
.db_dict
= {"filter": {"_id": "id"}}
203 self
.local_bundle
= "bundle"
204 self
.cs_bundle
= "cs:bundle"
205 self
.http_bundle
= "https://example.com/bundle.yaml"
206 self
.kdu_name
= "kdu_name"
207 self
.cluster_uuid
= "cluster"
208 self
.kdu_instance
= "{}-{}".format(self
.kdu_name
, "id")
209 self
.k8s_juju_conn
.libjuju
.add_model
= AsyncMock()
210 self
.k8s_juju_conn
.libjuju
.deploy
= AsyncMock()
212 def test_success_local(self
, mock_chdir
):
213 self
.loop
.run_until_complete(
214 self
.k8s_juju_conn
.install(
219 kdu_name
=self
.kdu_name
,
220 db_dict
=self
.db_dict
,
224 self
.assertEqual(mock_chdir
.call_count
, 2)
225 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
226 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
227 "local:{}".format(self
.local_bundle
),
228 model_name
=self
.kdu_instance
,
233 def test_success_cs(self
, mock_chdir
):
234 self
.loop
.run_until_complete(
235 self
.k8s_juju_conn
.install(
240 kdu_name
=self
.kdu_name
,
241 db_dict
=self
.db_dict
,
245 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
246 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
248 model_name
=self
.kdu_instance
,
253 def test_success_http(self
, mock_chdir
):
254 self
.loop
.run_until_complete(
255 self
.k8s_juju_conn
.install(
260 kdu_name
=self
.kdu_name
,
261 db_dict
=self
.db_dict
,
265 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
266 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
268 model_name
=self
.kdu_instance
,
273 def test_success_not_kdu_name(self
, mock_chdir
):
274 self
.loop
.run_until_complete(
275 self
.k8s_juju_conn
.install(
280 db_dict
=self
.db_dict
,
284 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
285 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
287 model_name
=self
.kdu_instance
,
292 def test_missing_db_dict(self
, mock_chdir
):
294 with self
.assertRaises(K8sException
):
295 self
.loop
.run_until_complete(
296 self
.k8s_juju_conn
.install(
301 kdu_name
=self
.kdu_name
,
305 self
.assertIsNone(kdu_instance
)
306 self
.k8s_juju_conn
.libjuju
.add_model
.assert_not_called()
307 self
.k8s_juju_conn
.libjuju
.deploy
.assert_not_called()
309 @asynctest.mock
.patch("os.getcwd")
310 def test_getcwd_exception(self
, mock_getcwd
, mock_chdir
):
311 mock_getcwd
.side_effect
= FileNotFoundError()
312 self
.loop
.run_until_complete(
313 self
.k8s_juju_conn
.install(
318 kdu_name
=self
.kdu_name
,
319 db_dict
=self
.db_dict
,
323 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
324 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
326 model_name
=self
.kdu_instance
,
331 def test_missing_bundle(self
, mock_chdir
):
332 with self
.assertRaises(K8sException
):
333 self
.loop
.run_until_complete(
334 self
.k8s_juju_conn
.install(
339 kdu_name
=self
.kdu_name
,
341 db_dict
=self
.db_dict
,
344 self
.k8s_juju_conn
.libjuju
.add_model
.assert_not_called()
345 self
.k8s_juju_conn
.libjuju
.deploy
.assert_not_called()
347 def test_missing_exception(self
, mock_chdir
):
348 self
.k8s_juju_conn
.libjuju
.deploy
.side_effect
= Exception()
349 with self
.assertRaises(Exception):
350 self
.loop
.run_until_complete(
351 self
.k8s_juju_conn
.install(
356 kdu_name
=self
.kdu_name
,
357 db_dict
=self
.db_dict
,
361 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
362 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
363 "local:{}".format(self
.local_bundle
),
364 model_name
=self
.kdu_instance
,
370 class UninstallTest(K8sJujuConnTestCase
):
372 super(UninstallTest
, self
).setUp()
373 self
.k8s_juju_conn
.libjuju
.destroy_model
= AsyncMock()
375 def test_success(self
):
376 destroyed
= self
.loop
.run_until_complete(
377 self
.k8s_juju_conn
.uninstall("cluster_uuid", "model_name")
379 self
.assertTrue(destroyed
)
380 self
.k8s_juju_conn
.libjuju
.destroy_model
.assert_called_once()
382 def test_exception(self
):
384 self
.k8s_juju_conn
.libjuju
.destroy_model
.side_effect
= Exception()
385 with self
.assertRaises(Exception):
386 destroyed
= self
.loop
.run_until_complete(
387 self
.k8s_juju_conn
.uninstall("cluster_uuid", "model_name")
389 self
.assertIsNone(destroyed
)
390 self
.k8s_juju_conn
.libjuju
.destroy_model
.assert_called_once()
393 class ExecPrimitivesTest(K8sJujuConnTestCase
):
395 super(ExecPrimitivesTest
, self
).setUp()
396 self
.action_name
= "touch"
397 self
.application_name
= "myapp"
398 self
.model_name
= "model"
399 self
.k8s_juju_conn
.libjuju
.get_actions
= AsyncMock()
400 self
.k8s_juju_conn
.libjuju
.execute_action
= AsyncMock()
402 def test_success(self
):
403 params
= {"application-name": self
.application_name
}
404 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
405 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (
410 output
= self
.loop
.run_until_complete(
411 self
.k8s_juju_conn
.exec_primitive(
412 "cluster", self
.model_name
, self
.action_name
, params
=params
416 self
.assertEqual(output
, "success")
417 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
418 self
.application_name
, self
.model_name
420 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
421 self
.application_name
, self
.model_name
, self
.action_name
, **params
424 def test_exception(self
):
425 params
= {"application-name": self
.application_name
}
426 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
427 self
.k8s_juju_conn
.libjuju
.execute_action
.side_effect
= Exception()
430 with self
.assertRaises(Exception):
431 output
= self
.loop
.run_until_complete(
432 self
.k8s_juju_conn
.exec_primitive(
433 "cluster", self
.model_name
, self
.action_name
, params
=params
437 self
.assertIsNone(output
)
438 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
439 self
.application_name
, self
.model_name
441 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
442 self
.application_name
, self
.model_name
, self
.action_name
, **params
445 def test_missing_application_name_in_params(self
):
449 with self
.assertRaises(K8sException
):
450 output
= self
.loop
.run_until_complete(
451 self
.k8s_juju_conn
.exec_primitive(
452 "cluster", self
.model_name
, self
.action_name
, params
=params
456 self
.assertIsNone(output
)
457 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called()
458 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
460 def test_missing_params(self
):
462 with self
.assertRaises(K8sException
):
463 output
= self
.loop
.run_until_complete(
464 self
.k8s_juju_conn
.exec_primitive(
465 "cluster", self
.model_name
, self
.action_name
469 self
.assertIsNone(output
)
470 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called()
471 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
473 def test_missing_action(self
):
475 params
= {"application-name": self
.application_name
}
476 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
477 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (
481 with self
.assertRaises(K8sException
):
482 output
= self
.loop
.run_until_complete(
483 self
.k8s_juju_conn
.exec_primitive(
484 "cluster", self
.model_name
, "non-existing-action", params
=params
488 self
.assertIsNone(output
)
489 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
490 self
.application_name
, self
.model_name
492 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
494 def test_missing_not_completed(self
):
496 params
= {"application-name": self
.application_name
}
497 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
498 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (None, "failed")
499 with self
.assertRaises(K8sException
):
500 output
= self
.loop
.run_until_complete(
501 self
.k8s_juju_conn
.exec_primitive(
502 "cluster", self
.model_name
, self
.action_name
, params
=params
506 self
.assertIsNone(output
)
507 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
508 self
.application_name
, self
.model_name
510 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
511 self
.application_name
, self
.model_name
, self
.action_name
, **params
515 class InspectKduTest(K8sJujuConnTestCase
):
517 super(InspectKduTest
, self
).setUp()
519 @asynctest.mock
.patch("builtins.open")
520 @asynctest.mock
.patch("os.path.exists")
521 def test_existing_file(self
, mock_exists
, mock_open
):
522 mock_exists
.return_value
= True
524 'description': 'test bundle',
525 'bundle': 'kubernetes',
526 'applications': {'app':{ }, 'app2': { }}
528 mock_open
.return_value
= FakeFileWrapper(content
=content
)
529 kdu
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.inspect_kdu("model"))
530 self
.assertEqual(kdu
, {"app": {}, "app2": {}})
531 mock_exists
.assert_called_once()
532 mock_open
.assert_called_once()
534 @asynctest.mock
.patch("builtins.open")
535 @asynctest.mock
.patch("os.path.exists")
536 def test_not_existing_file(self
, mock_exists
, mock_open
):
538 mock_exists
.return_value
= False
539 with self
.assertRaises(K8sException
):
540 kdu
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.inspect_kdu("model"))
541 self
.assertEqual(kdu
, None)
542 mock_exists
.assert_called_once_with("model")
543 mock_open
.assert_not_called()
546 class HelpKduTest(K8sJujuConnTestCase
):
548 super(HelpKduTest
, self
).setUp()
550 @asynctest.mock
.patch("builtins.open")
551 @asynctest.mock
.patch("os.listdir")
552 def test_existing_file(self
, mock_listdir
, mock_open
):
553 content
= "Readme file content"
554 mock_open
.return_value
= FakeFileWrapper(content
=content
)
555 for file in ["README.md", "README.txt", "README"]:
556 mock_listdir
.return_value
= [file]
557 help = self
.loop
.run_until_complete(
558 self
.k8s_juju_conn
.help_kdu("kdu_instance")
560 self
.assertEqual(help, content
)
562 self
.assertEqual(mock_listdir
.call_count
, 3)
563 self
.assertEqual(mock_open
.call_count
, 3)
565 @asynctest.mock
.patch("builtins.open")
566 @asynctest.mock
.patch("os.listdir")
567 def test_not_existing_file(self
, mock_listdir
, mock_open
):
568 for file in ["src/charm.py", "tox.ini", "requirements.txt"]:
569 mock_listdir
.return_value
= [file]
570 help = self
.loop
.run_until_complete(
571 self
.k8s_juju_conn
.help_kdu("kdu_instance")
573 self
.assertEqual(help, None)
575 self
.assertEqual(mock_listdir
.call_count
, 3)
576 self
.assertEqual(mock_open
.call_count
, 0)
579 class StatusKduTest(K8sJujuConnTestCase
):
581 super(StatusKduTest
, self
).setUp()
582 self
.k8s_juju_conn
.libjuju
.get_model_status
= AsyncMock()
584 def test_success(self
):
585 applications
= {"app": {"status": {"status": "active"}}}
586 model
= FakeModel(applications
=applications
)
587 self
.k8s_juju_conn
.libjuju
.get_model_status
.return_value
= model
588 status
= self
.loop
.run_until_complete(
589 self
.k8s_juju_conn
.status_kdu("cluster", "kdu_instance")
591 self
.assertEqual(status
, {"app": {"status": "active"}})
592 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
594 def test_exception(self
):
595 self
.k8s_juju_conn
.libjuju
.get_model_status
.side_effect
= Exception()
597 with self
.assertRaises(Exception):
598 status
= self
.loop
.run_until_complete(
599 self
.k8s_juju_conn
.status_kdu("cluster", "kdu_instance")
601 self
.assertIsNone(status
)
602 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
605 class GetServicesTest(K8sJujuConnTestCase
):
607 super(GetServicesTest
, self
).setUp()
609 @asynctest.mock
.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
610 def test_success(self
, mock_get_credentials
):
611 mock_get_credentials
.return_value
= kubeconfig
612 self
.loop
.run_until_complete(self
.k8s_juju_conn
.get_services("", "", ""))
613 mock_get_credentials
.assert_called_once()
614 self
.kubectl
.get_services
.assert_called_once()
617 class GetServiceTest(K8sJujuConnTestCase
):
619 super(GetServiceTest
, self
).setUp()
621 @asynctest.mock
.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
622 def test_success(self
, mock_get_credentials
):
623 mock_get_credentials
.return_value
= kubeconfig
624 self
.loop
.run_until_complete(self
.k8s_juju_conn
.get_service("", "", ""))
625 mock_get_credentials
.assert_called_once()
626 self
.kubectl
.get_services
.assert_called_once()
629 class GetCredentialsTest(K8sJujuConnTestCase
):
631 super(GetCredentialsTest
, self
).setUp()
633 @asynctest.mock
.patch("yaml.safe_dump")
634 def test_success(self
, mock_safe_dump
):
635 self
.k8s_juju_conn
.db
.get_one
.return_value
= {
637 "credentials": "credentials",
638 "schema_version": "2",
640 self
.k8s_juju_conn
.get_credentials("cluster_uuid")
641 self
.k8s_juju_conn
.db
.get_one
.assert_called_once()
642 self
.k8s_juju_conn
.db
.encrypt_decrypt_fields
.assert_called_once()
643 mock_safe_dump
.assert_called_once()
646 class UpdateVcaStatusTest(K8sJujuConnTestCase
):
648 super(UpdateVcaStatusTest
, self
).setUp()
649 self
.vcaStatus
= {"model": {"applications": {"app": {"actions": {}}}}}
650 self
.kdu_name
= "kdu_name"
651 self
.kdu_instance
= "{}-{}".format(self
.kdu_name
, "id")
652 self
.k8s_juju_conn
.libjuju
.get_executed_actions
= AsyncMock()
653 self
.k8s_juju_conn
.libjuju
.get_actions
= AsyncMock()
654 self
.k8s_juju_conn
.libjuju
.get_application_configs
= AsyncMock()
656 def test_success(self
):
657 self
.loop
.run_until_complete(
658 self
.k8s_juju_conn
.update_vca_status(self
.vcaStatus
, self
.kdu_instance
)
660 self
.k8s_juju_conn
.libjuju
.get_executed_actions
.assert_called_once()
661 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once()
662 self
.k8s_juju_conn
.libjuju
.get_application_configs
.assert_called_once()
664 def test_exception(self
):
665 self
.k8s_juju_conn
.libjuju
.get_model
.return_value
= None
666 self
.k8s_juju_conn
.libjuju
.get_executed_actions
.side_effect
= Exception()
667 with self
.assertRaises(Exception):
668 self
.loop
.run_until_complete(
669 self
.k8s_juju_conn
.update_vca_status(self
.vcaStatus
, self
.kdu_instance
)
671 self
.k8s_juju_conn
.libjuju
.get_executed_actions
.assert_not_called()
672 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called_once()
673 self
.k8s_juju_conn
.libjuju
.get_application_configs
.assert_not_called_once()
676 class ScaleTest(K8sJujuConnTestCase
):
678 super(ScaleTest
, self
).setUp()
679 self
.application_name
= "app"
680 self
.kdu_name
= "kdu-instance"
682 self
.k8s_juju_conn
.libjuju
.scale_application
= AsyncMock()
684 def test_success(self
):
685 self
.loop
.run_until_complete(
686 self
.k8s_juju_conn
.scale(self
.kdu_name
, self
._scale
, self
.application_name
)
688 self
.k8s_juju_conn
.libjuju
.scale_application
.assert_called_once()
690 def test_exception(self
):
691 self
.k8s_juju_conn
.libjuju
.scale_application
.side_effect
= Exception()
692 with self
.assertRaises(Exception):
693 self
.loop
.run_until_complete(
694 self
.k8s_juju_conn
.scale(
695 self
.kdu_name
, self
._scale
, self
.application_name
698 self
.k8s_juju_conn
.libjuju
.scale_application
.assert_called_once()
701 class GetScaleCount(K8sJujuConnTestCase
):
703 super(GetScaleCount
, self
).setUp()
704 self
.k8s_juju_conn
.libjuju
.get_model_status
= AsyncMock()
706 def test_success(self
):
707 applications
= {"app": FakeApplication()}
708 model
= FakeModel(applications
=applications
)
709 self
.k8s_juju_conn
.libjuju
.get_model_status
.return_value
= model
710 status
= self
.loop
.run_until_complete(
711 self
.k8s_juju_conn
.get_scale_count("app", "kdu_instance")
713 self
.assertEqual(status
, 2)
714 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
716 def test_exception(self
):
717 self
.k8s_juju_conn
.libjuju
.get_model_status
.side_effect
= Exception()
719 with self
.assertRaises(Exception):
720 status
= self
.loop
.run_until_complete(
721 self
.k8s_juju_conn
.status_kdu("app", "kdu_instance")
723 self
.assertIsNone(status
)
724 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
727 class AddRelationTest(K8sJujuConnTestCase
):
729 super(AddRelationTest
, self
).setUp()
730 self
.k8s_juju_conn
.libjuju
.add_relation
= AsyncMock()
731 self
.k8s_juju_conn
.libjuju
.offer
= AsyncMock()
732 self
.k8s_juju_conn
.libjuju
.get_controller
= AsyncMock()
733 self
.k8s_juju_conn
.libjuju
.consume
= AsyncMock()
735 def test_standard_relation(self
):
736 relation_endpoint_1
= RelationEndpoint("model-1.app1.0", None, "endpoint")
737 relation_endpoint_2
= RelationEndpoint("model-1.app2.1", None, "endpoint")
738 self
.loop
.run_until_complete(
739 self
.k8s_juju_conn
.add_relation(relation_endpoint_1
, relation_endpoint_2
)
741 self
.k8s_juju_conn
.libjuju
.add_relation
.assert_called_once_with(
742 model_name
="model-1", endpoint_1
="app1:endpoint", endpoint_2
="app2:endpoint"
744 self
.k8s_juju_conn
.libjuju
.offer
.assert_not_called()
745 self
.k8s_juju_conn
.libjuju
.consume
.assert_not_called()
747 def test_cmr_relation_same_controller(self
):
748 relation_endpoint_1
= RelationEndpoint("model-1.app1.0", None, "endpoint")
749 relation_endpoint_2
= RelationEndpoint("model-2.app2.1", None, "endpoint")
750 offer
= Offer("admin/model-1.app1")
751 self
.k8s_juju_conn
.libjuju
.offer
.return_value
= offer
752 self
.k8s_juju_conn
.libjuju
.consume
.return_value
= "saas"
753 self
.loop
.run_until_complete(
754 self
.k8s_juju_conn
.add_relation(relation_endpoint_1
, relation_endpoint_2
)
756 self
.k8s_juju_conn
.libjuju
.offer
.assert_called_once_with(relation_endpoint_1
)
757 self
.k8s_juju_conn
.libjuju
.consume
.assert_called_once()
758 self
.k8s_juju_conn
.libjuju
.add_relation
.assert_called_once_with(
759 "model-2", "app2:endpoint", "saas"
762 def test_relation_exception(self
):
763 relation_endpoint_1
= RelationEndpoint("model-1.app1.0", None, "endpoint")
764 relation_endpoint_2
= RelationEndpoint("model-2.app2.1", None, "endpoint")
765 self
.k8s_juju_conn
.libjuju
.offer
.side_effect
= Exception()
766 with self
.assertRaises(Exception):
767 self
.loop
.run_until_complete(
768 self
.k8s_juju_conn
.add_relation(
769 relation_endpoint_1
, relation_endpoint_2