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
.kdu_name
= "kdu_name"
71 self
.kdu_instance
= "{}-{}".format(self
.kdu_name
, "id")
72 self
.default_namespace
= self
.kdu_instance
74 self
.k8s_juju_conn
= K8sJujuConnector(
81 self
.k8s_juju_conn
._store
.get_vca_id
.return_value
= None
82 self
.k8s_juju_conn
.libjuju
= Mock()
85 self
.kubectl
.get_secret_data
= AsyncMock()
86 self
.kubectl
.get_secret_data
.return_value
= ("token", "cacert")
87 self
.kubectl
.get_services
.return_value
= [{}]
88 self
.k8s_juju_conn
._get
_kubectl
= Mock()
89 self
.k8s_juju_conn
._get
_kubectl
.return_value
= self
.kubectl
90 self
.k8s_juju_conn
._obtain
_namespace
_from
_db
= Mock(
91 return_value
=self
.default_namespace
95 class InitEnvTest(K8sJujuConnTestCase
):
97 super(InitEnvTest
, self
).setUp()
98 self
.k8s_juju_conn
.libjuju
.add_k8s
= AsyncMock()
100 def test_with_cluster_uuid(
103 reuse_cluster_uuid
= "uuid"
104 uuid
, created
= self
.loop
.run_until_complete(
105 self
.k8s_juju_conn
.init_env(
106 k8s_creds
=kubeconfig
, reuse_cluster_uuid
=reuse_cluster_uuid
110 self
.assertTrue(created
)
111 self
.assertEqual(uuid
, reuse_cluster_uuid
)
112 self
.kubectl
.get_default_storage_class
.assert_called_once()
113 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
115 def test_with_no_cluster_uuid(
118 uuid
, created
= self
.loop
.run_until_complete(
119 self
.k8s_juju_conn
.init_env(k8s_creds
=kubeconfig
)
122 self
.assertTrue(created
)
123 self
.assertTrue(isinstance(uuid
, str))
124 self
.kubectl
.get_default_storage_class
.assert_called_once()
125 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
127 def test_init_env_exception(
130 self
.k8s_juju_conn
.libjuju
.add_k8s
.side_effect
= Exception()
133 with self
.assertRaises(Exception):
134 uuid
, created
= self
.loop
.run_until_complete(
135 self
.k8s_juju_conn
.init_env(k8s_creds
=kubeconfig
)
138 self
.assertIsNone(created
)
139 self
.assertIsNone(uuid
)
140 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
143 class NotImplementedTest(K8sJujuConnTestCase
):
145 super(NotImplementedTest
, self
).setUp()
147 def test_repo_add(self
):
148 with self
.assertRaises(MethodNotImplemented
):
149 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_add("", ""))
151 def test_repo_list(self
):
152 with self
.assertRaises(MethodNotImplemented
):
153 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_list())
155 def test_repo_remove(self
):
156 with self
.assertRaises(MethodNotImplemented
):
157 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_remove(""))
159 def test_synchronize_repos(self
):
161 self
.loop
.run_until_complete(self
.k8s_juju_conn
.synchronize_repos("", ""))
164 def test_upgrade(self
):
165 with self
.assertRaises(MethodNotImplemented
):
166 self
.loop
.run_until_complete(self
.k8s_juju_conn
.upgrade("", ""))
168 def test_rollback(self
):
169 with self
.assertRaises(MethodNotImplemented
):
170 self
.loop
.run_until_complete(self
.k8s_juju_conn
.rollback("", ""))
172 def test_get_namespace(self
):
173 self
.assertIsNone(self
.k8s_juju_conn
.get_namespace(""))
175 def test_instances_list(self
):
176 res
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.instances_list(""))
177 self
.assertEqual(res
, [])
180 class ResetTest(K8sJujuConnTestCase
):
182 super(ResetTest
, self
).setUp()
183 self
.k8s_juju_conn
.libjuju
.remove_cloud
= AsyncMock()
184 self
.k8s_juju_conn
.libjuju
.get_cloud_credentials
= AsyncMock()
186 cloud_creds
.result
= {"attrs": {RBAC_LABEL_KEY_NAME
: "asd"}}
187 self
.k8s_juju_conn
.libjuju
.get_cloud_credentials
.return_value
= [cloud_creds
]
188 self
.k8s_juju_conn
.get_credentials
= Mock()
189 self
.k8s_juju_conn
.get_credentials
.return_value
= kubeconfig
191 def test_success(self
):
192 removed
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.reset("uuid"))
193 self
.assertTrue(removed
)
194 self
.k8s_juju_conn
.libjuju
.remove_cloud
.assert_called_once()
196 def test_exception(self
):
198 self
.k8s_juju_conn
.libjuju
.remove_cloud
.side_effect
= Exception()
199 with self
.assertRaises(Exception):
200 removed
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.reset("uuid"))
201 self
.assertIsNone(removed
)
202 self
.k8s_juju_conn
.libjuju
.remove_cloud
.assert_called_once()
205 @asynctest.mock
.patch("os.chdir")
206 class InstallTest(K8sJujuConnTestCase
):
208 super(InstallTest
, self
).setUp()
209 self
.db_dict
= {"filter": {"_id": "id"}}
210 self
.local_bundle
= "bundle"
211 self
.cs_bundle
= "cs:bundle"
212 self
.http_bundle
= "https://example.com/bundle.yaml"
213 self
.cluster_uuid
= "cluster"
214 self
.k8s_juju_conn
.libjuju
.add_model
= AsyncMock()
215 self
.k8s_juju_conn
.libjuju
.deploy
= AsyncMock()
217 def test_success_local(self
, mock_chdir
):
218 self
.loop
.run_until_complete(
219 self
.k8s_juju_conn
.install(
224 kdu_name
=self
.kdu_name
,
225 db_dict
=self
.db_dict
,
229 self
.assertEqual(mock_chdir
.call_count
, 2)
230 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
231 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
232 "local:{}".format(self
.local_bundle
),
233 model_name
=self
.default_namespace
,
238 def test_success_cs(self
, mock_chdir
):
239 self
.loop
.run_until_complete(
240 self
.k8s_juju_conn
.install(
245 kdu_name
=self
.kdu_name
,
246 db_dict
=self
.db_dict
,
250 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
251 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
253 model_name
=self
.default_namespace
,
258 def test_success_http(self
, mock_chdir
):
259 self
.loop
.run_until_complete(
260 self
.k8s_juju_conn
.install(
265 kdu_name
=self
.kdu_name
,
266 db_dict
=self
.db_dict
,
270 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
271 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
273 model_name
=self
.default_namespace
,
278 def test_success_not_kdu_name(self
, mock_chdir
):
279 self
.loop
.run_until_complete(
280 self
.k8s_juju_conn
.install(
285 db_dict
=self
.db_dict
,
289 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
290 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
292 model_name
=self
.default_namespace
,
297 def test_missing_db_dict(self
, mock_chdir
):
299 with self
.assertRaises(K8sException
):
300 self
.loop
.run_until_complete(
301 self
.k8s_juju_conn
.install(
306 kdu_name
=self
.kdu_name
,
310 self
.assertIsNone(kdu_instance
)
311 self
.k8s_juju_conn
.libjuju
.add_model
.assert_not_called()
312 self
.k8s_juju_conn
.libjuju
.deploy
.assert_not_called()
314 @asynctest.mock
.patch("os.getcwd")
315 def test_getcwd_exception(self
, mock_getcwd
, mock_chdir
):
316 mock_getcwd
.side_effect
= FileNotFoundError()
317 self
.loop
.run_until_complete(
318 self
.k8s_juju_conn
.install(
323 kdu_name
=self
.kdu_name
,
324 db_dict
=self
.db_dict
,
328 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
329 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
331 model_name
=self
.default_namespace
,
336 def test_missing_bundle(self
, mock_chdir
):
337 with self
.assertRaises(K8sException
):
338 self
.loop
.run_until_complete(
339 self
.k8s_juju_conn
.install(
344 kdu_name
=self
.kdu_name
,
346 db_dict
=self
.db_dict
,
349 self
.k8s_juju_conn
.libjuju
.add_model
.assert_not_called()
350 self
.k8s_juju_conn
.libjuju
.deploy
.assert_not_called()
352 def test_missing_exception(self
, mock_chdir
):
353 self
.k8s_juju_conn
.libjuju
.deploy
.side_effect
= Exception()
354 with self
.assertRaises(Exception):
355 self
.loop
.run_until_complete(
356 self
.k8s_juju_conn
.install(
361 kdu_name
=self
.kdu_name
,
362 db_dict
=self
.db_dict
,
366 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
367 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
368 "local:{}".format(self
.local_bundle
),
369 model_name
=self
.default_namespace
,
375 class UninstallTest(K8sJujuConnTestCase
):
377 super(UninstallTest
, self
).setUp()
378 self
.k8s_juju_conn
.libjuju
.destroy_model
= AsyncMock()
380 def test_success(self
):
381 destroyed
= self
.loop
.run_until_complete(
382 self
.k8s_juju_conn
.uninstall("cluster_uuid", "model_name")
384 self
.assertTrue(destroyed
)
385 self
.k8s_juju_conn
.libjuju
.destroy_model
.assert_called_once()
387 def test_exception(self
):
389 self
.k8s_juju_conn
.libjuju
.destroy_model
.side_effect
= Exception()
390 with self
.assertRaises(Exception):
391 destroyed
= self
.loop
.run_until_complete(
392 self
.k8s_juju_conn
.uninstall("cluster_uuid", "model_name")
394 self
.assertIsNone(destroyed
)
395 self
.k8s_juju_conn
.libjuju
.destroy_model
.assert_called_once()
398 class ExecPrimitivesTest(K8sJujuConnTestCase
):
400 super(ExecPrimitivesTest
, self
).setUp()
401 self
.action_name
= "touch"
402 self
.application_name
= "myapp"
403 self
.k8s_juju_conn
.libjuju
.get_actions
= AsyncMock()
404 self
.k8s_juju_conn
.libjuju
.execute_action
= AsyncMock()
406 def test_success(self
):
407 params
= {"application-name": self
.application_name
}
408 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
409 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (
414 output
= self
.loop
.run_until_complete(
415 self
.k8s_juju_conn
.exec_primitive(
416 "cluster", self
.kdu_instance
, self
.action_name
, params
=params
420 self
.assertEqual(output
, "success")
421 self
.k8s_juju_conn
._obtain
_namespace
_from
_db
.assert_called_once_with(
422 kdu_instance
=self
.kdu_instance
424 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
425 application_name
=self
.application_name
, model_name
=self
.default_namespace
427 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
428 application_name
=self
.application_name
,
429 model_name
=self
.default_namespace
,
430 action_name
=self
.action_name
,
434 def test_exception(self
):
435 params
= {"application-name": self
.application_name
}
436 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
437 self
.k8s_juju_conn
.libjuju
.execute_action
.side_effect
= Exception()
440 with self
.assertRaises(Exception):
441 output
= self
.loop
.run_until_complete(
442 self
.k8s_juju_conn
.exec_primitive(
443 "cluster", self
.kdu_instance
, self
.action_name
, params
=params
447 self
.assertIsNone(output
)
448 self
.k8s_juju_conn
._obtain
_namespace
_from
_db
.assert_called_once_with(
449 kdu_instance
=self
.kdu_instance
451 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
452 application_name
=self
.application_name
, model_name
=self
.default_namespace
454 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
455 application_name
=self
.application_name
,
456 model_name
=self
.default_namespace
,
457 action_name
=self
.action_name
,
461 def test_missing_application_name_in_params(self
):
465 with self
.assertRaises(K8sException
):
466 output
= self
.loop
.run_until_complete(
467 self
.k8s_juju_conn
.exec_primitive(
468 "cluster", self
.kdu_instance
, self
.action_name
, params
=params
472 self
.assertIsNone(output
)
473 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called()
474 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
476 def test_missing_params(self
):
478 with self
.assertRaises(K8sException
):
479 output
= self
.loop
.run_until_complete(
480 self
.k8s_juju_conn
.exec_primitive(
481 "cluster", self
.kdu_instance
, self
.action_name
485 self
.assertIsNone(output
)
486 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called()
487 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
489 def test_missing_action(self
):
491 params
= {"application-name": self
.application_name
}
492 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
493 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (
497 with self
.assertRaises(K8sException
):
498 output
= self
.loop
.run_until_complete(
499 self
.k8s_juju_conn
.exec_primitive(
500 "cluster", self
.kdu_instance
, "non-existing-action", params
=params
504 self
.assertIsNone(output
)
505 self
.k8s_juju_conn
._obtain
_namespace
_from
_db
.assert_called_once_with(
506 kdu_instance
=self
.kdu_instance
508 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
509 application_name
=self
.application_name
, model_name
=self
.default_namespace
511 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
513 def test_missing_not_completed(self
):
515 params
= {"application-name": self
.application_name
}
516 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
517 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (None, "failed")
518 with self
.assertRaises(K8sException
):
519 output
= self
.loop
.run_until_complete(
520 self
.k8s_juju_conn
.exec_primitive(
521 "cluster", self
.kdu_instance
, self
.action_name
, params
=params
525 self
.assertIsNone(output
)
526 self
.k8s_juju_conn
._obtain
_namespace
_from
_db
.assert_called_once_with(
527 kdu_instance
=self
.kdu_instance
529 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
530 application_name
=self
.application_name
, model_name
=self
.default_namespace
532 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
533 application_name
=self
.application_name
,
534 model_name
=self
.default_namespace
,
535 action_name
=self
.action_name
,
540 class InspectKduTest(K8sJujuConnTestCase
):
542 super(InspectKduTest
, self
).setUp()
544 @asynctest.mock
.patch("builtins.open")
545 @asynctest.mock
.patch("os.path.exists")
546 def test_existing_file(self
, mock_exists
, mock_open
):
547 mock_exists
.return_value
= True
549 'description': 'test bundle',
550 'bundle': 'kubernetes',
551 'applications': {'app':{ }, 'app2': { }}
553 mock_open
.return_value
= FakeFileWrapper(content
=content
)
554 kdu
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.inspect_kdu("model"))
555 self
.assertEqual(kdu
, {"app": {}, "app2": {}})
556 mock_exists
.assert_called_once()
557 mock_open
.assert_called_once()
559 @asynctest.mock
.patch("builtins.open")
560 @asynctest.mock
.patch("os.path.exists")
561 def test_not_existing_file(self
, mock_exists
, mock_open
):
563 mock_exists
.return_value
= False
564 with self
.assertRaises(K8sException
):
565 kdu
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.inspect_kdu("model"))
566 self
.assertEqual(kdu
, None)
567 mock_exists
.assert_called_once_with("model")
568 mock_open
.assert_not_called()
571 class HelpKduTest(K8sJujuConnTestCase
):
573 super(HelpKduTest
, self
).setUp()
575 @asynctest.mock
.patch("builtins.open")
576 @asynctest.mock
.patch("os.listdir")
577 def test_existing_file(self
, mock_listdir
, mock_open
):
578 content
= "Readme file content"
579 mock_open
.return_value
= FakeFileWrapper(content
=content
)
580 for file in ["README.md", "README.txt", "README"]:
581 mock_listdir
.return_value
= [file]
582 help = self
.loop
.run_until_complete(
583 self
.k8s_juju_conn
.help_kdu("kdu_instance")
585 self
.assertEqual(help, content
)
587 self
.assertEqual(mock_listdir
.call_count
, 3)
588 self
.assertEqual(mock_open
.call_count
, 3)
590 @asynctest.mock
.patch("builtins.open")
591 @asynctest.mock
.patch("os.listdir")
592 def test_not_existing_file(self
, mock_listdir
, mock_open
):
593 for file in ["src/charm.py", "tox.ini", "requirements.txt"]:
594 mock_listdir
.return_value
= [file]
595 help = self
.loop
.run_until_complete(
596 self
.k8s_juju_conn
.help_kdu("kdu_instance")
598 self
.assertEqual(help, None)
600 self
.assertEqual(mock_listdir
.call_count
, 3)
601 self
.assertEqual(mock_open
.call_count
, 0)
604 class StatusKduTest(K8sJujuConnTestCase
):
606 super(StatusKduTest
, self
).setUp()
607 self
.k8s_juju_conn
.libjuju
.get_model_status
= AsyncMock()
609 def test_success(self
):
610 applications
= {"app": {"status": {"status": "active"}}}
611 model
= FakeModel(applications
=applications
)
612 self
.k8s_juju_conn
.libjuju
.get_model_status
.return_value
= model
613 status
= self
.loop
.run_until_complete(
614 self
.k8s_juju_conn
.status_kdu("cluster", "kdu_instance")
616 self
.assertEqual(status
, {"app": {"status": "active"}})
617 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
619 def test_exception(self
):
620 self
.k8s_juju_conn
.libjuju
.get_model_status
.side_effect
= Exception()
622 with self
.assertRaises(Exception):
623 status
= self
.loop
.run_until_complete(
624 self
.k8s_juju_conn
.status_kdu("cluster", "kdu_instance")
626 self
.assertIsNone(status
)
627 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
630 class GetServicesTest(K8sJujuConnTestCase
):
632 super(GetServicesTest
, self
).setUp()
634 @asynctest.mock
.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
635 def test_success(self
, mock_get_credentials
):
636 mock_get_credentials
.return_value
= kubeconfig
637 self
.loop
.run_until_complete(self
.k8s_juju_conn
.get_services("", "", ""))
638 mock_get_credentials
.assert_called_once()
639 self
.kubectl
.get_services
.assert_called_once()
642 class GetServiceTest(K8sJujuConnTestCase
):
644 super(GetServiceTest
, self
).setUp()
646 @asynctest.mock
.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
647 def test_success(self
, mock_get_credentials
):
648 mock_get_credentials
.return_value
= kubeconfig
649 self
.loop
.run_until_complete(self
.k8s_juju_conn
.get_service("", "", ""))
650 mock_get_credentials
.assert_called_once()
651 self
.kubectl
.get_services
.assert_called_once()
654 class GetCredentialsTest(K8sJujuConnTestCase
):
656 super(GetCredentialsTest
, self
).setUp()
658 @asynctest.mock
.patch("yaml.safe_dump")
659 def test_success(self
, mock_safe_dump
):
660 self
.k8s_juju_conn
.db
.get_one
.return_value
= {
662 "credentials": "credentials",
663 "schema_version": "2",
665 self
.k8s_juju_conn
.get_credentials("cluster_uuid")
666 self
.k8s_juju_conn
.db
.get_one
.assert_called_once()
667 self
.k8s_juju_conn
.db
.encrypt_decrypt_fields
.assert_called_once()
668 mock_safe_dump
.assert_called_once()
671 class UpdateVcaStatusTest(K8sJujuConnTestCase
):
673 super(UpdateVcaStatusTest
, self
).setUp()
674 self
.vcaStatus
= {"model": {"applications": {"app": {"actions": {}}}}}
675 self
.k8s_juju_conn
.libjuju
.get_executed_actions
= AsyncMock()
676 self
.k8s_juju_conn
.libjuju
.get_actions
= AsyncMock()
677 self
.k8s_juju_conn
.libjuju
.get_application_configs
= AsyncMock()
679 def test_success(self
):
680 self
.loop
.run_until_complete(
681 self
.k8s_juju_conn
.update_vca_status(self
.vcaStatus
, self
.kdu_instance
)
683 self
.k8s_juju_conn
.libjuju
.get_executed_actions
.assert_called_once()
684 self
.k8s_juju_conn
.libjuju
.get_application_configs
.assert_called_once()
686 def test_exception(self
):
687 self
.k8s_juju_conn
.libjuju
.get_model
.return_value
= None
688 self
.k8s_juju_conn
.libjuju
.get_executed_actions
.side_effect
= Exception()
689 with self
.assertRaises(Exception):
690 self
.loop
.run_until_complete(
691 self
.k8s_juju_conn
.update_vca_status(self
.vcaStatus
, self
.kdu_instance
)
693 self
.k8s_juju_conn
.libjuju
.get_executed_actions
.assert_not_called()
694 self
.k8s_juju_conn
.libjuju
.get_application_configs
.assert_not_called_once()
697 class ScaleTest(K8sJujuConnTestCase
):
699 super(ScaleTest
, self
).setUp()
700 self
.application_name
= "app"
701 self
.kdu_name
= "kdu-instance"
703 self
.k8s_juju_conn
.libjuju
.scale_application
= AsyncMock()
705 def test_success(self
):
706 self
.loop
.run_until_complete(
707 self
.k8s_juju_conn
.scale(self
.kdu_name
, self
._scale
, self
.application_name
)
709 self
.k8s_juju_conn
.libjuju
.scale_application
.assert_called_once()
711 def test_exception(self
):
712 self
.k8s_juju_conn
.libjuju
.scale_application
.side_effect
= Exception()
713 with self
.assertRaises(Exception):
714 self
.loop
.run_until_complete(
715 self
.k8s_juju_conn
.scale(
716 self
.kdu_name
, self
._scale
, self
.application_name
719 self
.k8s_juju_conn
.libjuju
.scale_application
.assert_called_once()
722 class GetScaleCount(K8sJujuConnTestCase
):
724 super(GetScaleCount
, self
).setUp()
725 self
.k8s_juju_conn
.libjuju
.get_model_status
= AsyncMock()
727 def test_success(self
):
728 applications
= {"app": FakeApplication()}
729 model
= FakeModel(applications
=applications
)
730 self
.k8s_juju_conn
.libjuju
.get_model_status
.return_value
= model
731 status
= self
.loop
.run_until_complete(
732 self
.k8s_juju_conn
.get_scale_count("app", "kdu_instance")
734 self
.assertEqual(status
, 2)
735 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
737 def test_exception(self
):
738 self
.k8s_juju_conn
.libjuju
.get_model_status
.side_effect
= Exception()
740 with self
.assertRaises(Exception):
741 status
= self
.loop
.run_until_complete(
742 self
.k8s_juju_conn
.status_kdu("app", "kdu_instance")
744 self
.assertIsNone(status
)
745 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
748 class AddRelationTest(K8sJujuConnTestCase
):
750 super(AddRelationTest
, self
).setUp()
751 self
.k8s_juju_conn
.libjuju
.add_relation
= AsyncMock()
752 self
.k8s_juju_conn
.libjuju
.offer
= AsyncMock()
753 self
.k8s_juju_conn
.libjuju
.get_controller
= AsyncMock()
754 self
.k8s_juju_conn
.libjuju
.consume
= AsyncMock()
756 def test_standard_relation(self
):
757 relation_endpoint_1
= RelationEndpoint("model-1.app1.0", None, "endpoint")
758 relation_endpoint_2
= RelationEndpoint("model-1.app2.1", None, "endpoint")
759 self
.loop
.run_until_complete(
760 self
.k8s_juju_conn
.add_relation(relation_endpoint_1
, relation_endpoint_2
)
762 self
.k8s_juju_conn
.libjuju
.add_relation
.assert_called_once_with(
763 model_name
="model-1", endpoint_1
="app1:endpoint", endpoint_2
="app2:endpoint"
765 self
.k8s_juju_conn
.libjuju
.offer
.assert_not_called()
766 self
.k8s_juju_conn
.libjuju
.consume
.assert_not_called()
768 def test_cmr_relation_same_controller(self
):
769 relation_endpoint_1
= RelationEndpoint("model-1.app1.0", None, "endpoint")
770 relation_endpoint_2
= RelationEndpoint("model-2.app2.1", None, "endpoint")
771 offer
= Offer("admin/model-1.app1")
772 self
.k8s_juju_conn
.libjuju
.offer
.return_value
= offer
773 self
.k8s_juju_conn
.libjuju
.consume
.return_value
= "saas"
774 self
.loop
.run_until_complete(
775 self
.k8s_juju_conn
.add_relation(relation_endpoint_1
, relation_endpoint_2
)
777 self
.k8s_juju_conn
.libjuju
.offer
.assert_called_once_with(relation_endpoint_1
)
778 self
.k8s_juju_conn
.libjuju
.consume
.assert_called_once()
779 self
.k8s_juju_conn
.libjuju
.add_relation
.assert_called_once_with(
780 "model-2", "app2:endpoint", "saas"
783 def test_relation_exception(self
):
784 relation_endpoint_1
= RelationEndpoint("model-1.app1.0", None, "endpoint")
785 relation_endpoint_2
= RelationEndpoint("model-2.app2.1", None, "endpoint")
786 self
.k8s_juju_conn
.libjuju
.offer
.side_effect
= Exception()
787 with self
.assertRaises(Exception):
788 self
.loop
.run_until_complete(
789 self
.k8s_juju_conn
.add_relation(
790 relation_endpoint_1
, relation_endpoint_2