From 15f4e0c5a610224a676411ed91f2f3da9e948c35 Mon Sep 17 00:00:00 2001 From: "Josephine.Rutten" Date: Fri, 8 Nov 2024 12:56:51 +0100 Subject: [PATCH] solve severity high issues of sonar cloud --- src/cnaas_nms/db/git.py | 6 +- src/cnaas_nms/db/joblock.py | 2 +- src/cnaas_nms/db/mgmtdomain.py | 4 +- src/cnaas_nms/db/settings.py | 96 ++++++++++++--------- src/cnaas_nms/devicehandler/sync_devices.py | 2 +- 5 files changed, 61 insertions(+), 49 deletions(-) diff --git a/src/cnaas_nms/db/git.py b/src/cnaas_nms/db/git.py index 8401d274..e77858c9 100644 --- a/src/cnaas_nms/db/git.py +++ b/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() diff --git a/src/cnaas_nms/db/joblock.py b/src/cnaas_nms/db/joblock.py index c4334af2..8333e4d2 100644 --- a/src/cnaas_nms/db/joblock.py +++ b/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: diff --git a/src/cnaas_nms/db/mgmtdomain.py b/src/cnaas_nms/db/mgmtdomain.py index 39bcdf21..9baab573 100644 --- a/src/cnaas_nms/db/mgmtdomain.py +++ b/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 diff --git a/src/cnaas_nms/db/settings.py b/src/cnaas_nms/db/settings.py index 77e7319e..c72ae99a 100644 --- a/src/cnaas_nms/db/settings.py +++ b/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 diff --git a/src/cnaas_nms/devicehandler/sync_devices.py b/src/cnaas_nms/devicehandler/sync_devices.py index 5b098824..7e46d269 100644 --- a/src/cnaas_nms/devicehandler/sync_devices.py +++ b/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)))