How to use list_role_assignments method in tempest

Best Python code snippet using tempest_python

test_identity.py

Source:test_identity.py

...99        self.assertTrue(self.operator_cloud.delete_role(role_name))100    # TODO(Shrews): Once we can support assigning roles within shade, we101    # need to make this test a little more specific, and add more for testing102    # filtering functionality.103    def test_list_role_assignments(self):104        if self.identity_version in ('2', '2.0'):105            self.skipTest("Identity service does not support role assignments")106        assignments = self.operator_cloud.list_role_assignments()107        self.assertIsInstance(assignments, list)108        self.assertGreater(len(assignments), 0)109    def test_list_role_assignments_v2(self):110        user = self.operator_cloud.get_user('demo')111        project = self.operator_cloud.get_project('demo')112        assignments = self.operator_cloud.list_role_assignments(113            filters={'user': user['id'], 'project': project['id']})114        self.assertIsInstance(assignments, list)115        self.assertGreater(len(assignments), 0)116    def test_grant_revoke_role_user_project(self):117        user_name = self.user_prefix + '_user_project'118        user_email = 'nobody@nowhere.com'119        role_name = self.role_prefix + '_grant_user_project'120        role = self.operator_cloud.create_role(role_name)121        user = self._create_user(name=user_name,122                                 email=user_email,123                                 default_project='demo')124        self.assertTrue(self.operator_cloud.grant_role(125            role_name, user=user['id'], project='demo', wait=True))126        assignments = self.operator_cloud.list_role_assignments({127            'role': role['id'],128            'user': user['id'],129            'project': self.operator_cloud.get_project('demo')['id']130        })131        self.assertIsInstance(assignments, list)132        self.assertEqual(1, len(assignments))133        self.assertTrue(self.operator_cloud.revoke_role(134            role_name, user=user['id'], project='demo', wait=True))135        assignments = self.operator_cloud.list_role_assignments({136            'role': role['id'],137            'user': user['id'],138            'project': self.operator_cloud.get_project('demo')['id']139        })140        self.assertIsInstance(assignments, list)141        self.assertEqual(0, len(assignments))142    def test_grant_revoke_role_group_project(self):143        if self.identity_version in ('2', '2.0'):144            self.skipTest("Identity service does not support group")145        role_name = self.role_prefix + '_grant_group_project'146        role = self.operator_cloud.create_role(role_name)147        group_name = self.group_prefix + '_group_project'148        group = self.operator_cloud.create_group(149            name=group_name,150            description='test group',151            domain='default')152        self.assertTrue(self.operator_cloud.grant_role(153            role_name, group=group['id'], project='demo'))154        assignments = self.operator_cloud.list_role_assignments({155            'role': role['id'],156            'group': group['id'],157            'project': self.operator_cloud.get_project('demo')['id']158        })159        self.assertIsInstance(assignments, list)160        self.assertEqual(1, len(assignments))161        self.assertTrue(self.operator_cloud.revoke_role(162            role_name, group=group['id'], project='demo'))163        assignments = self.operator_cloud.list_role_assignments({164            'role': role['id'],165            'group': group['id'],166            'project': self.operator_cloud.get_project('demo')['id']167        })168        self.assertIsInstance(assignments, list)169        self.assertEqual(0, len(assignments))170    def test_grant_revoke_role_user_domain(self):171        if self.identity_version in ('2', '2.0'):172            self.skipTest("Identity service does not support domain")173        role_name = self.role_prefix + '_grant_user_domain'174        role = self.operator_cloud.create_role(role_name)175        user_name = self.user_prefix + '_user_domain'176        user_email = 'nobody@nowhere.com'177        user = self._create_user(name=user_name,178                                 email=user_email,179                                 default_project='demo')180        self.assertTrue(self.operator_cloud.grant_role(181            role_name, user=user['id'], domain='default'))182        assignments = self.operator_cloud.list_role_assignments({183            'role': role['id'],184            'user': user['id'],185            'domain': self.operator_cloud.get_domain('default')['id']186        })187        self.assertIsInstance(assignments, list)188        self.assertEqual(1, len(assignments))189        self.assertTrue(self.operator_cloud.revoke_role(190            role_name, user=user['id'], domain='default'))191        assignments = self.operator_cloud.list_role_assignments({192            'role': role['id'],193            'user': user['id'],194            'domain': self.operator_cloud.get_domain('default')['id']195        })196        self.assertIsInstance(assignments, list)197        self.assertEqual(0, len(assignments))198    def test_grant_revoke_role_group_domain(self):199        if self.identity_version in ('2', '2.0'):200            self.skipTest("Identity service does not support domain or group")201        role_name = self.role_prefix + '_grant_group_domain'202        role = self.operator_cloud.create_role(role_name)203        group_name = self.group_prefix + '_group_domain'204        group = self.operator_cloud.create_group(205            name=group_name,206            description='test group',207            domain='default')208        self.assertTrue(self.operator_cloud.grant_role(209            role_name, group=group['id'], domain='default'))210        assignments = self.operator_cloud.list_role_assignments({211            'role': role['id'],212            'group': group['id'],213            'domain': self.operator_cloud.get_domain('default')['id']214        })215        self.assertIsInstance(assignments, list)216        self.assertEqual(1, len(assignments))217        self.assertTrue(self.operator_cloud.revoke_role(218            role_name, group=group['id'], domain='default'))219        assignments = self.operator_cloud.list_role_assignments({220            'role': role['id'],221            'group': group['id'],222            'domain': self.operator_cloud.get_domain('default')['id']223        })224        self.assertIsInstance(assignments, list)...

