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 MethodNotImplemented
, K8sException
25 from n2vc
.vca
.connection_data
import ConnectionData
28 class K8sJujuConnTestCase(asynctest
.TestCase
):
29 @asynctest.mock
.patch("n2vc.k8s_juju_conn.Libjuju")
30 @asynctest.mock
.patch("n2vc.k8s_juju_conn.MotorStore")
31 @asynctest.mock
.patch("n2vc.k8s_juju_conn.get_connection")
32 @asynctest.mock
.patch("n2vc.vca.connection_data.base64_to_cacert")
35 mock_base64_to_cacert
=None,
36 mock_get_connection
=None,
40 self
.loop
= asyncio
.get_event_loop()
42 mock_base64_to_cacert
.return_value
= """
43 -----BEGIN CERTIFICATE-----
45 -----END CERTIFICATE-----"""
46 mock_libjuju
.return_value
= Mock()
47 mock_store
.return_value
= AsyncMock()
48 mock_vca_connection
= Mock()
49 mock_get_connection
.return_value
= mock_vca_connection
50 mock_vca_connection
.data
.return_value
= ConnectionData(
52 "endpoints": ["1.2.3.4:17070"],
58 "lxd-credentials": "credentials",
59 "k8s-cloud": "k8s_cloud",
60 "k8s-credentials": "k8s_credentials",
62 "api-proxy": "api_proxy",
65 logging
.disable(logging
.CRITICAL
)
67 self
.kdu_name
= "kdu_name"
68 self
.kdu_instance
= "{}-{}".format(self
.kdu_name
, "id")
69 self
.default_namespace
= self
.kdu_instance
71 self
.k8s_juju_conn
= K8sJujuConnector(
78 self
.k8s_juju_conn
._store
.get_vca_id
.return_value
= None
79 self
.k8s_juju_conn
.libjuju
= Mock()
82 self
.kubectl
.get_secret_data
= AsyncMock()
83 self
.kubectl
.get_secret_data
.return_value
= ("token", "cacert")
84 self
.kubectl
.get_services
.return_value
= [{}]
85 self
.k8s_juju_conn
._get
_kubectl
= Mock()
86 self
.k8s_juju_conn
._get
_kubectl
.return_value
= self
.kubectl
87 self
.k8s_juju_conn
._obtain
_namespace
_from
_db
= Mock(
88 return_value
=self
.default_namespace
92 class InitEnvTest(K8sJujuConnTestCase
):
94 super(InitEnvTest
, self
).setUp()
95 self
.k8s_juju_conn
.libjuju
.add_k8s
= AsyncMock()
97 def test_with_cluster_uuid(
100 reuse_cluster_uuid
= "uuid"
101 uuid
, created
= self
.loop
.run_until_complete(
102 self
.k8s_juju_conn
.init_env(
103 k8s_creds
=kubeconfig
, reuse_cluster_uuid
=reuse_cluster_uuid
107 self
.assertTrue(created
)
108 self
.assertEqual(uuid
, reuse_cluster_uuid
)
109 self
.kubectl
.get_default_storage_class
.assert_called_once()
110 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
112 def test_with_no_cluster_uuid(
115 uuid
, created
= self
.loop
.run_until_complete(
116 self
.k8s_juju_conn
.init_env(k8s_creds
=kubeconfig
)
119 self
.assertTrue(created
)
120 self
.assertTrue(isinstance(uuid
, str))
121 self
.kubectl
.get_default_storage_class
.assert_called_once()
122 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
124 def test_init_env_exception(
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
)
134 self
.assertIsNone(created
)
135 self
.assertIsNone(uuid
)
136 self
.kubectl
.create_cluster_role
.assert_called_once()
137 self
.kubectl
.create_service_account
.assert_called_once()
138 self
.kubectl
.create_cluster_role_binding
.assert_called_once()
139 self
.kubectl
.get_default_storage_class
.assert_called_once()
140 self
.kubectl
.delete_cluster_role
.assert_called_once()
141 self
.kubectl
.delete_service_account
.assert_called_once()
142 self
.kubectl
.delete_cluster_role_binding
.assert_called_once()
143 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
146 class NotImplementedTest(K8sJujuConnTestCase
):
148 super(NotImplementedTest
, self
).setUp()
150 def test_repo_add(self
):
151 with self
.assertRaises(MethodNotImplemented
):
152 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_add("", ""))
154 def test_repo_list(self
):
155 with self
.assertRaises(MethodNotImplemented
):
156 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_list())
158 def test_repo_remove(self
):
159 with self
.assertRaises(MethodNotImplemented
):
160 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_remove(""))
162 def test_synchronize_repos(self
):
164 self
.loop
.run_until_complete(self
.k8s_juju_conn
.synchronize_repos("", ""))
167 def test_upgrade(self
):
168 with self
.assertRaises(MethodNotImplemented
):
169 self
.loop
.run_until_complete(self
.k8s_juju_conn
.upgrade("", ""))
171 def test_rollback(self
):
172 with self
.assertRaises(MethodNotImplemented
):
173 self
.loop
.run_until_complete(self
.k8s_juju_conn
.rollback("", ""))
175 def test_get_namespace(self
):
176 self
.assertIsNone(self
.k8s_juju_conn
.get_namespace(""))
178 def test_instances_list(self
):
179 res
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.instances_list(""))
180 self
.assertEqual(res
, [])
183 class ResetTest(K8sJujuConnTestCase
):
185 super(ResetTest
, self
).setUp()
186 self
.k8s_juju_conn
.libjuju
.remove_cloud
= AsyncMock()
187 self
.k8s_juju_conn
.libjuju
.get_cloud_credentials
= AsyncMock()
189 cloud_creds
.result
= {"attrs": {RBAC_LABEL_KEY_NAME
: "asd"}}
190 self
.k8s_juju_conn
.libjuju
.get_cloud_credentials
.return_value
= [cloud_creds
]
191 self
.k8s_juju_conn
.get_credentials
= Mock()
192 self
.k8s_juju_conn
.get_credentials
.return_value
= kubeconfig
194 def test_success(self
):
195 removed
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.reset("uuid"))
196 self
.assertTrue(removed
)
197 self
.k8s_juju_conn
.libjuju
.remove_cloud
.assert_called_once()
199 def test_exception(self
):
201 self
.k8s_juju_conn
.libjuju
.remove_cloud
.side_effect
= Exception()
202 with self
.assertRaises(Exception):
203 removed
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.reset("uuid"))
204 self
.assertIsNone(removed
)
205 self
.k8s_juju_conn
.libjuju
.remove_cloud
.assert_called_once()
208 @asynctest.mock
.patch("os.chdir")
209 class InstallTest(K8sJujuConnTestCase
):
211 super(InstallTest
, self
).setUp()
212 self
.db_dict
= {"filter": {"_id": "id"}}
213 self
.local_bundle
= "bundle"
214 self
.cs_bundle
= "cs:bundle"
215 self
.http_bundle
= "https://example.com/bundle.yaml"
216 self
.cluster_uuid
= "cluster"
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
,
233 self
.assertEqual(mock_chdir
.call_count
, 2)
234 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
235 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
236 "local:{}".format(self
.local_bundle
),
237 model_name
=self
.default_namespace
,
240 instantiation_params
=None,
243 def test_success_cs(self
, mock_chdir
):
244 self
.loop
.run_until_complete(
245 self
.k8s_juju_conn
.install(
250 kdu_name
=self
.kdu_name
,
251 db_dict
=self
.db_dict
,
256 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
257 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
259 model_name
=self
.default_namespace
,
262 instantiation_params
=None,
265 def test_success_http(self
, mock_chdir
):
266 params
= {"overlay": {"applications": {"squid": {"scale": 2}}}}
267 self
.loop
.run_until_complete(
268 self
.k8s_juju_conn
.install(
273 kdu_name
=self
.kdu_name
,
274 db_dict
=self
.db_dict
,
279 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
280 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
282 model_name
=self
.default_namespace
,
285 instantiation_params
=params
.get("overlay"),
288 def test_success_not_kdu_name(self
, mock_chdir
):
289 params
= {"some_key": {"applications": {"squid": {"scale": 2}}}}
290 self
.loop
.run_until_complete(
291 self
.k8s_juju_conn
.install(
296 db_dict
=self
.db_dict
,
301 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
302 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
304 model_name
=self
.default_namespace
,
307 instantiation_params
=None,
310 def test_missing_db_dict(self
, mock_chdir
):
312 with self
.assertRaises(K8sException
):
313 self
.loop
.run_until_complete(
314 self
.k8s_juju_conn
.install(
319 kdu_name
=self
.kdu_name
,
323 self
.assertIsNone(kdu_instance
)
324 self
.k8s_juju_conn
.libjuju
.add_model
.assert_not_called()
325 self
.k8s_juju_conn
.libjuju
.deploy
.assert_not_called()
327 @asynctest.mock
.patch("os.getcwd")
328 def test_getcwd_exception(self
, mock_getcwd
, mock_chdir
):
329 mock_getcwd
.side_effect
= FileNotFoundError()
330 self
.loop
.run_until_complete(
331 self
.k8s_juju_conn
.install(
336 kdu_name
=self
.kdu_name
,
337 db_dict
=self
.db_dict
,
341 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
342 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
344 model_name
=self
.default_namespace
,
347 instantiation_params
=None,
350 def test_missing_bundle(self
, mock_chdir
):
351 with self
.assertRaises(K8sException
):
352 self
.loop
.run_until_complete(
353 self
.k8s_juju_conn
.install(
358 kdu_name
=self
.kdu_name
,
360 db_dict
=self
.db_dict
,
363 self
.k8s_juju_conn
.libjuju
.add_model
.assert_not_called()
364 self
.k8s_juju_conn
.libjuju
.deploy
.assert_not_called()
366 def test_missing_exception(self
, mock_chdir
):
367 self
.k8s_juju_conn
.libjuju
.deploy
.side_effect
= Exception()
368 with self
.assertRaises(Exception):
369 self
.loop
.run_until_complete(
370 self
.k8s_juju_conn
.install(
375 kdu_name
=self
.kdu_name
,
376 db_dict
=self
.db_dict
,
380 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
381 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
382 "local:{}".format(self
.local_bundle
),
383 model_name
=self
.default_namespace
,
386 instantiation_params
=None,
390 class UninstallTest(K8sJujuConnTestCase
):
392 super(UninstallTest
, self
).setUp()
393 self
.k8s_juju_conn
.libjuju
.destroy_model
= AsyncMock()
395 def test_success(self
):
396 destroyed
= self
.loop
.run_until_complete(
397 self
.k8s_juju_conn
.uninstall("cluster_uuid", "model_name")
399 self
.assertTrue(destroyed
)
400 self
.k8s_juju_conn
.libjuju
.destroy_model
.assert_called_once()
402 def test_exception(self
):
404 self
.k8s_juju_conn
.libjuju
.destroy_model
.side_effect
= Exception()
405 with self
.assertRaises(Exception):
406 destroyed
= self
.loop
.run_until_complete(
407 self
.k8s_juju_conn
.uninstall("cluster_uuid", "model_name")
409 self
.assertIsNone(destroyed
)
410 self
.k8s_juju_conn
.libjuju
.destroy_model
.assert_called_once()
413 class ExecPrimitivesTest(K8sJujuConnTestCase
):
415 super(ExecPrimitivesTest
, self
).setUp()
416 self
.action_name
= "touch"
417 self
.application_name
= "myapp"
418 self
.k8s_juju_conn
.libjuju
.get_actions
= AsyncMock()
419 self
.k8s_juju_conn
.libjuju
.execute_action
= AsyncMock()
421 def test_success(self
):
422 params
= {"application-name": self
.application_name
}
423 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
424 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (
429 output
= self
.loop
.run_until_complete(
430 self
.k8s_juju_conn
.exec_primitive(
431 "cluster", self
.kdu_instance
, self
.action_name
, params
=params
435 self
.assertEqual(output
, "success")
436 self
.k8s_juju_conn
._obtain
_namespace
_from
_db
.assert_called_once_with(
437 kdu_instance
=self
.kdu_instance
439 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
440 application_name
=self
.application_name
, model_name
=self
.default_namespace
442 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
443 application_name
=self
.application_name
,
444 model_name
=self
.default_namespace
,
445 action_name
=self
.action_name
,
449 def test_exception(self
):
450 params
= {"application-name": self
.application_name
}
451 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
452 self
.k8s_juju_conn
.libjuju
.execute_action
.side_effect
= Exception()
455 with self
.assertRaises(Exception):
456 output
= self
.loop
.run_until_complete(
457 self
.k8s_juju_conn
.exec_primitive(
458 "cluster", self
.kdu_instance
, self
.action_name
, params
=params
462 self
.assertIsNone(output
)
463 self
.k8s_juju_conn
._obtain
_namespace
_from
_db
.assert_called_once_with(
464 kdu_instance
=self
.kdu_instance
466 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
467 application_name
=self
.application_name
, model_name
=self
.default_namespace
469 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
470 application_name
=self
.application_name
,
471 model_name
=self
.default_namespace
,
472 action_name
=self
.action_name
,
476 def test_missing_application_name_in_params(self
):
480 with self
.assertRaises(K8sException
):
481 output
= self
.loop
.run_until_complete(
482 self
.k8s_juju_conn
.exec_primitive(
483 "cluster", self
.kdu_instance
, self
.action_name
, params
=params
487 self
.assertIsNone(output
)
488 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called()
489 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
491 def test_missing_params(self
):
493 with self
.assertRaises(K8sException
):
494 output
= self
.loop
.run_until_complete(
495 self
.k8s_juju_conn
.exec_primitive(
496 "cluster", self
.kdu_instance
, self
.action_name
500 self
.assertIsNone(output
)
501 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called()
502 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
504 def test_missing_action(self
):
506 params
= {"application-name": self
.application_name
}
507 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
508 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (
512 with self
.assertRaises(K8sException
):
513 output
= self
.loop
.run_until_complete(
514 self
.k8s_juju_conn
.exec_primitive(
515 "cluster", self
.kdu_instance
, "non-existing-action", params
=params
519 self
.assertIsNone(output
)
520 self
.k8s_juju_conn
._obtain
_namespace
_from
_db
.assert_called_once_with(
521 kdu_instance
=self
.kdu_instance
523 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
524 application_name
=self
.application_name
, model_name
=self
.default_namespace
526 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
528 def test_missing_not_completed(self
):
530 params
= {"application-name": self
.application_name
}
531 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
532 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (None, "failed")
533 with self
.assertRaises(K8sException
):
534 output
= self
.loop
.run_until_complete(
535 self
.k8s_juju_conn
.exec_primitive(
536 "cluster", self
.kdu_instance
, self
.action_name
, params
=params
540 self
.assertIsNone(output
)
541 self
.k8s_juju_conn
._obtain
_namespace
_from
_db
.assert_called_once_with(
542 kdu_instance
=self
.kdu_instance
544 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
545 application_name
=self
.application_name
, model_name
=self
.default_namespace
547 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
548 application_name
=self
.application_name
,
549 model_name
=self
.default_namespace
,
550 action_name
=self
.action_name
,
555 class InspectKduTest(K8sJujuConnTestCase
):
557 super(InspectKduTest
, self
).setUp()
559 @asynctest.mock
.patch("builtins.open")
560 @asynctest.mock
.patch("os.path.exists")
561 def test_existing_file(self
, mock_exists
, mock_open
):
562 mock_exists
.return_value
= True
564 'description': 'test bundle',
565 'bundle': 'kubernetes',
566 'applications': {'app':{ }, 'app2': { }}
568 mock_open
.return_value
= FakeFileWrapper(content
=content
)
569 kdu
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.inspect_kdu("model"))
570 self
.assertEqual(kdu
, {"app": {}, "app2": {}})
571 mock_exists
.assert_called_once()
572 mock_open
.assert_called_once()
574 @asynctest.mock
.patch("builtins.open")
575 @asynctest.mock
.patch("os.path.exists")
576 def test_not_existing_file(self
, mock_exists
, mock_open
):
578 mock_exists
.return_value
= False
579 with self
.assertRaises(K8sException
):
580 kdu
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.inspect_kdu("model"))
581 self
.assertEqual(kdu
, None)
582 mock_exists
.assert_called_once_with("model")
583 mock_open
.assert_not_called()
586 class HelpKduTest(K8sJujuConnTestCase
):
588 super(HelpKduTest
, self
).setUp()
590 @asynctest.mock
.patch("builtins.open")
591 @asynctest.mock
.patch("os.listdir")
592 def test_existing_file(self
, mock_listdir
, mock_open
):
593 content
= "Readme file content"
594 mock_open
.return_value
= FakeFileWrapper(content
=content
)
595 for file in ["README.md", "README.txt", "README"]:
596 mock_listdir
.return_value
= [file]
597 help = self
.loop
.run_until_complete(
598 self
.k8s_juju_conn
.help_kdu("kdu_instance")
600 self
.assertEqual(help, content
)
602 self
.assertEqual(mock_listdir
.call_count
, 3)
603 self
.assertEqual(mock_open
.call_count
, 3)
605 @asynctest.mock
.patch("builtins.open")
606 @asynctest.mock
.patch("os.listdir")
607 def test_not_existing_file(self
, mock_listdir
, mock_open
):
608 for file in ["src/charm.py", "tox.ini", "requirements.txt"]:
609 mock_listdir
.return_value
= [file]
610 help = self
.loop
.run_until_complete(
611 self
.k8s_juju_conn
.help_kdu("kdu_instance")
613 self
.assertEqual(help, None)
615 self
.assertEqual(mock_listdir
.call_count
, 3)
616 self
.assertEqual(mock_open
.call_count
, 0)
619 class StatusKduTest(K8sJujuConnTestCase
):
621 super(StatusKduTest
, self
).setUp()
622 self
.k8s_juju_conn
.libjuju
.get_model_status
= AsyncMock()
624 def test_success(self
):
625 applications
= {"app": {"status": {"status": "active"}}}
626 model
= FakeModel(applications
=applications
)
627 self
.k8s_juju_conn
.libjuju
.get_model_status
.return_value
= model
628 status
= self
.loop
.run_until_complete(
629 self
.k8s_juju_conn
.status_kdu("cluster", "kdu_instance")
631 self
.assertEqual(status
, {"app": {"status": "active"}})
632 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
634 def test_exception(self
):
635 self
.k8s_juju_conn
.libjuju
.get_model_status
.side_effect
= Exception()
637 with self
.assertRaises(Exception):
638 status
= self
.loop
.run_until_complete(
639 self
.k8s_juju_conn
.status_kdu("cluster", "kdu_instance")
641 self
.assertIsNone(status
)
642 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
645 class GetServicesTest(K8sJujuConnTestCase
):
647 super(GetServicesTest
, self
).setUp()
649 @asynctest.mock
.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
650 def test_success(self
, mock_get_credentials
):
651 mock_get_credentials
.return_value
= kubeconfig
652 self
.loop
.run_until_complete(self
.k8s_juju_conn
.get_services("", "", ""))
653 mock_get_credentials
.assert_called_once()
654 self
.kubectl
.get_services
.assert_called_once()
657 class GetServiceTest(K8sJujuConnTestCase
):
659 super(GetServiceTest
, self
).setUp()
661 @asynctest.mock
.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
662 def test_success(self
, mock_get_credentials
):
663 mock_get_credentials
.return_value
= kubeconfig
664 self
.loop
.run_until_complete(self
.k8s_juju_conn
.get_service("", "", ""))
665 mock_get_credentials
.assert_called_once()
666 self
.kubectl
.get_services
.assert_called_once()
669 class GetCredentialsTest(K8sJujuConnTestCase
):
671 super(GetCredentialsTest
, self
).setUp()
673 @asynctest.mock
.patch("yaml.safe_dump")
674 def test_success(self
, mock_safe_dump
):
675 self
.k8s_juju_conn
.db
.get_one
.return_value
= {
677 "credentials": "credentials",
678 "schema_version": "2",
680 self
.k8s_juju_conn
.get_credentials("cluster_uuid")
681 self
.k8s_juju_conn
.db
.get_one
.assert_called_once()
682 self
.k8s_juju_conn
.db
.encrypt_decrypt_fields
.assert_called_once()
683 mock_safe_dump
.assert_called_once()
686 class UpdateVcaStatusTest(K8sJujuConnTestCase
):
688 super(UpdateVcaStatusTest
, self
).setUp()
689 self
.vcaStatus
= {"model": {"applications": {"app": {"actions": {}}}}}
690 self
.k8s_juju_conn
.libjuju
.get_executed_actions
= AsyncMock()
691 self
.k8s_juju_conn
.libjuju
.get_actions
= AsyncMock()
692 self
.k8s_juju_conn
.libjuju
.get_application_configs
= AsyncMock()
694 def test_success(self
):
695 self
.loop
.run_until_complete(
696 self
.k8s_juju_conn
.update_vca_status(self
.vcaStatus
, self
.kdu_instance
)
698 self
.k8s_juju_conn
.libjuju
.get_executed_actions
.assert_called_once()
699 self
.k8s_juju_conn
.libjuju
.get_application_configs
.assert_called_once()
701 def test_exception(self
):
702 self
.k8s_juju_conn
.libjuju
.get_model
.return_value
= None
703 self
.k8s_juju_conn
.libjuju
.get_executed_actions
.side_effect
= Exception()
704 with self
.assertRaises(Exception):
705 self
.loop
.run_until_complete(
706 self
.k8s_juju_conn
.update_vca_status(self
.vcaStatus
, self
.kdu_instance
)
708 self
.k8s_juju_conn
.libjuju
.get_executed_actions
.assert_not_called()
709 self
.k8s_juju_conn
.libjuju
.get_application_configs
.assert_not_called_once()
712 class ScaleTest(K8sJujuConnTestCase
):
714 super(ScaleTest
, self
).setUp()
715 self
.application_name
= "app"
716 self
.kdu_name
= "kdu-instance"
718 self
.k8s_juju_conn
.libjuju
.scale_application
= AsyncMock()
720 def test_success(self
):
721 self
.loop
.run_until_complete(
722 self
.k8s_juju_conn
.scale(self
.kdu_name
, self
._scale
, self
.application_name
)
724 self
.k8s_juju_conn
.libjuju
.scale_application
.assert_called_once()
726 def test_exception(self
):
727 self
.k8s_juju_conn
.libjuju
.scale_application
.side_effect
= Exception()
728 with self
.assertRaises(Exception):
729 self
.loop
.run_until_complete(
730 self
.k8s_juju_conn
.scale(
731 self
.kdu_name
, self
._scale
, self
.application_name
734 self
.k8s_juju_conn
.libjuju
.scale_application
.assert_called_once()
737 class GetScaleCount(K8sJujuConnTestCase
):
739 super(GetScaleCount
, self
).setUp()
740 self
.k8s_juju_conn
.libjuju
.get_model_status
= AsyncMock()
742 def test_success(self
):
743 applications
= {"app": FakeApplication()}
744 model
= FakeModel(applications
=applications
)
745 self
.k8s_juju_conn
.libjuju
.get_model_status
.return_value
= model
746 status
= self
.loop
.run_until_complete(
747 self
.k8s_juju_conn
.get_scale_count("app", "kdu_instance")
749 self
.assertEqual(status
, 2)
750 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
752 def test_exception(self
):
753 self
.k8s_juju_conn
.libjuju
.get_model_status
.side_effect
= Exception()
755 with self
.assertRaises(Exception):
756 status
= self
.loop
.run_until_complete(
757 self
.k8s_juju_conn
.status_kdu("app", "kdu_instance")
759 self
.assertIsNone(status
)
760 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
763 class AddRelationTest(K8sJujuConnTestCase
):
765 super(AddRelationTest
, self
).setUp()
766 self
.k8s_juju_conn
.libjuju
.add_relation
= AsyncMock()
767 self
.k8s_juju_conn
.libjuju
.offer
= AsyncMock()
768 self
.k8s_juju_conn
.libjuju
.get_controller
= AsyncMock()
769 self
.k8s_juju_conn
.libjuju
.consume
= AsyncMock()
771 def test_standard_relation_same_model_and_controller(self
):
772 relation_endpoint_1
= RelationEndpoint("model-1.app1.0", None, "endpoint1")
773 relation_endpoint_2
= RelationEndpoint("model-1.app2.1", None, "endpoint2")
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
.add_relation
.assert_called_once_with(
778 model_name
="model-1",
779 endpoint_1
="app1:endpoint1",
780 endpoint_2
="app2:endpoint2",
782 self
.k8s_juju_conn
.libjuju
.offer
.assert_not_called()
783 self
.k8s_juju_conn
.libjuju
.consume
.assert_not_called()
785 def test_cmr_relation_same_controller(self
):
786 relation_endpoint_1
= RelationEndpoint("model-1.app1.0", None, "endpoint")
787 relation_endpoint_2
= RelationEndpoint("model-2.app2.1", None, "endpoint")
788 offer
= Offer("admin/model-1.app1")
789 self
.k8s_juju_conn
.libjuju
.offer
.return_value
= offer
790 self
.k8s_juju_conn
.libjuju
.consume
.return_value
= "saas"
791 self
.loop
.run_until_complete(
792 self
.k8s_juju_conn
.add_relation(relation_endpoint_1
, relation_endpoint_2
)
794 self
.k8s_juju_conn
.libjuju
.offer
.assert_called_once_with(relation_endpoint_1
)
795 self
.k8s_juju_conn
.libjuju
.consume
.assert_called_once()
796 self
.k8s_juju_conn
.libjuju
.add_relation
.assert_called_once_with(
797 "model-2", "app2:endpoint", "saas"
800 def test_cmr_relation_different_controller(self
):
801 self
.k8s_juju_conn
._get
_libjuju
= AsyncMock(
802 return_value
=self
.k8s_juju_conn
.libjuju
804 relation_endpoint_1
= RelationEndpoint("model-1.app1.0", "vca-id-1", "endpoint")
805 relation_endpoint_2
= RelationEndpoint("model-1.app2.1", "vca-id-2", "endpoint")
806 offer
= Offer("admin/model-1.app1")
807 self
.k8s_juju_conn
.libjuju
.offer
.return_value
= offer
808 self
.k8s_juju_conn
.libjuju
.consume
.return_value
= "saas"
809 self
.loop
.run_until_complete(
810 self
.k8s_juju_conn
.add_relation(relation_endpoint_1
, relation_endpoint_2
)
812 self
.k8s_juju_conn
.libjuju
.offer
.assert_called_once_with(relation_endpoint_1
)
813 self
.k8s_juju_conn
.libjuju
.consume
.assert_called_once()
814 self
.k8s_juju_conn
.libjuju
.add_relation
.assert_called_once_with(
815 "model-1", "app2:endpoint", "saas"
818 def test_relation_exception(self
):
819 relation_endpoint_1
= RelationEndpoint("model-1.app1.0", None, "endpoint")
820 relation_endpoint_2
= RelationEndpoint("model-2.app2.1", None, "endpoint")
821 self
.k8s_juju_conn
.libjuju
.offer
.side_effect
= Exception()
822 with self
.assertRaises(Exception):
823 self
.loop
.run_until_complete(
824 self
.k8s_juju_conn
.add_relation(
825 relation_endpoint_1
, relation_endpoint_2