Skip to content

Commit

Permalink
Remove pulp-smash from test_rbac_sync_repositories & test_sync_signat…
Browse files Browse the repository at this point in the history
…ures
  • Loading branch information
gerrod3 committed Feb 6, 2025
1 parent 1c7a390 commit de15665
Show file tree
Hide file tree
Showing 2 changed files with 67 additions and 78 deletions.
82 changes: 39 additions & 43 deletions pulp_container/tests/functional/api/test_rbac_sync_repositories.py
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
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

0 comments on commit de15665

Please sign in to comment.