Skip to content

Commit

Permalink
test rbac
Browse files Browse the repository at this point in the history
No-Issue
  • Loading branch information
jerabekjiri committed Jul 31, 2024
1 parent 982174d commit a3e88c7
Show file tree
Hide file tree
Showing 2 changed files with 142 additions and 142 deletions.
18 changes: 9 additions & 9 deletions .github/workflows/ci_oci-env-integration.yml
Original file line number Diff line number Diff line change
Expand Up @@ -15,16 +15,16 @@ jobs:
fail-fast: false
matrix:
env:
- TEST_PROFILE: ldap
- TEST_PROFILE: keycloak
- TEST_PROFILE: standalone
# - TEST_PROFILE: ldap
# - TEST_PROFILE: keycloak
# - TEST_PROFILE: standalone
- TEST_PROFILE: rbac
- TEST_PROFILE: certified-sync
- TEST_PROFILE: insights
- TEST_PROFILE: iqe_rbac
- TEST_PROFILE: x_repo_search
- TEST_PROFILE: community
- TEST_PROFILE: dab_jwt
# - TEST_PROFILE: certified-sync
# - TEST_PROFILE: insights
# - TEST_PROFILE: iqe_rbac
# - TEST_PROFILE: x_repo_search
# - TEST_PROFILE: community
# - TEST_PROFILE: dab_jwt
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
Expand Down
266 changes: 133 additions & 133 deletions galaxy_ng/tests/integration/api/test_rbac_roles.py
Original file line number Diff line number Diff line change
Expand Up @@ -238,31 +238,31 @@
# TODO: Update object tests to include delete actions
OBJECT_ACTIONS = {
# ansible
change_collection_namespace,
upload_collection_to_namespace,
deprecate_collections,
undeprecate_collections,
upload_collection_to_custom_repo,
upload_collection_to_custom_staging_repo,
collection_repo_list_roles,
collection_repo_add_role,
collection_repo_remove_role,
private_repo_list,
private_distro_list,
private_collection_version_list,
view_private_repository_version,
private_repo_v3,

# ee
change_ee_description,
change_ee_readme,
# change_collection_namespace,
# upload_collection_to_namespace,
# deprecate_collections,
# undeprecate_collections,
# upload_collection_to_custom_repo,
# upload_collection_to_custom_staging_repo,
# collection_repo_list_roles,
# collection_repo_add_role,
# collection_repo_remove_role,
# private_repo_list,
# private_distro_list,
# private_collection_version_list,
# view_private_repository_version,
# private_repo_v3,

# # ee
# change_ee_description,
# change_ee_readme,
create_ee_in_existing_namespace,
push_updates_to_existing_ee,
change_ee_tags,
sync_remote_ee,
ee_namespace_list_roles,
ee_namespace_add_role,
ee_namespace_remove_role
# change_ee_tags,
# sync_remote_ee,
# ee_namespace_list_roles,
# ee_namespace_add_role,
# ee_namespace_remove_role
}

