Merge pull request #1291 from okorach:config-import-scloud-v0.1

Config import sonarcloud alpha

Author: okorach

cli/config.py

@@ -149,76 +149,68 @@ def __export_config(endpoint: platform.Platform, what: list[str], **kwargs) -> N
         if len(non_existing_projects) > 0:
             utilities.exit_fatal(f"Project key(s) '{','.join(non_existing_projects)}' do(es) not exist", errcodes.NO_SUCH_KEY)
 
+    calls = {
+        options.WHAT_SETTINGS: [__JSON_KEY_SETTINGS, platform.export],
+        options.WHAT_RULES: [__JSON_KEY_RULES, rules.export],
+        options.WHAT_PROFILES: [__JSON_KEY_PROFILES, qualityprofiles.export],
+        options.WHAT_GATES: [__JSON_KEY_GATES, qualitygates.export],
+        options.WHAT_PROJECTS: [__JSON_KEY_PROJECTS, projects.export],
+        options.WHAT_APPS: [__JSON_KEY_APPS, applications.export],
+        options.WHAT_PORTFOLIOS: [__JSON_KEY_PORTFOLIOS, portfolios.export],
+        options.WHAT_USERS: [__JSON_KEY_USERS, users.export],
+        options.WHAT_GROUPS: [__JSON_KEY_GROUPS, groups.export],
+    }
+
     log.info("Exporting configuration from %s", kwargs[options.URL])
     key_list = kwargs[options.KEYS]
     sq_settings = {__JSON_KEY_PLATFORM: endpoint.basics()}
-    if options.WHAT_SETTINGS in what:
-        sq_settings[__JSON_KEY_SETTINGS] = endpoint.export(export_settings=export_settings)
-    if options.WHAT_RULES in what or options.WHAT_PROFILES in what:
-        sq_settings[__JSON_KEY_RULES] = rules.export(endpoint, export_settings=export_settings)
-    if options.WHAT_PROFILES in what:
-        sq_settings[__JSON_KEY_PROFILES] = qualityprofiles.export(endpoint, export_settings=export_settings)
-    if options.WHAT_GATES in what:
-        sq_settings[__JSON_KEY_GATES] = qualitygates.export(endpoint, export_settings=export_settings)
-    if options.WHAT_PROJECTS in what:
-        sq_settings[__JSON_KEY_PROJECTS] = projects.export(endpoint, key_list=key_list, export_settings=export_settings)
-    if options.WHAT_APPS in what:
-        try:
-            sq_settings[__JSON_KEY_APPS] = applications.export(endpoint, key_list=key_list, export_settings=export_settings)
-        except exceptions.UnsupportedOperation as e:
-            log.warning(e.message)
-    if options.WHAT_PORTFOLIOS in what:
+    for what_item, call_data in calls.items():
+        if what_item not in what:
+            continue
+        ndx, func = call_data
         try:
-            sq_settings[__JSON_KEY_PORTFOLIOS] = portfolios.export(endpoint, key_list=key_list, export_settings=export_settings)
+            sq_settings[ndx] = func(endpoint, export_settings=export_settings, key_list=key_list)
         except exceptions.UnsupportedOperation as e:
             log.warning(e.message)
-    if options.WHAT_USERS in what:
-        sq_settings[__JSON_KEY_USERS] = users.export(endpoint, export_settings=export_settings)
-    if options.WHAT_GROUPS in what:
-        sq_settings[__JSON_KEY_GROUPS] = groups.export(endpoint, export_settings=export_settings)
-
     sq_settings = utilities.remove_empties(sq_settings)
     if not kwargs["dontInlineLists"]:
         sq_settings = utilities.inline_lists(sq_settings, exceptions=("conditions",))
     __write_export(sq_settings, kwargs[options.REPORT_FILE], kwargs[options.FORMAT])
     log.info("Exporting configuration from %s completed", kwargs["url"])
 
 
-def __import_config(endpoint: platform.Platform, what: list[str], **kwargs) -> None:
-    """Imports a platform configuration from a JSON file"""
-    log.info("Importing configuration to %s", kwargs[options.URL])
-    key_list = kwargs[options.KEYS]
+def __read_input_file(file: str) -> dict[str, any]:
     try:
-        with open(kwargs[options.REPORT_FILE], "r", encoding="utf-8") as fd:
+        with open(file, "r", encoding="utf-8") as fd:
             data = json.loads(fd.read())
     except FileNotFoundError as e:
         utilities.exit_fatal(f"OS error while reading file: {e}", exit_code=errcodes.OS_ERROR)
