Merge pull request pulp#676 from gerrod3/test-modernize

Modernize tests to use new fixtures

Author: gerrod3

CHANGES/pulpcore.removal

@@ -0,0 +1 @@
+Raised the minimum `pulpcore` bound to `>=3.49` and dropped support for `python 3.8`.

functest_requirements.txt

@@ -1,7 +1,10 @@
-git+https://github.com/pulp/pulp-smash.git#egg=pulp-smash
 pytest<8
 lxml
 twine
 pypi-simple
 pulpcore-client
 pulp-python-client
+pytest-custom_exit_code
+pytest-xdist
+proxy.py~=2.4.4
+trustme~=1.1.0

pulp_python/pytest_plugin.py

@@ -0,0 +1,194 @@
+import pytest
+import uuid
+import subprocess
+
+from pulpcore.tests.functional.utils import BindingsNamespace
+from pulp_python.tests.functional.constants import (
+    PYTHON_FIXTURE_URL,
+    PYTHON_XS_PROJECT_SPECIFIER,
+    PYTHON_EGG_FILENAME,
+    PYTHON_URL,
+)
+
+
+# Bindings API Fixtures
+
+
+@pytest.fixture(scope="session")
+def python_bindings(_api_client_set, bindings_cfg):
+    """
+    A namespace providing preconfigured pulp_python api clients.
+
+    e.g. `python_bindings.RepositoriesPythonApi.list()`.
+    """
+    from pulpcore.client import pulp_python as python_bindings_module
+
+    api_client = python_bindings_module.ApiClient(bindings_cfg)
+    _api_client_set.add(api_client)
+    yield BindingsNamespace(python_bindings_module, api_client)
+    _api_client_set.remove(api_client)
+
+
+# Object Generation Fixtures
+
+
+@pytest.fixture
+def python_repo_factory(python_bindings, gen_object_with_cleanup):
+    """A factory to generate a Python Repository with auto-cleanup."""
+    def _gen_python_repo(remote=None, **body):
+        body.setdefault("name", str(uuid.uuid4()))
+        if remote:
+            body["remote"] = remote if isinstance(remote, str) else remote.pulp_href
+        return gen_object_with_cleanup(python_bindings.RepositoriesPythonApi, body)
+
+    return _gen_python_repo
+
+
+@pytest.fixture
+def python_repo(python_repo_factory):
+    """Creates a Python Repository and deletes it at test cleanup time."""
+    return python_repo_factory()
+
+
+@pytest.fixture
+def python_distribution_factory(python_bindings, gen_object_with_cleanup):
+    """A factory to generate a Python Distribution with auto-cleanup."""
+    def _gen_python_distribution(publication=None, repository=None, version=None, **body):
+        name = str(uuid.uuid4())
+        body.setdefault("name", name)
+        body.setdefault("base_path", name)
+        if publication:
+            body["publication"] = get_href(publication)
+        elif repository:
+            repo_href = get_href(repository)
+            if version:
+                if version.isnumeric():
+                    ver_href = f"{repo_href}versions/{version}/"
+                else:
+                    ver_href = get_href(version)
+                body = {"repository_version": ver_href}
+            else:
+                body["repository"] = repo_href
+        return gen_object_with_cleanup(python_bindings.DistributionsPypiApi, body)
+
+    yield _gen_python_distribution
+
+
+@pytest.fixture
+def python_publication_factory(python_bindings, gen_object_with_cleanup):
+    """A factory to generate a Python Publication with auto-cleanup."""
+    def _gen_python_publication(repository, version=None):
+        repo_href = get_href(repository)
+        if version:
+            if version.isnumeric():
+                ver_href = f"{repo_href}versions/{version}/"
+            else:
+                ver_href = get_href(version)
+            body = {"repository_version": ver_href}
+        else:
+            body = {"repository": repo_href}
+        return gen_object_with_cleanup(python_bindings.PublicationsPypiApi, body)
+
+    yield _gen_python_publication
+
+
+@pytest.fixture
+def python_remote_factory(python_bindings, gen_object_with_cleanup):
+    """A factory to generate a Python Remote with auto-cleanup."""
+    def _gen_python_remote(url=PYTHON_FIXTURE_URL, includes=None, **body):
+        body.setdefault("name", str(uuid.uuid4()))
+        body.setdefault("url", url)
+        if includes is None:
+            includes = PYTHON_XS_PROJECT_SPECIFIER
+        body["includes"] = includes
+        return gen_object_with_cleanup(python_bindings.RemotesPythonApi, body)
+
+    yield _gen_python_remote
+
+
+@pytest.fixture
+def python_repo_with_sync(
+    python_bindings, python_repo_factory, python_remote_factory, monitor_task
+):
+    """A factory to generate a Python Repository synced with the passed in Remote."""
+    def _gen_python_repo_sync(remote=None, mirror=False, repository=None, **body):
+        remote = remote or python_remote_factory()
+        repo = repository or python_repo_factory(**body)
+        sync_body = {"mirror": mirror, "remote": remote.pulp_href}
+        monitor_task(python_bindings.RepositoriesPythonApi.sync(repo.pulp_href, sync_body).task)
+        return python_bindings.RepositoriesPythonApi.read(repo.pulp_href)
+
+    yield _gen_python_repo_sync
+
+
+@pytest.fixture
+def download_python_file(tmp_path, http_get):
+    """Download a Python file and return its path."""
+    def _download_python_file(relative_path, url):
+        file_path = tmp_path / relative_path
+        with open(file_path, mode="wb") as f:
+            f.write(http_get(url))
+        return file_path
+
+    yield _download_python_file
+
+
+@pytest.fixture
+def python_file(download_python_file):
+    """Get a default (shelf-reader.tar.gz) Python file."""
+    return download_python_file(PYTHON_EGG_FILENAME, PYTHON_URL)
+
+
+@pytest.fixture
+def python_content_factory(python_bindings, download_python_file, monitor_task):
+    """A factory to create a Python Package Content."""
+    def _gen_python_content(relative_path=PYTHON_EGG_FILENAME, url=None, **body):
+        body["relative_path"] = relative_path
+        if url:
+            body["file"] = download_python_file(relative_path, url)
+        elif not any(x in body for x in ("artifact", "file", "upload")):
+            body["file"] = download_python_file(PYTHON_EGG_FILENAME, PYTHON_URL)
+        if repo := body.get("repository"):
+            repo_href = repo if isinstance(repo, str) else repo.pulp_href
+            body["repository"] = repo_href
+
+        task = python_bindings.ContentPackagesApi.create(**body).task
+        response = monitor_task(task)
+        return python_bindings.ContentPackagesApi.read(response.created_resources[0])
+
+    yield _gen_python_content
+
+
+# Utility fixtures
+
+
+@pytest.fixture
+def shelf_reader_cleanup():
+    """Take care of uninstalling shelf-reader before/after the test."""
+    cmd = ("pip", "uninstall", "shelf-reader", "-y")
+    subprocess.run(cmd)
+    yield
+    subprocess.run(cmd)
+
+
+@pytest.fixture
+def python_content_summary(python_bindings):
+    """Get a summary of the repository version's content."""
+    def _gen_summary(repository_version=None, repository=None, version=None):
+        if repository_version is None:
+            repo_href = get_href(repository)
+            if version:
+                repo_ver_href = f"{repo_href}versions/{version}/"
+            else:
+                repo_api = python_bindings.RepositoriesPythonApi
+                repo_ver_href = repo_api.read(repo_href).latest_version_href
+        else:
+            repo_ver_href = get_href(repository_version)
+        return python_bindings.RepositoriesPythonVersionsApi.read(repo_ver_href).content_summary
+
+    yield _gen_summary
+
+
+def get_href(item):
+    """Tries to get the href from the given item, wether it is a string or object."""
+    return item if isinstance(item, str) else item.pulp_href

