Use helpers and change conversion function names

Author: okorach-sonar

cli/config.py

@@ -32,6 +32,11 @@
 from cli import options
 from sonar import exceptions, errcodes, utilities, version
 from sonar.util import types, constants as c
+from sonar.util import platform_helper as pfhelp
+from sonar.util import project_helper as pjhelp
+from sonar.util import portfolio_helper as foliohelp
+from sonar.util import qualityprofile_helper as qphelp
+
 import sonar.logging as log
 from sonar import platform, rules, qualityprofiles, qualitygates, users, groups
 from sonar import projects, portfolios, applications
@@ -309,16 +314,16 @@ def convert_json(**kwargs) -> dict[str, Any]:
     with open(kwargs["convertFrom"], encoding="utf-8") as fd:
         old_json = json.loads(fd.read())
     mapping = {
-        "platform": platform.old_to_new_json,
-        "globalSettings": platform.global_settings_old_to_new_json,
-        "qualityProfiles": qualityprofiles.old_to_new_json,
-        "qualityGates": qualitygates.old_to_new_json,
-        "projects": projects.old_to_new_json,
-        "portfolios": portfolios.old_to_new_json,
+        "platform": pfhelp.convert_basics_json,
+        "globalSettings": pfhelp.convert_global_settings_json,
+        "qualityProfiles": qphelp.convert_qps_json,
+        "qualityGates": qualitygates.convert_qgs_json,
+        "projects": pjhelp.convert_projects_json,
+        "portfolios": foliohelp.convert_portfolios_json,
         "applications": applications.old_to_new_json,
-        "users": users.old_to_new_json,
-        "groups": groups.old_to_new_json,
-        "rules": rules.old_to_new_json,
+        "users": users.convert_users_json,
+        "groups": groups.convert_groups_json,
+        "rules": rules.convert_rules_json,
     }
     new_json = {}
     for k, func in mapping.items():

sonar/applications.py