-    if options.WHAT_GROUPS in what:
-        groups.import_config(endpoint, data)
-    if options.WHAT_USERS in what:
-        users.import_config(endpoint, data)
-    if options.WHAT_GATES in what:
-        qualitygates.import_config(endpoint, data)
-    if options.WHAT_RULES in what:
-        rules.import_config(endpoint, data)
-    if options.WHAT_PROFILES in what:
-        if options.WHAT_RULES not in what:
-            rules.import_config(endpoint, data)
-        qualityprofiles.import_config(endpoint, data)
-    if options.WHAT_SETTINGS in what:
-        endpoint.import_config(data)
-    if options.WHAT_PROJECTS in what:
-        projects.import_config(endpoint, data, key_list=key_list)
-    if options.WHAT_APPS in what:
-        try:
-            applications.import_config(endpoint, data, key_list=key_list)
-        except exceptions.UnsupportedOperation as e:
-            log.warning(e.message)
-    if options.WHAT_PORTFOLIOS in what:
-        try:
-            portfolios.import_config(endpoint, data, key_list=key_list)
-        except exceptions.UnsupportedOperation as e:
-            log.warning(e.message)
+    return data
+
+
+def __import_config(endpoint: platform.Platform, what: list[str], data: dict[str, any], **kwargs) -> None:
+    """Imports a platform configuration from a JSON file"""
+    log.info("Importing configuration to %s", kwargs[options.URL])
+    key_list = kwargs[options.KEYS]
+
+    calls = {
+        options.WHAT_GROUPS: groups.import_config,
+        options.WHAT_USERS: users.import_config,
+        options.WHAT_GATES: qualitygates.import_config,
+        options.WHAT_RULES: rules.import_config,
+        options.WHAT_PROFILES: qualityprofiles.import_config,
+        options.WHAT_SETTINGS: platform.import_config,
+        options.WHAT_PROJECTS: projects.import_config,
+        options.WHAT_APPS: applications.import_config,
+        options.WHAT_PORTFOLIOS: portfolios.import_config,
+    }
+
+    for what_item, func in calls.items():
+        if what_item in what:
+            try:
+                func(endpoint, data, key_list=key_list)
+            except exceptions.UnsupportedOperation as e:
+                log.warning(e.message)
     log.info("Importing configuration to %s completed", kwargs[options.URL])
 
 
@@ -235,6 +227,8 @@ def main() -> None:
         utilities.exit_fatal(f"One of --{options.EXPORT} or --{options.IMPORT} option must be chosen", exit_code=errcodes.ARGS_ERROR)
 
     what = utilities.check_what(kwargs.pop(options.WHAT, None), _EVERYTHING, "exported or imported")
+    if options.WHAT_PROFILES in what and options.WHAT_RULES not in what:
+        what.append(options.WHAT_RULES)
     kwargs[options.FORMAT] = utilities.deduct_format(kwargs[options.FORMAT], kwargs[options.REPORT_FILE], allowed_formats=("json", "yaml"))
     if kwargs[options.EXPORT]:
         try:
@@ -246,7 +240,7 @@ def main() -> None:
     if kwargs["import"]:
         if kwargs["file"] is None:
             utilities.exit_fatal("--file is mandatory to import configuration", errcodes.ARGS_ERROR)
-        __import_config(endpoint, what, **kwargs)
+        __import_config(endpoint, what, __read_input_file(kwargs[options.REPORT_FILE]), **kwargs)
     utilities.stop_clock(start_time)
     sys.exit(0)
 

sonar/applications.py