pulp_python/tests/functional/api/test_auto_publish.py

@@ -1,63 +1,41 @@
-# coding=utf-8
-"""Tests automatic updating of publications and distributions."""
-from pulp_smash.pulp3.bindings import monitor_task
-from pulp_smash.pulp3.utils import download_content_unit
-
-from pulp_python.tests.functional.utils import (
-    cfg,
-    gen_python_remote,
-    TestCaseUsingBindings,
-    TestHelpersMixin
-)
-from pulp_python.tests.functional.utils import set_up_module as setUpModule  # noqa:F401
-from pulpcore.client.pulp_python import RepositorySyncURL
-
-
-class AutoPublishDistributeTestCase(TestCaseUsingBindings, TestHelpersMixin):
-    """Test auto-publish and auto-distribution"""
-
-    def setUp(self):
-        """Create remote, repo, publish settings, and distribution."""
-        self.remote = self.remote_api.create(gen_python_remote(policy="immediate"))
-        self.addCleanup(self.remote_api.delete, self.remote.pulp_href)
-        self.repo, self.distribution = self._create_empty_repo_and_distribution(autopublish=True)
-
-    def test_01_sync(self):
-        """Assert that syncing the repository triggers auto-publish and auto-distribution."""
-        self.assertEqual(self.publications_api.list().count, 0)
-        self.assertTrue(self.distribution.publication is None)
-
-        # Sync the repository.
-        repository_sync_data = RepositorySyncURL(remote=self.remote.pulp_href)
-        sync_response = self.repo_api.sync(self.repo.pulp_href, repository_sync_data)
-        task = monitor_task(sync_response.task)
-
-        # Check that all the appropriate resources were created
-        self.assertGreater(len(task.created_resources), 1)
-        self.assertEqual(self.publications_api.list().count, 1)
-        download_content_unit(cfg, self.distribution.to_dict(), "simple/")
-
-        # Sync the repository again. Since there should be no new repository version, there
-        # should be no new publications or distributions either.
-        sync_response = self.repo_api.sync(self.repo.pulp_href, repository_sync_data)
-        task = monitor_task(sync_response.task)
-
-        self.assertEqual(len(task.created_resources), 0)
-        self.assertEqual(self.publications_api.list().count, 1)
-
-    def test_02_modify(self):
-        """Assert that modifying the repository triggers auto-publish and auto-distribution."""
-        self.assertEqual(self.publications_api.list().count, 0)
-        self.assertTrue(self.distribution.publication is None)
-
-        # Modify the repository by adding a content unit
-        content = self.content_api.list().results[0].pulp_href
-        modify_response = self.repo_api.modify(
-            self.repo.pulp_href, {"add_content_units": [content]}
-        )
-        task = monitor_task(modify_response.task)
-
-        # Check that all the appropriate resources were created
-        self.assertGreater(len(task.created_resources), 1)
-        self.assertEqual(self.publications_api.list().count, 1)
-        download_content_unit(cfg, self.distribution.to_dict(), "simple/")
+import pytest
+
+
+@pytest.mark.parallel
+def test_autopublish_sync(
+    python_bindings, python_repo_factory, python_remote_factory, monitor_task
+):
+    """Assert that syncing the repository triggers auto-publish."""
+    remote = python_remote_factory()
+    repo = python_repo_factory(remote=remote, autopublish=True)
+
+    # Sync repository
+    task = monitor_task(python_bindings.RepositoriesPythonApi.sync(repo.pulp_href, {}).task)
+    assert len(task.created_resources) == 2
+    results = python_bindings.PublicationsPypiApi.list(repository=repo.pulp_href)
+    assert results.count == 1
+    assert results.results[0].pulp_href in task.created_resources
+
+    # Sync the repository again. Since there should be no new repository version, there
+    # should be no new publications
+    task = monitor_task(python_bindings.RepositoriesPythonApi.sync(repo.pulp_href, {}).task)
+    assert len(task.created_resources) == 0
+    results = python_bindings.PublicationsPypiApi.list(repository=repo.pulp_href)
+    assert results.count == 1
+
+
+@pytest.mark.parallel
+def test_autopublish_modify(
+    python_bindings, python_repo_factory, python_content_factory, monitor_task
+):
+    """Assert that modifying the repository triggers auto-publish."""
+    repo = python_repo_factory(autopublish=True)
+    content = python_content_factory()
+
+    # Modify the repository by adding a content unit
+    body = {"add_content_units": [content.pulp_href]}
+    task = monitor_task(python_bindings.RepositoriesPythonApi.modify(repo.pulp_href, body).task)
+    assert len(task.created_resources) == 2
+    results = python_bindings.PublicationsPypiApi.list(repository=repo.pulp_href)
+    assert results.count == 1
+    assert results.results[0].pulp_href in task.created_resources