Merge "Identity: Migrate 'role assignment' commands to SDK"

openstackclient/identity/v3/role_assignment.py

@@ -13,13 +13,53 @@
 
 """Identity v3 Assignment action implementations"""
 
+from openstack import exceptions as sdk_exceptions
 from osc_lib.command import command
-from osc_lib import utils
 
 from openstackclient.i18n import _
 from openstackclient.identity import common
 
 
+def _format_role_assignment_(assignment, include_names):
+    def _get_names(attr):
+        return (
+            (
+                attr['name']
+                + (
+                    "@" + domain['name']
+                    if (domain := attr.get('domain'))
+                    else ''
+                )
+            )
+            or ''
+            if attr
+            else ''
+        )
+
+    def _get_ids(attr):
+        return attr['id'] or '' if attr else ''
+
+    func = _get_names if include_names else _get_ids
+    return (
+        func(assignment.role),
+        func(assignment.user),
+        func(assignment.group),
+        func(assignment.scope.get('project')),
+        func(assignment.scope.get('domain')),
+        'all' if assignment.scope.get("system") else '',
+        assignment.scope.get("OS-INHERIT:inherited_to") == 'projects',
+    )
+
+
+def _find_sdk_id(find_command, name_or_id, **kwargs):
+    try:
+        return find_command(
+            name_or_id=name_or_id, ignore_missing=False, **kwargs
+        ).id
+    except sdk_exceptions.ForbiddenException:
+        return name_or_id
+
+
 class ListRoleAssignment(command.Lister):
     _description = _("List role assignments")
 
@@ -28,7 +68,7 @@ class ListRoleAssignment(command.Lister):
         parser.add_argument(
             '--effective',
             action="store_true",
-            default=False,
+            default=None,
             help=_('Returns only effective role assignments'),
         )
         parser.add_argument(
@@ -88,81 +128,74 @@ class ListRoleAssignment(command.Lister):
         )
         return parser
 
-    def _as_tuple(self, assignment):
-        return (
-            assignment.role,
-            assignment.user,
-            assignment.group,
-            assignment.project,
-            assignment.domain,
-            assignment.system,
-            assignment.inherited,
-        )
-
     def take_action(self, parsed_args):
-        identity_client = self.app.client_manager.identity
+        identity_client = self.app.client_manager.sdk_connection.identity
         auth_ref = self.app.client_manager.auth_ref
 
-        role = None
+        role_id = None
         role_domain_id = None
         if parsed_args.role_domain:
-            role_domain_id = common.find_domain(
+            role_domain_id = _find_sdk_id(
-                identity_client, parsed_args.role_domain
+                identity_client.find_domain,
-            ).id
+                name_or_id=parsed_args.role_domain,
+            )
         if parsed_args.role:
