solve severity high issues of sonar cloud

src/cnaas_nms/db/git.py

@@ -186,7 +186,7 @@ def _refresh_repo_task_settings(job_id: Optional[int] = None) -> str:
     logger = get_logger()
     local_repo_path = app_settings.SETTINGS_LOCAL
     remote_repo_path = app_settings.SETTINGS_REMOTE
-    ret, changed_files = _refresh_repo_task(local_repo_path, remote_repo_path, job_id)
+    ret, changed_files = _refresh_repo_task(local_repo_path, remote_repo_path)
 
     try:
         rebuild_settings_cache()
@@ -239,7 +239,7 @@ def _refresh_repo_task_templates(job_id: Optional[int] = None) -> str:
     logger = get_logger()
     local_repo_path = app_settings.TEMPLATES_LOCAL
     remote_repo_path = app_settings.TEMPLATES_REMOTE
-    ret, changed_files = _refresh_repo_task(local_repo_path, remote_repo_path, job_id)
+    ret, changed_files = _refresh_repo_task(local_repo_path, remote_repo_path)
 
     logger.debug("Files changed in template repository: {}".format(changed_files or "None"))
     updated_devtypes = template_syncstatus(updated_templates=changed_files)
@@ -256,7 +256,7 @@ def _refresh_repo_task_templates(job_id: Optional[int] = None) -> str:
     return ret
 
 
-def _refresh_repo_task(local_repo_path, remote_repo_path, job_id: Optional[int] = None) -> Tuple[str, Set[str]]:
+def _refresh_repo_task(local_repo_path, remote_repo_path) -> Tuple[str, Set[str]]:
     """Should only be called by refresh_repo function."""
     logger = get_logger()
 

src/cnaas_nms/db/joblock.py

@@ -17,7 +17,7 @@ class Joblock(cnaas_nms.db.base.Base):
     job_id = mapped_column(Integer, ForeignKey("job.id"), unique=True, primary_key=True)
     job = relationship("Job", foreign_keys=[job_id])
     name = mapped_column(String(32), unique=True, nullable=False)
-    start_time = mapped_column(DateTime, default=datetime.datetime.now)  # onupdate=now
+    start_time = mapped_column(DateTime, default=datetime.datetime.now)
     abort = mapped_column(Boolean, default=False)
 
     def as_dict(self) -> dict:

src/cnaas_nms/db/mgmtdomain.py

@@ -3,7 +3,7 @@
 import ipaddress
 from ipaddress import IPv4Address, IPv6Address, ip_interface
 from itertools import dropwhile, islice
-from typing import List, Optional, Set, Union
+from typing import Optional, Set, Union
 
 from sqlalchemy import ForeignKey, Integer, String, Unicode, UniqueConstraint
 from sqlalchemy.orm import load_only, mapped_column, relationship
@@ -116,7 +116,7 @@ def is_taken(addr):
         else:
             mgmt_net = ip_interface(intf_addr).network
         candidates = islice(mgmt_net.hosts(), api_settings.MGMTDOMAIN_RESERVED_COUNT, None)
-        free_ips: List[IPAddress] = dropwhile(is_taken, candidates)  # type: ignore
+        free_ips = dropwhile(is_taken, candidates)  # type: ignore
         return next(free_ips, None)  # type: ignore
 
     @staticmethod

src/cnaas_nms/db/settings.py

@@ -470,6 +470,58 @@ def filter_yamldata(data: Union[List, dict], groups: List[str], hostname: str) -
     return filtered_yaml_data
 
 