@@ -518,7 +518,7 @@ def export(endpoint: pf.Platform, export_settings: types.ConfigSettings, **kwarg
     """
     check_supported(endpoint)
     write_q = kwargs.get("write_q", None)
-    key_regexp = kwargs.get("key_list", ".*")
+    key_regexp = kwargs.get("key_list", ".+")
 
     app_list = {k: v for k, v in get_list(endpoint).items() if not key_regexp or re.match(key_regexp, k)}
     apps_settings = []
@@ -547,7 +547,7 @@ def audit(endpoint: pf.Platform, audit_settings: types.ConfigSettings, **kwargs)
         return []
     log.info("--- Auditing applications ---")
     problems = []
-    key_regexp = kwargs.get("key_list", None) or ".*"
+    key_regexp = kwargs.get("key_list", ".+")
     for obj in [o for o in get_list(endpoint).values() if not key_regexp or re.match(key_regexp, o.key)]:
         problems += obj.audit(audit_settings, **kwargs)
     return problems

sonar/groups.py

@@ -434,6 +434,6 @@ def exists(endpoint: pf.Platform, name: str) -> bool:
     return Group.get_object(endpoint=endpoint, name=name) is not None
 
 
-def old_to_new_json(old_json: dict[str, Any]) -> dict[str, Any]:
+def convert_groups_json(old_json: dict[str, Any]) -> dict[str, Any]:
     """Converts sonar-config old groups JSON report format to new format"""
     return util.dict_to_list(old_json, "name", "description")

sonar/portfolios.py

@@ -455,26 +455,20 @@ def set_manual_mode(self) -> Portfolio:
 
     def set_tags_mode(self, tags: list[str], branch: Optional[str] = None) -> Portfolio:
         """Sets a portfolio to tags mode"""
-        if branch is None:
-            branch = c.DEFAULT_BRANCH
-        self.post("views/set_tags_mode", params={"portfolio": self.key, "tags": util.list_to_csv(tags), "branch": get_api_branch(branch)})
-        self._selection_mode = {_SELECTION_MODE_TAGS: tags, "branch": branch}
+        self.post("views/set_tags_mode", params={"portfolio": self.key, "tags": util.list_to_csv(tags), "branch": branch})
+        self._selection_mode = {_SELECTION_MODE_TAGS: tags, "branch": branch or c.DEFAULT_BRANCH}
         return self
 
     def set_regexp_mode(self, regexp: str, branch: Optional[str] = None) -> Portfolio:
         """Sets a portfolio to regexp mode"""
-        if branch is None:
-            branch = c.DEFAULT_BRANCH
-        self.post("views/set_regexp_mode", params={"portfolio": self.key, "regexp": regexp, "branch": get_api_branch(branch)})
-        self._selection_mode = {_SELECTION_MODE_REGEXP: regexp, "branch": branch}
+        self.post("views/set_regexp_mode", params={"portfolio": self.key, "regexp": regexp, "branch": branch})
+        self._selection_mode = {_SELECTION_MODE_REGEXP: regexp, "branch": branch or c.DEFAULT_BRANCH}
         return self
 
     def set_remaining_projects_mode(self, branch: Optional[str] = None) -> Portfolio:
         """Sets a portfolio to remaining projects mode"""
-        if branch is None:
-            branch = c.DEFAULT_BRANCH
-        self.post("views/set_remaining_projects_mode", params={"portfolio": self.key, "branch": get_api_branch(branch)})
-        self._selection_mode = {"rest": True, "branch": branch}
+        self.post("views/set_remaining_projects_mode", params={"portfolio": self.key, "branch": branch})
+        self._selection_mode = {"rest": True, "branch": branch or c.DEFAULT_BRANCH}
         return self
 
     def set_none_mode(self) -> Portfolio:
@@ -808,17 +802,6 @@ def recompute(endpoint: pf.Platform) -> None:
     endpoint.post(Portfolio.API["REFRESH"])
 
 
-def _find_sub_portfolio(key: str, data: types.ApiPayload) -> types.ApiPayload:
-    """Finds a subportfolio in a JSON hierarchy"""
-    for subp in data.get("subViews", []):
-        if subp["key"] == key:
-            return subp
-        child = _find_sub_portfolio(key, subp)
-        if child is not None:
-            return child
-    return {}
-
-
 def __create_portfolio_hierarchy(endpoint: pf.Platform, data: types.ApiPayload, parent_key: str) -> int:
     """Creates the hierarchy of portfolios that are new defined by reference"""
     nbr_creations = 0
@@ -840,34 +823,3 @@ def __create_portfolio_hierarchy(endpoint: pf.Platform, data: types.ApiPayload,
         o.root_portfolio = o_parent.root_portfolio
         nbr_creations += __create_portfolio_hierarchy(endpoint, subp, parent_key=key)
     return nbr_creations
-
-
-def get_api_branch(branch: str) -> str:
-    """Returns the value to pass to the API for the branch parameter"""
-    return branch if branch != c.DEFAULT_BRANCH else None
-
-
-def clear_cache(endpoint: pf.Platform) -> None:
-    """Clears the cache of an endpoint"""
-    Portfolio.clear_cache(endpoint)
-
-
-def old_to_new_json_one(old_json: dict[str, Any]) -> dict[str, Any]:
-    """Converts the sonar-config old JSON report format for a single portfolio to the new one"""
-    new_json = old_json.copy()
-    for key in "children", "portfolios":
-        if key in new_json:
-            new_json[key] = old_to_new_json(new_json[key])
-    if "permissions" in old_json:
-        new_json["permissions"] = util.perms_to_list(old_json["permissions"])
-    if "branches" in old_json:
-        new_json["branches"] = util.dict_to_list(old_json["branches"], "name")
-    return new_json
-
-
-def old_to_new_json(old_json: dict[str, Any]) -> dict[str, Any]:
-    """Converts the sonar-config portfolios old JSON report format to the new one"""
-    new_json = old_json.copy()
-    for k, v in new_json.items():
-        new_json[k] = old_to_new_json_one(v)
-    return util.dict_to_list(new_json, "key")

sonar/projects.py

@@ -52,6 +52,7 @@
 from sonar.audit.rules import get_rule, RuleId
 from sonar.audit.problem import Problem
 import sonar.util.constants as c
+import sonar.util.project_helper as phelp
 
 _CLASS_LOCK = Lock()
 
@@ -87,45 +88,6 @@
 
 _PROJECT_QUALIFIER = "qualifier=TRK"
 
-_UNNEEDED_CONTEXT_DATA = (
-    "sonar.announcement.message",
-    "sonar.auth.github.allowUsersToSignUp",
-    "sonar.auth.github.apiUrl",
-    "sonar.auth.github.appId",
-    "sonar.auth.github.enabled",
-    "sonar.auth.github.groupsSync",
-    "sonar.auth.github.organizations",
-    "sonar.auth.github.webUrl",
-    "sonar.builtInQualityProfiles.disableNotificationOnUpdate",
-    "sonar.core.id",
-    "sonar.core.serverBaseURL",
-    "sonar.core.startTime",
-    "sonar.dbcleaner.branchesToKeepWhenInactive",
-    "sonar.forceAuthentication",
-    "sonar.host.url",
-    "sonar.java.jdkHome",
-    "sonar.links.ci",
-    "sonar.links.homepage",
-    "sonar.links.issue",
-    "sonar.links.scm",
-    "sonar.links.scm_dev",
-    "sonar.plugins.risk.consent",
-)
-
-_UNNEEDED_TASK_DATA = (
-    "analysisId",
-    "componentId",
-    "hasScannerContext",
-    "id",
-    "warningCount",
-    "componentQualifier",
-    "nodeName",
-    "componentName",
-    "componentKey",
-    "submittedAt",
-    "executedAt",
-    "type",
-)
 
 # Keys to exclude when applying settings in update()
 _SETTINGS_WITH_SPECIFIC_IMPORT = (
@@ -980,10 +942,10 @@ def migration_export(self, export_settings: types.ConfigSettings) -> types.Objec
         if last_task:
             ctxt = last_task.scanner_context()
             if ctxt:
-                ctxt = {k: v for k, v in ctxt.items() if k not in _UNNEEDED_CONTEXT_DATA}
+                ctxt = {k: v for k, v in ctxt.items() if k not in phelp.UNNEEDED_CONTEXT_DATA}
             t_hist = []
             for t in self.task_history():
-                t_hist.append({k: v for k, v in t.sq_json.items() if k not in _UNNEEDED_TASK_DATA})
+                t_hist.append({k: v for k, v in t.sq_json.items() if k not in phelp.UNNEEDED_TASK_DATA})
             json_data["backgroundTasks"] = {
                 "lastTaskScannerContext": ctxt,
                 # "lastTaskWarnings": last_task.warnings(),
@@ -1343,9 +1305,6 @@ def count(endpoint: pf.Platform, params: types.ApiParams = None) -> int:
     """Counts projects
 
     :param params: list of parameters to filter projects to search
-    :type params: dict
-    :return: Count of projects
-    :rtype: int
     """
     new_params = {} if params is None else params.copy()
     new_params.update({"ps": 1, "p": 1})
@@ -1696,23 +1655,3 @@ def import_zips(endpoint: pf.Platform, project_list: list[str], threads: int = 2
             log.info("%d/%d imports (%d%%) - Latest: %s - %s", i, nb_projects, int(i * 100 / nb_projects), proj_key, status)
             log.info("%s", ", ".join([f"{k}:{v}" for k, v in statuses_count.items()]))
     return statuses
-
-
-def old_to_new_json_one(old_json: dict[str, Any]) -> dict[str, Any]:
-    """Converts the sonar-config projects old JSON report format for a single project to the new one"""
-    new_json = old_json.copy()
-    if "permissions" in old_json:
-        new_json["permissions"] = util.perms_to_list(old_json["permissions"])
-    if "branches" in old_json:
-        new_json["branches"] = util.dict_to_list(old_json["branches"], "name")
-    if "settings" in old_json:
-        new_json["settings"] = util.dict_to_list(old_json["settings"], "key")
-    return new_json
-
-
-def old_to_new_json(old_json: dict[str, Any]) -> dict[str, Any]:
-    """Converts the sonar-config projects old JSON report format to the new one"""
-    new_json = old_json.copy()
-    for k, v in new_json.items():
-        new_json[k] = old_to_new_json_one(v)
-    return util.dict_to_list(new_json, "key")

sonar/qualitygates.py

@@ -563,6 +563,6 @@ def search_by_name(endpoint: pf.Platform, name: str) -> dict[str, QualityGate]:
     return util.search_by_name(endpoint, name, QualityGate.API[c.LIST], "qualitygates")
 
 
-def old_to_new_json(old_json: dict[str, Any]) -> dict[str, Any]:
+def convert_qgs_json(old_json: dict[str, Any]) -> dict[str, Any]:
     """Converts the sonar-config quality gates old JSON report format to the new one"""
     return util.dict_to_list(old_json, "name")

sonar/qualityprofiles.py

@@ -32,6 +32,7 @@
 import sonar.logging as log
 import sonar.platform as pf
 from sonar.util import types, cache, constants as c
+from sonar.util import qualityprofile_helper as qphelp
 from sonar import exceptions
 from sonar import rules, languages
 import sonar.permissions.qualityprofile_permissions as permissions
@@ -41,9 +42,6 @@
 from sonar.audit.rules import get_rule, RuleId
 from sonar.audit.problem import Problem
 
-_KEY_PARENT = "parent"
-_CHILDREN_KEY = "children"
-
 _IMPORTABLE_PROPERTIES = ("name", "language", "parentName", "isBuiltIn", "isDefault", "rules", "permissions", "prioritizedRules")
 
 _CLASS_LOCK = Lock()
@@ -363,20 +361,20 @@ def update(self, data: types.ObjectJsonRepr) -> QualityProfile:
                 QualityProfile.CACHE.pop(self)
                 self.name = data["name"]
                 QualityProfile.CACHE.put(self)
-            self.set_parent(data.pop(_KEY_PARENT, None))
+            self.set_parent(data.pop(qphelp.KEY_PARENT, None))
             self.set_rules(data.get("rules", []) + data.get("addedRules", []))
             self.activate_rules(data.get("modifiedRules", []))
             self.set_permissions(data.get("permissions", []))
             self.is_built_in = data.get("isBuiltIn", False)
             if data.get("isDefault", False):
                 self.set_as_default()
 
-        for child_name, child_data in data.get(_CHILDREN_KEY, {}).items():
+        for child_name, child_data in data.get(qphelp.KEY_CHILDREN, {}).items():
             try:
                 child_qp = get_object(self.endpoint, child_name, self.language)
             except exceptions.ObjectNotFound:
                 child_qp = QualityProfile.create(self.endpoint, child_name, self.language)
-            child_qp.update(child_data | {_KEY_PARENT: self.name})
+            child_qp.update(child_data | {qphelp.KEY_PARENT: self.name})
         return self
 
     def to_json(self, export_settings: types.ConfigSettings) -> types.ObjectJsonRepr:
@@ -430,7 +428,6 @@ def rule_impacts(self, rule_key: str, substitute_with_default: bool = True) -> d
 
         :param str rule_key: The rule key to get severities for
         :return: The impacts of the rule in the quality profile
-        :rtype: dict[str, str]
         """
         return rules.Rule.get_object(self.endpoint, rule_key).impacts(self.key, substitute_with_default=substitute_with_default)
 
@@ -439,7 +436,6 @@ def rule_severity(self, rule_key: str, substitute_with_default: bool = True) ->
 
         :param str rule_key: The rule key to get severities for
         :return: The severity
-        :rtype: str
         """
         return rules.Rule.get_object(self.endpoint, rule_key).rule_severity(self.key, substitute_with_default=substitute_with_default)
 
@@ -742,12 +738,12 @@ def hierarchize_language(qp_list: dict[str, str], endpoint: pf.Platform, languag
                 continue
             parent_qp_name = qp_json_data.pop("parentName")
             parent_qp = hierarchy[parent_qp_name]
-            if _CHILDREN_KEY not in parent_qp:
-                parent_qp[_CHILDREN_KEY] = {}
+            if qphelp.KEY_CHILDREN not in parent_qp:
+                parent_qp[qphelp.KEY_CHILDREN] = {}
             this_qp = get_object(endpoint=endpoint, name=qp_name, language=language)
             qp_json_data |= this_qp.diff(get_object(endpoint=endpoint, name=parent_qp_name, language=language))
             qp_json_data.pop("rules", None)
-            parent_qp[_CHILDREN_KEY][qp_name] = qp_json_data
+            parent_qp[qphelp.KEY_CHILDREN][qp_name] = qp_json_data
             to_remove.append(qp_name)
     for qp_name in to_remove:
         hierarchy.pop(qp_name)
@@ -766,29 +762,6 @@ def hierarchize(qp_list: types.ObjectJsonRepr, endpoint: pf.Platform) -> types.O
     return {lang: hierarchize_language(lang_qp_list, endpoint=endpoint, language=lang) for lang, lang_qp_list in qp_list.items()}
 
 
-def flatten_language(language: str, qp_list: types.ObjectJsonRepr) -> types.ObjectJsonRepr:
-    """Converts a hierarchical list of QP of a given language into a flat list"""
-    flat_list = {}
-    for qp_name, qp_data in qp_list.copy().items():
-        if _CHILDREN_KEY in qp_data:
-            children = flatten_language(language, qp_data[_CHILDREN_KEY])
-            for child in children.values():
-                if "parent" not in child:
-                    child["parent"] = f"{language}:{qp_name}"
-            qp_data.pop(_CHILDREN_KEY)
-            flat_list.update(children)
-        flat_list[f"{language}:{qp_name}"] = qp_data
-    return flat_list
-
-
-def flatten(qp_list: types.ObjectJsonRepr) -> types.ObjectJsonRepr:
-    """Organize a hierarchical list of QP in a flat list"""
-    flat_list = {}
-    for lang, lang_qp_list in qp_list.items():
-        flat_list.update(flatten_language(lang, lang_qp_list))
-    return flat_list
-
-
 def export(endpoint: pf.Platform, export_settings: types.ConfigSettings, **kwargs) -> types.ObjectJsonRepr:
     """Exports all or a list of quality profiles configuration as dict
 
@@ -807,7 +780,7 @@ def export(endpoint: pf.Platform, export_settings: types.ConfigSettings, **kwarg
             qp_list[lang] = {}
         qp_list[lang][name] = json_data
     qp_list = hierarchize(qp_list, endpoint=endpoint)
-    qp_list = __convert_profiles_to_list(qp_list)
+    qp_list = qphelp.convert_qps_json(qp_list)
     if write_q := kwargs.get("write_q", None):
         write_q.put(qp_list)
         write_q.put(util.WRITE_END)
@@ -893,27 +866,3 @@ def exists(endpoint: pf.Platform, name: str, language: str) -> bool:
         return True
     except exceptions.ObjectNotFound:
         return False
-
-
-def convert_one_qp_yaml(qp: types.ObjectJsonRepr) -> types.ObjectJsonRepr:
-    """Converts a QP in a modified version more suitable for YAML export"""
-
-    if _CHILDREN_KEY in qp:
-        qp[_CHILDREN_KEY] = {k: convert_one_qp_yaml(q) for k, q in qp[_CHILDREN_KEY].items()}
-        qp[_CHILDREN_KEY] = util.dict_to_list(qp[_CHILDREN_KEY], "name")
-    return qp
-
-
-def __convert_children_to_list(qp_json: dict[str, Any]) -> list[dict[str, Any]]:
-    """Converts a profile's children profiles to list"""
-    for v in qp_json.values():
-        if "children" in v:
-            v["children"] = __convert_children_to_list(v["children"])
-    return util.dict_to_list(qp_json, "name")
-
-
-def old_to_new_json(qp_json: dict[str, Any]) -> list[dict[str, Any]]:
-    """Converts a language top level list of profiles to list"""
-    for k, v in qp_json.items():
-        qp_json[k] = __convert_children_to_list(v)
-    return util.dict_to_list(qp_json, "language", "profiles")