add test for new CLI batch update permissions + requests code path of permissions from config

Author: fmigneault

magpie/cli/batch_update_permissions.py

@@ -2,9 +2,11 @@
 """
 Magpie helper to create or delete a set of permissions.
 
-When parsing permissions to create, any underlying user, group, service or resource
-that are missing, but that can be resolved with reasonable defaults, will be dynamically
-created prior to setting the corresponding permission on it.
+When parsing permissions to create, any underlying user, group, or intermediate resource
+that are missing, but that can be resolved with reasonable defaults or with an explicit
+definition in the configuration file, will be dynamically created prior to setting the
+corresponding permission on it. All referenced services should exist beforehand. Consider
+using the 'register_providers' utility to register services beforehand as needed.
 
 See https://pavics-magpie.readthedocs.io/en/latest/configuration.html#file-permissions-cfg for more details.
 """
@@ -40,7 +42,7 @@ def make_parser():
     parser.add_argument("-P", "--password", "--magpie-admin-password", help=(
         "Admin password for magpie login (if omitted, will try using 'MAGPIE_ADMIN_PASSWORD' environment variable)."
     ))
-    parser.add_argument("-c", "--config", required=True, nargs="+", help=(
+    parser.add_argument("-c", "--config", required=True, nargs="+", action="append", help=(
         "Path to a single configuration file or a directory containing configuration file that contains permissions. "
         "The option can be specified multiple times to provide multiple lookup directories or specific files to load. "
         "Configuration files must be in JSON or YAML format, with their respective extensions, or the '.cfg' extension."
@@ -56,22 +58,11 @@ def main(args=None, parser=None, namespace=None):
     ns = parser.parse_args(args=args, namespace=namespace)
     setup_logger_from_options(LOGGER, ns)
 
-    all_configs = []
-    for cfg in ns.config:
-        configs = get_all_configs(cfg, "permissions", allow_missing=True)
-        all_configs.extend(configs)
-
-    if ns.verbose:
-        LOGGER.info(
-            "Resolved permissions to update:\n\n%s\n",
-            yaml.safe_dump(all_configs, allow_unicode=True, encoding="utf-8", indent=4, sort_keys=False)
-        )
-
-    if not all_configs or all(not cfg for cfg in all_configs):
-        LOGGER.error("Could not find any permissions configuration under specified locations.")
-        return ERROR_PARAMS
+    all_configs = [cfg for cfg_args in ns.config for cfg in cfg_args]
     try:
-        magpie_register_permissions_from_config(all_configs)
+        for config in all_configs:
+            LOGGER.info("Processing: [%s]", config)
+            magpie_register_permissions_from_config(config)
     except Exception as exc:
         LOGGER.error("Failed permissions parsing and update from specified configurations [%s].", str(exc))
         return ERROR_EXEC

magpie/register.py

@@ -5,7 +5,7 @@
 import subprocess  # nosec
 import time
 from tempfile import NamedTemporaryFile
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, overload
 
 import requests
 import six
@@ -61,14 +61,16 @@
         CookiesOrSessionType,
         GroupsConfig,
         GroupsSettings,
+        Literal,
         MultiConfigs,
         PermissionConfigItem,
         PermissionsConfig,
         ServicesConfig,
         ServicesSettings,
         Str,
         UsersConfig,
-        UsersSettings
+        UsersSettings,
+        WebhooksConfig
     )
 
 
@@ -573,6 +575,36 @@ def _load_config(path_or_dict, section, allow_missing=False):
 CONFIG_KNOWN_EXTENSIONS = frozenset([".cfg", ".json", ".yml", ".yaml"])
 
 
+@overload
+def get_all_configs(path_or_dict, section, allow_missing=False):
+    # type: (Union[Str, CombinedConfig], Literal["groups"], bool) -> GroupsConfig
+    ...
+
+
+@overload
+def get_all_configs(path_or_dict, section, allow_missing=False):
+    # type: (Union[Str, CombinedConfig], Literal["users"], bool) -> UsersConfig
+    ...
+
+
+@overload
+def get_all_configs(path_or_dict, section, allow_missing=False):
+    # type: (Union[Str, CombinedConfig], Literal["permissions"], bool) -> PermissionsConfig
+    ...
+
+
+@overload
+def get_all_configs(path_or_dict, section, allow_missing=False):
+    # type: (Union[Str, CombinedConfig], Literal["services"], bool) -> ServicesConfig
+    ...
+
+
+@overload
+def get_all_configs(path_or_dict, section, allow_missing=False):
+    # type: (Union[Str, CombinedConfig], Literal["webhooks"], bool) -> WebhooksConfig
+    ...
+
+
 def get_all_configs(path_or_dict, section, allow_missing=False):
     # type: (Union[Str, CombinedConfig], Str, bool) -> MultiConfigs
     """
@@ -1020,19 +1052,36 @@ def magpie_register_permissions_from_config(
         cookies_or_session = db_session
 
     LOGGER.debug("Loading configurations.")
-    permissions = get_all_configs(permissions_config, "permissions")  # type: List[PermissionsConfig]
+    if isinstance(permissions_config, list):
+        permissions = [permissions_config]
+    else:
+        permissions = get_all_configs(permissions_config, "permissions")
     perms_cfg_count = len(permissions)
     LOGGER.log(logging.INFO if perms_cfg_count else logging.WARNING,
                "Found %s permissions configurations.", perms_cfg_count)
     users_settings = groups_settings = None
     if perms_cfg_count:
-        users = get_all_configs(permissions_config, "users", allow_missing=True)    # type: List[UsersConfig]
-        groups = get_all_configs(permissions_config, "groups", allow_missing=True)  # type: List[GroupsConfig]
+        if isinstance(permissions_config, str):
+            users = get_all_configs(permissions_config, "users", allow_missing=True)
+        else:
+            users = []
+        if isinstance(permissions_config, str):
+            groups = get_all_configs(permissions_config, "groups", allow_missing=True)
+        else:
+            groups = []
         users_settings = _resolve_config_registry(users, "username") or {}
         groups_settings = _resolve_config_registry(groups, "name") or {}
     for i, perms in enumerate(permissions):
         LOGGER.info("Processing permissions from configuration (%s/%s).", i + 1, perms_cfg_count)
-        _process_permissions(perms, magpie_url, cookies_or_session, users_settings, groups_settings, raise_errors)
+        _process_permissions(
+            perms,
+            magpie_url,
+            cookies_or_session,
+            users_settings,
+            groups_settings,
+            settings,
+            raise_errors,
+        )
     LOGGER.info("All permissions processed.")
 
 
@@ -1061,16 +1110,23 @@ def _resolve_config_registry(config_files, key):
     return config_map
 
 
-def _process_permissions(permissions, magpie_url, cookies_or_session, users=None, groups=None, raise_errors=False):
-    # type: (PermissionsConfig, Str, Session, Optional[UsersSettings], Optional[GroupsSettings], bool) -> None
+def _process_permissions(
+    permissions,            # type: PermissionsConfig
+    magpie_url,             # type: Str
+    cookies_or_session,     # type: Session
+    users=None,             # type: Optional[UsersSettings]
+    groups=None,            # type: Optional[GroupsSettings]
+    settings=None,          # type: Optional[AnySettingsContainer]
+    raise_errors=False,     # type: bool
+):                          # type: (...) -> None
     """
     Processes a single `permissions` configuration.
     """
     if not permissions:
         LOGGER.warning("Permissions configuration are empty.")
         return
 
-    anon_user = get_constant("MAGPIE_ANONYMOUS_USER")
+    anon_user = get_constant("MAGPIE_ANONYMOUS_USER", settings)
     perm_count = len(permissions)
     LOGGER.log(logging.INFO if perm_count else logging.WARNING,
                "Found %s permissions to evaluate from configuration.", perm_count)

tests/test_magpie_cli.py

@@ -12,18 +12,24 @@
 import os
 import subprocess
 import tempfile
+from typing import TYPE_CHECKING
+from urllib.parse import urlparse
 
 import mock
 import six
 
 from magpie.cli import batch_update_permissions, batch_update_users, magpie_helper_cli
 from magpie.constants import get_constant
+from magpie.permissions import Permission, PermissionType
+from magpie.services import ServiceAPI, ServiceWPS
 from tests import runner, utils
 
 if six.PY2:
     from backports import tempfile as tempfile2  # noqa  # pylint: disable=E0611,no-name-in-module  # Python 2
 else:
     tempfile2 = tempfile  # pylint: disable=C0103,invalid-name
+if TYPE_CHECKING:
+    from magpie.typedefs import JSON
 
 KNOWN_HELPERS = [
     "batch_update_users",
@@ -120,34 +126,34 @@ def test_magpie_batch_update_permissions_command():
         "-P", test_admin_pwd,
     ]
 
-    # cleanup in case of previous failure
+    def mock_request(method, url, *args, **kwargs):
+        _path = urlparse(url).path
+        # because CLI utility does multiple login tests, we must force TestApp logout to forget session
+        # otherwise, error is raised because of user session mismatch between previous login and new one requested
+        if _path.startswith("/signin"):
+            utils.check_or_try_logout_user(test_app)
+        return utils.test_request(test_app, method, _path, *args, **kwargs)
+
     _, test_admin_cookies = utils.check_or_try_login_user(test_app, username=test_admin_usr, password=test_admin_pwd)
+
+    # cleanup in case of previous run
     utils.TestSetup.delete_TestUser(test_app, override_user_name=test_usr, override_cookies=test_admin_cookies)
     utils.TestSetup.delete_TestGroup(test_app, override_group_name=test_grp, override_cookies=test_admin_cookies)
     utils.TestSetup.delete_TestService(test_app, override_service_name=test_api, override_cookies=test_admin_cookies)
     utils.TestSetup.delete_TestService(test_app, override_service_name=test_wps, override_cookies=test_admin_cookies)
 
-    def mock_request(*args, **kwargs):
-        method, url, args = args[0], args[1], args[2:]
-        path = url.replace(test_url, "")
-        # because CLI utility does multiple login tests, we must force TestApp logout to forget session
-        # otherwise, error is raised because of user session mismatch between previous login and new one requested
-        if path.startswith("/signin"):
-            utils.check_or_try_logout_user(test_app)
-        return utils.test_request(test_app, method, path, *args, **kwargs)
+    utils.TestSetup.create_TestService(test_app,
+                                       override_service_name=test_api,
+                                       override_service_type=ServiceAPI.service_type)
+    utils.TestSetup.create_TestService(test_app,
+                                       override_service_name=test_wps,
+                                       override_service_type=ServiceWPS.service_type)
 
-    def run_command(operation_name, operation_args):
-        with tempfile2.TemporaryDirectory() as tmpdir:
-            with mock.patch("requests.Session.request", side_effect=mock_request):
-                with mock.patch("requests.request", side_effect=mock_request):
-                    err_code = batch_update_permissions.main(operation_args)
-                    assert err_code == 0, "failed execution due to invalid arguments"
-                    assert len(os.listdir(tmpdir)) == 1, "utility should have produced 1 output file"
-                    file = os.path.join(tmpdir, os.listdir(tmpdir)[0])
-                    utils.check_val_is_in(operation_name, file)
-                    assert os.path.isfile(file)
-                    with open(file, mode="r", encoding="utf-8") as fd:
-                        file_text = fd.read()
+    def run_command(__operation_name, operation_args):
+        with mock.patch("requests.Session.request", side_effect=mock_request):
+            with mock.patch("requests.request", side_effect=mock_request):
+                err_code = batch_update_permissions.main(operation_args)
+                assert err_code == 0, "failed execution due to invalid arguments"
 
     with contextlib.ExitStack() as stack:
         tmp_file = stack.enter_context(tempfile.NamedTemporaryFile(mode="w", suffix=".json"))
@@ -156,8 +162,8 @@ def run_command(operation_name, operation_args):
                 {
                     "group": test_grp,
                     "service": test_api,
-                    "resource": "/api/v1",
-                    "permission": "read",
+                    "resource": "/v1",
+                    "permission": Permission.READ.value,
                     "action": "create",
                 }
             ]
@@ -173,7 +179,7 @@ def run_command(operation_name, operation_args):
                 {
                     "user": test_usr,
                     "service": test_wps,
-                    "permission": "getcapabilities",
+                    "permission": Permission.GET_CAPABILITIES.value,
                     "action": "create",
                 }
             ]
@@ -186,6 +192,34 @@ def run_command(operation_name, operation_args):
         test_args += ["--verbose"]
         run_command("batch_update_permissions", test_args)
 
+    path = f"/groups/{test_grp}/resources"
+    resp = utils.test_request(test_app, "GET", path)
+    body = utils.check_response_basic_info(resp)
+    res_api = body["resources"][ServiceAPI.service_type][test_api]  # type: JSON
+    res_wps = body["resources"][ServiceWPS.service_type][test_wps]  # type: JSON
+    perms_res_api = res_api["permissions"]
+    perms_res_wps = res_wps["permissions"]
+    utils.check_val_equal(len(perms_res_wps), 0)
+    utils.check_val_equal(len(perms_res_api), 0, msg="Permission should not be created directly on the service.")
+    res_res_api = res_api["resources"]
+    utils.check_val_equal(len(res_res_api), 1, msg="Resource should have been created dynamically.")
+    perms_sub_api = res_res_api[list(res_res_api)[0]]["permissions"]  # type: JSON
+    utils.check_val_equal(len(perms_sub_api), 1)
+    utils.check_val_equal(perms_sub_api[0]["name"], Permission.READ.value)
+    utils.check_val_equal(perms_sub_api[0]["type"], PermissionType.INHERITED.value)
+
+    path = f"/users/{test_usr}/resources"
+    resp = utils.test_request(test_app, "GET", path)
+    body = utils.check_response_basic_info(resp)
+    res_api = body["resources"][ServiceAPI.service_type][test_api]
+    res_wps = body["resources"][ServiceWPS.service_type][test_wps]
+    perms_res_api = res_api["permissions"]
+    perms_res_wps = res_wps["permissions"]
+    utils.check_val_equal(len(perms_res_api), 0)
+    utils.check_val_equal(len(perms_res_wps), 1)
+    utils.check_val_equal(perms_res_wps[0]["name"], Permission.GET_CAPABILITIES.value)
+    utils.check_val_equal(perms_res_wps[0]["type"], PermissionType.DIRECT.value)
+
 
 @runner.MAGPIE_TEST_CLI
 @runner.MAGPIE_TEST_LOCAL