+def recursive_filter_yamldata_dictionary(
+    data: dict, groups: List[str], hostname: str, recdepth=100
+) -> Union[List, dict, None]:
+    ret_d = {}
+    group_match = False
+    hostname_match = False
+    do_filter_group = False
+    do_filter_hostname = False
+    for key, value in data.items():
+        if not value:
+            ret_d[key] = value
+            continue
+        if key == "groups":
+            if not isinstance(value, list):  # Should already be checked by pydantic now
+                raise SettingsSyntaxError(
+                    "Groups field must be a list or empty (currently {}) in: {}".format(type(value).__name__, data)
+                )
+            do_filter_group = True
+            ret_d[key] = value
+            for group in value:
+                if group in groups:
+                    group_match = True
+        elif key == "devices":
+            if not isinstance(value, list):  # Should already be checked by pydantic now
+                raise SettingsSyntaxError(
+                    "Devices field must be a list or empty (currently {}) in: {}".format(type(value).__name__, data)
+                )
+            do_filter_hostname = True
+            ret_d[key] = value
+            if hostname in value:
+                hostname_match = True
+        else:
+            ret_v = recursive_filter_yamldata(value, groups, hostname, recdepth - 1)
+            if ret_v:
+                ret_d[key] = ret_v
+    if (do_filter_group or do_filter_hostname) and not group_match and not hostname_match:
+        return None
+    else:
+        return ret_d
+
+
+def recursive_filter_yamldata_list(
+    data: List, groups: List[str], hostname: str, recdepth=100
+) -> Union[List, dict, None]:
+    ret_l = []
+    for item in data:
+        f_item = recursive_filter_yamldata(item, groups, hostname, recdepth - 1)
+        if f_item:
+            ret_l.append(f_item)
+    return ret_l
+
+
 def recursive_filter_yamldata(
     data: Union[List, dict], groups: List[str], hostname: str, recdepth=100
 ) -> Union[List, dict, None]:
@@ -489,49 +541,9 @@ def recursive_filter_yamldata(
     if recdepth < 1:
         return data
     elif isinstance(data, list):
-        ret_l = []
-        for item in data:
-            f_item = recursive_filter_yamldata(item, groups, hostname, recdepth - 1)
-            if f_item:
-                ret_l.append(f_item)
-        return ret_l
+        return recursive_filter_yamldata_list(data, groups, hostname, recdepth)
     elif isinstance(data, dict):
-        ret_d = {}
-        group_match = False
-        hostname_match = False
-        do_filter_group = False
-        do_filter_hostname = False
-        for key, value in data.items():
-            if not value:
-                ret_d[key] = value
-                continue
-            if key == "groups":
-                if not isinstance(value, list):  # Should already be checked by pydantic now
-                    raise SettingsSyntaxError(
-                        "Groups field must be a list or empty (currently {}) in: {}".format(type(value).__name__, data)
-                    )
-                do_filter_group = True
-                ret_d[key] = value
-                for group in value:
-                    if group in groups:
-                        group_match = True
-            elif key == "devices":
-                if not isinstance(value, list):  # Should already be checked by pydantic now
-                    raise SettingsSyntaxError(
-                        "Devices field must be a list or empty (currently {}) in: {}".format(type(value).__name__, data)
-                    )
-                do_filter_hostname = True
-                ret_d[key] = value
-                if hostname in value:
-                    hostname_match = True
-            else:
-                ret_v = recursive_filter_yamldata(value, groups, hostname, recdepth - 1)
-                if ret_v:
-                    ret_d[key] = ret_v
-        if (do_filter_group or do_filter_hostname) and not group_match and not hostname_match:
-            return None
-        else:
-            return ret_d
+        return recursive_filter_yamldata_dictionary(data, groups, hostname, recdepth)
     else:
         return data
 

src/cnaas_nms/devicehandler/sync_devices.py

@@ -773,7 +773,7 @@ def confirm_devices(
     logger = get_logger()
     nr = cnaas_init()
 
-    nr_filtered, dev_count, skipped_hostnames = select_devices(nr, hostnames, resync=resync)
+    nr_filtered, dev_count, _ = select_devices(nr, hostnames, resync=resync)
 
     device_list = list(nr_filtered.inventory.hosts.keys())
     logger.info("Device(s) selected for commit-confirm ({}): {}".format(dev_count, ", ".join(device_list)))