Add test to ensure that artifact downloading in _sign_rpm_content works

Create a test that syncs a repository, then syncs a completely different
repository, and then re-syncs the first.  Artifacts will be created during
the first sync that will be orphaned (since signing creates a new
artifact), and then, on the third sync, packages will be deduped to these
artifacts.  This test verifies that the artifact downloading in
_sign_rpm_content works with these deduped artifacts.

Signed-off-by: Jonathan Dieter <jdieter@ciq.com>

Author: jdieter

pulp_rpm/tests/functional/api/test_package_signing.py

@@ -14,6 +14,7 @@
     RPM_PACKAGE_FILENAME,
     RPM_UNSIGNED_URL,
     RPM_UNSIGNED_MODIFIED_FIXTURE_URL,
+    RPM_COMPLEX_FIXTURE_URL,
     RPM_PACKAGE_CONTENT_NAME,
 )
 from pulp_rpm.tests.functional.utils import get_package_repo_path
@@ -509,3 +510,121 @@ def test_no_resync_of_packages_on_second_sync(
     }
 
     assert original_packages["zebra"] == mutated_packages["zebra"]
+
+
+def test_do_resync_of_packages_on_third_sync(
+    init_and_sync,
+    tmp_path,
+    gen_object_with_cleanup,
+    download_content_unit,
+    signing_gpg_extra,
+    rpm_package_signing_service,
+    rpm_package_api,
+    rpm_repository_api,
+    rpm_repository_factory,
+    rpm_publication_factory,
+    rpm_distribution_factory,
+    delete_orphans_pre,
+    get_content,
+):
+    """
+    Ensure that a third sync where the second sync is completely different does successfully sign
+    all packages again.  This is primarily to test that the artifact downloading in
+    _sign_rpm_content is working correctly.
+    """
+    from pulp_rpm.app.shared_utils import RpmTool
+
+    # Setup GPG and RPM tool
+    gpg_a, _ = signing_gpg_extra
+
+    rpm_tool = RpmTool(tmp_path)
+    rpm_tool.import_pubkey_string(gpg_a.pubkey)
+
+    # Create repository with package signing service configured
+    repository = rpm_repository_factory(
+        package_signing_service=rpm_package_signing_service.pulp_href,
+        package_signing_fingerprint=gpg_a.fingerprint,
+    )
+    init_and_sync(
+        repository=repository,
+        sync_policy="mirror_content_only",
+    )
+
+    # Get synced packages - refresh repository to get latest version
+    updated_repository = rpm_repository_api.read(repository.pulp_href)
+    packages = rpm_package_api.list(repository_version=updated_repository.latest_version_href)
+    assert packages.count > 0, "No packages were synced"
+
+    # Test the first package to verify it was signed during sync
+    test_package = packages.results[0]
+
+    # Verify that the final served package is signed
+    publication = rpm_publication_factory(repository=repository.pulp_href)
+    distribution = rpm_distribution_factory(publication=publication.pulp_href)
+    downloaded_package = tmp_path / "package.rpm"
+    downloaded_package.write_bytes(
+        download_content_unit(
+            distribution.base_path, get_package_repo_path(test_package.location_href)
+        )
+    )
+    assert rpm_tool.verify_signature(downloaded_package)
+
+    # Save the content information of the packages from the original sync
+    original_packages = {
+        content["name"]: content
+        for content in get_content(updated_repository)["present"][RPM_PACKAGE_CONTENT_NAME]
+    }
+
+    # Second sync with completely different content
+    init_and_sync(
+        repository=repository,
+        url=RPM_COMPLEX_FIXTURE_URL,
+        sync_policy="mirror_content_only",
+    )
+    # Get synced packages - refresh repository to get latest version
+    updated_repository = rpm_repository_api.read(repository.pulp_href)
+    packages = rpm_package_api.list(repository_version=updated_repository.latest_version_href)
+    assert packages.count > 0, "No packages were synced"
+
+    # Test the first package to verify it was signed during sync
+    test_package = packages.results[0]
+
+    # Verify that the final served package is signed
+    publication = rpm_publication_factory(repository=repository.pulp_href)
+    distribution = rpm_distribution_factory(publication=publication.pulp_href)
+    downloaded_package = tmp_path / "package.rpm"
+    downloaded_package.write_bytes(
+        download_content_unit(
+            distribution.base_path, get_package_repo_path(test_package.location_href)
+        )
+    )
+    assert rpm_tool.verify_signature(downloaded_package)
+
+    # Third sync with the same content as the first sync
+    init_and_sync(repository=repository, sync_policy="mirror_content_only")
+    # Get synced packages - refresh repository to get latest version
+    updated_repository = rpm_repository_api.read(repository.pulp_href)
+    packages = rpm_package_api.list(repository_version=updated_repository.latest_version_href)
+    assert packages.count > 0, "No packages were synced"
+
+    # Test the first package to verify it was signed during sync
+    test_package = packages.results[0]
+
+    # Verify that the final served package is signed
+    publication = rpm_publication_factory(repository=repository.pulp_href)
+    distribution = rpm_distribution_factory(publication=publication.pulp_href)
+    downloaded_package = tmp_path / "package.rpm"
+    downloaded_package.write_bytes(
+        download_content_unit(
+            distribution.base_path, get_package_repo_path(test_package.location_href)
+        )
+    )
+    assert rpm_tool.verify_signature(downloaded_package)
+
+    # Test that the zebra package was re-synced this time
+    mutated_packages = {
+        content["name"]: content
+        for content in get_content(updated_repository)["present"][RPM_PACKAGE_CONTENT_NAME]
+    }
+
+    assert original_packages["zebra"] != mutated_packages["zebra"]