[PR #3050/25dfd999 backport][3.18] Refactor test_auto_publish to use pytest

CHANGES/2870.misc

@@ -0,0 +1 @@
+Refactored test_auto_publish to use pytest fixtures.

pulp_rpm/tests/functional/api/test_auto_publish.py

@@ -1,100 +1,73 @@
 # coding=utf-8
-"""Tests that sync file plugin repositories."""
-import unittest
+"""Tests that sync rpm plugin repositories."""
+import pytest
 
-from pulp_smash import config
-from pulp_smash.pulp3.bindings import monitor_task
-from pulp_smash.pulp3.utils import gen_repo
-
-from pulp_rpm.tests.functional.utils import gen_rpm_client, gen_rpm_remote
-from pulp_rpm.tests.functional.utils import set_up_module as setUpModule  # noqa:F401
 
 from pulpcore.client.pulp_rpm import (
-    ContentPackagesApi,
-    DistributionsRpmApi,
-    PublicationsRpmApi,
-    RepositoriesRpmApi,
     RpmRepositorySyncURL,
-    RemotesRpmApi,
 )
 
 
-class AutoPublishDistributeTestCase(unittest.TestCase):
-    """Test auto-publish and auto-distribution."""
-
-    @classmethod
-    def setUpClass(cls):
-        """Create class-wide variables."""
-        cls.cfg = config.get_config()
-        cls.client = gen_rpm_client()
-
-        cls.content_api = ContentPackagesApi(cls.client)
-        cls.repo_api = RepositoriesRpmApi(cls.client)
-        cls.remote_api = RemotesRpmApi(cls.client)
-        cls.publications_api = PublicationsRpmApi(cls.client)
-        cls.distributions_api = DistributionsRpmApi(cls.client)
-
-    def setUp(self):
-        """Create remote, repo, publish settings, and distribution."""
-        self.remote = self.remote_api.create(gen_rpm_remote())
-        self.repo = self.repo_api.create(gen_repo(autopublish=True, sqlite_metadata=True))
-        response = self.distributions_api.create(
-            {"name": "foo", "base_path": "bar/foo", "repository": self.repo.pulp_href}
-        )
-        distribution_href = monitor_task(response.task).created_resources[0]
-        self.distribution = self.distributions_api.read(distribution_href)
-
-    def tearDown(self):
-        """Clean up."""
-        monitor_task(self.repo_api.delete(self.repo.pulp_href).task)
-        monitor_task(self.remote_api.delete(self.remote.pulp_href).task)
-        monitor_task(self.distributions_api.delete(self.distribution.pulp_href).task)
-
-    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 = RpmRepositorySyncURL(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)
-        publications = self.publications_api.list()
-        self.assertEqual(publications.count, 1)
-
-        # Check that the publish settings were used
-        publication = publications.results[0]
-        self.assertEqual(publication.sqlite_metadata, True)
-
-        # 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 coment 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)
-        publications = self.publications_api.list()
-        self.assertEqual(publications.count, 1)
-
-        # Check that the publish settings were used
-        publication = publications.results[0]
-        self.assertEqual(publication.sqlite_metadata, True)
+@pytest.fixture
+def setup_autopublish(rpm_repository_factory, rpm_rpmremote_factory, rpm_distribution_factory):
+    """Create remote, repo, publish settings, and distribution."""
+    remote = rpm_rpmremote_factory()
+    repo = rpm_repository_factory(autopublish=True, sqlite_metadata=True)
+    distribution = rpm_distribution_factory(repository=repo.pulp_href)
+
+    return repo, remote, distribution
+
+
+@pytest.mark.parallel
+def test_01_sync(setup_autopublish, rpm_repository_api, rpm_publication_api, monitor_task):
+    """Assert that syncing the repository triggers auto-publish and auto-distribution."""
+    repo, remote, distribution = setup_autopublish
+    assert rpm_publication_api.list(repository=repo.pulp_href).count == 0
+    assert distribution.publication is None
+
+    # Sync the repository.
+    repository_sync_data = RpmRepositorySyncURL(remote=remote.pulp_href)
+    sync_response = rpm_repository_api.sync(repo.pulp_href, repository_sync_data)
+    task = monitor_task(sync_response.task)
+
+    # Check that all the appropriate resources were created
+    assert len(task.created_resources) > 1
+    publications = rpm_publication_api.list(repository=repo.pulp_href)
+    assert publications.count == 1
+
+    # Check that the publish settings were used
+    publication = publications.results[0]
+    assert publication.sqlite_metadata is True
+
+    # Sync the repository again. Since there should be no new repository version, there
+    # should be no new publications or distributions either.
+    sync_response = rpm_repository_api.sync(repo.pulp_href, repository_sync_data)
+    task = monitor_task(sync_response.task)
+
+    assert len(task.created_resources) == 0
+    assert rpm_publication_api.list(repository=repo.pulp_href).count == 1
+
+
+@pytest.mark.parallel
+def test_02_modify(
+    setup_autopublish, rpm_repository_api, rpm_package_api, rpm_publication_api, monitor_task
+):
+    """Assert that modifying the repository triggers auto-publish and auto-distribution."""
+    repo, remote, distribution = setup_autopublish
+    assert rpm_publication_api.list(repository=repo.pulp_href).count == 0
+    assert distribution.publication is None
+
+    # Modify the repository by adding a content unit
+    content = rpm_package_api.list().results[0].pulp_href
+
+    modify_response = rpm_repository_api.modify(repo.pulp_href, {"add_content_units": [content]})
+    task = monitor_task(modify_response.task)
+
+    # Check that all the appropriate resources were created
+    assert len(task.created_resources) > 1
+    publications = rpm_publication_api.list(repository=repo.pulp_href)
+    assert publications.count == 1
+
+    # Check that the publish settings were used
+    publication = publications.results[0]
+    assert publication.sqlite_metadata is True