Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Remove pulp-smash from test_rbac_sync_repositories & test_sync_signatures #1915

Merged
merged 1 commit into from
Feb 6, 2025
Merged

Remove pulp-smash from test_rbac_sync_repositories & test_sync_signatures #1915

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
82 changes: 39 additions & 43 deletions pulp_container/tests/functional/api/test_rbac_sync_repositories.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,19 +1,13 @@
"""Tests that container sync repositories have RBAC."""

import pytest

from django.conf import settings

from pulp_smash import utils
from pulp_smash.pulp3.bindings import monitor_task

from pulpcore.client.pulp_container.exceptions import ApiException
import uuid


@pytest.mark.parallel
def test_rbac_sync_repositories(gen_user, container_repository_api):
def test_rbac_sync_repositories(gen_user, container_bindings, pulp_settings, monitor_task):
"""RBAC sync repositories."""
if settings.TOKEN_AUTH_DISABLED:
if pulp_settings.TOKEN_AUTH_DISABLED:
pytest.skip("RBAC cannot be tested when token authentication is disabled")

user1 = gen_user(model_roles=["container.containerrepository_creator"])
Expand All @@ -22,64 +16,66 @@ def test_rbac_sync_repositories(gen_user, container_repository_api):
repository = None

"""Create a repository."""
body = {"name": utils.uuid4()}
with user2, pytest.raises(ApiException):
container_repository_api.create(body)
with user3, pytest.raises(ApiException):
container_repository_api.create(body)
body = {"name": str(uuid.uuid4())}
with user2, pytest.raises(container_bindings.ApiException):
container_bindings.RepositoriesContainerApi.create(body)
with user3, pytest.raises(container_bindings.ApiException):
container_bindings.RepositoriesContainerApi.create(body)
with user1:
repository = container_repository_api.create(body)
repository = container_bindings.RepositoriesContainerApi.create(body)

"""Read a repository by its href."""
with user1:
container_repository_api.read(repository.pulp_href)
container_bindings.RepositoriesContainerApi.read(repository.pulp_href)
with user2:
# read with global read permission
container_repository_api.read(repository.pulp_href)
with user3, pytest.raises(ApiException):
container_bindings.RepositoriesContainerApi.read(repository.pulp_href)
with user3, pytest.raises(container_bindings.ApiException):
# read without read permission
container_repository_api.read(repository.pulp_href)
container_bindings.RepositoriesContainerApi.read(repository.pulp_href)

"""Read a repository by its name."""
with user1:
page = container_repository_api.list(name=repository.name)
page = container_bindings.RepositoriesContainerApi.list(name=repository.name)
assert len(page.results) == 1
with user2:
page = container_repository_api.list(name=repository.name)
page = container_bindings.RepositoriesContainerApi.list(name=repository.name)
assert len(page.results) == 1
with user3:
page = container_repository_api.list(name=repository.name)
page = container_bindings.RepositoriesContainerApi.list(name=repository.name)
assert len(page.results) == 0

"""Update a repository using HTTP PATCH."""
body = {"name": utils.uuid4()}
with user2, pytest.raises(ApiException):
container_repository_api.partial_update(repository.pulp_href, body)
with user3, pytest.raises(ApiException):
container_repository_api.partial_update(repository.pulp_href, body)
body = {"name": str(uuid.uuid4())}
with user2, pytest.raises(container_bindings.ApiException):
container_bindings.RepositoriesContainerApi.partial_update(repository.pulp_href, body)
with user3, pytest.raises(container_bindings.ApiException):
container_bindings.RepositoriesContainerApi.partial_update(repository.pulp_href, body)
with user1:
response = container_repository_api.partial_update(repository.pulp_href, body)
response = container_bindings.RepositoriesContainerApi.partial_update(
repository.pulp_href, body
)
monitor_task(response.task)
repository = container_repository_api.read(repository.pulp_href)
repository = container_bindings.RepositoriesContainerApi.read(repository.pulp_href)

"""Update a repository using HTTP PUT."""
body = {"name": utils.uuid4()}
with user2, pytest.raises(ApiException):
container_repository_api.update(repository.pulp_href, body)
with user3, pytest.raises(ApiException):
container_repository_api.update(repository.pulp_href, body)
body = {"name": str(uuid.uuid4())}
with user2, pytest.raises(container_bindings.ApiException):
container_bindings.RepositoriesContainerApi.update(repository.pulp_href, body)
with user3, pytest.raises(container_bindings.ApiException):
container_bindings.RepositoriesContainerApi.update(repository.pulp_href, body)
with user1:
response = container_repository_api.update(repository.pulp_href, body)
response = container_bindings.RepositoriesContainerApi.update(repository.pulp_href, body)
monitor_task(response.task)
repository = container_repository_api.read(repository.pulp_href)
repository = container_bindings.RepositoriesContainerApi.read(repository.pulp_href)

"""Delete a repository."""
with user2, pytest.raises(ApiException):
container_repository_api.delete(repository.pulp_href)
with user3, pytest.raises(ApiException):
container_repository_api.delete(repository.pulp_href)
with user2, pytest.raises(container_bindings.ApiException):
container_bindings.RepositoriesContainerApi.delete(repository.pulp_href)
with user3, pytest.raises(container_bindings.ApiException):
container_bindings.RepositoriesContainerApi.delete(repository.pulp_href)
with user1:
response = container_repository_api.delete(repository.pulp_href)
response = container_bindings.RepositoriesContainerApi.delete(repository.pulp_href)
monitor_task(response.task)
with pytest.raises(ApiException):
container_repository_api.read(repository.pulp_href)
with pytest.raises(container_bindings.ApiException):
container_bindings.RepositoriesContainerApi.read(repository.pulp_href)
63 changes: 28 additions & 35 deletions pulp_container/tests/functional/api/test_sync_signatures.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,7 @@
import pytest

