diff --git a/pulp_ansible/tests/functional/api/collection/v3/test_sync.py b/pulp_ansible/tests/functional/api/collection/v3/test_sync.py index 1a33e54a1..944789e11 100644 --- a/pulp_ansible/tests/functional/api/collection/v3/test_sync.py +++ b/pulp_ansible/tests/functional/api/collection/v3/test_sync.py @@ -1,196 +1,139 @@ """Tests related to sync ansible plugin collection content type.""" -from pulpcore.client.pulp_ansible import AnsibleRepositorySyncURL +import pytest -from pulpcore.tests.functional.utils import PulpTaskError - -from pulp_ansible.tests.functional.utils import ( - gen_ansible_remote, - monitor_task, -) -from pulp_ansible.tests.functional.utils import TestCaseUsingBindings REQUIREMENTS_FILE = "collections:\n - testing.k8s_demo_collection" -class SyncCollectionsFromPulpServerTestCase(TestCaseUsingBindings): - """ - Test whether one can sync collections from a Pulp server. +@pytest.fixture(scope="class") +def galaxy_remote(ansible_collection_remote_factory): + return ansible_collection_remote_factory( + url="https://galaxy.ansible.com", + requirements_file=REQUIREMENTS_FILE, + sync_dependencies=False, + ) + - This performs two sync's, the first uses the V2 API and galaxy.ansible.com. The second is from - Pulp using the V3 API and uses the content brought in from the first sync. +@pytest.fixture(scope="class") +def galaxy_repository(ansible_bindings, monitor_task, ansible_repository_factory, galaxy_remote): + repository = ansible_repository_factory(remote=galaxy_remote.pulp_href) + result = ansible_bindings.RepositoriesAnsibleApi.sync(repository.pulp_href, {}) + monitor_task(result.task) + return repository + +@pytest.fixture(scope="class") +def galaxy_distribution(ansible_distribution_factory, galaxy_repository): + """ + Provide a distribution that serves the testing.k8s_demo_collection. """ + return ansible_distribution_factory(repository=galaxy_repository) - def setUp(self): - """Set up the Sync tests.""" - self.requirements_file = "collections:\n - testing.k8s_demo_collection" - body = gen_ansible_remote( - url="https://galaxy.ansible.com", - requirements_file=REQUIREMENTS_FILE, - sync_dependencies=False, - ) - self.remote = self.remote_collection_api.create(body) - self.addCleanup(self.remote_collection_api.delete, self.remote.pulp_href) - self.first_repo = self._create_repo_and_sync_with_remote(self.remote) - self.distribution = self._create_distribution_from_repo(self.first_repo) +class TestSync: + def test_synced_collection_is_findable(self, ansible_bindings, galaxy_distribution): + collection = ansible_bindings.PulpAnsibleApiV3CollectionsApi.read( + "k8s_demo_collection", "testing", galaxy_distribution.base_path + ) - def test_sync_collections_from_pulp(self): - """Test sync collections from pulp server.""" - second_body = gen_ansible_remote( - url=self.distribution.client_url, - requirements_file=REQUIREMENTS_FILE, + assert collection.name == "k8s_demo_collection" + assert collection.namespace == "testing" + + @pytest.mark.parametrize("requirements_file", (None, REQUIREMENTS_FILE)) + @pytest.mark.parametrize("mirror", (True, False)) + def test_sync_collection_from_pulp( + self, + ansible_bindings, + monitor_task, + ansible_repository_factory, + ansible_collection_remote_factory, + galaxy_distribution, + requirements_file, + mirror, + ): + remote = ansible_collection_remote_factory( + url=galaxy_distribution.client_url, + requirements_file=requirements_file, sync_dependencies=False, include_pulp_auth=True, ) - second_remote = self.remote_collection_api.create(second_body) - self.addCleanup(self.remote_collection_api.delete, second_remote.pulp_href) - - second_repo = self._create_repo_and_sync_with_remote(second_remote) - - first_content = self.cv_api.list( - repository_version=f"{self.first_repo.pulp_href}versions/1/" + repository = ansible_repository_factory(remote=remote.pulp_href) + result = ansible_bindings.RepositoriesAnsibleApi.sync( + repository.pulp_href, {"mirror": mirror} ) - assert len(first_content.results) >= 1 - second_content = self.cv_api.list(repository_version=f"{second_repo.pulp_href}versions/1/") - assert len(second_content.results) >= 1 - - def test_sync_collections_from_pulp_using_mirror_second_time(self): - """Test sync collections from pulp server using a mirror option the second time.""" - body = gen_ansible_remote( - url="https://galaxy.ansible.com", - requirements_file=REQUIREMENTS_FILE, - sync_dependencies=False, - ) - remote = self.remote_collection_api.create(body) - self.addCleanup(self.remote_collection_api.delete, remote.pulp_href) - - first_repo = self._create_repo_and_sync_with_remote(remote) - distribution = self._create_distribution_from_repo(first_repo) - - second_body = gen_ansible_remote(url=distribution.client_url, include_pulp_auth=True) - second_remote = self.remote_collection_api.create(second_body) - self.addCleanup(self.remote_collection_api.delete, second_remote.pulp_href) - - second_repo = self._create_repo_and_sync_with_remote(second_remote) + monitor_task(result.task) - first_content = self.cv_api.list(repository_version=f"{first_repo.pulp_href}versions/1/") - assert len(first_content.results) >= 1 - second_content = self.cv_api.list(repository_version=f"{second_repo.pulp_href}versions/1/") - assert len(second_content.results) >= 1 - - def test_sync_collection_named_api(self): - """Test sync collections from pulp server.""" - body = gen_ansible_remote( - url="https://galaxy.ansible.com", - requirements_file="collections:\n - rswaf.api", - sync_dependencies=False, + content = ansible_bindings.ContentCollectionVersionsApi.list( + repository_version=f"{repository.pulp_href}versions/1/" ) - remote = self.remote_collection_api.create(body) - self.addCleanup(self.remote_collection_api.delete, remote.pulp_href) - - repo = self._create_repo_and_sync_with_remote(remote) - distribution = self._create_distribution_from_repo(repo) - - collection = self.collections_v3api.read("api", "rswaf", distribution.base_path) - - assert "api" == collection.name - assert "rswaf" == collection.namespace - - def test_noop_resync_collections_from_pulp(self): - """Test whether sync yields no-op when repo hasn't changed since last sync.""" - second_body = gen_ansible_remote( - url=self.distribution.client_url, - requirements_file=REQUIREMENTS_FILE, + assert len(content.results) >= 1 + + @pytest.mark.parametrize("mirror", (True, False)) + def test_sync_absent_collection_from_pulp_fails( + self, + ansible_bindings, + monitor_task, + ansible_repository_factory, + ansible_collection_remote_factory, + galaxy_distribution, + mirror, + ): + remote = ansible_collection_remote_factory( + url=galaxy_distribution.client_url, + requirements_file="collections:\n - absent.not_present", sync_dependencies=False, include_pulp_auth=True, ) - second_remote = self.remote_collection_api.create(second_body) - self.addCleanup(self.remote_collection_api.delete, second_remote.pulp_href) - - second_repo = self._create_repo_with_attached_remote_and_sync(second_remote) - - second_content = self.cv_api.list(repository_version=f"{second_repo.pulp_href}versions/1/") - assert len(second_content.results) >= 1 - - # Resync - repository_sync_data = AnsibleRepositorySyncURL( - remote=second_remote.pulp_href, optimize=True + repository = ansible_repository_factory(remote=remote.pulp_href) + result = ansible_bindings.RepositoriesAnsibleApi.sync( + repository.pulp_href, {"mirror": mirror} ) - sync_response = self.repo_api.sync(second_repo.pulp_href, repository_sync_data) - task = monitor_task(sync_response.task) - second_repo = self.repo_api.read(second_repo.pulp_href) - - msg = "no-op: {url} did not change since last sync".format(url=second_remote.url) - messages = [r.message for r in task.progress_reports] - assert msg in str(messages) - - def test_noop_resync_with_mirror_from_pulp(self): - """Test whether no-op sync with mirror=True doesn't remove repository content.""" - second_body = gen_ansible_remote( - url=self.distribution.client_url, + message = "Collection absent.not_present does not exist on" + with pytest.raises(Exception, match=message): + monitor_task(result.task) + + @pytest.mark.parametrize("mirror", (True, False)) + def test_resync_is_noop( + self, + ansible_bindings, + monitor_task, + ansible_repository_factory, + ansible_collection_remote_factory, + galaxy_distribution, + mirror, + ): + remote = ansible_collection_remote_factory( + url=galaxy_distribution.client_url, requirements_file=REQUIREMENTS_FILE, sync_dependencies=False, include_pulp_auth=True, ) - second_remote = self.remote_collection_api.create(second_body) - self.addCleanup(self.remote_collection_api.delete, second_remote.pulp_href) - - second_repo = self._create_repo_with_attached_remote_and_sync(second_remote) - - second_content = self.cv_api.list(repository_version=f"{second_repo.pulp_href}versions/1/") - assert len(second_content.results) >= 1 + repository = ansible_repository_factory(remote=remote.pulp_href) + result = ansible_bindings.RepositoriesAnsibleApi.sync( + repository.pulp_href, {"mirror": mirror} + ) + monitor_task(result.task) # Resync - repository_sync_data = AnsibleRepositorySyncURL( - remote=second_remote.pulp_href, optimize=True, mirror=True + result = ansible_bindings.RepositoriesAnsibleApi.sync( + repository.pulp_href, {"mirror": mirror, "optimize": True} ) - sync_response = self.repo_api.sync(second_repo.pulp_href, repository_sync_data) - task = monitor_task(sync_response.task) - second_repo = self.repo_api.read(second_repo.pulp_href) - assert int(second_repo.latest_version_href[-2]) == 1 - - msg = "no-op: {url} did not change since last sync".format(url=second_remote.url) - messages = [r.message for r in task.progress_reports] - assert msg in str(messages) - - def test_update_requirements_file(self): - """Test requirements_file update.""" - body = gen_ansible_remote( - url=self.distribution.client_url, - requirements_file=REQUIREMENTS_FILE, - sync_dependencies=False, - include_pulp_auth=True, - ) - remote = self.remote_collection_api.create(body) - self.addCleanup(self.remote_collection_api.delete, remote.pulp_href) - - repo = self._create_repo_with_attached_remote_and_sync(remote) - assert repo.last_synced_metadata_time is not None - - response = self.remote_collection_api.partial_update( - remote.pulp_href, {"requirements_file": "collections:\n - ansible.posix"} + task = monitor_task(result.task) + msg = f"no-op: {galaxy_distribution.client_url} did not change since last sync" + messages = " ".join([r.message for r in task.progress_reports]) + assert msg in messages + + def test_updating_requirements_file_resets_last_synced( + self, ansible_bindings, monitor_task, galaxy_repository, galaxy_remote + ): + repository = ansible_bindings.RepositoriesAnsibleApi.read(galaxy_repository.pulp_href) + assert repository.last_synced_metadata_time is not None + + response = ansible_bindings.RemotesCollectionApi.partial_update( + galaxy_remote.pulp_href, {"requirements_file": "collections:\n - ansible.posix"} ) monitor_task(response.task) - repo = self.repo_api.read(repo.pulp_href) - assert repo.last_synced_metadata_time is None - - def test_sync_with_missing_collection(self): - """Test that syncing with a non-present collection gives a useful error.""" - body = gen_ansible_remote( - url=self.distribution.client_url, - requirements_file="collections:\n - absent.not_present", - sync_dependencies=False, - include_pulp_auth=True, - ) - remote = self.remote_collection_api.create(body) - self.addCleanup(self.remote_collection_api.delete, remote.pulp_href) - - with self.assertRaises(PulpTaskError) as cm: - self._create_repo_and_sync_with_remote(remote) - - task_result = cm.exception.task.to_dict() - msg = "absent.not_present does not exist" - assert msg in task_result["error"]["description"], task_result["error"]["description"] + repository = ansible_bindings.RepositoriesAnsibleApi.read(galaxy_repository.pulp_href) + assert repository.last_synced_metadata_time is None