role_assignments.py

Source:role_assignments.py

1from flask import Blueprint, g, request, current_app, jsonify2from werkzeug.exceptions import Forbidden, NotFound3from ..models import Org, Repo, User4from .helpers import authorize, authorized_resources, oso5bp = Blueprint("routes.role_assignments", __name__, url_prefix="/orgs/<int:org_id>")6@bp.route("/unassigned_users", methods=["GET"])7def org_unassigned_users_index(org_id):8    org = g.session.get_or_404(Org, id=org_id)9    if not authorize("list_role_assignments", org):10        raise Forbidden11    existing = oso.get("has_role", User, None, org)12    existing_ids = {e["args"][0]["id"] for e in existing}13    unassigned = g.session.query(User).filter(User.id.notin_(existing_ids))14    return jsonify([u.repr() for u in unassigned])15@bp.route("/role_assignments", methods=["GET"])16def org_index(org_id):17    org = g.session.get_or_404(Org, id=org_id)18    if not authorize("list_role_assignments", org):19        raise NotFound20    assignments = oso.get("has_role", User, None, org)21    assignments = {(a["args"][0]["id"], a["args"][1]["id"]) for a in assignments}22    # TODO(gj): fetch users in bulk23    assignments = [24        {25            "user": g.session.get_or_404(User, id=id).repr(),26            "role": role,27        }28        for (id, role) in assignments29    ]30    return jsonify(assignments)31@bp.route("/role_assignments", methods=["POST"])32def org_create(org_id):33    payload = request.get_json(force=True)34    org = g.session.get_or_404(Org, id=org_id)35    if not authorize("list_role_assignments", org):36        raise NotFound37    if not authorize("create_role_assignments", org):38        raise Forbidden39    user = g.session.get_or_404(User, id=payload["user_id"])40    if not authorize("read", user):41        raise NotFound42    oso.tell("has_role", user, payload["role"], org)43    return {"user": user.repr(), "role": payload["role"]}, 20144@bp.route("/role_assignments", methods=["PATCH"])45def org_update(org_id):46    payload = request.get_json(force=True)47    org = g.session.get_or_404(Org, id=org_id)48    if not authorize("list_role_assignments", org):49        raise NotFound50    if not authorize("update_role_assignments", org):51        raise Forbidden52    user = g.session.get_or_404(User, id=payload["user_id"])53    if not authorize("read", user):54        raise NotFound55    oso.bulk_delete(56        [57            ["has_role", user, role["args"][1]["id"], org]58            for role in oso.get("has_role", user, None, org)59        ]60    )61    oso.tell("has_role", user, payload["role"], org)62    return {"user": user.repr(), "role": payload["role"]}63@bp.route("/role_assignments", methods=["DELETE"])64def org_delete(org_id):65    payload = request.get_json(force=True)66    org = g.session.get_or_404(Org, id=org_id)67    if not authorize("list_role_assignments", org):68        raise NotFound69    if not authorize("delete_role_assignments", org):70        raise Forbidden71    user = g.session.get_or_404(User, id=payload["user_id"])72    if not authorize("read", user):73        raise NotFound74    oso.bulk_delete(75        [76            ["has_role", user, role["args"][1]["id"], org]77            for role in oso.get("has_role", user, None, org)78        ]79    )80    return current_app.response_class(status=204, mimetype="application/json")81@bp.route("/repos/<int:repo_id>/unassigned_users", methods=["GET"])82def repo_unassigned_users_index(org_id, repo_id):83    repo = g.session.get_or_404(Repo, id=repo_id)84    if not authorize("list_role_assignments", repo):85        raise NotFound86    if not authorize("create_role_assignments", repo):87        raise Forbidden88    existing = oso.get("has_role", User, None, repo)89    existing_ids = {e["args"][0]["id"] for e in existing}90    unassigned = g.session.query(User).filter(User.id.notin_(existing_ids))91    return jsonify([u.repr() for u in unassigned])92@bp.route("/repos/<int:repo_id>/role_assignments", methods=["GET"])93def repo_index(org_id, repo_id):94    repo = g.session.get_or_404(Repo, id=repo_id)95    if not authorize("list_role_assignments", repo):96        raise Forbidden97    assignments = oso.get("has_role", User, None, repo)98    assignments = {(a["args"][0]["id"], a["args"][1]["id"]) for a in assignments}99    # TODO(gj): fetch users in bulk100    assignments = [101        {102            "user": g.session.get_or_404(User, id=id).repr(),103            "role": role,104        }105        for (id, role) in assignments106    ]107    return jsonify(assignments)108@bp.route("/repos/<int:repo_id>/role_assignments", methods=["POST"])109def repo_create(org_id, repo_id):110    payload = request.get_json(force=True)111    repo = g.session.get_or_404(Repo, id=repo_id)112    if not authorize("list_role_assignments", repo):113        raise NotFound114    if not authorize("create_role_assignments", repo):115        raise Forbidden116    user = g.session.get_or_404(User, id=payload["user_id"])117    if not authorize("read", user):118        raise NotFound119    oso.tell("has_role", user, payload["role"], repo)120    return {"user": user.repr(), "role": payload["role"]}, 201121@bp.route("/repos/<int:repo_id>/role_assignments", methods=["PATCH"])122def repo_update(org_id, repo_id):123    payload = request.get_json(force=True)124    repo = g.session.get_or_404(Repo, id=repo_id)125    if not authorize("list_role_assignments", repo):126        raise NotFound127    if not authorize("update_role_assignments", repo):128        raise Forbidden129    user = g.session.get_or_404(User, id=payload["user_id"])130    if not authorize("read", user):131        raise NotFound132    oso.bulk_delete(133        [134            ["has_role", user, role["args"][1]["id"], repo]135            for role in oso.get("has_role", user, None, repo)136        ]137    )138    oso.tell("has_role", user, payload["role"], repo)139    return {"user": user.repr(), "role": payload["role"]}140@bp.route("/repos/<int:repo_id>/role_assignments", methods=["DELETE"])141def repo_delete(org_id, repo_id):142    payload = request.get_json(force=True)143    repo = g.session.get_or_404(Repo, id=repo_id)144    if not authorize("list_role_assignments", repo):145        raise NotFound146    if not authorize("delete_role_assignments", repo):147        raise Forbidden148    user = g.session.get_or_404(User, id=payload["user_id"])149    if not authorize("read", user):150        raise NotFound151    oso.bulk_delete(152        [153            ["has_role", user, role["args"][1]["id"], repo]154            for role in oso.get("has_role", user, None, repo)155        ]156    )...

test_role_assignments_rbac.py

Source:test_role_assignments_rbac.py

...18class IdentityRoleAssignmentsV3RbacTest(rbac_base.BaseIdentityV3RbacTest):19    @decorators.idempotent_id('afe57adb-1b9c-43d9-84a9-f0cf4c94e416')20    @rbac_rule_validation.action(service="keystone",21                                 rule="identity:list_role_assignments")22    def test_list_role_assignments(self):23        self.rbac_utils.switch_role(self, toggle_rbac_role=True)24        self.role_assignments_client.\25            list_role_assignments()['role_assignments']26    @decorators.idempotent_id('36c7a990-857e-415c-8717-38d7200a9894')27    @rbac_rule_validation.action(28        service="keystone",29        rule="identity:list_role_assignments_for_tree")30    def test_list_role_assignments_for_tree(self):31        project = self.setup_test_project()32        self.rbac_utils.switch_role(self, toggle_rbac_role=True)33        self.role_assignments_client.list_role_assignments(34            include_subtree=True,...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.