n2vc/tests/unit/test_k8s_helm3_conn.py

   1 ##
   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
   5 #
   6 #         http://www.apache.org/licenses/LICENSE-2.0
   7 #
   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
  12 # under the License.
  13 #
  14 # For those usages not covered by the Apache License, Version 2.0 please
  15 # contact: alfonso.tiernosepulveda@telefonica.com
  16 ##
  17
  18 import asynctest
  19 import logging
  20
  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
  25
  26 __author__ = "Isabel Lloret <illoret@indra.es>"
  27
  28
  29 class TestK8sHelm3Conn(asynctest.TestCase):
  30     logging.basicConfig(level=logging.DEBUG)
  31     logger = logging.getLogger(__name__)
  32     logger.setLevel(logging.DEBUG)
  33
  34     @patch("n2vc.k8s_helm_base_conn.EnvironConfig")
  35     async def setUp(self, mock_env):
  36         mock_env.return_value = {
  37             "stablerepourl": "https://charts.helm.sh/stable"
  38         }
  39         self.db = Mock(DbMemory())
  40         self.fs = asynctest.Mock(FsLocal())
  41         self.fs.path = "./tmp/"
  42         self.namespace = "testk8s"
  43         self.cluster_id = "helm3_cluster_id"
  44         self.cluster_uuid = "{}:{}".format(self.namespace, self.cluster_id)
  45         # pass fake kubectl and helm commands to make sure it does not call actual commands
  46         K8sHelm3Connector._check_file_exists = asynctest.Mock(return_value=True)
  47         cluster_dir = self.fs.path + self.cluster_id
  48         self.env = {
  49             "HELM_CACHE_HOME": "{}/.cache/helm".format(cluster_dir),
  50             "HELM_CONFIG_HOME": "{}/.config/helm".format(cluster_dir),
  51             "HELM_DATA_HOME": "{}/.local/share/helm".format(cluster_dir),
  52             "KUBECONFIG": "{}/.kube/config".format(cluster_dir),
  53         }
  54         self.helm_conn = K8sHelm3Connector(self.fs, self.db, log=self.logger)
  55         self.logger.debug("Set up executed")
  56
  57     @asynctest.fail_on(active_handles=True)
  58     async def test_init_env(self):
  59         k8s_creds = "false_credentials_string"
  60         self.helm_conn._get_namespaces = asynctest.CoroutineMock(return_value=[])
  61         self.helm_conn._create_namespace = asynctest.CoroutineMock()
  62         self.helm_conn.repo_list = asynctest.CoroutineMock(return_value=[])
  63         self.helm_conn.repo_add = asynctest.CoroutineMock()
  64
  65         k8scluster_uuid, installed = await self.helm_conn.init_env(
  66             k8s_creds, namespace=self.namespace, reuse_cluster_uuid=self.cluster_id
  67         )
  68
  69         self.assertEqual(
  70             k8scluster_uuid,
  71             "{}:{}".format(self.namespace, self.cluster_id),
  72             "Check cluster_uuid format: <namespace>.<cluster_id>",
  73         )
  74         self.helm_conn._get_namespaces.assert_called_once_with(self.cluster_id)
  75         self.helm_conn._create_namespace.assert_called_once_with(
  76             self.cluster_id, self.namespace
  77         )
  78         self.helm_conn.repo_list.assert_called_once_with(k8scluster_uuid)
  79         self.helm_conn.repo_add.assert_called_once_with(
  80             k8scluster_uuid, "stable", "https://charts.helm.sh/stable"
  81         )
  82         self.helm_conn.fs.reverse_sync.assert_called_once_with(
  83             from_path=self.cluster_id
  84         )
  85         self.logger.debug(f"cluster_uuid: {k8scluster_uuid}")
  86
  87     @asynctest.fail_on(active_handles=True)
  88     async def test_repo_add(self):
  89         repo_name = "bitnami"
  90         repo_url = "https://charts.bitnami.com/bitnami"
  91         self.helm_conn._local_async_exec = asynctest.CoroutineMock(return_value=("", 0))
  92
  93         await self.helm_conn.repo_add(self.cluster_uuid, repo_name, repo_url)
  94
  95         self.helm_conn.fs.sync.assert_called_once_with(from_path=self.cluster_id)
  96         self.helm_conn.fs.reverse_sync.assert_called_once_with(
  97             from_path=self.cluster_id
  98         )
  99         self.assertEqual(
 100             self.helm_conn._local_async_exec.call_count,
 101             2,
 102             "local_async_exec expected 2 calls, called {}".format(
 103                 self.helm_conn._local_async_exec.call_count
 104             ),
 105         )
 106
 107         repo_update_command = "/usr/bin/helm3 repo update"
 108         repo_add_command = "/usr/bin/helm3 repo add {} {}".format(repo_name, repo_url)
 109         calls = self.helm_conn._local_async_exec.call_args_list
 110         call0_kargs = calls[0][1]
 111         self.assertEqual(
 112             call0_kargs.get("command"),
 113             repo_update_command,
 114             "Invalid repo update command: {}".format(call0_kargs.get("command")),
 115         )
 116         self.assertEqual(
 117             call0_kargs.get("env"),
 118             self.env,
 119             "Invalid env for update command: {}".format(call0_kargs.get("env")),
 120         )
 121         call1_kargs = calls[1][1]
 122         self.assertEqual(
 123             call1_kargs.get("command"),
 124             repo_add_command,
 125             "Invalid repo add command: {}".format(call1_kargs.get("command")),
 126         )
 127         self.assertEqual(
 128             call1_kargs.get("env"),
 129             self.env,
 130             "Invalid env for add command: {}".format(call1_kargs.get("env")),
 131         )
 132
 133     @asynctest.fail_on(active_handles=True)
 134     async def test_repo_list(self):
 135
 136         self.helm_conn._local_async_exec = asynctest.CoroutineMock(return_value=("", 0))
 137
 138         await self.helm_conn.repo_list(self.cluster_uuid)
 139
 140         self.helm_conn.fs.sync.assert_called_once_with(from_path=self.cluster_id)
 141         self.helm_conn.fs.reverse_sync.assert_called_once_with(
 142             from_path=self.cluster_id
 143         )
 144         command = "/usr/bin/helm3 repo list --output yaml"
 145         self.helm_conn._local_async_exec.assert_called_with(
 146             command=command, env=self.env, raise_exception_on_error=False
 147         )
 148
 149     @asynctest.fail_on(active_handles=True)
 150     async def test_repo_remove(self):
 151
 152         self.helm_conn._local_async_exec = asynctest.CoroutineMock(return_value=("", 0))
 153         repo_name = "bitnami"
 154         await self.helm_conn.repo_remove(self.cluster_uuid, repo_name)
 155
 156         self.helm_conn.fs.sync.assert_called_once_with(from_path=self.cluster_id)
 157         self.helm_conn.fs.reverse_sync.assert_called_once_with(
 158             from_path=self.cluster_id
 159         )
 160         command = "/usr/bin/helm3 repo remove {}".format(repo_name)
 161         self.helm_conn._local_async_exec.assert_called_with(
 162             command=command, env=self.env, raise_exception_on_error=True
 163         )
 164
 165     @asynctest.fail_on(active_handles=True)
 166     async def test_install(self):
 167         kdu_model = "stable/openldap:1.2.2"
 168         kdu_instance = "stable-openldap-0005399828"
 169         db_dict = {}
 170         self.helm_conn._local_async_exec = asynctest.CoroutineMock(return_value=("", 0))
 171         self.helm_conn._status_kdu = asynctest.CoroutineMock(return_value=None)
 172         self.helm_conn._store_status = asynctest.CoroutineMock()
 173         self.kdu_instance = "stable-openldap-0005399828"
 174         self.helm_conn.generate_kdu_instance_name = Mock(return_value=self.kdu_instance)
 175         self.helm_conn._get_namespaces = asynctest.CoroutineMock(return_value=[])
 176         self.helm_conn._create_namespace = asynctest.CoroutineMock()
 177
 178         await self.helm_conn.install(
 179             self.cluster_uuid,
 180             kdu_model,
 181             self.kdu_instance,
 182             atomic=True,
 183             namespace=self.namespace,
 184             db_dict=db_dict,
 185         )
 186
 187         self.helm_conn._get_namespaces.assert_called_once()
 188         self.helm_conn._create_namespace.assert_called_once_with(
 189             self.cluster_id, self.namespace
 190         )
 191         self.helm_conn.fs.sync.assert_called_once_with(from_path=self.cluster_id)
 192         self.helm_conn.fs.reverse_sync.assert_called_once_with(
 193             from_path=self.cluster_id
 194         )
 195         self.helm_conn._store_status.assert_called_with(
 196             cluster_id=self.cluster_id,
 197             kdu_instance=kdu_instance,
 198             namespace=self.namespace,
 199             db_dict=db_dict,
 200             operation="install",
 201             run_once=True,
 202             check_every=0,
 203         )
 204         command = (
 205             "/usr/bin/helm3 install stable-openldap-0005399828 --atomic --output yaml   "
 206             "--timeout 300s --namespace testk8s stable/openldap --version 1.2.2"
 207         )
 208         self.helm_conn._local_async_exec.assert_called_once_with(
 209             command=command, env=self.env, raise_exception_on_error=False
 210         )
 211
 212     @asynctest.fail_on(active_handles=True)
 213     async def test_upgrade(self):
 214         kdu_model = "stable/openldap:1.2.3"
 215         kdu_instance = "stable-openldap-0005399828"
 216         db_dict = {}
 217         instance_info = {
 218             "chart": "openldap-1.2.2",
 219             "name": kdu_instance,
 220             "namespace": self.namespace,
 221             "revision": 1,
 222             "status": "DEPLOYED",
 223         }
 224         self.helm_conn._local_async_exec = asynctest.CoroutineMock(return_value=("", 0))
 225         self.helm_conn._store_status = asynctest.CoroutineMock()
 226         self.helm_conn.get_instance_info = asynctest.CoroutineMock(
 227             return_value=instance_info
 228         )
 229
 230         await self.helm_conn.upgrade(
 231             self.cluster_uuid, kdu_instance, kdu_model, atomic=True, db_dict=db_dict
 232         )
 233         self.helm_conn.fs.sync.assert_called_once_with(from_path=self.cluster_id)
 234         self.helm_conn.fs.reverse_sync.assert_called_once_with(
 235             from_path=self.cluster_id
 236         )
 237         self.helm_conn._store_status.assert_called_with(
 238             cluster_id=self.cluster_id,
 239             kdu_instance=kdu_instance,
 240             namespace=self.namespace,
 241             db_dict=db_dict,
 242             operation="upgrade",
 243             run_once=True,
 244             check_every=0,
 245         )
 246         command = (
 247             "/usr/bin/helm3 upgrade stable-openldap-0005399828 stable/openldap "
 248             "--namespace testk8s --atomic --output yaml  --timeout 300s  "
 249             "--version 1.2.3"
 250         )
 251         self.helm_conn._local_async_exec.assert_called_once_with(
 252             command=command, env=self.env, raise_exception_on_error=False
 253         )
 254
 255     @asynctest.fail_on(active_handles=True)
 256     async def test_rollback(self):
 257         kdu_instance = "stable-openldap-0005399828"
 258         db_dict = {}
 259         instance_info = {
 260             "chart": "openldap-1.2.3",
 261             "name": kdu_instance,
 262             "namespace": self.namespace,
 263             "revision": 2,
 264             "status": "DEPLOYED",
 265         }
 266         self.helm_conn._local_async_exec = asynctest.CoroutineMock(return_value=("", 0))
 267         self.helm_conn._store_status = asynctest.CoroutineMock()
 268         self.helm_conn.get_instance_info = asynctest.CoroutineMock(
 269             return_value=instance_info
 270         )
 271
 272         await self.helm_conn.rollback(
 273             self.cluster_uuid, kdu_instance=kdu_instance, revision=1, db_dict=db_dict
 274         )
 275         self.helm_conn.fs.sync.assert_called_once_with(from_path=self.cluster_id)
 276         self.helm_conn.fs.reverse_sync.assert_called_once_with(
 277             from_path=self.cluster_id
 278         )
 279         self.helm_conn._store_status.assert_called_with(
 280             cluster_id=self.cluster_id,
 281             kdu_instance=kdu_instance,
 282             namespace=self.namespace,
 283             db_dict=db_dict,
 284             operation="rollback",
 285             run_once=True,
 286             check_every=0,
 287         )
 288         command = "/usr/bin/helm3 rollback stable-openldap-0005399828 1 --namespace=testk8s --wait"
 289         self.helm_conn._local_async_exec.assert_called_once_with(
 290             command=command, env=self.env, raise_exception_on_error=False
 291         )
 292
 293     @asynctest.fail_on(active_handles=True)
 294     async def test_uninstall(self):
 295         kdu_instance = "stable-openldap-0005399828"
 296         instance_info = {
 297             "chart": "openldap-1.2.2",
 298             "name": kdu_instance,
 299             "namespace": self.namespace,
 300             "revision": 3,
 301             "status": "DEPLOYED",
 302         }
 303         self.helm_conn._local_async_exec = asynctest.CoroutineMock(return_value=("", 0))
 304         self.helm_conn._store_status = asynctest.CoroutineMock()
 305         self.helm_conn.get_instance_info = asynctest.CoroutineMock(
 306             return_value=instance_info
 307         )
 308
 309         await self.helm_conn.uninstall(self.cluster_uuid, kdu_instance)
 310         self.helm_conn.fs.sync.assert_called_once_with(from_path=self.cluster_id)
 311         self.helm_conn.fs.reverse_sync.assert_called_once_with(
 312             from_path=self.cluster_id
 313         )
 314         command = "/usr/bin/helm3 uninstall {} --namespace={}".format(
 315             kdu_instance, self.namespace
 316         )
 317         self.helm_conn._local_async_exec.assert_called_once_with(
 318             command=command, env=self.env, raise_exception_on_error=True
 319         )
 320
 321     @asynctest.fail_on(active_handles=True)
 322     async def test_get_services(self):
 323         kdu_instance = "test_services_1"
 324         service = {"name": "testservice", "type": "LoadBalancer"}
 325         self.helm_conn._local_async_exec_pipe = asynctest.CoroutineMock(
 326             return_value=("", 0)
 327         )
 328         self.helm_conn._parse_services = Mock(return_value=["testservice"])
 329         self.helm_conn._get_service = asynctest.CoroutineMock(return_value=service)
 330
 331         services = await self.helm_conn.get_services(
 332             self.cluster_uuid, kdu_instance, self.namespace
 333         )
 334         self.helm_conn.fs.sync.assert_called_once_with(from_path=self.cluster_id)
 335         self.helm_conn.fs.reverse_sync.assert_called_once_with(
 336             from_path=self.cluster_id
 337         )
 338         self.helm_conn._parse_services.assert_called_once()
 339         command1 = "/usr/bin/helm3 get manifest {} --namespace=testk8s".format(
 340             kdu_instance
 341         )
 342         command2 = "/usr/bin/kubectl get --namespace={} -f -".format(self.namespace)
 343         self.helm_conn._local_async_exec_pipe.assert_called_once_with(
 344             command1, command2, env=self.env, raise_exception_on_error=True
 345         )
 346         self.assertEqual(
 347             services, [service], "Invalid service returned from get_service"
 348         )
 349
 350     @asynctest.fail_on(active_handles=True)
 351     async def test_get_service(self):
 352         service_name = "service1"
 353
 354         self.helm_conn._local_async_exec = asynctest.CoroutineMock(return_value=("", 0))
 355         await self.helm_conn.get_service(
 356             self.cluster_uuid, service_name, self.namespace
 357         )
 358
 359         self.helm_conn.fs.sync.assert_called_once_with(from_path=self.cluster_id)
 360         self.helm_conn.fs.reverse_sync.assert_called_once_with(
 361             from_path=self.cluster_id
 362         )
 363         command = (
 364             "/usr/bin/kubectl --kubeconfig=./tmp/helm3_cluster_id/.kube/config "
 365             "--namespace=testk8s get service service1 -o=yaml"
 366         )
 367         self.helm_conn._local_async_exec.assert_called_once_with(
 368             command=command, env=self.env, raise_exception_on_error=True
 369         )
 370
 371     @asynctest.fail_on(active_handles=True)
 372     async def test_inspect_kdu(self):
 373         self.helm_conn._local_async_exec = asynctest.CoroutineMock(return_value=("", 0))
 374
 375         kdu_model = "stable/openldap:1.2.4"
 376         repo_url = "https://kubernetes-charts.storage.googleapis.com/"
 377         await self.helm_conn.inspect_kdu(kdu_model, repo_url)
 378
 379         command = (
 380             "/usr/bin/helm3 show all openldap --repo "
 381             "https://kubernetes-charts.storage.googleapis.com/ "
 382             "--version 1.2.4"
 383         )
 384         self.helm_conn._local_async_exec.assert_called_with(
 385             command=command, encode_utf8=True
 386         )
 387
 388     @asynctest.fail_on(active_handles=True)
 389     async def test_help_kdu(self):
 390         self.helm_conn._local_async_exec = asynctest.CoroutineMock(return_value=("", 0))
 391
 392         kdu_model = "stable/openldap:1.2.4"
 393         repo_url = "https://kubernetes-charts.storage.googleapis.com/"
 394         await self.helm_conn.help_kdu(kdu_model, repo_url)
 395
 396         command = (
 397             "/usr/bin/helm3 show readme openldap --repo "
 398             "https://kubernetes-charts.storage.googleapis.com/ "
 399             "--version 1.2.4"
 400         )
 401         self.helm_conn._local_async_exec.assert_called_with(
 402             command=command, encode_utf8=True
 403         )
 404
 405     @asynctest.fail_on(active_handles=True)
 406     async def test_values_kdu(self):
 407         self.helm_conn._local_async_exec = asynctest.CoroutineMock(return_value=("", 0))
 408
 409         kdu_model = "stable/openldap:1.2.4"
 410         repo_url = "https://kubernetes-charts.storage.googleapis.com/"
 411         await self.helm_conn.values_kdu(kdu_model, repo_url)
 412
 413         command = (
 414             "/usr/bin/helm3 show values openldap --repo "
 415             "https://kubernetes-charts.storage.googleapis.com/ "
 416             "--version 1.2.4"
 417         )
 418         self.helm_conn._local_async_exec.assert_called_with(
 419             command=command, encode_utf8=True
 420         )
 421
 422     @asynctest.fail_on(active_handles=True)
 423     async def test_instances_list(self):
 424         self.helm_conn._local_async_exec = asynctest.CoroutineMock(return_value=("", 0))
 425
 426         await self.helm_conn.instances_list(self.cluster_uuid)
 427         self.helm_conn.fs.sync.assert_called_once_with(from_path=self.cluster_id)
 428         self.helm_conn.fs.reverse_sync.assert_called_once_with(
 429             from_path=self.cluster_id
 430         )
 431         command = "/usr/bin/helm3 list --all-namespaces  --output yaml"
 432         self.helm_conn._local_async_exec.assert_called_once_with(
 433             command=command, env=self.env, raise_exception_on_error=True
 434         )
 435
 436     @asynctest.fail_on(active_handles=True)
 437     async def test_status_kdu(self):
 438         kdu_instance = "stable-openldap-0005399828"
 439         self.helm_conn._local_async_exec = asynctest.CoroutineMock(return_value=("", 0))
 440
 441         await self.helm_conn._status_kdu(
 442             self.cluster_id, kdu_instance, self.namespace, return_text=True
 443         )
 444         command = "/usr/bin/helm3 status {} --namespace={} --output yaml".format(
 445             kdu_instance, self.namespace
 446         )
 447         self.helm_conn._local_async_exec.assert_called_once_with(
 448             command=command,
 449             env=self.env,
 450             raise_exception_on_error=True,
 451             show_error_log=False,
 452         )
 453
 454     @asynctest.fail_on(active_handles=True)
 455     async def test_store_status(self):
 456         kdu_instance = "stable-openldap-0005399828"
 457         db_dict = {}
 458         status = {
 459             "info": {
 460                 "description": "Install complete",
 461                 "status": {
 462                     "code": "1",
 463                     "notes": "The openldap helm chart has been installed",
 464                 },
 465             }
 466         }
 467         self.helm_conn._status_kdu = asynctest.CoroutineMock(return_value=status)
 468         self.helm_conn.write_app_status_to_db = asynctest.CoroutineMock(
 469             return_value=status
 470         )
 471
 472         await self.helm_conn._store_status(
 473             cluster_id=self.cluster_id,
 474             kdu_instance=kdu_instance,
 475             namespace=self.namespace,
 476             db_dict=db_dict,
 477             operation="install",
 478             run_once=True,
 479             check_every=0,
 480         )
 481         self.helm_conn._status_kdu.assert_called_once_with(
 482             cluster_id=self.cluster_id,
 483             kdu_instance=kdu_instance,
 484             namespace=self.namespace,
 485             return_text=False,
 486         )
 487         self.helm_conn.write_app_status_to_db.assert_called_once_with(
 488             db_dict=db_dict,
 489             status="Install complete",
 490             detailed_status=str(status),
 491             operation="install",
 492         )
 493
 494     @asynctest.fail_on(active_handles=True)
 495     async def test_reset_uninstall_false(self):
 496         self.helm_conn._uninstall_sw = asynctest.CoroutineMock()
 497
 498         await self.helm_conn.reset(self.cluster_uuid, force=False, uninstall_sw=False)
 499         self.helm_conn.fs.sync.assert_called_once_with(from_path=self.cluster_id)
 500         self.helm_conn.fs.file_delete.assert_called_once_with(
 501             self.cluster_id, ignore_non_exist=True
 502         )
 503         self.helm_conn._uninstall_sw.assert_not_called()
 504
 505     @asynctest.fail_on(active_handles=True)
 506     async def test_reset_uninstall(self):
 507         kdu_instance = "stable-openldap-0021099429"
 508         instances = [
 509             {
 510                 "app_version": "2.4.48",
 511                 "chart": "openldap-1.2.3",
 512                 "name": kdu_instance,
 513                 "namespace": self.namespace,
 514                 "revision": "1",
 515                 "status": "deployed",
 516                 "updated": "2020-10-30 11:11:20.376744191 +0000 UTC",
 517             }
 518         ]
 519         self.helm_conn._uninstall_sw = asynctest.CoroutineMock()
 520         self.helm_conn.instances_list = asynctest.CoroutineMock(return_value=instances)
 521         self.helm_conn.uninstall = asynctest.CoroutineMock()
 522
 523         await self.helm_conn.reset(self.cluster_uuid, force=True, uninstall_sw=True)
 524         self.helm_conn.fs.sync.assert_called_once_with(from_path=self.cluster_id)
 525         self.helm_conn.fs.file_delete.assert_called_once_with(
 526             self.cluster_id, ignore_non_exist=True
 527         )
 528         self.helm_conn.instances_list.assert_called_once_with(
 529             cluster_uuid=self.cluster_uuid
 530         )
 531         self.helm_conn.uninstall.assert_called_once_with(
 532             cluster_uuid=self.cluster_uuid, kdu_instance=kdu_instance
 533         )
 534         self.helm_conn._uninstall_sw.assert_called_once_with(
 535             self.cluster_id, self.namespace
 536         )
 537
 538     @asynctest.fail_on(active_handles=True)
 539     async def test_sync_repos_add(self):
 540         repo_list = [
 541             {
 542                 "name": "stable",
 543                 "url": "https://kubernetes-charts.storage.googleapis.com/",
 544             }
 545         ]
 546         self.helm_conn.repo_list = asynctest.CoroutineMock(return_value=repo_list)
 547
 548         def get_one_result(*args, **kwargs):
 549             if args[0] == "k8sclusters":
 550                 return {
 551                     "_admin": {
 552                         "helm_chart_repos": ["4b5550a9-990d-4d95-8a48-1f4614d6ac9c"]
 553                     }
 554                 }
 555             elif args[0] == "k8srepos":
 556                 return {
 557                     "_id": "4b5550a9-990d-4d95-8a48-1f4614d6ac9c",
 558                     "type": "helm-chart",
 559                     "name": "bitnami",
 560                     "url": "https://charts.bitnami.com/bitnami",
 561                 }
 562
 563         self.helm_conn.db.get_one = asynctest.Mock()
 564         self.helm_conn.db.get_one.side_effect = get_one_result
 565
 566         self.helm_conn.repo_add = asynctest.CoroutineMock()
 567         self.helm_conn.repo_remove = asynctest.CoroutineMock()
 568
 569         deleted_repo_list, added_repo_dict = await self.helm_conn.synchronize_repos(
 570             self.cluster_uuid
 571         )
 572         self.helm_conn.repo_remove.assert_not_called()
 573         self.helm_conn.repo_add.assert_called_once_with(
 574             self.cluster_uuid, "bitnami", "https://charts.bitnami.com/bitnami"
 575         )
 576         self.assertEqual(deleted_repo_list, [], "Deleted repo list should be empty")
 577         self.assertEqual(
 578             added_repo_dict,
 579             {"4b5550a9-990d-4d95-8a48-1f4614d6ac9c": "bitnami"},
 580             "Repos added should include only one bitnami",
 581         )
 582
 583     @asynctest.fail_on(active_handles=True)
 584     async def test_sync_repos_delete(self):
 585         repo_list = [
 586             {
 587                 "name": "stable",
 588                 "url": "https://kubernetes-charts.storage.googleapis.com/",
 589             },
 590             {"name": "bitnami", "url": "https://charts.bitnami.com/bitnami"},
 591         ]
 592         self.helm_conn.repo_list = asynctest.CoroutineMock(return_value=repo_list)
 593
 594         def get_one_result(*args, **kwargs):
 595             if args[0] == "k8sclusters":
 596                 return {"_admin": {"helm_chart_repos": []}}
 597
 598         self.helm_conn.db.get_one = asynctest.Mock()
 599         self.helm_conn.db.get_one.side_effect = get_one_result
 600
 601         self.helm_conn.repo_add = asynctest.CoroutineMock()
 602         self.helm_conn.repo_remove = asynctest.CoroutineMock()
 603
 604         deleted_repo_list, added_repo_dict = await self.helm_conn.synchronize_repos(
 605             self.cluster_uuid
 606         )
 607         self.helm_conn.repo_add.assert_not_called()
 608         self.helm_conn.repo_remove.assert_called_once_with(self.cluster_uuid, "bitnami")
 609         self.assertEqual(
 610             deleted_repo_list, ["bitnami"], "Deleted repo list should be bitnami"
 611         )
 612         self.assertEqual(added_repo_dict, {}, "No repos should be added")