2 # Licensed under the Apache License, Version 2.0 (the "License"); you may
3 # not use this file except in compliance with the License. You may obtain
4 # a copy of the License at
6 # http://www.apache.org/licenses/LICENSE-2.0
8 # Unless required by applicable law or agreed to in writing, software
9 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
10 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
11 # License for the specific language governing permissions and limitations
14 # For those usages not covered by the Apache License, Version 2.0 please
15 # contact: alfonso.tiernosepulveda@telefonica.com
21 from asynctest
.mock
import Mock
, patch
22 from osm_common
.dbmemory
import DbMemory
23 from osm_common
.fslocal
import FsLocal
24 from n2vc
.k8s_helm3_conn
import K8sHelm3Connector
, K8sException
26 __author__
= "Isabel Lloret <illoret@indra.es>"
29 class TestK8sHelm3Conn(asynctest
.TestCase
):
30 logging
.basicConfig(level
=logging
.DEBUG
)
31 logger
= logging
.getLogger(__name__
)
32 logger
.setLevel(logging
.DEBUG
)
34 @patch("n2vc.k8s_helm_base_conn.EnvironConfig")
35 async def setUp(self
, mock_env
):
36 mock_env
.return_value
= {"stablerepourl": "https://charts.helm.sh/stable"}
37 self
.db
= Mock(DbMemory())
38 self
.fs
= asynctest
.Mock(FsLocal())
39 self
.fs
.path
= "./tmp/"
40 self
.namespace
= "testk8s"
41 self
.cluster_id
= "helm3_cluster_id"
42 self
.cluster_uuid
= "{}:{}".format(self
.namespace
, self
.cluster_id
)
43 # pass fake kubectl and helm commands to make sure it does not call actual commands
44 K8sHelm3Connector
._check
_file
_exists
= asynctest
.Mock(return_value
=True)
45 cluster_dir
= self
.fs
.path
+ self
.cluster_id
47 "HELM_CACHE_HOME": "{}/.cache/helm".format(cluster_dir
),
48 "HELM_CONFIG_HOME": "{}/.config/helm".format(cluster_dir
),
49 "HELM_DATA_HOME": "{}/.local/share/helm".format(cluster_dir
),
50 "KUBECONFIG": "{}/.kube/config".format(cluster_dir
),
52 self
.helm_conn
= K8sHelm3Connector(self
.fs
, self
.db
, log
=self
.logger
)
53 self
.logger
.debug("Set up executed")
55 @asynctest.fail_on(active_handles
=True)
56 async def test_init_env(self
):
57 k8s_creds
= "false_credentials_string"
58 self
.helm_conn
._get
_namespaces
= asynctest
.CoroutineMock(return_value
=[])
59 self
.helm_conn
._create
_namespace
= asynctest
.CoroutineMock()
60 self
.helm_conn
.repo_list
= asynctest
.CoroutineMock(return_value
=[])
61 self
.helm_conn
.repo_add
= asynctest
.CoroutineMock()
63 k8scluster_uuid
, installed
= await self
.helm_conn
.init_env(
64 k8s_creds
, namespace
=self
.namespace
, reuse_cluster_uuid
=self
.cluster_id
69 "{}:{}".format(self
.namespace
, self
.cluster_id
),
70 "Check cluster_uuid format: <namespace>.<cluster_id>",
72 self
.helm_conn
._get
_namespaces
.assert_called_once_with(self
.cluster_id
)
73 self
.helm_conn
._create
_namespace
.assert_called_once_with(
74 self
.cluster_id
, self
.namespace
76 self
.helm_conn
.repo_list
.assert_called_once_with(k8scluster_uuid
)
77 self
.helm_conn
.repo_add
.assert_called_once_with(
78 k8scluster_uuid
, "stable", "https://charts.helm.sh/stable"
80 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(
81 from_path
=self
.cluster_id
83 self
.logger
.debug(f
"cluster_uuid: {k8scluster_uuid}")
85 @asynctest.fail_on(active_handles
=True)
86 async def test_repo_add(self
):
88 repo_url
= "https://charts.bitnami.com/bitnami"
89 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
91 await self
.helm_conn
.repo_add(self
.cluster_uuid
, repo_name
, repo_url
)
93 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
94 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(
95 from_path
=self
.cluster_id
98 self
.helm_conn
._local
_async
_exec
.call_count
,
100 "local_async_exec expected 2 calls, called {}".format(
101 self
.helm_conn
._local
_async
_exec
.call_count
105 repo_update_command
= "/usr/bin/helm3 repo update"
106 repo_add_command
= "/usr/bin/helm3 repo add {} {}".format(repo_name
, repo_url
)
107 calls
= self
.helm_conn
._local
_async
_exec
.call_args_list
108 call0_kargs
= calls
[0][1]
110 call0_kargs
.get("command"),
112 "Invalid repo update command: {}".format(call0_kargs
.get("command")),
115 call0_kargs
.get("env"),
117 "Invalid env for update command: {}".format(call0_kargs
.get("env")),
119 call1_kargs
= calls
[1][1]
121 call1_kargs
.get("command"),
123 "Invalid repo add command: {}".format(call1_kargs
.get("command")),
126 call1_kargs
.get("env"),
128 "Invalid env for add command: {}".format(call1_kargs
.get("env")),
131 @asynctest.fail_on(active_handles
=True)
132 async def test_repo_list(self
):
134 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
136 await self
.helm_conn
.repo_list(self
.cluster_uuid
)
138 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
139 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(
140 from_path
=self
.cluster_id
142 command
= "/usr/bin/helm3 repo list --output yaml"
143 self
.helm_conn
._local
_async
_exec
.assert_called_with(
144 command
=command
, env
=self
.env
, raise_exception_on_error
=False
147 @asynctest.fail_on(active_handles
=True)
148 async def test_repo_remove(self
):
150 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
151 repo_name
= "bitnami"
152 await self
.helm_conn
.repo_remove(self
.cluster_uuid
, repo_name
)
154 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
155 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(
156 from_path
=self
.cluster_id
158 command
= "/usr/bin/helm3 repo remove {}".format(repo_name
)
159 self
.helm_conn
._local
_async
_exec
.assert_called_with(
160 command
=command
, env
=self
.env
, raise_exception_on_error
=True
163 @asynctest.fail_on(active_handles
=True)
164 async def test_install(self
):
165 kdu_model
= "stable/openldap:1.2.2"
166 kdu_instance
= "stable-openldap-0005399828"
168 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
169 self
.helm_conn
._status
_kdu
= asynctest
.CoroutineMock(return_value
=None)
170 self
.helm_conn
._store
_status
= asynctest
.CoroutineMock()
171 self
.kdu_instance
= "stable-openldap-0005399828"
172 self
.helm_conn
.generate_kdu_instance_name
= Mock(return_value
=self
.kdu_instance
)
173 self
.helm_conn
._get
_namespaces
= asynctest
.CoroutineMock(return_value
=[])
174 self
.helm_conn
._namespace
_exists
= asynctest
.CoroutineMock(
175 side_effect
=self
.helm_conn
._namespace
_exists
177 self
.helm_conn
._create
_namespace
= asynctest
.CoroutineMock()
179 await self
.helm_conn
.install(
184 namespace
=self
.namespace
,
188 self
.helm_conn
._namespace
_exists
.assert_called_once()
189 self
.helm_conn
._get
_namespaces
.assert_called_once()
190 self
.helm_conn
._create
_namespace
.assert_called_once_with(
191 self
.cluster_id
, self
.namespace
193 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
194 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(
195 from_path
=self
.cluster_id
197 self
.helm_conn
._store
_status
.assert_called_with(
198 cluster_id
=self
.cluster_id
,
199 kdu_instance
=kdu_instance
,
200 namespace
=self
.namespace
,
207 "/usr/bin/helm3 install stable-openldap-0005399828 --atomic --output yaml "
208 "--timeout 300s --namespace testk8s stable/openldap --version 1.2.2"
210 self
.helm_conn
._local
_async
_exec
.assert_called_once_with(
211 command
=command
, env
=self
.env
, raise_exception_on_error
=False
214 # Exception test if namespace could not being created for some reason
215 self
.helm_conn
._namespace
_exists
.return_value
= False
216 self
.helm_conn
._create
_namespace
.side_effect
= Exception()
218 with self
.assertRaises(K8sException
):
219 await self
.helm_conn
.install(
224 namespace
=self
.namespace
,
228 @asynctest.fail_on(active_handles
=True)
229 async def test_namespace_exists(self
):
230 self
.helm_conn
._get
_namespaces
= asynctest
.CoroutineMock()
232 self
.helm_conn
._get
_namespaces
.return_value
= ["testk8s", "kube-system"]
233 result
= await self
.helm_conn
._namespace
_exists
(self
.cluster_id
, self
.namespace
)
234 self
.helm_conn
._get
_namespaces
.assert_called_once()
235 self
.assertEqual(result
, True)
237 self
.helm_conn
._get
_namespaces
.reset_mock()
238 result
= await self
.helm_conn
._namespace
_exists
(
239 self
.cluster_id
, "none-exists-namespace"
241 self
.helm_conn
._get
_namespaces
.assert_called_once()
242 self
.assertEqual(result
, False)
244 @asynctest.fail_on(active_handles
=True)
245 async def test_upgrade(self
):
246 kdu_model
= "stable/openldap:1.2.3"
247 kdu_instance
= "stable-openldap-0005399828"
250 "chart": "openldap-1.2.2",
251 "name": kdu_instance
,
252 "namespace": self
.namespace
,
254 "status": "DEPLOYED",
256 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
257 self
.helm_conn
._store
_status
= asynctest
.CoroutineMock()
258 self
.helm_conn
.get_instance_info
= asynctest
.CoroutineMock(
259 return_value
=instance_info
262 await self
.helm_conn
.upgrade(
263 self
.cluster_uuid
, kdu_instance
, kdu_model
, atomic
=True, db_dict
=db_dict
265 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
266 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(
267 from_path
=self
.cluster_id
269 self
.helm_conn
._store
_status
.assert_called_with(
270 cluster_id
=self
.cluster_id
,
271 kdu_instance
=kdu_instance
,
272 namespace
=self
.namespace
,
279 "/usr/bin/helm3 upgrade stable-openldap-0005399828 stable/openldap "
280 "--namespace testk8s --atomic --output yaml --timeout 300s "
283 self
.helm_conn
._local
_async
_exec
.assert_called_once_with(
284 command
=command
, env
=self
.env
, raise_exception_on_error
=False
287 @asynctest.fail_on(active_handles
=True)
288 async def test_rollback(self
):
289 kdu_instance
= "stable-openldap-0005399828"
292 "chart": "openldap-1.2.3",
293 "name": kdu_instance
,
294 "namespace": self
.namespace
,
296 "status": "DEPLOYED",
298 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
299 self
.helm_conn
._store
_status
= asynctest
.CoroutineMock()
300 self
.helm_conn
.get_instance_info
= asynctest
.CoroutineMock(
301 return_value
=instance_info
304 await self
.helm_conn
.rollback(
305 self
.cluster_uuid
, kdu_instance
=kdu_instance
, revision
=1, db_dict
=db_dict
307 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
308 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(
309 from_path
=self
.cluster_id
311 self
.helm_conn
._store
_status
.assert_called_with(
312 cluster_id
=self
.cluster_id
,
313 kdu_instance
=kdu_instance
,
314 namespace
=self
.namespace
,
316 operation
="rollback",
320 command
= "/usr/bin/helm3 rollback stable-openldap-0005399828 1 --namespace=testk8s --wait"
321 self
.helm_conn
._local
_async
_exec
.assert_called_once_with(
322 command
=command
, env
=self
.env
, raise_exception_on_error
=False
325 @asynctest.fail_on(active_handles
=True)
326 async def test_uninstall(self
):
327 kdu_instance
= "stable-openldap-0005399828"
329 "chart": "openldap-1.2.2",
330 "name": kdu_instance
,
331 "namespace": self
.namespace
,
333 "status": "DEPLOYED",
335 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
336 self
.helm_conn
._store
_status
= asynctest
.CoroutineMock()
337 self
.helm_conn
.get_instance_info
= asynctest
.CoroutineMock(
338 return_value
=instance_info
341 await self
.helm_conn
.uninstall(self
.cluster_uuid
, kdu_instance
)
342 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
343 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(
344 from_path
=self
.cluster_id
346 command
= "/usr/bin/helm3 uninstall {} --namespace={}".format(
347 kdu_instance
, self
.namespace
349 self
.helm_conn
._local
_async
_exec
.assert_called_once_with(
350 command
=command
, env
=self
.env
, raise_exception_on_error
=True
353 @asynctest.fail_on(active_handles
=True)
354 async def test_get_services(self
):
355 kdu_instance
= "test_services_1"
356 service
= {"name": "testservice", "type": "LoadBalancer"}
357 self
.helm_conn
._local
_async
_exec
_pipe
= asynctest
.CoroutineMock(
360 self
.helm_conn
._parse
_services
= Mock(return_value
=["testservice"])
361 self
.helm_conn
._get
_service
= asynctest
.CoroutineMock(return_value
=service
)
363 services
= await self
.helm_conn
.get_services(
364 self
.cluster_uuid
, kdu_instance
, self
.namespace
366 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
367 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(
368 from_path
=self
.cluster_id
370 self
.helm_conn
._parse
_services
.assert_called_once()
371 command1
= "/usr/bin/helm3 get manifest {} --namespace=testk8s".format(
374 command2
= "/usr/bin/kubectl get --namespace={} -f -".format(self
.namespace
)
375 self
.helm_conn
._local
_async
_exec
_pipe
.assert_called_once_with(
376 command1
, command2
, env
=self
.env
, raise_exception_on_error
=True
379 services
, [service
], "Invalid service returned from get_service"
382 @asynctest.fail_on(active_handles
=True)
383 async def test_get_service(self
):
384 service_name
= "service1"
386 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
387 await self
.helm_conn
.get_service(
388 self
.cluster_uuid
, service_name
, self
.namespace
391 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
392 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(
393 from_path
=self
.cluster_id
396 "/usr/bin/kubectl --kubeconfig=./tmp/helm3_cluster_id/.kube/config "
397 "--namespace=testk8s get service service1 -o=yaml"
399 self
.helm_conn
._local
_async
_exec
.assert_called_once_with(
400 command
=command
, env
=self
.env
, raise_exception_on_error
=True
403 @asynctest.fail_on(active_handles
=True)
404 async def test_inspect_kdu(self
):
405 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
407 kdu_model
= "stable/openldap:1.2.4"
408 repo_url
= "https://kubernetes-charts.storage.googleapis.com/"
409 await self
.helm_conn
.inspect_kdu(kdu_model
, repo_url
)
412 "/usr/bin/helm3 show all openldap --repo "
413 "https://kubernetes-charts.storage.googleapis.com/ "
416 self
.helm_conn
._local
_async
_exec
.assert_called_with(
417 command
=command
, encode_utf8
=True
420 @asynctest.fail_on(active_handles
=True)
421 async def test_help_kdu(self
):
422 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
424 kdu_model
= "stable/openldap:1.2.4"
425 repo_url
= "https://kubernetes-charts.storage.googleapis.com/"
426 await self
.helm_conn
.help_kdu(kdu_model
, repo_url
)
429 "/usr/bin/helm3 show readme openldap --repo "
430 "https://kubernetes-charts.storage.googleapis.com/ "
433 self
.helm_conn
._local
_async
_exec
.assert_called_with(
434 command
=command
, encode_utf8
=True
437 @asynctest.fail_on(active_handles
=True)
438 async def test_values_kdu(self
):
439 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
441 kdu_model
= "stable/openldap:1.2.4"
442 repo_url
= "https://kubernetes-charts.storage.googleapis.com/"
443 await self
.helm_conn
.values_kdu(kdu_model
, repo_url
)
446 "/usr/bin/helm3 show values openldap --repo "
447 "https://kubernetes-charts.storage.googleapis.com/ "
450 self
.helm_conn
._local
_async
_exec
.assert_called_with(
451 command
=command
, encode_utf8
=True
454 @asynctest.fail_on(active_handles
=True)
455 async def test_instances_list(self
):
456 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
458 await self
.helm_conn
.instances_list(self
.cluster_uuid
)
459 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
460 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(
461 from_path
=self
.cluster_id
463 command
= "/usr/bin/helm3 list --all-namespaces --output yaml"
464 self
.helm_conn
._local
_async
_exec
.assert_called_once_with(
465 command
=command
, env
=self
.env
, raise_exception_on_error
=True
468 @asynctest.fail_on(active_handles
=True)
469 async def test_status_kdu(self
):
470 kdu_instance
= "stable-openldap-0005399828"
471 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
473 await self
.helm_conn
._status
_kdu
(
474 self
.cluster_id
, kdu_instance
, self
.namespace
, return_text
=True
476 command
= "/usr/bin/helm3 status {} --namespace={} --output yaml".format(
477 kdu_instance
, self
.namespace
479 self
.helm_conn
._local
_async
_exec
.assert_called_once_with(
482 raise_exception_on_error
=True,
483 show_error_log
=False,
486 @asynctest.fail_on(active_handles
=True)
487 async def test_store_status(self
):
488 kdu_instance
= "stable-openldap-0005399828"
492 "description": "Install complete",
495 "notes": "The openldap helm chart has been installed",
499 self
.helm_conn
._status
_kdu
= asynctest
.CoroutineMock(return_value
=status
)
500 self
.helm_conn
.write_app_status_to_db
= asynctest
.CoroutineMock(
504 await self
.helm_conn
._store
_status
(
505 cluster_id
=self
.cluster_id
,
506 kdu_instance
=kdu_instance
,
507 namespace
=self
.namespace
,
513 self
.helm_conn
._status
_kdu
.assert_called_once_with(
514 cluster_id
=self
.cluster_id
,
515 kdu_instance
=kdu_instance
,
516 namespace
=self
.namespace
,
519 self
.helm_conn
.write_app_status_to_db
.assert_called_once_with(
521 status
="Install complete",
522 detailed_status
=str(status
),
526 @asynctest.fail_on(active_handles
=True)
527 async def test_reset_uninstall_false(self
):
528 self
.helm_conn
._uninstall
_sw
= asynctest
.CoroutineMock()
530 await self
.helm_conn
.reset(self
.cluster_uuid
, force
=False, uninstall_sw
=False)
531 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
532 self
.helm_conn
.fs
.file_delete
.assert_called_once_with(
533 self
.cluster_id
, ignore_non_exist
=True
535 self
.helm_conn
._uninstall
_sw
.assert_not_called()
537 @asynctest.fail_on(active_handles
=True)
538 async def test_reset_uninstall(self
):
539 kdu_instance
= "stable-openldap-0021099429"
542 "app_version": "2.4.48",
543 "chart": "openldap-1.2.3",
544 "name": kdu_instance
,
545 "namespace": self
.namespace
,
547 "status": "deployed",
548 "updated": "2020-10-30 11:11:20.376744191 +0000 UTC",
551 self
.helm_conn
._uninstall
_sw
= asynctest
.CoroutineMock()
552 self
.helm_conn
.instances_list
= asynctest
.CoroutineMock(return_value
=instances
)
553 self
.helm_conn
.uninstall
= asynctest
.CoroutineMock()
555 await self
.helm_conn
.reset(self
.cluster_uuid
, force
=True, uninstall_sw
=True)
556 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
557 self
.helm_conn
.fs
.file_delete
.assert_called_once_with(
558 self
.cluster_id
, ignore_non_exist
=True
560 self
.helm_conn
.instances_list
.assert_called_once_with(
561 cluster_uuid
=self
.cluster_uuid
563 self
.helm_conn
.uninstall
.assert_called_once_with(
564 cluster_uuid
=self
.cluster_uuid
, kdu_instance
=kdu_instance
566 self
.helm_conn
._uninstall
_sw
.assert_called_once_with(
567 self
.cluster_id
, self
.namespace
570 @asynctest.fail_on(active_handles
=True)
571 async def test_sync_repos_add(self
):
575 "url": "https://kubernetes-charts.storage.googleapis.com/",
578 self
.helm_conn
.repo_list
= asynctest
.CoroutineMock(return_value
=repo_list
)
580 def get_one_result(*args
, **kwargs
):
581 if args
[0] == "k8sclusters":
584 "helm_chart_repos": ["4b5550a9-990d-4d95-8a48-1f4614d6ac9c"]
587 elif args
[0] == "k8srepos":
589 "_id": "4b5550a9-990d-4d95-8a48-1f4614d6ac9c",
590 "type": "helm-chart",
592 "url": "https://charts.bitnami.com/bitnami",
595 self
.helm_conn
.db
.get_one
= asynctest
.Mock()
596 self
.helm_conn
.db
.get_one
.side_effect
= get_one_result
598 self
.helm_conn
.repo_add
= asynctest
.CoroutineMock()
599 self
.helm_conn
.repo_remove
= asynctest
.CoroutineMock()
601 deleted_repo_list
, added_repo_dict
= await self
.helm_conn
.synchronize_repos(
604 self
.helm_conn
.repo_remove
.assert_not_called()
605 self
.helm_conn
.repo_add
.assert_called_once_with(
606 self
.cluster_uuid
, "bitnami", "https://charts.bitnami.com/bitnami"
608 self
.assertEqual(deleted_repo_list
, [], "Deleted repo list should be empty")
611 {"4b5550a9-990d-4d95-8a48-1f4614d6ac9c": "bitnami"},
612 "Repos added should include only one bitnami",
615 @asynctest.fail_on(active_handles
=True)
616 async def test_sync_repos_delete(self
):
620 "url": "https://kubernetes-charts.storage.googleapis.com/",
622 {"name": "bitnami", "url": "https://charts.bitnami.com/bitnami"},
624 self
.helm_conn
.repo_list
= asynctest
.CoroutineMock(return_value
=repo_list
)
626 def get_one_result(*args
, **kwargs
):
627 if args
[0] == "k8sclusters":
628 return {"_admin": {"helm_chart_repos": []}}
630 self
.helm_conn
.db
.get_one
= asynctest
.Mock()
631 self
.helm_conn
.db
.get_one
.side_effect
= get_one_result
633 self
.helm_conn
.repo_add
= asynctest
.CoroutineMock()
634 self
.helm_conn
.repo_remove
= asynctest
.CoroutineMock()
636 deleted_repo_list
, added_repo_dict
= await self
.helm_conn
.synchronize_repos(
639 self
.helm_conn
.repo_add
.assert_not_called()
640 self
.helm_conn
.repo_remove
.assert_called_once_with(self
.cluster_uuid
, "bitnami")
642 deleted_repo_list
, ["bitnami"], "Deleted repo list should be bitnami"
644 self
.assertEqual(added_repo_dict
, {}, "No repos should be added")