OBJECT_ROLES_TO_TEST = {
Expand Down Expand Up @@ -648,96 +648,96 @@ def _get_reusable_extras(gc):
return REUSABLE_EXTRA


# @pytest.mark.rbac_roles
@pytest.mark.parametrize("role", ROLES_TO_TEST)
def test_global_role_actions(role, galaxy_client):
USERNAME = f"{NAMESPACE}_user_{gen_string()}"

user = create_user(USERNAME, PASSWORD)
group = create_group_for_user(user, role)
group_id = group['id']

expected_allows = ROLES_TO_TEST[role]
gc = galaxy_client("admin", ignore_cache=True)
extra = _get_reusable_extras(gc)

failures = []
# Test global actions
for action in GLOBAL_ACTIONS:
expect_pass = action in expected_allows or action in ACTIONS_FOR_ALL_USERS
try:
action(user, PASSWORD, expect_pass, extra)
except AssertionError:
failures.append(action.__name__)

# cleanup user, group
requests.delete(f"{API_ROOT}_ui/v1/users/{user['id']}/", auth=ADMIN_CREDENTIALS)
requests.delete(f"{API_ROOT}_ui/v1/groups/{group_id}/", auth=ADMIN_CREDENTIALS)

assert failures == []


@pytest.mark.rbac_roles
@pytest.mark.parametrize("role", OBJECT_ROLES_TO_TEST)
def test_object_role_actions(role, galaxy_client):
USERNAME = f"{NAMESPACE}_user_{gen_string()}"

gc = galaxy_client("admin", ignore_cache=True)
extra = _get_reusable_extras(gc)

namespace_href = extra["collection"].get_namespace()["pulp_href"]
repo_href = extra["custom_repo"].get_repo()["pulp_href"]
private_repo_href = extra["private_repo"].get_repo()["pulp_href"]
local_ee_href = extra["local_ee"].get_namespace()["pulp_href"]
remote_ee_href = extra["remote_ee"].get_namespace()["pulp_href"]

user = create_user(USERNAME, PASSWORD)
# create group without any global roles
group = create_group_for_user(user)
group_id = group['id']

def _apply_roles():
# assign object roles
if "collection" in role:
add_group_role(group["pulp_href"], role, namespace_href)

if "execution_environment" in role:
add_group_role(group["pulp_href"], role, local_ee_href)
add_group_role(group["pulp_href"], role, remote_ee_href)

# for the repo owner role, grant them collection namespace permissions
# too so that they can upload collections to their repository.
if role == "galaxy.ansible_repository_owner":
add_group_role(group["pulp_href"], "galaxy.collection_namespace_owner", namespace_href)
add_group_role(group["pulp_href"], role, repo_href)
add_group_role(group["pulp_href"], role, private_repo_href)

failures = []
expected_allows = OBJECT_ROLES_TO_TEST[role]

# since we're also applying the namespace owner role to test if the user can
# upload to repositories they own (when they have namespace perms), we also
# need to add the namespace owner actions to the list of expected allows
if role == "galaxy.ansible_repository_owner":
expected_allows = expected_allows.union({upload_collection_to_custom_repo})
expected_allows = expected_allows.union(
OBJECT_ROLES_TO_TEST["galaxy.collection_namespace_owner"])

# Test global actions
for action in OBJECT_ACTIONS:
# re apply roles in case they get reset
_apply_roles()
expect_pass = action in expected_allows or action in ACTIONS_FOR_ALL_USERS
try:
action(user, PASSWORD, expect_pass, extra)
except AssertionError:
failures.append(action.__name__)

# cleanup user, group
requests.delete(f"{API_ROOT}_ui/v1/users/{user['id']}/", auth=ADMIN_CREDENTIALS)
requests.delete(f"{API_ROOT}_ui/v1/groups/{group_id}/", auth=ADMIN_CREDENTIALS)

assert failures == []
# # @pytest.mark.rbac_roles
# @pytest.mark.parametrize("role", ROLES_TO_TEST)
# def test_global_role_actions(role, galaxy_client):
# USERNAME = f"{NAMESPACE}_user_{gen_string()}"

# user = create_user(USERNAME, PASSWORD)
# group = create_group_for_user(user, role)
# group_id = group['id']

# expected_allows = ROLES_TO_TEST[role]
# gc = galaxy_client("admin", ignore_cache=True)
# extra = _get_reusable_extras(gc)

# failures = []
# # Test global actions
# for action in GLOBAL_ACTIONS:
# expect_pass = action in expected_allows or action in ACTIONS_FOR_ALL_USERS
# try:
# action(user, PASSWORD, expect_pass, extra)
# except AssertionError:
# failures.append(action.__name__)

# # cleanup user, group
# requests.delete(f"{API_ROOT}_ui/v1/users/{user['id']}/", auth=ADMIN_CREDENTIALS)
# requests.delete(f"{API_ROOT}_ui/v1/groups/{group_id}/", auth=ADMIN_CREDENTIALS)

# assert failures == []


# # @pytest.mark.rbac_roles
# @pytest.mark.parametrize("role", OBJECT_ROLES_TO_TEST)
# def test_object_role_actions(role, galaxy_client):
# USERNAME = f"{NAMESPACE}_user_{gen_string()}"

# gc = galaxy_client("admin", ignore_cache=True)
# extra = _get_reusable_extras(gc)

# namespace_href = extra["collection"].get_namespace()["pulp_href"]
# repo_href = extra["custom_repo"].get_repo()["pulp_href"]
# private_repo_href = extra["private_repo"].get_repo()["pulp_href"]
# local_ee_href = extra["local_ee"].get_namespace()["pulp_href"]
# remote_ee_href = extra["remote_ee"].get_namespace()["pulp_href"]

# user = create_user(USERNAME, PASSWORD)
# # create group without any global roles
# group = create_group_for_user(user)
# group_id = group['id']

# def _apply_roles():
# # assign object roles
# if "collection" in role:
# add_group_role(group["pulp_href"], role, namespace_href)

# if "execution_environment" in role:
# add_group_role(group["pulp_href"], role, local_ee_href)
# add_group_role(group["pulp_href"], role, remote_ee_href)

# # for the repo owner role, grant them collection namespace permissions
# # too so that they can upload collections to their repository.
# if role == "galaxy.ansible_repository_owner":
# add_group_role(group["pulp_href"], "galaxy.collection_namespace_owner", namespace_href)
# add_group_role(group["pulp_href"], role, repo_href)
# add_group_role(group["pulp_href"], role, private_repo_href)

# failures = []
# expected_allows = OBJECT_ROLES_TO_TEST[role]

# # since we're also applying the namespace owner role to test if the user can
# # upload to repositories they own (when they have namespace perms), we also
# # need to add the namespace owner actions to the list of expected allows
# if role == "galaxy.ansible_repository_owner":
# expected_allows = expected_allows.union({upload_collection_to_custom_repo})
# expected_allows = expected_allows.union(
# OBJECT_ROLES_TO_TEST["galaxy.collection_namespace_owner"])

# # Test global actions
# for action in OBJECT_ACTIONS:
# # re apply roles in case they get reset
# _apply_roles()
# expect_pass = action in expected_allows or action in ACTIONS_FOR_ALL_USERS
# try:
# action(user, PASSWORD, expect_pass, extra)
# except AssertionError:
# failures.append(action.__name__)

# # cleanup user, group
# requests.delete(f"{API_ROOT}_ui/v1/users/{user['id']}/", auth=ADMIN_CREDENTIALS)
# requests.delete(f"{API_ROOT}_ui/v1/groups/{group_id}/", auth=ADMIN_CREDENTIALS)

# assert failures == []


@pytest.mark.rbac_roles
Expand All @@ -758,23 +758,23 @@ def test_role_actions_for_admin(galaxy_client):


# @pytest.mark.rbac_roles
def test_all_actions_are_tested():
"""
Ensures that all of the actions defined in ROLES_TO_TEST and OBJECT_ROLES_TO_TEST
are also included in GLOBAL_ACTIONS
"""

tested_actions = {action.__name__ for action in GLOBAL_ACTIONS}
role_actions = set()

for role in ROLES_TO_TEST:
role_actions = role_actions.union([action.__name__ for action in ROLES_TO_TEST[role]])

for role in OBJECT_ROLES_TO_TEST:
role_actions = role_actions.union(
[action.__name__ for action in OBJECT_ROLES_TO_TEST[role]])

# assert that all of the actions declared on the roles are also declared
# in the global set of tests
diff = role_actions.difference(tested_actions)
assert diff == set()
# def test_all_actions_are_tested():
# """
# Ensures that all of the actions defined in ROLES_TO_TEST and OBJECT_ROLES_TO_TEST
# are also included in GLOBAL_ACTIONS
# """

# tested_actions = {action.__name__ for action in GLOBAL_ACTIONS}
# role_actions = set()

# for role in ROLES_TO_TEST:
# role_actions = role_actions.union([action.__name__ for action in ROLES_TO_TEST[role]])

# for role in OBJECT_ROLES_TO_TEST:
# role_actions = role_actions.union(
# [action.__name__ for action in OBJECT_ROLES_TO_TEST[role]])

# # assert that all of the actions declared on the roles are also declared
# # in the global set of tests
# diff = role_actions.difference(tested_actions)
# assert diff == set()

0 comments on commit a3e88c7

Please sign in to comment.