Skip to content

Commit

Permalink
Fix tests grouped under a class
Browse files Browse the repository at this point in the history
  • Loading branch information
gerrod3 committed Feb 10, 2025
1 parent 41105b3 commit 3063689
Show file tree
Hide file tree
Showing 4 changed files with 137 additions and 98 deletions.
64 changes: 32 additions & 32 deletions pulp_container/tests/functional/api/test_push_content.py
Original file line number Diff line number Diff line change
Expand Up @@ -415,50 +415,50 @@ def test_push_to_existing_regular_repository(
local_registry.tag_and_push(image_path, local_url)


class PushManifestListTestCase:
class TestPushManifestList:
"""A test case that verifies if a container client can push manifest lists to the registry."""

manifest_a = f"{REGISTRY_V2_REPO_PULP}:manifest_a"
manifest_b = f"{REGISTRY_V2_REPO_PULP}:manifest_b"
manifest_c = f"{REGISTRY_V2_REPO_PULP}:manifest_c"
v2s2_tag = "manifest_list"
v2s2_image_path = f"foo_v2s2:{v2s2_tag}"
oci_tag = "manifest_list_oci"
oci_image_path = f"foo_oci:{oci_tag}"
empty_image_tag = "empty_manifest_list"
empty_image_path = f"foo_empty:{empty_image_tag}"

@pytest.fixture(scope="class", autouse=True)
def setup(self, local_registry):
def setup(self, registry_client):
"""Initialize a new manifest list that will be pushed to the registry."""
self.manifest_a = f"{REGISTRY_V2_REPO_PULP}:manifest_a"
self.manifest_b = f"{REGISTRY_V2_REPO_PULP}:manifest_b"
self.manifest_c = f"{REGISTRY_V2_REPO_PULP}:manifest_c"
local_registry.pull(self.manifest_a)
local_registry.pull(self.manifest_b)
local_registry.pull(self.manifest_c)
self.registry_name = local_registry.name
registry_client.pull(self.manifest_a)
registry_client.pull(self.manifest_b)
registry_client.pull(self.manifest_c)

# create a new manifest list composed of the pulled manifest images
self.v2s2_tag = "manifest_list"
self.v2s2_image_path = f"{self.registry_name}/foo_v2s2:{self.v2s2_tag}"
local_registry._dispatch_command("manifest", "create", self.v2s2_tag)
local_registry._dispatch_command("manifest", "add", self.v2s2_tag, self.manifest_a)
local_registry._dispatch_command("manifest", "add", self.v2s2_tag, self.manifest_b)
local_registry._dispatch_command("manifest", "add", self.v2s2_tag, self.manifest_c)
registry_client._dispatch_command("manifest", "create", self.v2s2_tag)
registry_client._dispatch_command("manifest", "add", self.v2s2_tag, self.manifest_a)
registry_client._dispatch_command("manifest", "add", self.v2s2_tag, self.manifest_b)
registry_client._dispatch_command("manifest", "add", self.v2s2_tag, self.manifest_c)

# create a new manifest list composed of the pulled manifest images
self.oci_tag = "manifest_list_oci"
self.oci_image_path = f"{self.registry_name}/foo_oci:{self.oci_tag}"
local_registry._dispatch_command("manifest", "create", self.oci_tag)
local_registry._dispatch_command("manifest", "add", self.oci_tag, self.manifest_a)
local_registry._dispatch_command("manifest", "add", self.oci_tag, self.manifest_b)
local_registry._dispatch_command("manifest", "add", self.oci_tag, self.manifest_c)
registry_client._dispatch_command("manifest", "create", self.oci_tag)
registry_client._dispatch_command("manifest", "add", self.oci_tag, self.manifest_a)
registry_client._dispatch_command("manifest", "add", self.oci_tag, self.manifest_b)
registry_client._dispatch_command("manifest", "add", self.oci_tag, self.manifest_c)

# create an empty manifest list
self.empty_image_tag = "empty_manifest_list"
self.empty_image_path = f"{self.registry_name}/foo_empty:{self.empty_image_tag}"
local_registry._dispatch_command("manifest", "create", self.empty_image_tag)
registry_client._dispatch_command("manifest", "create", self.empty_image_tag)

yield

local_registry._dispatch_command("manifest", "rm", self.v2s2_tag)
local_registry._dispatch_command("manifest", "rm", self.oci_tag)
local_registry._dispatch_command("manifest", "rm", self.empty_image_tag)
registry_client._dispatch_command("manifest", "rm", self.v2s2_tag)
registry_client._dispatch_command("manifest", "rm", self.oci_tag)
registry_client._dispatch_command("manifest", "rm", self.empty_image_tag)

local_registry._dispatch_command("image", "rm", self.manifest_a)
local_registry._dispatch_command("image", "rm", self.manifest_b)
local_registry._dispatch_command("image", "rm", self.manifest_c)
registry_client._dispatch_command("image", "rm", self.manifest_a)
registry_client._dispatch_command("image", "rm", self.manifest_b)
registry_client._dispatch_command("image", "rm", self.manifest_c)

def test_push_manifest_list_v2s2(self, local_registry, container_bindings, add_to_cleanup):
"""Push the created manifest list in the v2s2 format."""
Expand Down Expand Up @@ -498,7 +498,7 @@ def test_push_manifest_list_v2s2(self, local_registry, container_bindings, add_t

# load manifest_list.json
image_path = "/v2/{}/manifests/{}".format(distribution.base_path, latest_tag.name)
latest_image_url = urljoin(container_bindings.cfg.get_base_url(), image_path)
latest_image_url = urljoin(container_bindings.client.configuration.host, image_path)

auth = get_auth_for_url(latest_image_url)
content_response = requests.get(
Expand Down Expand Up @@ -548,7 +548,7 @@ def test_push_manifest_list_oci(self, local_registry, container_bindings, add_to

# load manifest_list.json
image_path = "/v2/{}/manifests/{}".format(distribution.base_path, latest_tag.name)
latest_image_url = urljoin(container_bindings.cfg.get_base_url(), image_path)
latest_image_url = urljoin(container_bindings.client.configuration.host, image_path)

auth = get_auth_for_url(latest_image_url)
content_response = requests.get(
Expand Down
140 changes: 81 additions & 59 deletions pulp_container/tests/functional/api/test_tagging_images.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,14 +2,15 @@

import pytest
import subprocess
from urllib.parse import urlparse
from pulp_container.tests.functional.constants import (
CONTAINER_TAG_PATH,
PULP_FIXTURE_1,
REGISTRY_V2_REPO_PULP,
)


@pytest.fixture
@pytest.fixture(scope="class")
def tagger_helper(container_bindings, monitor_task):
class TaggingTestCommons:
"""Common utilities for tagging and untagging images."""
Expand Down Expand Up @@ -51,29 +52,33 @@ def untag_image(self, tag_name):
return TaggingTestCommons


class RepositoryTaggingTestCase:
class TestRepositoryTagging:
"""A test case for standard a container repository."""

@pytest.fixture(scope="class", autouse=True)
def setup(self, tagger_helper, container_repo, container_remote_factory, container_sync):
@pytest.fixture(scope="class")
def setup(
self, tagger_helper, container_repository_factory, container_remote_factory, container_sync
):
"""Create class wide-variables."""
self.tagger = tagger_helper(container_repo)
self.repository = container_repo
repository = container_repository_factory()
tagger = tagger_helper(repository)

remote = container_remote_factory(url=PULP_FIXTURE_1)
container_sync(container_repo, remote)
remote = container_remote_factory(upstream_name=PULP_FIXTURE_1)
container_sync(repository, remote)
return repository, tagger

def test_01_tag_first_image(self, container_bindings):
def test_01_tag_first_image(self, container_bindings, setup):
"""
Create a new test for manifest.
This test checks if the tag was created in a new repository version.
"""
manifest_a = self.tagger.get_manifest_by_tag("manifest_a")
self.tagger.tag_image(manifest_a, "new_tag")
repository, tagger = setup
manifest_a = tagger.get_manifest_by_tag("manifest_a")
tagger.tag_image(manifest_a, "new_tag")

new_repository_version_href = "{repository_href}versions/{new_version}/".format(
repository_href=self.repository.pulp_href, new_version="2"
repository_href=repository.pulp_href, new_version="2"
)
created_tag = container_bindings.ContentTagsApi.list(
repository_version_added=new_repository_version_href
Expand All @@ -91,38 +96,40 @@ def test_01_tag_first_image(self, container_bindings):
removed_content = repository_version.content_summary.removed
assert removed_content == {}

def test_02_tag_first_image_with_same_tag(self, container_bindings):
def test_02_tag_first_image_with_same_tag(self, container_bindings, setup):
"""
Tag the same manifest with the same name.
This test checks if a new repository version was created with no content added.
"""
repository, tagger = setup
latest_version_before = container_bindings.RepositoriesContainerApi.read(
self.repository.pulp_href
repository.pulp_href
).latest_version_href

manifest_a = self.tagger.get_manifest_by_tag("manifest_a")
self.tagger.tag_image(manifest_a, "new_tag")
manifest_a = tagger.get_manifest_by_tag("manifest_a")
tagger.tag_image(manifest_a, "new_tag")

latest_version_after = container_bindings.RepositoriesContainerApi.read(
self.repository.pulp_href
repository.pulp_href
).latest_version_href

assert latest_version_before == latest_version_after

def test_03_tag_second_image_with_same_tag(self, container_bindings):
def test_03_tag_second_image_with_same_tag(self, container_bindings, setup):
"""
Tag a different manifest with the same name.
This test checks if a new repository version was created with a new content added
and the old removed.
"""
manifest_a = self.tagger.get_manifest_by_tag("manifest_a")
manifest_b = self.tagger.get_manifest_by_tag("manifest_b")
self.tagger.tag_image(manifest_b, "new_tag")
repository, tagger = setup
manifest_a = tagger.get_manifest_by_tag("manifest_a")
manifest_b = tagger.get_manifest_by_tag("manifest_b")
tagger.tag_image(manifest_b, "new_tag")

new_repository_version_href = "{repository_href}versions/{new_version}/".format(
repository_href=self.repository.pulp_href, new_version="3"
repository_href=repository.pulp_href, new_version="3"
)
created_tag = container_bindings.ContentTagsApi.list(
repository_version_added=new_repository_version_href
Expand Down Expand Up @@ -156,12 +163,13 @@ def test_03_tag_second_image_with_same_tag(self, container_bindings):
removed_tags = removed_content["container.tag"]["count"]
assert removed_tags == 1

def test_04_untag_second_image(self, container_bindings):
def test_04_untag_second_image(self, container_bindings, setup):
"""Untag the manifest and check if the tag was added in a new repository version."""
self.tagger.untag_image("new_tag")
repository, tagger = setup
tagger.untag_image("new_tag")

new_repository_version_href = "{repository_href}versions/{new_version}/".format(
repository_href=self.repository.pulp_href, new_version="4"
repository_href=repository.pulp_href, new_version="4"
)

removed_tags_href = "{unit_path}?{filters}".format(
Expand All @@ -185,67 +193,81 @@ def test_04_untag_second_image(self, container_bindings):
).results[0]
assert removed_tag.name == "new_tag"

def test_05_untag_second_image_again(self, container_bindings):
def test_05_untag_second_image_again(self, container_bindings, setup):
"""Untag the manifest that was already untagged."""
repository, tagger = setup
with pytest.raises(container_bindings.ApiException):
self.tagger.untag_image("new_tag")
tagger.untag_image("new_tag")


class PushRepositoryTaggingTestCase:
class TestPushRepositoryTagging:
"""A test case for a container push repository."""

@pytest.fixture(scope="class", autouse=True)
def setup(self, tagger_helper, container_bindings, local_registry):
repository_name = "namespace/tags"

@pytest.fixture(scope="class")
def setup(self, tagger_helper, container_bindings, registry_client, add_to_cleanup):
"""Define APIs to use and pull images needed later in tests."""
self.repository_name = "namespace/tags"
self.registry_repository_name = f"{local_registry.name}/{self.repository_name}"
cfg = container_bindings.client.configuration
registry_name = urlparse(cfg.host).netloc
registry_repository_name = f"{registry_name}/{self.repository_name}"
manifest_a = f"{REGISTRY_V2_REPO_PULP}:manifest_a"
tagged_registry_manifest_a = f"{self.registry_repository_name}:manifest_a"
tagged_registry_manifest_a = f"{registry_repository_name}:manifest_a"
manifest_b = f"{REGISTRY_V2_REPO_PULP}:manifest_b"
tagged_registry_manifest_b = f"{self.registry_repository_name}:manifest_b"

local_registry.pull(manifest_a)
local_registry.pull(manifest_b)
local_registry.tag(manifest_a, tagged_registry_manifest_a)
local_registry.tag(manifest_b, tagged_registry_manifest_b)
local_registry.push(tagged_registry_manifest_a)
local_registry.push(tagged_registry_manifest_b)

self.repository = container_bindings.RepositoriesContainerPushApi.list(
tagged_registry_manifest_b = f"{registry_repository_name}:manifest_b"

registry_client.pull(manifest_a)
registry_client.pull(manifest_b)
registry_client.tag(manifest_a, tagged_registry_manifest_a)
registry_client.tag(manifest_b, tagged_registry_manifest_b)
registry_client.login("-u", cfg.username, "-p", cfg.password, registry_name)
registry_client.push(tagged_registry_manifest_a)
registry_client.push(tagged_registry_manifest_b)
registry_client.logout(registry_name)

repository = container_bindings.RepositoriesContainerPushApi.list(
name=self.repository_name
).results[0]
tagger = tagger_helper(repository)
distro = container_bindings.DistributionsContainerApi.list(
name=self.repository_name
).results[0]
self.tagger = tagger_helper(self.repository)
add_to_cleanup(container_bindings.DistributionsContainerApi, distro.pulp_href)
return repository, tagger

def test_01_tag_first_image(self, local_registry):
def test_01_tag_first_image(self, local_registry, setup):
"""Check if a tag was created and correctly pulled from a repository."""
manifest_a = self.tagger.get_manifest_by_tag("manifest_a")
self.tagger.tag_image(manifest_a, "new_tag")
repository, tagger = setup
manifest_a = tagger.get_manifest_by_tag("manifest_a")
tagger.tag_image(manifest_a, "new_tag")

tagged_image = f"{self.registry_repository_name}:new_tag"
tagged_image = f"{self.repository_name}:new_tag"
local_registry.pull(tagged_image)
local_registry.rmi(tagged_image)
local_registry._dispatch_command("rmi", tagged_image)

def test_02_tag_second_image_with_same_tag(self, local_registry):
def test_02_tag_second_image_with_same_tag(self, local_registry, setup):
"""Check if the existing tag correctly references a new manifest."""
tagged_image = f"{self.registry_repository_name}:manifest_b"
repository, tagger = setup
tagged_image = f"{self.repository_name}:manifest_b"
local_registry.pull(tagged_image)
local_image_b = local_registry.inspect(tagged_image)
local_registry.rmi(tagged_image)
local_registry._dispatch_command("rmi", tagged_image)

manifest_b = self.tagger.get_manifest_by_tag("manifest_b")
self.tagger.tag_image(manifest_b, "new_tag")
tagged_image = f"{self.registry_repository_name}:new_tag"
manifest_b = tagger.get_manifest_by_tag("manifest_b")
tagger.tag_image(manifest_b, "new_tag")
tagged_image = f"{self.repository_name}:new_tag"
local_registry.pull(tagged_image)
local_image_b_tagged = local_registry.inspect(tagged_image)

assert local_image_b[0]["Id"] == local_image_b_tagged[0]["Id"]

local_registry.rmi(tagged_image)
local_registry._dispatch_command("rmi", tagged_image)

def test_03_remove_tag(self, local_registry):
def test_03_remove_tag(self, local_registry, setup):
"""Check if the client cannot pull by the removed tag."""
self.tagger.untag_image("new_tag")
repository, tagger = setup
tagger.untag_image("new_tag")

non_existing_tagged_image = f"{self.registry_repository_name}:new_tag"
non_existing_tagged_image = f"{self.repository_name}:new_tag"
with pytest.raises(subprocess.CalledProcessError):
local_registry.pull(non_existing_tagged_image)
Original file line number Diff line number Diff line change
Expand Up @@ -11,9 +11,10 @@
get_auth_for_url,
)
from pulp_container.constants import MEDIA_TYPE
from pulp_container.tests.functional.constants import PULP_FIXTURE_1


class TokenAuthenticationTestCase:
class TestTokenAuthentication:
"""
A test case for authenticating users via Bearer token.
Expand All @@ -25,12 +26,14 @@ class TokenAuthenticationTestCase:
@pytest.fixture(scope="class")
def setup(
self,
container_repo,
container_remote,
container_repository_factory,
container_remote_factory,
container_sync,
container_distribution_factory,
container_bindings,
):
container_repo = container_repository_factory()
container_remote = container_remote_factory(upstream_name=PULP_FIXTURE_1)
container_sync(container_repo, container_remote)
distro = container_distribution_factory(repository=container_repo.pulp_href)
tag_response = container_bindings.ContentTagsApi.list(name="manifest_a")
Expand Down
Loading

0 comments on commit 3063689

Please sign in to comment.