diff --git a/galaxy_ng/tests/functional/__init__.py b/galaxy_ng/tests/functional/__init__.py deleted file mode 100644 index b9f2906b8e..0000000000 --- a/galaxy_ng/tests/functional/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Tests for galaxy plugin.""" diff --git a/galaxy_ng/tests/functional/api/__init__.py b/galaxy_ng/tests/functional/api/__init__.py deleted file mode 100644 index 4bd27a158d..0000000000 --- a/galaxy_ng/tests/functional/api/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Tests that communicate with Galaxy NG via the API.""" diff --git a/galaxy_ng/tests/functional/api/test_container_repository_tags.py b/galaxy_ng/tests/functional/api/test_container_repository_tags.py deleted file mode 100644 index 7ec3ce4e91..0000000000 --- a/galaxy_ng/tests/functional/api/test_container_repository_tags.py +++ /dev/null @@ -1,69 +0,0 @@ -import unittest -import pytest - -from urllib.parse import urlparse - -from pulpcore.client.galaxy_ng.exceptions import ApiException -from pulp_container.tests.functional.api import rbac_base -from pulp_container.tests.functional.constants import PULP_FIXTURE_1 -from pulp_smash import cli - -from galaxy_ng.tests.functional.utils import TestCaseUsingBindings - - -@pytest.mark.skip(reason="does not work currently in CI") -class ContainerRepositoryTagsTestCase(TestCaseUsingBindings, rbac_base.BaseRegistryTest): - """Test whether a container repository's tags can be listed. - - When running functional tests in dev environment please ensure that - Pulp Smash can either execute commands as root or can successfully - execute "sudo" on the localhost. - .. note:: When running against a non-https registry the client config - `insecure-registries` must be enabled. - For docker it is located in `/etc/docker/daemon.json` and content is:: - {"insecure-registries": ["0.0.0.0:5001"]} - For podman it is located in `/etc/containers/registries.conf` with:: - [registries.insecure] - registries = ['0.0.0.0:5001'] - """ - - @classmethod - def setUpClass(cls): - """ - Define APIs to use and pull images needed later in tests - """ - super().setUpClass() - - cls.registry = cli.RegistryClient(cls.cfg) - cls.registry.raise_if_unsupported(unittest.SkipTest, "Tests require podman/docker") - cls.registry_name = urlparse(cls.cfg.get_base_url()).netloc - admin_user, admin_password = cls.cfg.pulp_auth - cls.user_admin = {"username": admin_user, "password": admin_password} - cls._pull(f"{PULP_FIXTURE_1}:manifest_a") - - def test_list_container_repository_tags(self): - image_name = "foo/bar" - image_path = f"{PULP_FIXTURE_1}:manifest_a" - local_url = "/".join([self.registry_name, f"{image_name}:1.0"]) - - # expect the Container Repository Tags not to exist yet - with self.assertRaises(ApiException) as context: - self.container_repo_tags_api.list(base_path=image_name) - - # Bad request - assert context.exception.status == 404 - - self._push(image_path, local_url, self.user_admin) - - response = self.container_repo_tags_api.list(base_path=image_name) - - self.assertEqual(response.meta.count, 1) - for entry in response.data: - self.assertIn(entry.name, ["1.0"]) - - # Delete created Execution Environment - # api does not currently support delete - ee_delete_response = self.smash_client.delete( - f"{self.galaxy_api_prefix}/v3/plugin/execution-environments/repositories/{image_name}/" - ) - print(f"Delete execution environment: {ee_delete_response['state']}") diff --git a/galaxy_ng/tests/functional/api/test_container_sync_and_manifest_lists.py b/galaxy_ng/tests/functional/api/test_container_sync_and_manifest_lists.py deleted file mode 100644 index 5525be7788..0000000000 --- a/galaxy_ng/tests/functional/api/test_container_sync_and_manifest_lists.py +++ /dev/null @@ -1,121 +0,0 @@ -from pulp_smash.pulp3.bindings import monitor_task - -from galaxy_ng.tests.functional.utils import TestCaseUsingBindings - - -class ContainerSyncandManifestListTestCase(TestCaseUsingBindings): - """Test container sync and manifest lists. - - At the moment, the only way to get a container with a manifest list is to sync it from another - registy. Because of that, this will test both here. - - Test manifest list + sync - - set up docker registry - - set up remote container in docker registry for pulp/test-fixture-1 - - set tags to: ml_i, manifest_a - - perform sync - - verify that the repo has two images with tags ml_i and manifest_a, - that manifest_a is a manifest and that ml_i is a manifest list with - two manifests in it - - Test multiple repositories in a registry - - set up remote container in docker registry for pulp/test-fixture-1 with just tag manifest_a - - set up remote container in docker registry for pulp/test-fixture-1 with just tag manifest_b - - sync the entire registry - - verify that both repos have synced - """ - - def setUp(self): - self.docker_registry = self.container_registries_api.create({ - "name": "Docker Hub", - "url": "https://registry.hub.docker.com", - }) - self.addCleanup( - self.smash_client.delete, - ( - f"{self.galaxy_api_prefix}/_ui/v1/execution-environments/" - f"registries/{self.docker_registry.id}/" - ) - ) - - def _delete_remote_repo(self, remote): - self.smash_client.delete( - "{}/v3/plugin/execution-environments/repositories/{}/".format( - self.galaxy_api_prefix, - remote.name - ) - ) - - def test_manifests_and_remote_sync(self): - remote_repo = self.container_remotes_api.create({ - "name": "test-repo1", - "upstream_name": "pulp/test-fixture-1", - "include_tags": ["ml_i", "manifest_b"], - "registry": self.docker_registry.id, - }) - - self.addCleanup(self._delete_remote_repo, remote_repo) - - # the galaxy_ng client doesn't seem to return anything with the sync function, so we're - # using the api directly instead - self.smash_client.post( - "{}/v3/plugin/execution-environments/repositories/{}/_content/sync/".format( - self.galaxy_api_prefix, - remote_repo.name - ) - ) - - tags_list = self.container_repo_tags_api.list(remote_repo.name) - self.assertEqual(tags_list.meta.count, 2) - - tags_list = self.container_repo_tags_api.list(remote_repo.name, name="manifest_b") - self.assertEqual(tags_list.meta.count, 1) - - tagged_ml = tags_list.data[0] - self.assertIn("manifest.v2", tagged_ml.tagged_manifest.media_type) - - # Note, the manifest_a tag in the pulp test repo points to a - # manifest that's part of the ml_i manifest list - # so it won't show up in here since exclude_child_manifests - # filters out any manifest that is part of a manifest list. - image_list = self.container_images_api.list(remote_repo.name, exclude_child_manifests=True) - self.assertEqual(image_list.meta.count, 1) - - manifest_list = image_list.data[0] - - self.assertEqual(len(manifest_list.image_manifests), 2) - for img in manifest_list.image_manifests: - self.assertEqual(img.architecture, "amd64") - - self.assertIn("manifest.list.v2", manifest_list.media_type) - - def test_registry_sync(self): - remote_repo1 = self.container_remotes_api.create({ - "name": "test-repo1", - "upstream_name": "pulp/test-fixture-1", - "include_tags": ["manifest_b"], - "registry": self.docker_registry.id, - }) - - remote_repo2 = self.container_remotes_api.create({ - "name": "test-repo2", - "upstream_name": "pulp/test-fixture-1", - "include_tags": ["manifest_b"], - "registry": self.docker_registry.id, - }) - - self.addCleanup(self._delete_remote_repo, remote_repo1) - self.addCleanup(self._delete_remote_repo, remote_repo2) - - sync_task = self.smash_client.post(( - f"{self.galaxy_api_prefix}/_ui/v1/execution-environments/" - f"registries/{self.docker_registry.id}/sync/")) - - for task in sync_task['child_tasks']: - monitor_task(task) - - tags_list = self.container_repo_tags_api.list(remote_repo1.name) - self.assertEqual(tags_list.meta.count, 1) - - tags_list = self.container_repo_tags_api.list(remote_repo2.name) - self.assertEqual(tags_list.meta.count, 1) diff --git a/galaxy_ng/tests/functional/api/test_namespace.py b/galaxy_ng/tests/functional/api/test_namespace.py deleted file mode 100644 index c758e55640..0000000000 --- a/galaxy_ng/tests/functional/api/test_namespace.py +++ /dev/null @@ -1,51 +0,0 @@ -import string -import random - -from pulpcore.client.galaxy_ng.exceptions import ApiException -from galaxy_ng.tests.functional.utils import TestCaseUsingBindings -from galaxy_ng.tests.functional.utils import set_up_module as setUpModule # noqa:F401 - - -class CreateNamespaceTestCase(TestCaseUsingBindings): - """Test whether a namespace can be created.""" - - def delete_namespace(self, namespace_name): - # delete namespace - # namespace_api does not support delete, so we can use the smash_client directly - response = self.smash_client.delete( - f"{self.galaxy_api_prefix}/v3/namespaces/{namespace_name}" - ) - self.assertEqual(response.status_code, 204) - - def test_create_and_delete_namespace(self): - # generate name formed by 10 random ascii lowercase letters - random_name = ''.join(random.choices(string.ascii_lowercase, k=10)) - namespace_data = {"name": random_name, "groups": []} - - # create namespace - namespace = self.namespace_api.create(namespace=namespace_data) - self.assertEqual(namespace.name, random_name) - - # ensure namespace is available - namespaces = self.namespace_api.list(limit=100) - self.assertIn(namespace.name, [item.name for item in namespaces.data]) - - # delete namespace - self.delete_namespace(namespace.name) - - # ensure namespace is NO MORE available - namespaces = self.namespace_api.list(limit=100) - self.assertNotIn(namespace.name, [item.name for item in namespaces.data]) - - def test_negative_create_namespace_with_invalid_name(self): - # generate name formed by 10 random ascii lowercase letters - random_name = ''.join(random.choices(string.ascii_lowercase, k=10)) - random_name = f"ABC-{random_name}-$@" - namespace_data = {"name": random_name, "groups": []} - - # expect the namespace is not created because of invalid name - with self.assertRaises(ApiException) as context: - self.namespace_api.create(namespace=namespace_data) - - # Bad request - assert context.exception.status == 400 diff --git a/galaxy_ng/tests/functional/cli/__init__.py b/galaxy_ng/tests/functional/cli/__init__.py deleted file mode 100644 index a0c2eca4a5..0000000000 --- a/galaxy_ng/tests/functional/cli/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Tests that communicate with Galaxy NG via the ansible-galaxy CLI.""" diff --git a/galaxy_ng/tests/functional/cli/test_collection_install.py b/galaxy_ng/tests/functional/cli/test_collection_install.py deleted file mode 100644 index 76e3e3e321..0000000000 --- a/galaxy_ng/tests/functional/cli/test_collection_install.py +++ /dev/null @@ -1,54 +0,0 @@ -"""Tests that Collections hosted by Pulp can be installed by ansible-galaxy.""" -from os import path -import subprocess -import tempfile -import pytest - -from galaxy_ng.tests.functional.utils import TestCaseUsingBindings -from galaxy_ng.tests.functional.utils import set_up_module as setUpModule # noqa:F401 - - -class InstallCollectionTestCase(TestCaseUsingBindings): - """Test whether ansible-galaxy can install a Collection hosted by Pulp.""" - - def install_collection(self, collection_name, auth=True, repo_name="community"): - """Install given collection.""" - - # Preapare ansible.cfg for ansible-galaxy CLI - self.update_ansible_cfg(repo_name, auth) - - # In a temp dir, install a collection using ansible-galaxy CLI - with tempfile.TemporaryDirectory() as temp_dir: - cmd = "ansible-galaxy collection install -vvv {} -c -p {}".format( - collection_name, temp_dir - ) - - directory = "{}/ansible_collections/{}".format( - temp_dir, collection_name.replace(".", "/") - ) - - self.assertTrue( - not path.exists(directory), "Directory {} already exists".format(directory) - ) - - subprocess.run(cmd.split()) - - self.assertTrue(path.exists(directory), "Could not find directory {}".format(directory)) - - def test_install_collection(self): - """Test whether ansible-galaxy can install a Collection hosted by Pulp.""" - collection_name = "ansible.posix" - # Sync the repository. - self.sync_repo(f"collections:\n - {collection_name}") - self.install_collection(collection_name) - - def test_install_collection_unauthenticated(self): - """Test whether ansible-galaxy can download a Collection without auth.""" - - pytest.skip( - "GALAXY_ENABLE_UNAUTHENTICATED_COLLECTION_DOWNLOAD is not configurable yet.") - - collection_name = "ansible.posix" - # Sync the repository. - self.sync_repo(f"collections:\n - {collection_name}") - self.install_collection(collection_name, auth=False) diff --git a/galaxy_ng/tests/functional/cli/test_collection_upload.py b/galaxy_ng/tests/functional/cli/test_collection_upload.py deleted file mode 100644 index fbf0848914..0000000000 --- a/galaxy_ng/tests/functional/cli/test_collection_upload.py +++ /dev/null @@ -1,112 +0,0 @@ -"""Tests that Collections can be uploaded to Pulp with the ansible-galaxy CLI.""" - -import subprocess -import tempfile - -from pulp_smash.pulp3.bindings import delete_orphans -from pulp_smash.utils import http_get - -from galaxy_ng.tests.functional.utils import TestCaseUsingBindings - - -class UploadCollectionTestCase(TestCaseUsingBindings): - """Test whether ansible-galaxy can upload a Collection to Pulp.""" - - def test_upload_collection(self): - """Test whether ansible-galaxy can upload a Collection to Pulp.""" - delete_orphans() - - # Create namespace if it doesn't exist - data = self.namespace_api.list(name="pulp").data - if len(data) == 0: - self.namespace_api.create(namespace={"name": "pulp", "groups": []}) - - # Prepare ansible.cfg for ansible-galaxy CLI - self.update_ansible_cfg("published") - - # In a temp dir, publish a collection using ansible-galaxy CLI - with tempfile.TemporaryDirectory() as tmp_dir: - content = http_get( - "https://beta-galaxy.ansible.com" - "/api/v3/plugin/ansible/content/published/collections/artifacts" - "/pulp-squeezer-0.0.9.tar.gz" - ) - collection_path = f"{tmp_dir}/pulp-squeezer-0.0.9.tar.gz" - with open(collection_path, "wb") as f: - f.write(content) - - cmd = "ansible-galaxy collection publish -vvv -c {}".format(collection_path) - - subprocess.check_output(cmd.split()) - - # Verify that the collection was published - collections = self.collections_api.list("published") - self.assertEqual(collections.meta.count, 1) - - collection = self.collections_api.read(path="published", namespace="pulp", name="squeezer") - self.assertEqual(collection.namespace, "pulp") - self.assertEqual(collection.name, "squeezer") - self.assertEqual(collection.highest_version["version"], "0.0.9") - - # Cleanup - self.delete_collection(collection.namespace, collection.name) - self.delete_namespace(collection.namespace) - - def test_uploaded_collection_logged(self): - """ - Test whether a Collection uploaded via ansible-galaxy is - logged correctly in API Access Log. - """ - delete_orphans() - - # Create namespace if it doesn't exist - data = self.namespace_api.list(name="pulp").data - if len(data) == 0: - self.namespace_api.create(namespace={"name": "pulp", "groups": []}) - - # Preapare ansible.cfg for ansible-galaxy CLI - self.update_ansible_cfg("published") - - # In a temp dir, publish a collection using ansible-galaxy CLI - with tempfile.TemporaryDirectory() as tmp_dir: - content = http_get( - "https://beta-galaxy.ansible.com" - "/api/v3/plugin/ansible/content/published/collections/artifacts" - "/pulp-squeezer-0.0.9.tar.gz" - ) - collection_path = f"{tmp_dir}/pulp-squeezer-0.0.9.tar.gz" - with open(collection_path, "wb") as f: - f.write(content) - - cmd = "ansible-galaxy collection publish -vvv -c {}".format(collection_path) - - subprocess.check_output(cmd.split()) - - # Try to copy access log from container if tests are running outside of a container - # else copy direct because tests are running in the container - try: - cmd = f"docker cp pulp:/var/log/galaxy_api_access.log {tmp_dir}" - subprocess.check_output(cmd.split()) - except (subprocess.CalledProcessError, FileNotFoundError): - cmd = f"cp /var/log/galaxy_api_access.log {tmp_dir}" - subprocess.check_output(cmd.split()) - - with open(f"{tmp_dir}/galaxy_api_access.log") as f: - log_contents = f.readlines() - print(log_contents) - - # Verify that the collection was published - collections = self.collections_api.list("published") - self.assertEqual(collections.meta.count, 1) - - # Verify that the colletion publishing was logged in the api access log - collection = self.collections_api.read(path="published", namespace="pulp", name="squeezer") - for line in log_contents: - if "INFO: Collection uploaded by user 'admin': " in line and 'pulp' in line: - self.assertIn(collection.namespace, line) - self.assertIn(collection.name, line) - self.assertIn(collection.highest_version["version"], line) - - # Cleanup - self.delete_collection(collection.namespace, collection.name) - self.delete_namespace(collection.namespace) diff --git a/galaxy_ng/tests/functional/constants.py b/galaxy_ng/tests/functional/constants.py deleted file mode 100644 index 2a691ca30e..0000000000 --- a/galaxy_ng/tests/functional/constants.py +++ /dev/null @@ -1,58 +0,0 @@ -"""Constants for Pulp Galaxy plugin tests.""" -from urllib.parse import urljoin - -from pulp_smash.constants import PULP_FIXTURES_BASE_URL -from pulp_smash.pulp3.constants import ( - BASE_DISTRIBUTION_PATH, - BASE_PUBLICATION_PATH, - BASE_REMOTE_PATH, - BASE_REPO_PATH, - BASE_CONTENT_PATH, -) - -# FIXME: list any download policies supported by your plugin type here. -# If your plugin supports all download policies, you can import this -# from pulp_smash.pulp3.constants instead. -# DOWNLOAD_POLICIES = ["immediate", "streamed", "on_demand"] -DOWNLOAD_POLICIES = ["immediate"] - -# FIXME: replace 'unit' with your own content type names, and duplicate as necessary for each type -GALAXY_CONTENT_NAME = "galaxy.unit" - -# FIXME: replace 'unit' with your own content type names, and duplicate as necessary for each type -GALAXY_CONTENT_PATH = urljoin(BASE_CONTENT_PATH, "galaxy/units/") - -GALAXY_REMOTE_PATH = urljoin(BASE_REMOTE_PATH, "galaxy/galaxy/") - -GALAXY_REPO_PATH = urljoin(BASE_REPO_PATH, "galaxy/galaxy/") - -GALAXY_PUBLICATION_PATH = urljoin(BASE_PUBLICATION_PATH, "galaxy/galaxy/") - -GALAXY_DISTRIBUTION_PATH = urljoin(BASE_DISTRIBUTION_PATH, "galaxy/galaxy/") - -# FIXME: replace this with your own fixture repository URL and metadata -GALAXY_FIXTURE_URL = urljoin(PULP_FIXTURES_BASE_URL, "galaxy/") -"""The URL to a galaxy repository.""" - -# FIXME: replace this with the actual number of content units in your test fixture -GALAXY_FIXTURE_COUNT = 3 -"""The number of content units available at :data:`GALAXY_FIXTURE_URL`.""" - -GALAXY_FIXTURE_SUMMARY = {GALAXY_CONTENT_NAME: GALAXY_FIXTURE_COUNT} -"""The desired content summary after syncing :data:`GALAXY_FIXTURE_URL`.""" - -# FIXME: replace this with the location of one specific content unit of your choosing -GALAXY_URL = urljoin(GALAXY_FIXTURE_URL, "") -"""The URL to an galaxy file at :data:`GALAXY_FIXTURE_URL`.""" - -# FIXME: replace this with your own fixture repository URL and metadata -GALAXY_INVALID_FIXTURE_URL = urljoin(PULP_FIXTURES_BASE_URL, "galaxy-invalid/") -"""The URL to an invalid galaxy repository.""" - -# FIXME: replace this with your own fixture repository URL and metadata -GALAXY_LARGE_FIXTURE_URL = urljoin(PULP_FIXTURES_BASE_URL, "galaxy_large/") -"""The URL to a galaxy repository containing a large number of content units.""" - -# FIXME: replace this with the actual number of content units in your test fixture -GALAXY_LARGE_FIXTURE_COUNT = 25 -"""The number of content units available at :data:`GALAXY_LARGE_FIXTURE_URL`.""" diff --git a/galaxy_ng/tests/functional/utils.py b/galaxy_ng/tests/functional/utils.py deleted file mode 100644 index 84e0522fb9..0000000000 --- a/galaxy_ng/tests/functional/utils.py +++ /dev/null @@ -1,269 +0,0 @@ -"""Utilities for tests for the galaxy plugin.""" -import os -from functools import partial -import requests -from unittest import SkipTest -from tempfile import NamedTemporaryFile - -from pulp_smash import api, config, selectors -from pulp_smash.pulp3.bindings import delete_orphans, monitor_task, PulpTestCase -from pulp_smash.pulp3.utils import ( - gen_remote, - gen_repo, - get_content, - require_pulp_3, - require_pulp_plugins, - sync, -) - -from galaxy_ng.tests.functional.constants import ( - GALAXY_CONTENT_NAME, - GALAXY_CONTENT_PATH, - GALAXY_FIXTURE_URL, - GALAXY_PUBLICATION_PATH, - GALAXY_REMOTE_PATH, - GALAXY_REPO_PATH, - GALAXY_URL, -) - -from pulpcore.client.pulpcore import ( - ApiClient as CoreApiClient, - ArtifactsApi, - TasksApi, -) -from pulpcore.client.galaxy_ng import ( - ApiClient as GalaxyApiClient, - ApiContentV3SyncApi, - ApiContentV3SyncConfigApi, - ApiV3NamespacesApi, - ApiV3PluginExecutionEnvironmentsRepositoriesContentTagsApi as ContainerRepositoryEndpointApi, - ApiV3PluginExecutionEnvironmentsRepositoriesApi as ContainerRepositoryApi, - ApiUiV1ExecutionEnvironmentsRemotesApi, - ApiUiV1ExecutionEnvironmentsRegistriesApi, - ApiV3PluginExecutionEnvironmentsRepositoriesContentSyncApi, - ApiUiV1ExecutionEnvironmentsRegistriesSyncApi, - ApiV3PluginExecutionEnvironmentsRepositoriesContentImagesApi as ContainerImagesAPI, -) -from pulpcore.client.pulp_ansible import ( - ApiClient as PulpAnsibleApiClient, - PulpAnsibleApiV3CollectionsApi, -) - -configuration = config.get_config().get_bindings_config() - - -def set_up_module(): - """Skip tests Pulp 3 isn't under test or if galaxy_ng isn't installed.""" - require_pulp_3(SkipTest) - require_pulp_plugins({"galaxy"}, SkipTest) - - -def gen_galaxy_client(): - """Return an OBJECT for galaxy client.""" - return GalaxyApiClient(configuration) - - -def gen_pulp_ansible_client(): - """Return an OBJECT for galaxy client.""" - return PulpAnsibleApiClient(configuration) - - -def gen_galaxy_remote(url=GALAXY_FIXTURE_URL, **kwargs): - """Return a semi-random dict for use in creating a galaxy Remote. - - :param url: The URL of an external content source. - """ - # FIXME: Add any fields specific to a galaxy remote here - return gen_remote(url, **kwargs) - - -def get_galaxy_content_paths(repo, version_href=None): - """Return the relative path of content units present in a galaxy repository. - - :param repo: A dict of information about the repository. - :param version_href: The repository version to read. - :returns: A dict of lists with the paths of units present in a given repository. - Paths are given as pairs with the remote and the local version for different content types. - """ - # FIXME: The "relative_path" is actually a file path and name - # It's just an example -- this needs to be replaced with an implementation that works - # for repositories of this content type. - return { - GALAXY_CONTENT_NAME: [ - (content_unit["relative_path"], content_unit["relative_path"]) - for content_unit in get_content(repo, version_href)[GALAXY_CONTENT_NAME] - ], - } - - -def gen_galaxy_content_attrs(artifact): - """Generate a dict with content unit attributes. - - :param artifact: A dict of info about the artifact. - :returns: A semi-random dict for use in creating a content unit. - """ - # FIXME: Add content specific metadata here. - return {"_artifact": artifact["pulp_href"]} - - -def populate_pulp(cfg, url=GALAXY_FIXTURE_URL): - """Add galaxy contents to Pulp. - - :param pulp_smash.config.PulpSmashConfig: Information about a Pulp application. - :param url: The galaxy repository URL. Defaults to - :data:`pulp_smash.constants.GALAXY_FIXTURE_URL` - :returns: A list of dicts, where each dict describes one galaxy content in Pulp. - """ - client = api.Client(cfg, api.json_handler) - remote = {} - repo = {} - try: - remote.update(client.post(GALAXY_REMOTE_PATH, gen_galaxy_remote(url))) - repo.update(client.post(GALAXY_REPO_PATH, gen_repo())) - sync(cfg, remote, repo) - finally: - if remote: - client.delete(remote["pulp_href"]) - if repo: - client.delete(repo["pulp_href"]) - return client.get(GALAXY_CONTENT_PATH)["results"] - - -def publish(cfg, repo, version_href=None): - """Publish a repository. - :param pulp_smash.config.PulpSmashConfig cfg: Information about the Pulp - host. - :param repo: A dict of information about the repository. - :param version_href: A href for the repo version to be published. - :returns: A publication. A dict of information about the just created - publication. - """ - if version_href: - body = {"repository_version": version_href} - else: - body = {"repository": repo["pulp_href"]} - - client = api.Client(cfg, api.json_handler) - call_report = client.post(GALAXY_PUBLICATION_PATH, body) - tasks = tuple(api.poll_spawned_tasks(cfg, call_report)) - return client.get(tasks[-1]["created_resources"][0]) - - -skip_if = partial(selectors.skip_if, exc=SkipTest) # pylint:disable=invalid-name -"""The ``@skip_if`` decorator, customized for unittest. - -:func:`pulp_smash.selectors.skip_if` is test runner agnostic. This function is -identical, except that ``exc`` has been set to ``unittest.SkipTest``. -""" - -core_client = CoreApiClient(configuration) -tasks = TasksApi(core_client) - - -def gen_artifact(url=GALAXY_URL): - """Creates an artifact.""" - response = requests.get(url) - with NamedTemporaryFile() as temp_file: - temp_file.write(response.content) - temp_file.flush() - artifact = ArtifactsApi(core_client).create(file=temp_file.name) - return artifact.to_dict() - - -class TestCaseUsingBindings(PulpTestCase): - """A parent TestCase that instantiates the various bindings used throughout tests.""" - - @classmethod - def setUpClass(cls): - """Create class-wide variables.""" - cls.cfg = config.get_config() - cls.client = gen_galaxy_client() - cls.pulp_ansible_client = gen_pulp_ansible_client() - cls.smash_client = api.Client(cls.cfg, api.smart_handler) - cls.namespace_api = ApiV3NamespacesApi(cls.client) - cls.collections_api = PulpAnsibleApiV3CollectionsApi(cls.pulp_ansible_client) - cls.sync_config_api = ApiContentV3SyncConfigApi(cls.client) - cls.sync_api = ApiContentV3SyncApi(cls.client) - cls.container_repo_tags_api = ContainerRepositoryEndpointApi(cls.client) - cls.container_repo_api = ContainerRepositoryApi(cls.client) - cls.container_remotes_api = ApiUiV1ExecutionEnvironmentsRemotesApi(cls.client) - cls.container_registries_api = ApiUiV1ExecutionEnvironmentsRegistriesApi(cls.client) - cls.container_remote_sync_api = \ - ApiV3PluginExecutionEnvironmentsRepositoriesContentSyncApi(cls.client) - cls.container_registry_sync_api = ApiUiV1ExecutionEnvironmentsRegistriesSyncApi(cls.client) - cls.container_images_api = ContainerImagesAPI(cls.client) - cls.get_ansible_cfg_before_test() - cls.galaxy_api_prefix = os.getenv( - "PULP_GALAXY_API_PATH_PREFIX", "/api/galaxy").rstrip("/") - - def tearDown(self): - """Clean class-wide variable.""" - with open("ansible.cfg", "w") as f: - f.write(self.previous_ansible_cfg) - delete_orphans() - - @classmethod - def get_token(cls): - """Get a Galaxy NG token.""" - return cls.smash_client.post(f"{cls.galaxy_api_prefix}/v3/auth/token/")["token"] - - @classmethod - def get_ansible_cfg_before_test(cls): - """Update ansible.cfg to use the given base_path.""" - try: - with open("ansible.cfg", "r") as f: - cls.previous_ansible_cfg = f.read() - except FileNotFoundError: - cls.previous_ansible_cfg = ( - "[defaults]\n" - "remote_tmp = /tmp/ansible\n" - "local_tmp = /tmp/ansible\n" - ) - - def update_ansible_cfg(self, base_path, auth=True): - """Update ansible.cfg to use the given base_path.""" - token = f"token={self.get_token()}" if auth else "" - ansible_cfg = ( - f"{self.previous_ansible_cfg}\n" - "[galaxy]\n" - "server_list = community_repo\n" - "\n" - "[galaxy_server.community_repo]\n" - f"url={self.cfg.get_base_url()}" - f"{self.galaxy_api_prefix}/content/{base_path}/\n" - f"{token}" - ) - with open("ansible.cfg", "w") as f: - f.write(ansible_cfg) - - def sync_repo(self, requirements_file, **kwargs): - """Sync a repository with a given requirements_file""" - repo_name = kwargs.get("repo_name", "community") - url = kwargs.get("url", "https://beta-galaxy.ansible.com/api/") - - self.sync_config_api.update( - repo_name, - { - "url": f"{url}", - "requirements_file": f"{requirements_file}", - }, - ) - - response = self.sync_api.sync(repo_name) - api_root = os.environ.get("PULP_API_ROOT", "/pulp/") - monitor_task(f"{api_root}api/v3/tasks/{response.task}/") - - def delete_namespace(self, namespace_name): - """Delete a Namespace""" - # namespace_api does not support delete, so we can use the smash_client directly - self.smash_client.delete( - f"{self.galaxy_api_prefix}/v3/namespaces/{namespace_name}" - ) - - def delete_collection(self, collection_namespace, collection_name): - """Delete a Collection""" - monitor_task(self.collections_api.delete( - namespace=collection_namespace, - name=collection_name, - path="published" - ).task)