from pulp_smash.pulp3.bindings import monitor_task
from pulp_smash.pulp3.utils import gen_repo

from pulpcore.client.pulp_container import (
ContainerContainerRepository,
ContainerRepositorySyncURL,
)
from pulp_container.tests.functional.utils import gen_container_remote
from pulp_container.tests.functional.conftest import gen_container_remote

REDHAT_REGISTRY_V2 = "https://registry.access.redhat.com"
DEPRECATED_REPOSITORY_NAME = "rhel7-rhel-minimal"
Expand All @@ -19,9 +13,10 @@
@pytest.fixture
def synced_repository(
delete_orphans_pre,
container_repository_api,
container_remote_api,
gen_object_with_cleanup,
container_repo,
container_remote_factory,
container_bindings,
monitor_task,
request,
):
"""A repository that contains signatures synced from sigstore, if specified."""
Expand All @@ -35,32 +30,29 @@ def synced_repository(
if request.param["sigstore"]:
data["sigstore"] = request.param["sigstore"]

remote = gen_object_with_cleanup(container_remote_api, data)

data = ContainerContainerRepository(**gen_repo())
repository = gen_object_with_cleanup(container_repository_api, data)
remote = container_remote_factory(**data)

signed_only = request.param["signed_only"]
data = ContainerRepositorySyncURL(remote=remote.pulp_href, signed_only=signed_only)
response = container_repository_api.sync(repository.pulp_href, data)
data = {"remote": remote.pulp_href, "signed_only": signed_only}
response = container_bindings.RepositoriesContainerApi.sync(container_repo.pulp_href, data)
monitor_task(response.task)

return container_repository_api.read(repository.pulp_href)
return container_bindings.RepositoriesContainerApi.read(container_repo.pulp_href)


@pytest.mark.parametrize(
"synced_repository", [{"sigstore": None, "signed_only": False}], indirect=True
)
def test_sync_images_without_signatures(
container_signature_api, container_tag_api, synced_repository
):
def test_sync_images_without_signatures(container_bindings, synced_repository):
"""Sync a repository without specifying sigstore."""
signatures = container_signature_api.list(
signatures = container_bindings.ContentSignaturesApi.list(
repository_version=synced_repository.latest_version_href
).results
assert len(signatures) == 0

tags = container_tag_api.list(repository_version=synced_repository.latest_version_href).results
tags = container_bindings.ContentTagsApi.list(
repository_version=synced_repository.latest_version_href
).results
assert len(tags) == 2


Expand All @@ -73,19 +65,19 @@ def test_sync_images_without_signatures(
],
indirect=True,
)
def test_sync_signed_images_from_sigstore(
container_signature_api, container_manifest_api, container_tag_api, synced_repository
):
def test_sync_signed_images_from_sigstore(container_bindings, synced_repository):
"""Sync a repository with specifying sigstore."""
signatures = container_signature_api.list(
signatures = container_bindings.ContentSignaturesApi.list(
repository_version=synced_repository.latest_version_href
).results
tags = container_bindings.ContentTagsApi.list(
repository_version=synced_repository.latest_version_href
).results
tags = container_tag_api.list(repository_version=synced_repository.latest_version_href).results

tags_dict = {tag.name: tag for tag in tags}

single_manifest_href = tags_dict[IMAGE_MANIFEST_TAG].tagged_manifest
manifest = container_manifest_api.read(single_manifest_href)
manifest = container_bindings.ContentManifestsApi.read(single_manifest_href)

single_manifest_signatures = list(
filter(lambda s: s.signed_manifest == manifest.pulp_href, signatures)
Expand All @@ -97,10 +89,11 @@ def test_sync_signed_images_from_sigstore(
assert all(s.name.startswith(manifest.digest) for s in single_manifest_signatures)

manifest_list_href = tags_dict[MANIFEST_LIST_TAG].tagged_manifest
manifest_list = container_manifest_api.read(manifest_list_href)
manifest_list = container_bindings.ContentManifestsApi.read(manifest_list_href)

listed_manifests = [
container_manifest_api.read(lm_href) for lm_href in manifest_list.listed_manifests
container_bindings.ContentManifestsApi.read(lm_href)
for lm_href in manifest_list.listed_manifests
]
for lm in listed_manifests:
manifest_signatures = list(filter(lambda s: lm.pulp_href == s.signed_manifest, signatures))
Expand All @@ -116,14 +109,14 @@ def test_sync_signed_images_from_sigstore(
@pytest.mark.parametrize(
"synced_repository", [{"sigstore": None, "signed_only": True}], indirect=True
)
def test_sync_images_without_sigstore_requiring_signatures(
container_signature_api, container_tag_api, synced_repository
):
def test_sync_images_without_sigstore_requiring_signatures(container_bindings, synced_repository):
"""Sync a repository with no sigstore but with the signed_only option enabled."""
signatures = container_signature_api.list(
signatures = container_bindings.ContentSignaturesApi.list(
repository_version=synced_repository.latest_version_href
).results
assert len(signatures) == 0

tags = container_tag_api.list(repository_version=synced_repository.latest_version_href).results
tags = container_bindings.ContentTagsApi.list(
repository_version=synced_repository.latest_version_href
).results
assert len(tags) == 0