-            role = utils.find_resource(
+            role_id = _find_sdk_id(
-                identity_client.roles,
+                identity_client.find_role,
-                parsed_args.role,
+                name_or_id=parsed_args.role,
                 domain_id=role_domain_id,
             )
 
-        user = None
+        user_id = None
         if parsed_args.user:
-            user = common.find_user(
+            user_id = _find_sdk_id(
-                identity_client,
+                identity_client.find_user,
-                parsed_args.user,
+                name_or_id=parsed_args.user,
-                parsed_args.user_domain,
+                domain_id=parsed_args.user_domain,
             )
         elif parsed_args.authuser:
             if auth_ref:
-                user = common.find_user(identity_client, auth_ref.user_id)
+                user_id = _find_sdk_id(
+                    identity_client.find_user,
+                    name_or_id=auth_ref.user_id,
+                )
 
         system = None
         if parsed_args.system:
             system = parsed_args.system
 
-        domain = None
+        domain_id = None
         if parsed_args.domain:
-            domain = common.find_domain(
+            domain_id = _find_sdk_id(
-                identity_client,
+                identity_client.find_domain,
-                parsed_args.domain,
+                name_or_id=parsed_args.domain,
             )
 
-        project = None
+        project_id = None
         if parsed_args.project:
-            project = common.find_project(
+            project_id = _find_sdk_id(
-                identity_client,
+                identity_client.find_project,
-                common._get_token_resource(
+                name_or_id=common._get_token_resource(
                     identity_client, 'project', parsed_args.project
                 ),
-                parsed_args.project_domain,
+                domain_id=parsed_args.project_domain,
             )
         elif parsed_args.authproject:
             if auth_ref:
-                project = common.find_project(
+                project_id = _find_sdk_id(
-                    identity_client, auth_ref.project_id
+                    identity_client.find_project,
+                    name_or_id=auth_ref.project_id,
                 )
 
-        group = None
+        group_id = None
         if parsed_args.group:
-            group = common.find_group(
+            group_id = _find_sdk_id(
-                identity_client,
+                identity_client.find_group,
-                parsed_args.group,
+                name_or_id=parsed_args.group,
-                parsed_args.group_domain,
+                domain_id=parsed_args.group_domain,
             )
 
-        include_names = True if parsed_args.names else False
+        include_names = True if parsed_args.names else None
-        effective = True if parsed_args.effective else False
         columns = (
             'Role',
             'User',
@@ -174,104 +207,23 @@ class ListRoleAssignment(command.Lister):
         )
 
         inherited_to = 'projects' if parsed_args.inherited else None
-        data = identity_client.role_assignments.list(
+
-            domain=domain,
+        data = identity_client.role_assignments(
-            user=user,
+            role_id=role_id,
-            group=group,
+            user_id=user_id,
-            project=project,
+            group_id=group_id,
-            system=system,
+            scope_project_id=project_id,
-            role=role,
+            scope_domain_id=domain_id,
-            effective=effective,
+            scope_system=system,
-            os_inherit_extension_inherited_to=inherited_to,
+            effective=parsed_args.effective,
             include_names=include_names,
+            inherited_to=inherited_to,
         )
 
         data_parsed = []
         for assignment in data:
-            # Removing the extra "scope" layer in the assignment json
+            data_parsed.append(
-            scope = assignment.scope
+                _format_role_assignment_(assignment, include_names)
-            if 'project' in scope:
+            )
-                if include_names:
-                    prj = '@'.join(
-                        [
-                            scope['project']['name'],
-                            scope['project']['domain']['name'],
-                        ]
-                    )
-                    setattr(assignment, 'project', prj)
-                else:
-                    setattr(assignment, 'project', scope['project']['id'])
-                assignment.domain = ''
-                assignment.system = ''
-            elif 'domain' in scope:
-                if include_names:
-                    setattr(assignment, 'domain', scope['domain']['name'])
-                else:
-                    setattr(assignment, 'domain', scope['domain']['id'])
-                assignment.project = ''
-                assignment.system = ''
-            elif 'system' in scope:
-                # NOTE(lbragstad): If, or when, keystone supports role
-                # assignments on subsets of a system, this will have to evolve
-                # to handle that case instead of hardcoding to the entire
-                # system.
-                setattr(assignment, 'system', 'all')
-                assignment.domain = ''
-                assignment.project = ''
-            else:
-                assignment.system = ''
-                assignment.domain = ''
-                assignment.project = ''
-
-            inherited = scope.get('OS-INHERIT:inherited_to') == 'projects'
-            assignment.inherited = inherited
-
-            del assignment.scope
-
-            if hasattr(assignment, 'user'):
-                if include_names:
-                    usr = '@'.join(
-                        [
-                            assignment.user['name'],
-                            assignment.user['domain']['name'],
-                        ]
-                    )
-                    setattr(assignment, 'user', usr)
-                else:
-                    setattr(assignment, 'user', assignment.user['id'])
-                assignment.group = ''
-            elif hasattr(assignment, 'group'):
-                if include_names:
-                    grp = '@'.join(
-                        [
-                            assignment.group['name'],
-                            assignment.group['domain']['name'],
-                        ]
-                    )
-                    setattr(assignment, 'group', grp)
-                else:
-                    setattr(assignment, 'group', assignment.group['id'])
-                assignment.user = ''
-            else:
-                assignment.user = ''
-                assignment.group = ''
-
-            if hasattr(assignment, 'role'):
-                if include_names:
-                    # TODO(henry-nash): If this is a domain specific role it
-                    # would be good show this as role@domain, although this
-                    # domain info is not yet included in the response from the
-                    # server. Although we could get it by re-reading the role
-                    # from the ID, let's wait until the server does the right
-                    # thing.
-                    setattr(assignment, 'role', assignment.role['name'])
-                else:
-                    setattr(assignment, 'role', assignment.role['id'])
-            else:
-                assignment.role = ''
-
-            # Creating a tuple from data object fields
-            # (including the blank ones)
-            data_parsed.append(self._as_tuple(assignment))
 
         return columns, tuple(data_parsed)

openstackclient/tests/functional/identity/v3/common.py

@@ -103,6 +103,15 @@ class IdentityTests(base.TestCase):
         'Implied Role ID',
         'Implied Role Name',
     ]
+    ROLE_ASSIGNMENT_LIST_HEADERS = [
+        'Role',
+        'User',
+        'Group',
+        'Project',
+        'Domain',
+        'System',
+        'Inherited',
+    ]
     REGISTERED_LIMIT_FIELDS = [
         'id',
         'service_id',

openstackclient/tests/functional/identity/v3/test_role_assignment.py

@@ -0,0 +1,210 @@
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from openstackclient.tests.functional.identity.v3 import common
+
+
+class RoleAssignmentTests(common.IdentityTests):
+    def test_role_assignment_list_no_filters(self):
+        raw_output = self.openstack('role assignment list')
+        items = self.parse_listing(raw_output)
+        self.assert_table_structure(items, self.ROLE_ASSIGNMENT_LIST_HEADERS)
+
+    def test_role_assignment_list_user_role_system(self):
+        role_name = self._create_dummy_role()
+        username = self._create_dummy_user()
+        system = 'all'
+        raw_output = self.openstack(
+            'role add '
+            '--user %(user)s '
+            '--system %(system)s '
+            '%(role)s'
+            % {
+                'user': username,
+                'system': system,
+                'role': role_name,
+            }
+        )
+        self.addCleanup(
+            self.openstack,
+            'role remove '
+            '--user %(user)s '
+            '--system  %(system)s '
+            '%(role)s'
+            % {
+                'user': username,
+                'system': system,
+                'role': role_name,
+            },
+        )
+        self.assertEqual(0, len(raw_output))
+
+        raw_output = self.openstack(
+            'role assignment list ' '--user %(user)s ' % {'user': username}
+        )
+        items = self.parse_listing(raw_output)
+        self.assert_table_structure(items, self.ROLE_ASSIGNMENT_LIST_HEADERS)
+
+        raw_output = self.openstack(
+            'role assignment list ' '--role %(role)s ' % {'role': role_name}
+        )
+        items = self.parse_listing(raw_output)
+        self.assert_table_structure(items, self.ROLE_ASSIGNMENT_LIST_HEADERS)
+
+        raw_output = self.openstack(
+            'role assignment list ' '--system %(system)s ' % {'system': system}
+        )
+        items = self.parse_listing(raw_output)
+        self.assert_table_structure(items, self.ROLE_ASSIGNMENT_LIST_HEADERS)
+
+    def test_role_assignment_list_group(self):
+        role_name = self._create_dummy_role()
+        group = self._create_dummy_group()
+        system = 'all'
+        raw_output = self.openstack(
+            'role add '
+            '--group %(group)s '
+            '--system %(system)s '
+            '%(role)s'
+            % {
+                'group': group,
+                'system': system,
+                'role': role_name,
+            }
+        )
+        self.addCleanup(
+            self.openstack,
+            'role remove '
+            '--group %(group)s '
+            '--system %(system)s '
+            '%(role)s'
+            % {
+                'group': group,
+                'system': system,
+                'role': role_name,
+            },
+        )
+        self.assertEqual(0, len(raw_output))
+        raw_output = self.openstack(
+            'role assignment list ' '--group %(group)s ' % {'group': group}
+        )
+        items = self.parse_listing(raw_output)
+        self.assert_table_structure(items, self.ROLE_ASSIGNMENT_LIST_HEADERS)
+
+    def test_role_assignment_list_domain(self):
+        role_name = self._create_dummy_role()
+        username = self._create_dummy_user()
+        raw_output = self.openstack(
+            'role add '
+            '--domain %(domain)s '
+            '--user %(user)s '
+            '%(role)s'
+            % {
+                'domain': self.domain_name,
+                'user': username,
+                'role': role_name,
+            }
+        )
+        self.addCleanup(
+            self.openstack,
+            'role remove '
+            '--domain %(domain)s '
+            '--user %(user)s '
+            '%(role)s'
+            % {
+                'domain': self.domain_name,
+                'user': username,
+                'role': role_name,
+            },
+        )
+        self.assertEqual(0, len(raw_output))
+        raw_output = self.openstack(
+            'role assignment list '
+            '--domain %(domain)s ' % {'domain': self.domain_name}
+        )
+        items = self.parse_listing(raw_output)
+        self.assert_table_structure(items, self.ROLE_ASSIGNMENT_LIST_HEADERS)
+
+    def test_role_assignment_list_project(self):
+        role_name = self._create_dummy_role()
+        username = self._create_dummy_user()
+        raw_output = self.openstack(
+            'role add '
+            '--project %(project)s '
+            '--user %(user)s '
+            '%(role)s'
+            % {
+                'project': self.project_name,
+                'user': username,
+                'role': role_name,
+            }
+        )
+        self.addCleanup(
+            self.openstack,
+            'role remove '
+            '--project %(project)s '
+            '--user %(user)s '
+            '%(role)s'
+            % {
+                'project': self.project_name,
+                'user': username,
+                'role': role_name,
+            },
+        )
+        self.assertEqual(0, len(raw_output))
+        raw_output = self.openstack(
+            'role assignment list '
+            '--project %(project)s ' % {'project': self.project_name}
+        )
+        items = self.parse_listing(raw_output)
+        self.assert_table_structure(items, self.ROLE_ASSIGNMENT_LIST_HEADERS)
+
+    def test_role_assignment_list_effective(self):
+        raw_output = self.openstack('role assignment list ' '--effective')
+        items = self.parse_listing(raw_output)
+        self.assert_table_structure(items, self.ROLE_ASSIGNMENT_LIST_HEADERS)
+
+    def test_role_assignment_list_auth_user(self):
+        raw_output = self.openstack('role assignment list ' '--auth-user')
+        items = self.parse_listing(raw_output)
+        self.assert_table_structure(items, self.ROLE_ASSIGNMENT_LIST_HEADERS)
+
+    def test_role_assignment_list_auth_project(self):
+        raw_output = self.openstack('role assignment list ' '--auth-project')
+        items = self.parse_listing(raw_output)
+        self.assert_table_structure(items, self.ROLE_ASSIGNMENT_LIST_HEADERS)
+
+    def test_role_assignment_list_inherited(self):
+        role_name = self._create_dummy_role()
+        username = self._create_dummy_user()
+        raw_output = self.openstack(
+            'role add '
+            '--project %(project)s '
+            '--user %(user)s '
+            '--inherited '
+            '%(role)s'
+            % {
+                'project': self.project_name,
+                'user': username,
+                'role': role_name,
+            }
+        )
+        self.assertEqual(0, len(raw_output))
+
+        raw_output = self.openstack('role assignment list --inherited')
+        items = self.parse_listing(raw_output)
+        self.assert_table_structure(items, self.ROLE_ASSIGNMENT_LIST_HEADERS)
+
+    def test_role_assignment_list_names(self):
+        raw_output = self.openstack('role assignment list --names')
+        items = self.parse_listing(raw_output)
+        self.assert_table_structure(items, self.ROLE_ASSIGNMENT_LIST_HEADERS)

releasenotes/notes/migrate-role-assignment-to-sdk-e6e52bef467b4e4c.yaml

@@ -0,0 +1,4 @@
+---
+features:
+  - |
+    Migrate ``role assignment`` commands from keystoneclient to SDK.