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
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 async def setUp(self
):
35 self
.db
= Mock(DbMemory())
36 self
.fs
= asynctest
.Mock(FsLocal())
37 self
.fs
.path
= "./tmp/"
38 self
.namespace
= "testk8s"
39 self
.cluster_id
= "helm3_cluster_id"
40 self
.cluster_uuid
= "{}:{}".format(self
.namespace
, self
.cluster_id
)
41 # pass fake kubectl and helm commands to make sure it does not call actual commands
42 K8sHelm3Connector
._check
_file
_exists
= asynctest
.Mock(return_value
=True)
43 cluster_dir
= self
.fs
.path
+ self
.cluster_id
45 "HELM_CACHE_HOME": "{}/.cache/helm".format(cluster_dir
),
46 "HELM_CONFIG_HOME": "{}/.config/helm".format(cluster_dir
),
47 "HELM_DATA_HOME": "{}/.local/share/helm".format(cluster_dir
),
48 "KUBECONFIG": "{}/.kube/config".format(cluster_dir
)
50 self
.helm_conn
= K8sHelm3Connector(self
.fs
, self
.db
,
52 self
.logger
.debug("Set up executed")
54 @asynctest.fail_on(active_handles
=True)
55 async def test_init_env(self
):
56 k8s_creds
= "false_credentials_string"
57 self
.helm_conn
._get
_namespaces
= asynctest
.CoroutineMock(return_value
=[])
58 self
.helm_conn
._create
_namespace
= asynctest
.CoroutineMock()
59 self
.helm_conn
.repo_list
= asynctest
.CoroutineMock(return_value
=[])
60 self
.helm_conn
.repo_add
= asynctest
.CoroutineMock()
62 k8scluster_uuid
, installed
= await self
.helm_conn
.init_env(
63 k8s_creds
, namespace
=self
.namespace
, reuse_cluster_uuid
=self
.cluster_id
)
65 self
.assertEqual(k8scluster_uuid
, "{}:{}".format(self
.namespace
, self
.cluster_id
),
66 "Check cluster_uuid format: <namespace>.<cluster_id>")
67 self
.helm_conn
._get
_namespaces
.assert_called_once_with(self
.cluster_id
)
68 self
.helm_conn
._create
_namespace
.assert_called_once_with(self
.cluster_id
, self
.namespace
)
69 self
.helm_conn
.repo_list
.assert_called_once_with(k8scluster_uuid
)
70 self
.helm_conn
.repo_add
.assert_called_once_with(
71 k8scluster_uuid
, "stable", "https://charts.helm.sh/stable")
72 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(from_path
=self
.cluster_id
)
73 self
.logger
.debug(f
"cluster_uuid: {k8scluster_uuid}")
75 @asynctest.fail_on(active_handles
=True)
76 async def test_repo_add(self
):
78 repo_url
= "https://charts.bitnami.com/bitnami"
79 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
81 await self
.helm_conn
.repo_add(self
.cluster_uuid
, repo_name
, repo_url
)
83 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
84 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(from_path
=self
.cluster_id
)
85 self
.assertEqual(self
.helm_conn
._local
_async
_exec
.call_count
, 2,
86 "local_async_exec expected 2 calls, called {}".format(
87 self
.helm_conn
._local
_async
_exec
.call_count
))
89 repo_update_command
= "/usr/bin/helm3 repo update"
90 repo_add_command
= "/usr/bin/helm3 repo add {} {}".format(repo_name
, repo_url
)
91 calls
= self
.helm_conn
._local
_async
_exec
.call_args_list
92 call0_kargs
= calls
[0][1]
93 self
.assertEqual(call0_kargs
.get("command"), repo_update_command
,
94 "Invalid repo update command: {}".format(call0_kargs
.get("command")))
95 self
.assertEqual(call0_kargs
.get("env"), self
.env
,
96 "Invalid env for update command: {}".format(call0_kargs
.get("env")))
97 call1_kargs
= calls
[1][1]
98 self
.assertEqual(call1_kargs
.get("command"), repo_add_command
,
99 "Invalid repo add command: {}".format(call1_kargs
.get("command")))
100 self
.assertEqual(call1_kargs
.get("env"), self
.env
,
101 "Invalid env for add command: {}".format(call1_kargs
.get("env")))
103 @asynctest.fail_on(active_handles
=True)
104 async def test_repo_list(self
):
106 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
108 await self
.helm_conn
.repo_list(self
.cluster_uuid
)
110 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
111 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(from_path
=self
.cluster_id
)
112 command
= "/usr/bin/helm3 repo list --output yaml"
113 self
.helm_conn
._local
_async
_exec
.assert_called_with(command
=command
, env
=self
.env
,
114 raise_exception_on_error
=False)
116 @asynctest.fail_on(active_handles
=True)
117 async def test_repo_remove(self
):
119 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
120 repo_name
= "bitnami"
121 await self
.helm_conn
.repo_remove(self
.cluster_uuid
, repo_name
)
123 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
124 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(from_path
=self
.cluster_id
)
125 command
= "/usr/bin/helm3 repo remove {}".format(repo_name
)
126 self
.helm_conn
._local
_async
_exec
.assert_called_with(command
=command
, env
=self
.env
,
127 raise_exception_on_error
=True)
129 @asynctest.fail_on(active_handles
=True)
130 async def test_install(self
):
131 kdu_model
= "stable/openldap:1.2.2"
132 kdu_instance
= "stable-openldap-0005399828"
134 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
135 self
.helm_conn
._status
_kdu
= asynctest
.CoroutineMock(return_value
=None)
136 self
.helm_conn
._store
_status
= asynctest
.CoroutineMock()
137 self
.kdu_instance
= "stable-openldap-0005399828"
138 self
.helm_conn
.generate_kdu_instance_name
= Mock(return_value
=self
.kdu_instance
)
139 self
.helm_conn
._get
_namespaces
= asynctest
.CoroutineMock(return_value
=[])
140 self
.helm_conn
._namespace
_exists
= asynctest
.CoroutineMock(side_effect
=self
.helm_conn
._namespace
_exists
)
141 self
.helm_conn
._create
_namespace
= asynctest
.CoroutineMock()
143 await self
.helm_conn
.install(
148 namespace
=self
.namespace
,
152 self
.helm_conn
._namespace
_exists
.assert_called_once()
153 self
.helm_conn
._get
_namespaces
.assert_called_once()
154 self
.helm_conn
._create
_namespace
.assert_called_once_with(self
.cluster_id
, self
.namespace
)
155 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
156 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(from_path
=self
.cluster_id
)
157 self
.helm_conn
._store
_status
.assert_called_with(cluster_id
=self
.cluster_id
,
158 kdu_instance
=kdu_instance
,
159 namespace
=self
.namespace
,
164 command
= "/usr/bin/helm3 install stable-openldap-0005399828 --atomic --output yaml " \
165 "--timeout 300s --namespace testk8s stable/openldap --version 1.2.2"
166 self
.helm_conn
._local
_async
_exec
.assert_called_once_with(command
=command
,
168 raise_exception_on_error
=False)
170 # Exception test if namespace could not being created for some reason
171 self
.helm_conn
._namespace
_exists
.return_value
= False
172 self
.helm_conn
._create
_namespace
.side_effect
= Exception()
174 with self
.assertRaises(K8sException
):
175 await self
.helm_conn
.install(
180 namespace
=self
.namespace
,
184 @asynctest.fail_on(active_handles
=True)
185 async def test_namespace_exists(self
):
186 self
.helm_conn
._get
_namespaces
= asynctest
.CoroutineMock()
188 self
.helm_conn
._get
_namespaces
.return_value
= ['testk8s', 'kube-system']
189 result
= await self
.helm_conn
._namespace
_exists
(self
.cluster_id
, self
.namespace
)
190 self
.helm_conn
._get
_namespaces
.assert_called_once()
191 self
.assertEqual(result
, True)
193 self
.helm_conn
._get
_namespaces
.reset_mock()
194 result
= await self
.helm_conn
._namespace
_exists
(self
.cluster_id
, 'none-exists-namespace')
195 self
.helm_conn
._get
_namespaces
.assert_called_once()
196 self
.assertEqual(result
, False)
198 @asynctest.fail_on(active_handles
=True)
199 async def test_upgrade(self
):
200 kdu_model
= "stable/openldap:1.2.3"
201 kdu_instance
= "stable-openldap-0005399828"
204 "chart": "openldap-1.2.2",
205 "name": kdu_instance
,
206 "namespace": self
.namespace
,
210 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
211 self
.helm_conn
._store
_status
= asynctest
.CoroutineMock()
212 self
.helm_conn
.get_instance_info
= asynctest
.CoroutineMock(return_value
=instance_info
)
214 await self
.helm_conn
.upgrade(self
.cluster_uuid
,
219 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
220 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(from_path
=self
.cluster_id
)
221 self
.helm_conn
._store
_status
.assert_called_with(cluster_id
=self
.cluster_id
,
222 kdu_instance
=kdu_instance
,
223 namespace
=self
.namespace
,
228 command
= "/usr/bin/helm3 upgrade stable-openldap-0005399828 stable/openldap " \
229 "--namespace testk8s --atomic --output yaml --timeout 300s " \
231 self
.helm_conn
._local
_async
_exec
.assert_called_once_with(command
=command
,
233 raise_exception_on_error
=False)
235 @asynctest.fail_on(active_handles
=True)
236 async def test_rollback(self
):
237 kdu_instance
= "stable-openldap-0005399828"
240 "chart": "openldap-1.2.3",
241 "name": kdu_instance
,
242 "namespace": self
.namespace
,
246 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
247 self
.helm_conn
._store
_status
= asynctest
.CoroutineMock()
248 self
.helm_conn
.get_instance_info
= asynctest
.CoroutineMock(return_value
=instance_info
)
250 await self
.helm_conn
.rollback(self
.cluster_uuid
,
251 kdu_instance
=kdu_instance
,
254 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
255 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(from_path
=self
.cluster_id
)
256 self
.helm_conn
._store
_status
.assert_called_with(cluster_id
=self
.cluster_id
,
257 kdu_instance
=kdu_instance
,
258 namespace
=self
.namespace
,
260 operation
="rollback",
263 command
= "/usr/bin/helm3 rollback stable-openldap-0005399828 1 --namespace=testk8s --wait"
264 self
.helm_conn
._local
_async
_exec
.assert_called_once_with(command
=command
,
266 raise_exception_on_error
=False)
268 @asynctest.fail_on(active_handles
=True)
269 async def test_uninstall(self
):
270 kdu_instance
= "stable-openldap-0005399828"
272 "chart": "openldap-1.2.2",
273 "name": kdu_instance
,
274 "namespace": self
.namespace
,
278 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
279 self
.helm_conn
._store
_status
= asynctest
.CoroutineMock()
280 self
.helm_conn
.get_instance_info
= asynctest
.CoroutineMock(return_value
=instance_info
)
282 await self
.helm_conn
.uninstall(self
.cluster_uuid
, kdu_instance
)
283 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
284 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(from_path
=self
.cluster_id
)
285 command
= "/usr/bin/helm3 uninstall {} --namespace={}".format(
286 kdu_instance
, self
.namespace
)
287 self
.helm_conn
._local
_async
_exec
.assert_called_once_with(command
=command
,
289 raise_exception_on_error
=True)
291 @asynctest.fail_on(active_handles
=True)
292 async def test_get_services(self
):
293 kdu_instance
= "test_services_1"
295 "name": "testservice",
296 "type": "LoadBalancer"
298 self
.helm_conn
._local
_async
_exec
_pipe
= asynctest
.CoroutineMock(return_value
=("", 0))
299 self
.helm_conn
._parse
_services
= Mock(return_value
=["testservice"])
300 self
.helm_conn
._get
_service
= asynctest
.CoroutineMock(return_value
=service
)
302 services
= await self
.helm_conn
.get_services(self
.cluster_uuid
, kdu_instance
,
304 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
305 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(from_path
=self
.cluster_id
)
306 self
.helm_conn
._parse
_services
.assert_called_once()
307 command1
= "/usr/bin/helm3 get manifest {} --namespace=testk8s".format(kdu_instance
)
308 command2
= "/usr/bin/kubectl get --namespace={} -f -".format(self
.namespace
)
309 self
.helm_conn
._local
_async
_exec
_pipe
.assert_called_once_with(command1
, command2
,
311 raise_exception_on_error
=True)
312 self
.assertEqual(services
, [service
], "Invalid service returned from get_service")
314 @asynctest.fail_on(active_handles
=True)
315 async def test_get_service(self
):
316 service_name
= "service1"
318 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
319 await self
.helm_conn
.get_service(self
.cluster_uuid
, service_name
, self
.namespace
)
321 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
322 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(from_path
=self
.cluster_id
)
323 command
= "/usr/bin/kubectl --kubeconfig=./tmp/helm3_cluster_id/.kube/config " \
324 "--namespace=testk8s get service service1 -o=yaml"
325 self
.helm_conn
._local
_async
_exec
.assert_called_once_with(command
=command
,
327 raise_exception_on_error
=True)
329 @asynctest.fail_on(active_handles
=True)
330 async def test_inspect_kdu(self
):
331 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
333 kdu_model
= "stable/openldap:1.2.4"
334 repo_url
= "https://kubernetes-charts.storage.googleapis.com/"
335 await self
.helm_conn
.inspect_kdu(kdu_model
, repo_url
)
337 command
= "/usr/bin/helm3 show all openldap --repo " \
338 "https://kubernetes-charts.storage.googleapis.com/ " \
340 self
.helm_conn
._local
_async
_exec
.assert_called_with(command
=command
, encode_utf8
=True)
342 @asynctest.fail_on(active_handles
=True)
343 async def test_help_kdu(self
):
344 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
346 kdu_model
= "stable/openldap:1.2.4"
347 repo_url
= "https://kubernetes-charts.storage.googleapis.com/"
348 await self
.helm_conn
.help_kdu(kdu_model
, repo_url
)
350 command
= "/usr/bin/helm3 show readme openldap --repo " \
351 "https://kubernetes-charts.storage.googleapis.com/ " \
353 self
.helm_conn
._local
_async
_exec
.assert_called_with(command
=command
, encode_utf8
=True)
355 @asynctest.fail_on(active_handles
=True)
356 async def test_values_kdu(self
):
357 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
359 kdu_model
= "stable/openldap:1.2.4"
360 repo_url
= "https://kubernetes-charts.storage.googleapis.com/"
361 await self
.helm_conn
.values_kdu(kdu_model
, repo_url
)
363 command
= "/usr/bin/helm3 show values openldap --repo " \
364 "https://kubernetes-charts.storage.googleapis.com/ " \
366 self
.helm_conn
._local
_async
_exec
.assert_called_with(command
=command
, encode_utf8
=True)
368 @asynctest.fail_on(active_handles
=True)
369 async def test_instances_list(self
):
370 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
372 await self
.helm_conn
.instances_list(self
.cluster_uuid
)
373 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
374 self
.helm_conn
.fs
.reverse_sync
.assert_called_once_with(from_path
=self
.cluster_id
)
375 command
= "/usr/bin/helm3 list --all-namespaces --output yaml"
376 self
.helm_conn
._local
_async
_exec
.assert_called_once_with(command
=command
,
378 raise_exception_on_error
=True)
380 @asynctest.fail_on(active_handles
=True)
381 async def test_status_kdu(self
):
382 kdu_instance
= "stable-openldap-0005399828"
383 self
.helm_conn
._local
_async
_exec
= asynctest
.CoroutineMock(return_value
=("", 0))
385 await self
.helm_conn
._status
_kdu
(self
.cluster_id
, kdu_instance
,
386 self
.namespace
, return_text
=True)
387 command
= "/usr/bin/helm3 status {} --namespace={} --output yaml".format(
388 kdu_instance
, self
.namespace
390 self
.helm_conn
._local
_async
_exec
.assert_called_once_with(command
=command
,
392 raise_exception_on_error
=True,
393 show_error_log
=False)
395 @asynctest.fail_on(active_handles
=True)
396 async def test_store_status(self
):
397 kdu_instance
= "stable-openldap-0005399828"
401 "description": "Install complete",
404 "notes": "The openldap helm chart has been installed"
408 self
.helm_conn
._status
_kdu
= asynctest
.CoroutineMock(return_value
=status
)
409 self
.helm_conn
.write_app_status_to_db
= asynctest
.CoroutineMock(return_value
=status
)
411 await self
.helm_conn
._store
_status
(cluster_id
=self
.cluster_id
,
412 kdu_instance
=kdu_instance
,
413 namespace
=self
.namespace
,
418 self
.helm_conn
._status
_kdu
.assert_called_once_with(cluster_id
=self
.cluster_id
,
419 kdu_instance
=kdu_instance
,
420 namespace
=self
.namespace
,
422 self
.helm_conn
.write_app_status_to_db
.assert_called_once_with(db_dict
=db_dict
,
423 status
="Install complete",
424 detailed_status
=str(status
),
427 @asynctest.fail_on(active_handles
=True)
428 async def test_reset_uninstall_false(self
):
429 self
.helm_conn
._uninstall
_sw
= asynctest
.CoroutineMock()
431 await self
.helm_conn
.reset(self
.cluster_uuid
, force
=False, uninstall_sw
=False)
432 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
433 self
.helm_conn
.fs
.file_delete
.assert_called_once_with(self
.cluster_id
,
434 ignore_non_exist
=True)
435 self
.helm_conn
._uninstall
_sw
.assert_not_called()
437 @asynctest.fail_on(active_handles
=True)
438 async def test_reset_uninstall(self
):
439 kdu_instance
= 'stable-openldap-0021099429'
442 'app_version': '2.4.48',
443 'chart': 'openldap-1.2.3',
444 'name': kdu_instance
,
445 'namespace': self
.namespace
,
447 'status': 'deployed',
448 'updated': '2020-10-30 11:11:20.376744191 +0000 UTC'
451 self
.helm_conn
._uninstall
_sw
= asynctest
.CoroutineMock()
452 self
.helm_conn
.instances_list
= asynctest
.CoroutineMock(return_value
=instances
)
453 self
.helm_conn
.uninstall
= asynctest
.CoroutineMock()
455 await self
.helm_conn
.reset(self
.cluster_uuid
, force
=True, uninstall_sw
=True)
456 self
.helm_conn
.fs
.sync
.assert_called_once_with(from_path
=self
.cluster_id
)
457 self
.helm_conn
.fs
.file_delete
.assert_called_once_with(self
.cluster_id
,
458 ignore_non_exist
=True)
459 self
.helm_conn
.instances_list
.assert_called_once_with(cluster_uuid
=self
.cluster_uuid
)
460 self
.helm_conn
.uninstall
.assert_called_once_with(cluster_uuid
=self
.cluster_uuid
,
461 kdu_instance
=kdu_instance
)
462 self
.helm_conn
._uninstall
_sw
.assert_called_once_with(self
.cluster_id
, self
.namespace
)
464 @asynctest.fail_on(active_handles
=True)
465 async def test_sync_repos_add(self
):
469 "url": "https://kubernetes-charts.storage.googleapis.com/"
472 self
.helm_conn
.repo_list
= asynctest
.CoroutineMock(return_value
=repo_list
)
474 def get_one_result(*args
, **kwargs
):
475 if args
[0] == "k8sclusters":
478 "helm_chart_repos": [
479 "4b5550a9-990d-4d95-8a48-1f4614d6ac9c"
483 elif args
[0] == "k8srepos":
485 "_id": "4b5550a9-990d-4d95-8a48-1f4614d6ac9c",
486 "type": "helm-chart",
488 "url": "https://charts.bitnami.com/bitnami"
490 self
.helm_conn
.db
.get_one
= asynctest
.Mock()
491 self
.helm_conn
.db
.get_one
.side_effect
= get_one_result
493 self
.helm_conn
.repo_add
= asynctest
.CoroutineMock()
494 self
.helm_conn
.repo_remove
= asynctest
.CoroutineMock()
496 deleted_repo_list
, added_repo_dict
= await self
.helm_conn
.synchronize_repos(
498 self
.helm_conn
.repo_remove
.assert_not_called()
499 self
.helm_conn
.repo_add
.assert_called_once_with(self
.cluster_uuid
, "bitnami",
500 "https://charts.bitnami.com/bitnami")
501 self
.assertEqual(deleted_repo_list
, [], "Deleted repo list should be empty")
502 self
.assertEqual(added_repo_dict
,
503 {"4b5550a9-990d-4d95-8a48-1f4614d6ac9c": "bitnami"},
504 "Repos added should include only one bitnami")
506 @asynctest.fail_on(active_handles
=True)
507 async def test_sync_repos_delete(self
):
511 "url": "https://kubernetes-charts.storage.googleapis.com/"
515 "url": "https://charts.bitnami.com/bitnami"
518 self
.helm_conn
.repo_list
= asynctest
.CoroutineMock(return_value
=repo_list
)
520 def get_one_result(*args
, **kwargs
):
521 if args
[0] == "k8sclusters":
524 "helm_chart_repos": []
528 self
.helm_conn
.db
.get_one
= asynctest
.Mock()
529 self
.helm_conn
.db
.get_one
.side_effect
= get_one_result
531 self
.helm_conn
.repo_add
= asynctest
.CoroutineMock()
532 self
.helm_conn
.repo_remove
= asynctest
.CoroutineMock()
534 deleted_repo_list
, added_repo_dict
= await self
.helm_conn
.synchronize_repos(
536 self
.helm_conn
.repo_add
.assert_not_called()
537 self
.helm_conn
.repo_remove
.assert_called_once_with(self
.cluster_uuid
, "bitnami")
538 self
.assertEqual(deleted_repo_list
, ["bitnami"], "Deleted repo list should be bitnami")
539 self
.assertEqual(added_repo_dict
, {}, "No repos should be added")