@@ -498,11 +498,10 @@ def export(endpoint: pf.Platform, export_settings: types.ConfigSettings, key_lis
     """Exports applications as JSON
 
     :param Platform endpoint: Reference to the Sonar platform
+    :param ConfigSetting export_settings: Options to use for export
     :param KeyList key_list: list of Application keys to export, defaults to all if None
-    :param full: Whether to export all attributes, including those that can't be set, defaults to False
-    :type full: bool
     :return: Dict of applications settings
-    :rtype: dict
+    :rtype: ObjectJsonRepr
     """
     if endpoint.is_sonarcloud():
         # log.info("Applications do not exist in SonarCloud, export skipped")

sonar/devops.py

@@ -209,6 +209,8 @@ def get_list(endpoint: platform.Platform) -> dict[str, DevopsPlatform]:
     :return: List of DevOps platforms
     :rtype: dict{<platformKey>: <DevopsPlatform>}
     """
+    if endpoint.is_sonarcloud():
+        raise exceptions.UnsupportedOperation("Can't get list of DevOps platforms on SonarCloud")
     if endpoint.edition() == "community":
         return _OBJECTS
     data = json.loads(endpoint.get(APIS["list"]).text)
@@ -253,12 +255,14 @@ def export(endpoint: platform.Platform, export_settings: types.ConfigSettings) -
     return json_data
 
 
-def import_config(endpoint: platform.Platform, config_data: types.ObjectJsonRepr) -> None:
+def import_config(endpoint: platform.Platform, config_data: types.ObjectJsonRepr, key_list: types.KeyList = None) -> None:
     """Imports DevOps platform configuration in SonarQube/Cloud"""
     devops_settings = config_data.get("devopsIntegration", {})
     if len(devops_settings) == 0:
         log.info("No devops integration settings in config, skipping import...")
         return
+    if endpoint.is_sonarcloud():
+        raise exceptions.UnsupportedOperation("Can't get import DevOps platforms in SonarCloud")
     log.info("Importing DevOps config %s", util.json_dump(devops_settings))
     if len(_OBJECTS) == 0:
         get_list(endpoint)

sonar/groups.py

@@ -256,15 +256,16 @@ def get_list(endpoint: pf.Platform) -> dict[str, Group]:
     return search(endpoint)
 
 
-def export(endpoint: pf.Platform, export_settings: types.ConfigSettings) -> types.ObjectJsonRepr:
-    """Exports all groups configuration as dict
-    Default groups (sonar-users) are not exported
-
-    :param endpoint: reference to the SonarQube platform
-    :type endpoint: pf.Platform
-    :return: list of groups
-    :rtype: dict{name: description}
+def export(endpoint: pf.Platform, export_settings: types.ConfigSettings, key_list: types.KeyList = None) -> types.ObjectJsonRepr:
+    """Exports groups representation in JSON
+
+    :param Platform endpoint: reference to the SonarQube platform
+    :param ConfigSettings export_settings: Export parameters
+    :param KeyList key_list: List of project keys to export, defaults to None (all projects)
+    :return: list of groups settings
+    :rtype: ObjectJsonRepr
     """
+
     log.info("Exporting groups")
     g_list = {}
     for g_name, g_obj in sorted(search(endpoint=endpoint).items()):
@@ -337,7 +338,7 @@ def create_or_update(endpoint: pf.Platform, name: str, description: str) -> Grou
         return Group.create(endpoint, name, description)
 
 
-def import_config(endpoint: pf.Platform, config_data: types.ObjectJsonRepr) -> None:
+def import_config(endpoint: pf.Platform, config_data: types.ObjectJsonRepr, key_list: types.KeyList = None) -> None:
     """Imports a group configuration in SonarQube
 
     :param Platform endpoint: reference to the SonarQube platform

sonar/permissions/permission_templates.py

@@ -21,6 +21,7 @@
 from __future__ import annotations
 
 import json
+from requests.exceptions import HTTPError
 
 import sonar.logging as log
 from sonar.util import types
@@ -136,7 +137,10 @@ def set_as_default(self, what_list: list[str]) -> None:
             if (ed == "community" and qual in ("VW", "APP")) or (ed == "developer" and qual == "VW"):
                 log.warning("Can't set permission template as default for %s on a %s edition", qual, ed)
                 continue
-            self.post("permissions/set_default_template", params={"templateId": self.key, "qualifier": qual})
+            try:
+                self.post("permissions/set_default_template", params={"templateId": self.key, "qualifier": qual})
+            except HTTPError as e:
+                log.error("HTTP Error: %s", utilities.sonar_error(e.response))
 
     def set_pattern(self, pattern: str) -> PermissionTemplate:
         """Sets a permission template pattern"""

sonar/permissions/permissions.py

@@ -24,8 +24,9 @@
 from typing import Optional
 
 import json
-from http import HTTPStatus
 from abc import ABC, abstractmethod
+from http import HTTPStatus
+from requests.exceptions import HTTPError
 
 import sonar.logging as log
 from sonar import utilities, errcodes
@@ -258,7 +259,10 @@ def _post_api(self, api: str, set_field: str, perms_dict: types.JsonPermissions,
             filtered_perms = self._filter_permissions_for_edition(perms)
             for p in filtered_perms:
                 params["permission"] = p
-                r = self.endpoint.post(api, params=params)
+                try:
+                    r = self.endpoint.post(api, params=params)
+                except HTTPError as e:
+                    log.error("HTTP Error: %s", utilities.sonar_error(e.response))
                 result = result and r.ok
         return result
 

sonar/permissions/quality_permissions.py

@@ -101,7 +101,9 @@ def _set_perms(self, new_perms, apis, field, diff_func, **kwargs):
 
     def _read_perms(self, apis, field, **kwargs):
         self.permissions = {p: [] for p in permissions.PERMISSION_TYPES}
-        if self.concerned_object.is_built_in:
+        if self.endpoint.is_sonarcloud():
+            log.debug("No permissions for %s because it's SonarCloud", str(self))
+        elif self.concerned_object.is_built_in:
             log.debug("No permissions for %s because it's built-in", str(self))
         else:
             for p in permissions.PERMISSION_TYPES:

sonar/permissions/qualitygate_permissions.py

@@ -38,14 +38,14 @@ def __str__(self):
         return f"permissions of {str(self.concerned_object)}"
 
     def read(self):
-        if self.endpoint.version() < (9, 2, 0):
+        if not self.endpoint.is_sonarcloud() and self.endpoint.version() < (9, 2, 0):
             log.debug("Can't read %s on SonarQube < 9.2", str(self))
             return self
         self._read_perms(QualityGatePermissions.APIS, QualityGatePermissions.API_GET_FIELD, gateName=self.concerned_object.name)
         return self
 
     def set(self, new_perms):
-        if self.endpoint.version() < (9, 2, 0):
+        if not self.endpoint.is_sonarcloud() and self.endpoint.version() < (9, 2, 0):
             raise exceptions.UnsupportedOperation(f"Can't set {str(self)} on SonarQube < 9.2")
 
         return self._set_perms(

sonar/platform.py

@@ -460,10 +460,16 @@ def import_config(self, config_data: types.ObjectJsonRepr) -> None:
 
         if settings.NEW_CODE_PERIOD in config_data["generalSettings"]:
             (nc_type, nc_val) = settings.decode(settings.NEW_CODE_PERIOD, config_data["generalSettings"][settings.NEW_CODE_PERIOD])
-            settings.set_new_code_period(self, nc_type, nc_val)
+            try:
+                settings.set_new_code_period(self, nc_type, nc_val)
+            except exceptions.UnsupportedOperation as e:
+                log.error(e.message)
         permission_templates.import_config(self, config_data)
         global_permissions.import_config(self, config_data)
-        devops.import_config(self, config_data)
+        try:
+            devops.import_config(self, config_data)
+        except exceptions.UnsupportedOperation as e:
+            log.warning(e.message)
 
     def audit(self, audit_settings: types.ConfigSettings) -> list[Problem]:
         """Audits a global platform configuration and returns the list of problems found
@@ -813,6 +819,17 @@ def latest() -> tuple[int, int, int]:
     return __lta_and_latest()[1]
 
 
+def import_config(endpoint: Platform, config_data: types.ObjectJsonRepr, key_list: types.KeyList = None) -> None:
+    """Imports a configuration in SonarQube
+
+    :param Platform endpoint: reference to the SonarQube platform
+    :param ObjectJsonRepr config_data: the configuration to import
+    :param KeyList key_list: Unused
+    :return: Nothing
+    """
+    endpoint.import_config(config_data)
+
+
 def _check_for_retry(response: requests.models.Response) -> tuple[bool, str]:
     """Verifies if a response had a 301 Moved permanently and if so provide the new location"""
     if len(response.history) > 0 and response.history[0].status_code == HTTPStatus.MOVED_PERMANENTLY:
@@ -836,3 +853,15 @@ def convert_for_yaml(original_json: types.ObjectJsonRepr) -> types.ObjectJsonRep
     if "devopsIntegration" in original_json:
         original_json["devopsIntegration"] = util.dict_to_list(original_json["devopsIntegration"], "name")
     return original_json
+
+
+def export(endpoint: Platform, export_settings: types.ConfigSettings, key_list: types.KeyList = None) -> types.ObjectJsonRepr:
+    """Exports all or a list of projects configuration as dict
+
+    :param Platform endpoint: reference to the SonarQube platform
+    :param ConfigSettings export_settings: Export parameters
+    :param KeyList key_list: Unused
+    :return: Platform settings
+    :rtype: ObjectJsonRepr
+    """
+    return endpoint.export(export_settings)

sonar/portfolios.py

@@ -728,10 +728,10 @@ def export(endpoint: pf.Platform, export_settings: types.ConfigSettings, key_lis
     """Exports portfolios as JSON
 
     :param Platform endpoint: Reference to the SonarQube platform
-    :param KeyList key_list: list of portfoliios keys to export as csv or list, defaults to all if None
     :param ConfigSetting export_settings: Options to use for export
-    :return: Dict of applications settings
-    :rtype: dict
+    :param KeyList key_list: list of portfoliios keys to export as csv or list, defaults to all if None
+    :return: Dict of portfolio settings
+    :rtype: ObjectJsonRepr
     """
     check_supported(endpoint)