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(
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
86 self
.k8s_juju_conn
._obtain
_namespace
_from
_db
= Mock(
87 return_value
=self
.default_namespace
91 class InitEnvTest(K8sJujuConnTestCase
):
93 super(InitEnvTest
, self
).setUp()
94 self
.k8s_juju_conn
.libjuju
.add_k8s
= AsyncMock()
96 def test_with_cluster_uuid(
99 reuse_cluster_uuid
= "uuid"
100 uuid
, created
= self
.loop
.run_until_complete(
101 self
.k8s_juju_conn
.init_env(
102 k8s_creds
=kubeconfig
, reuse_cluster_uuid
=reuse_cluster_uuid
106 self
.assertTrue(created
)
107 self
.assertEqual(uuid
, reuse_cluster_uuid
)
108 self
.kubectl
.get_default_storage_class
.assert_called_once()
109 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
111 def test_with_no_cluster_uuid(
114 uuid
, created
= self
.loop
.run_until_complete(
115 self
.k8s_juju_conn
.init_env(k8s_creds
=kubeconfig
)
118 self
.assertTrue(created
)
119 self
.assertTrue(isinstance(uuid
, str))
120 self
.kubectl
.get_default_storage_class
.assert_called_once()
121 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
123 def test_init_env_exception(
126 self
.k8s_juju_conn
.libjuju
.add_k8s
.side_effect
= Exception()
129 with self
.assertRaises(Exception):
130 uuid
, created
= self
.loop
.run_until_complete(
131 self
.k8s_juju_conn
.init_env(k8s_creds
=kubeconfig
)
133 self
.assertIsNone(created
)
134 self
.assertIsNone(uuid
)
135 self
.kubectl
.create_cluster_role
.assert_called_once()
136 self
.kubectl
.create_service_account
.assert_called_once()
137 self
.kubectl
.create_cluster_role_binding
.assert_called_once()
138 self
.kubectl
.get_default_storage_class
.assert_called_once()
139 self
.kubectl
.delete_cluster_role
.assert_called_once()
140 self
.kubectl
.delete_service_account
.assert_called_once()
141 self
.kubectl
.delete_cluster_role_binding
.assert_called_once()
142 self
.k8s_juju_conn
.libjuju
.add_k8s
.assert_called_once()
145 class NotImplementedTest(K8sJujuConnTestCase
):
147 super(NotImplementedTest
, self
).setUp()
149 def test_repo_add(self
):
150 with self
.assertRaises(MethodNotImplemented
):
151 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_add("", ""))
153 def test_repo_list(self
):
154 with self
.assertRaises(MethodNotImplemented
):
155 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_list())
157 def test_repo_remove(self
):
158 with self
.assertRaises(MethodNotImplemented
):
159 self
.loop
.run_until_complete(self
.k8s_juju_conn
.repo_remove(""))
161 def test_synchronize_repos(self
):
163 self
.loop
.run_until_complete(self
.k8s_juju_conn
.synchronize_repos("", ""))
166 def test_upgrade(self
):
167 with self
.assertRaises(MethodNotImplemented
):
168 self
.loop
.run_until_complete(self
.k8s_juju_conn
.upgrade("", ""))
170 def test_rollback(self
):
171 with self
.assertRaises(MethodNotImplemented
):
172 self
.loop
.run_until_complete(self
.k8s_juju_conn
.rollback("", ""))
174 def test_get_namespace(self
):
175 self
.assertIsNone(self
.k8s_juju_conn
.get_namespace(""))
177 def test_instances_list(self
):
178 res
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.instances_list(""))
179 self
.assertEqual(res
, [])
182 class ResetTest(K8sJujuConnTestCase
):
184 super(ResetTest
, self
).setUp()
185 self
.k8s_juju_conn
.libjuju
.remove_cloud
= AsyncMock()
186 self
.k8s_juju_conn
.libjuju
.get_cloud_credentials
= AsyncMock()
188 cloud_creds
.result
= {"attrs": {RBAC_LABEL_KEY_NAME
: "asd"}}
189 self
.k8s_juju_conn
.libjuju
.get_cloud_credentials
.return_value
= [cloud_creds
]
190 self
.k8s_juju_conn
.get_credentials
= Mock()
191 self
.k8s_juju_conn
.get_credentials
.return_value
= kubeconfig
193 def test_success(self
):
194 removed
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.reset("uuid"))
195 self
.assertTrue(removed
)
196 self
.k8s_juju_conn
.libjuju
.remove_cloud
.assert_called_once()
198 def test_exception(self
):
200 self
.k8s_juju_conn
.libjuju
.remove_cloud
.side_effect
= Exception()
201 with self
.assertRaises(Exception):
202 removed
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.reset("uuid"))
203 self
.assertIsNone(removed
)
204 self
.k8s_juju_conn
.libjuju
.remove_cloud
.assert_called_once()
207 @asynctest.mock
.patch("os.chdir")
208 class InstallTest(K8sJujuConnTestCase
):
210 super(InstallTest
, self
).setUp()
211 self
.db_dict
= {"filter": {"_id": "id"}}
212 self
.local_bundle
= "bundle"
213 self
.cs_bundle
= "cs:bundle"
214 self
.http_bundle
= "https://example.com/bundle.yaml"
215 self
.cluster_uuid
= "cluster"
216 self
.k8s_juju_conn
.libjuju
.add_model
= AsyncMock()
217 self
.k8s_juju_conn
.libjuju
.deploy
= AsyncMock()
219 def test_success_local(self
, mock_chdir
):
220 self
.loop
.run_until_complete(
221 self
.k8s_juju_conn
.install(
226 kdu_name
=self
.kdu_name
,
227 db_dict
=self
.db_dict
,
232 self
.assertEqual(mock_chdir
.call_count
, 2)
233 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
234 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
235 "local:{}".format(self
.local_bundle
),
236 model_name
=self
.default_namespace
,
239 instantiation_params
=None,
242 def test_success_cs(self
, mock_chdir
):
243 self
.loop
.run_until_complete(
244 self
.k8s_juju_conn
.install(
249 kdu_name
=self
.kdu_name
,
250 db_dict
=self
.db_dict
,
255 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
256 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
258 model_name
=self
.default_namespace
,
261 instantiation_params
=None,
264 def test_success_http(self
, mock_chdir
):
265 params
= {"overlay": {"applications": {"squid": {"scale": 2}}}}
266 self
.loop
.run_until_complete(
267 self
.k8s_juju_conn
.install(
272 kdu_name
=self
.kdu_name
,
273 db_dict
=self
.db_dict
,
278 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
279 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
281 model_name
=self
.default_namespace
,
284 instantiation_params
=params
.get("overlay"),
287 def test_success_not_kdu_name(self
, mock_chdir
):
288 params
= {"some_key": {"applications": {"squid": {"scale": 2}}}}
289 self
.loop
.run_until_complete(
290 self
.k8s_juju_conn
.install(
295 db_dict
=self
.db_dict
,
300 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
301 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
303 model_name
=self
.default_namespace
,
306 instantiation_params
=None,
309 def test_missing_db_dict(self
, mock_chdir
):
311 with self
.assertRaises(K8sException
):
312 self
.loop
.run_until_complete(
313 self
.k8s_juju_conn
.install(
318 kdu_name
=self
.kdu_name
,
322 self
.assertIsNone(kdu_instance
)
323 self
.k8s_juju_conn
.libjuju
.add_model
.assert_not_called()
324 self
.k8s_juju_conn
.libjuju
.deploy
.assert_not_called()
326 @asynctest.mock
.patch("os.getcwd")
327 def test_getcwd_exception(self
, mock_getcwd
, mock_chdir
):
328 mock_getcwd
.side_effect
= FileNotFoundError()
329 self
.loop
.run_until_complete(
330 self
.k8s_juju_conn
.install(
335 kdu_name
=self
.kdu_name
,
336 db_dict
=self
.db_dict
,
340 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
341 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
343 model_name
=self
.default_namespace
,
346 instantiation_params
=None,
349 def test_missing_bundle(self
, mock_chdir
):
350 with self
.assertRaises(K8sException
):
351 self
.loop
.run_until_complete(
352 self
.k8s_juju_conn
.install(
357 kdu_name
=self
.kdu_name
,
359 db_dict
=self
.db_dict
,
362 self
.k8s_juju_conn
.libjuju
.add_model
.assert_not_called()
363 self
.k8s_juju_conn
.libjuju
.deploy
.assert_not_called()
365 def test_missing_exception(self
, mock_chdir
):
366 self
.k8s_juju_conn
.libjuju
.deploy
.side_effect
= Exception()
367 with self
.assertRaises(Exception):
368 self
.loop
.run_until_complete(
369 self
.k8s_juju_conn
.install(
374 kdu_name
=self
.kdu_name
,
375 db_dict
=self
.db_dict
,
379 self
.k8s_juju_conn
.libjuju
.add_model
.assert_called_once()
380 self
.k8s_juju_conn
.libjuju
.deploy
.assert_called_once_with(
381 "local:{}".format(self
.local_bundle
),
382 model_name
=self
.default_namespace
,
385 instantiation_params
=None,
389 class UninstallTest(K8sJujuConnTestCase
):
391 super(UninstallTest
, self
).setUp()
392 self
.k8s_juju_conn
.libjuju
.destroy_model
= AsyncMock()
394 def test_success(self
):
395 destroyed
= self
.loop
.run_until_complete(
396 self
.k8s_juju_conn
.uninstall("cluster_uuid", "model_name")
398 self
.assertTrue(destroyed
)
399 self
.k8s_juju_conn
.libjuju
.destroy_model
.assert_called_once()
401 def test_exception(self
):
403 self
.k8s_juju_conn
.libjuju
.destroy_model
.side_effect
= Exception()
404 with self
.assertRaises(Exception):
405 destroyed
= self
.loop
.run_until_complete(
406 self
.k8s_juju_conn
.uninstall("cluster_uuid", "model_name")
408 self
.assertIsNone(destroyed
)
409 self
.k8s_juju_conn
.libjuju
.destroy_model
.assert_called_once()
412 class ExecPrimitivesTest(K8sJujuConnTestCase
):
414 super(ExecPrimitivesTest
, self
).setUp()
415 self
.action_name
= "touch"
416 self
.application_name
= "myapp"
417 self
.k8s_juju_conn
.libjuju
.get_actions
= AsyncMock()
418 self
.k8s_juju_conn
.libjuju
.execute_action
= AsyncMock()
420 def test_success(self
):
421 params
= {"application-name": self
.application_name
}
422 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
423 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (
428 output
= self
.loop
.run_until_complete(
429 self
.k8s_juju_conn
.exec_primitive(
430 "cluster", self
.kdu_instance
, self
.action_name
, params
=params
434 self
.assertEqual(output
, "success")
435 self
.k8s_juju_conn
._obtain
_namespace
_from
_db
.assert_called_once_with(
436 kdu_instance
=self
.kdu_instance
438 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
439 application_name
=self
.application_name
, model_name
=self
.default_namespace
441 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
442 application_name
=self
.application_name
,
443 model_name
=self
.default_namespace
,
444 action_name
=self
.action_name
,
448 def test_exception(self
):
449 params
= {"application-name": self
.application_name
}
450 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
451 self
.k8s_juju_conn
.libjuju
.execute_action
.side_effect
= Exception()
454 with self
.assertRaises(Exception):
455 output
= self
.loop
.run_until_complete(
456 self
.k8s_juju_conn
.exec_primitive(
457 "cluster", self
.kdu_instance
, self
.action_name
, params
=params
461 self
.assertIsNone(output
)
462 self
.k8s_juju_conn
._obtain
_namespace
_from
_db
.assert_called_once_with(
463 kdu_instance
=self
.kdu_instance
465 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
466 application_name
=self
.application_name
, model_name
=self
.default_namespace
468 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
469 application_name
=self
.application_name
,
470 model_name
=self
.default_namespace
,
471 action_name
=self
.action_name
,
475 def test_missing_application_name_in_params(self
):
479 with self
.assertRaises(K8sException
):
480 output
= self
.loop
.run_until_complete(
481 self
.k8s_juju_conn
.exec_primitive(
482 "cluster", self
.kdu_instance
, self
.action_name
, params
=params
486 self
.assertIsNone(output
)
487 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called()
488 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
490 def test_missing_params(self
):
492 with self
.assertRaises(K8sException
):
493 output
= self
.loop
.run_until_complete(
494 self
.k8s_juju_conn
.exec_primitive(
495 "cluster", self
.kdu_instance
, self
.action_name
499 self
.assertIsNone(output
)
500 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_not_called()
501 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
503 def test_missing_action(self
):
505 params
= {"application-name": self
.application_name
}
506 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
507 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (
511 with self
.assertRaises(K8sException
):
512 output
= self
.loop
.run_until_complete(
513 self
.k8s_juju_conn
.exec_primitive(
514 "cluster", self
.kdu_instance
, "non-existing-action", params
=params
518 self
.assertIsNone(output
)
519 self
.k8s_juju_conn
._obtain
_namespace
_from
_db
.assert_called_once_with(
520 kdu_instance
=self
.kdu_instance
522 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
523 application_name
=self
.application_name
, model_name
=self
.default_namespace
525 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_not_called()
527 def test_missing_not_completed(self
):
529 params
= {"application-name": self
.application_name
}
530 self
.k8s_juju_conn
.libjuju
.get_actions
.return_value
= [self
.action_name
]
531 self
.k8s_juju_conn
.libjuju
.execute_action
.return_value
= (None, "failed")
532 with self
.assertRaises(K8sException
):
533 output
= self
.loop
.run_until_complete(
534 self
.k8s_juju_conn
.exec_primitive(
535 "cluster", self
.kdu_instance
, self
.action_name
, params
=params
539 self
.assertIsNone(output
)
540 self
.k8s_juju_conn
._obtain
_namespace
_from
_db
.assert_called_once_with(
541 kdu_instance
=self
.kdu_instance
543 self
.k8s_juju_conn
.libjuju
.get_actions
.assert_called_once_with(
544 application_name
=self
.application_name
, model_name
=self
.default_namespace
546 self
.k8s_juju_conn
.libjuju
.execute_action
.assert_called_once_with(
547 application_name
=self
.application_name
,
548 model_name
=self
.default_namespace
,
549 action_name
=self
.action_name
,
554 class InspectKduTest(K8sJujuConnTestCase
):
556 super(InspectKduTest
, self
).setUp()
558 @asynctest.mock
.patch("builtins.open")
559 @asynctest.mock
.patch("os.path.exists")
560 def test_existing_file(self
, mock_exists
, mock_open
):
561 mock_exists
.return_value
= True
563 'description': 'test bundle',
564 'bundle': 'kubernetes',
565 'applications': {'app':{ }, 'app2': { }}
567 mock_open
.return_value
= FakeFileWrapper(content
=content
)
568 kdu
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.inspect_kdu("model"))
569 self
.assertEqual(kdu
, {"app": {}, "app2": {}})
570 mock_exists
.assert_called_once()
571 mock_open
.assert_called_once()
573 @asynctest.mock
.patch("builtins.open")
574 @asynctest.mock
.patch("os.path.exists")
575 def test_not_existing_file(self
, mock_exists
, mock_open
):
577 mock_exists
.return_value
= False
578 with self
.assertRaises(K8sException
):
579 kdu
= self
.loop
.run_until_complete(self
.k8s_juju_conn
.inspect_kdu("model"))
580 self
.assertEqual(kdu
, None)
581 mock_exists
.assert_called_once_with("model")
582 mock_open
.assert_not_called()
585 class HelpKduTest(K8sJujuConnTestCase
):
587 super(HelpKduTest
, self
).setUp()
589 @asynctest.mock
.patch("builtins.open")
590 @asynctest.mock
.patch("os.listdir")
591 def test_existing_file(self
, mock_listdir
, mock_open
):
592 content
= "Readme file content"
593 mock_open
.return_value
= FakeFileWrapper(content
=content
)
594 for file in ["README.md", "README.txt", "README"]:
595 mock_listdir
.return_value
= [file]
596 help = self
.loop
.run_until_complete(
597 self
.k8s_juju_conn
.help_kdu("kdu_instance")
599 self
.assertEqual(help, content
)
601 self
.assertEqual(mock_listdir
.call_count
, 3)
602 self
.assertEqual(mock_open
.call_count
, 3)
604 @asynctest.mock
.patch("builtins.open")
605 @asynctest.mock
.patch("os.listdir")
606 def test_not_existing_file(self
, mock_listdir
, mock_open
):
607 for file in ["src/charm.py", "tox.ini", "requirements.txt"]:
608 mock_listdir
.return_value
= [file]
609 help = self
.loop
.run_until_complete(
610 self
.k8s_juju_conn
.help_kdu("kdu_instance")
612 self
.assertEqual(help, None)
614 self
.assertEqual(mock_listdir
.call_count
, 3)
615 self
.assertEqual(mock_open
.call_count
, 0)
618 class StatusKduTest(K8sJujuConnTestCase
):
620 super(StatusKduTest
, self
).setUp()
621 self
.k8s_juju_conn
.libjuju
.get_model_status
= AsyncMock()
623 def test_success(self
):
624 applications
= {"app": {"status": {"status": "active"}}}
625 model
= FakeModel(applications
=applications
)
626 self
.k8s_juju_conn
.libjuju
.get_model_status
.return_value
= model
627 status
= self
.loop
.run_until_complete(
628 self
.k8s_juju_conn
.status_kdu("cluster", "kdu_instance")
630 self
.assertEqual(status
, {"app": {"status": "active"}})
631 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
633 def test_exception(self
):
634 self
.k8s_juju_conn
.libjuju
.get_model_status
.side_effect
= Exception()
636 with self
.assertRaises(Exception):
637 status
= self
.loop
.run_until_complete(
638 self
.k8s_juju_conn
.status_kdu("cluster", "kdu_instance")
640 self
.assertIsNone(status
)
641 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
644 class GetServicesTest(K8sJujuConnTestCase
):
646 super(GetServicesTest
, self
).setUp()
648 @asynctest.mock
.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
649 def test_success(self
, mock_get_credentials
):
650 mock_get_credentials
.return_value
= kubeconfig
651 self
.loop
.run_until_complete(self
.k8s_juju_conn
.get_services("", "", ""))
652 mock_get_credentials
.assert_called_once()
653 self
.kubectl
.get_services
.assert_called_once()
656 class GetServiceTest(K8sJujuConnTestCase
):
658 super(GetServiceTest
, self
).setUp()
660 @asynctest.mock
.patch("n2vc.k8s_juju_conn.K8sJujuConnector.get_credentials")
661 def test_success(self
, mock_get_credentials
):
662 mock_get_credentials
.return_value
= kubeconfig
663 self
.loop
.run_until_complete(self
.k8s_juju_conn
.get_service("", "", ""))
664 mock_get_credentials
.assert_called_once()
665 self
.kubectl
.get_services
.assert_called_once()
668 class GetCredentialsTest(K8sJujuConnTestCase
):
670 super(GetCredentialsTest
, self
).setUp()
672 @asynctest.mock
.patch("yaml.safe_dump")
673 def test_success(self
, mock_safe_dump
):
674 self
.k8s_juju_conn
.db
.get_one
.return_value
= {
676 "credentials": "credentials",
677 "schema_version": "2",
679 self
.k8s_juju_conn
.get_credentials("cluster_uuid")
680 self
.k8s_juju_conn
.db
.get_one
.assert_called_once()
681 self
.k8s_juju_conn
.db
.encrypt_decrypt_fields
.assert_called_once()
682 mock_safe_dump
.assert_called_once()
685 class UpdateVcaStatusTest(K8sJujuConnTestCase
):
687 super(UpdateVcaStatusTest
, self
).setUp()
688 self
.vcaStatus
= {"model": {"applications": {"app": {"actions": {}}}}}
689 self
.k8s_juju_conn
.libjuju
.get_executed_actions
= AsyncMock()
690 self
.k8s_juju_conn
.libjuju
.get_actions
= AsyncMock()
691 self
.k8s_juju_conn
.libjuju
.get_application_configs
= AsyncMock()
693 def test_success(self
):
694 self
.loop
.run_until_complete(
695 self
.k8s_juju_conn
.update_vca_status(self
.vcaStatus
, self
.kdu_instance
)
697 self
.k8s_juju_conn
.libjuju
.get_executed_actions
.assert_called_once()
698 self
.k8s_juju_conn
.libjuju
.get_application_configs
.assert_called_once()
700 def test_exception(self
):
701 self
.k8s_juju_conn
.libjuju
.get_model
.return_value
= None
702 self
.k8s_juju_conn
.libjuju
.get_executed_actions
.side_effect
= Exception()
703 with self
.assertRaises(Exception):
704 self
.loop
.run_until_complete(
705 self
.k8s_juju_conn
.update_vca_status(self
.vcaStatus
, self
.kdu_instance
)
707 self
.k8s_juju_conn
.libjuju
.get_executed_actions
.assert_not_called()
708 self
.k8s_juju_conn
.libjuju
.get_application_configs
.assert_not_called_once()
711 class ScaleTest(K8sJujuConnTestCase
):
713 super(ScaleTest
, self
).setUp()
714 self
.application_name
= "app"
715 self
.kdu_name
= "kdu-instance"
717 self
.k8s_juju_conn
.libjuju
.scale_application
= AsyncMock()
719 def test_success(self
):
720 self
.loop
.run_until_complete(
721 self
.k8s_juju_conn
.scale(self
.kdu_name
, self
._scale
, self
.application_name
)
723 self
.k8s_juju_conn
.libjuju
.scale_application
.assert_called_once()
725 def test_exception(self
):
726 self
.k8s_juju_conn
.libjuju
.scale_application
.side_effect
= Exception()
727 with self
.assertRaises(Exception):
728 self
.loop
.run_until_complete(
729 self
.k8s_juju_conn
.scale(
730 self
.kdu_name
, self
._scale
, self
.application_name
733 self
.k8s_juju_conn
.libjuju
.scale_application
.assert_called_once()
736 class GetScaleCount(K8sJujuConnTestCase
):
738 super(GetScaleCount
, self
).setUp()
739 self
.k8s_juju_conn
.libjuju
.get_model_status
= AsyncMock()
741 def test_success(self
):
742 applications
= {"app": FakeApplication()}
743 model
= FakeModel(applications
=applications
)
744 self
.k8s_juju_conn
.libjuju
.get_model_status
.return_value
= model
745 status
= self
.loop
.run_until_complete(
746 self
.k8s_juju_conn
.get_scale_count("app", "kdu_instance")
748 self
.assertEqual(status
, 2)
749 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
751 def test_exception(self
):
752 self
.k8s_juju_conn
.libjuju
.get_model_status
.side_effect
= Exception()
754 with self
.assertRaises(Exception):
755 status
= self
.loop
.run_until_complete(
756 self
.k8s_juju_conn
.status_kdu("app", "kdu_instance")
758 self
.assertIsNone(status
)
759 self
.k8s_juju_conn
.libjuju
.get_model_status
.assert_called_once()
762 class AddRelationTest(K8sJujuConnTestCase
):
764 super(AddRelationTest
, self
).setUp()
765 self
.k8s_juju_conn
.libjuju
.add_relation
= AsyncMock()
766 self
.k8s_juju_conn
.libjuju
.offer
= AsyncMock()
767 self
.k8s_juju_conn
.libjuju
.get_controller
= AsyncMock()
768 self
.k8s_juju_conn
.libjuju
.consume
= AsyncMock()
770 def test_standard_relation_same_model_and_controller(self
):
771 relation_endpoint_1
= RelationEndpoint("model-1.app1.0", None, "endpoint1")
772 relation_endpoint_2
= RelationEndpoint("model-1.app2.1", None, "endpoint2")
773 self
.loop
.run_until_complete(
774 self
.k8s_juju_conn
.add_relation(relation_endpoint_1
, relation_endpoint_2
)
776 self
.k8s_juju_conn
.libjuju
.add_relation
.assert_called_once_with(
777 model_name
="model-1",
778 endpoint_1
="app1:endpoint1",
779 endpoint_2
="app2:endpoint2",
781 self
.k8s_juju_conn
.libjuju
.offer
.assert_not_called()
782 self
.k8s_juju_conn
.libjuju
.consume
.assert_not_called()
784 def test_cmr_relation_same_controller(self
):
785 relation_endpoint_1
= RelationEndpoint("model-1.app1.0", None, "endpoint")
786 relation_endpoint_2
= RelationEndpoint("model-2.app2.1", None, "endpoint")
787 offer
= Offer("admin/model-1.app1")
788 self
.k8s_juju_conn
.libjuju
.offer
.return_value
= offer
789 self
.k8s_juju_conn
.libjuju
.consume
.return_value
= "saas"
790 self
.loop
.run_until_complete(
791 self
.k8s_juju_conn
.add_relation(relation_endpoint_1
, relation_endpoint_2
)
793 self
.k8s_juju_conn
.libjuju
.offer
.assert_called_once_with(relation_endpoint_1
)
794 self
.k8s_juju_conn
.libjuju
.consume
.assert_called_once()
795 self
.k8s_juju_conn
.libjuju
.add_relation
.assert_called_once_with(
796 "model-2", "app2:endpoint", "saas"
799 def test_cmr_relation_different_controller(self
):
800 self
.k8s_juju_conn
._get
_libjuju
= AsyncMock(
801 return_value
=self
.k8s_juju_conn
.libjuju
803 relation_endpoint_1
= RelationEndpoint("model-1.app1.0", "vca-id-1", "endpoint")
804 relation_endpoint_2
= RelationEndpoint("model-1.app2.1", "vca-id-2", "endpoint")
805 offer
= Offer("admin/model-1.app1")
806 self
.k8s_juju_conn
.libjuju
.offer
.return_value
= offer
807 self
.k8s_juju_conn
.libjuju
.consume
.return_value
= "saas"
808 self
.loop
.run_until_complete(
809 self
.k8s_juju_conn
.add_relation(relation_endpoint_1
, relation_endpoint_2
)
811 self
.k8s_juju_conn
.libjuju
.offer
.assert_called_once_with(relation_endpoint_1
)
812 self
.k8s_juju_conn
.libjuju
.consume
.assert_called_once()
813 self
.k8s_juju_conn
.libjuju
.add_relation
.assert_called_once_with(
814 "model-1", "app2:endpoint", "saas"
817 def test_relation_exception(self
):
818 relation_endpoint_1
= RelationEndpoint("model-1.app1.0", None, "endpoint")
819 relation_endpoint_2
= RelationEndpoint("model-2.app2.1", None, "endpoint")
820 self
.k8s_juju_conn
.libjuju
.offer
.side_effect
= Exception()
821 with self
.assertRaises(Exception):
822 self
.loop
.run_until_complete(
823 self
.k8s_juju_conn
.add_relation(
824 relation_endpoint_1
, relation_endpoint_2