Cleanup and iteration after review

[noissue]

pulp_rpm/app/tasks/signing.py

@@ -5,7 +5,6 @@
 from pulpcore.plugin.util import get_url
 
 from pulp_rpm.app.models.content import RpmPackageSigningService
-from pulp_rpm.app.shared_utils import get_sha256
 
 
 def sign_and_create(
@@ -24,21 +23,12 @@ def sign_and_create(
     package_signing_service = RpmPackageSigningService.objects.get(pk=signing_service_pk)
     uploaded_package = PulpTemporaryFile.objects.get(pk=temporary_file_pk)
     with NamedTemporaryFile(mode="wb", dir=".", delete=False) as final_package:
-        print("*" * 100)
-        fd = uploaded_package.file.open()
-        final_package.write(fd.read())
-        final_package.flush()
-        fd.close()
+        with uploaded_package.file.open() as fd:
+            final_package.write(fd.read())
+            final_package.flush()
 
-        final_package.seek(0)
-        print(f"{signing_fingerprint}")
-        print("digest_before", get_sha256(final_package.name))
         package_signing_service.sign(final_package.name, pubkey_fingerprint=signing_fingerprint)
-        final_package.seek(0)
         artifact = Artifact.init_and_validate(final_package.name)
-        print("local_pkg_diges", get_sha256(final_package.name))
-        print(f"{artifact.sha256=}")
-        print("*" * 100)
         artifact.save()
         resource = CreatedResource(content_object=artifact)
         resource.save()

pulp_rpm/app/viewsets/package.py

@@ -85,7 +85,7 @@ def create(self, request):
         if sign_package is not True:
             return super().create(request)
 
-        # signing case invariants
+        # signing case
         request.data.pop("file")
         request.data.pop("sign_package")
         temp_uploaded_file = serializer.validated_data.get("file")
@@ -104,9 +104,8 @@ def create(self, request):
         }
         task_payload = {k: v for k, v in request.data.items()}
         task_exclusive = [
-            item
-            for item in (serializer.validated_data.get(key) for key in ("upload", "repository"))
-            if item
+            serializer.validated_data.get("upload"),
+            serializer.validated_data.get("repository"),
         ]
         task = dispatch(
             rpm_tasks.signing.sign_and_create,

pulp_rpm/tests/functional/api/test_package_signing.py

@@ -1,3 +1,4 @@
+from dataclasses import dataclass
 from pathlib import Path
 
 import pytest
@@ -23,6 +24,13 @@ def test_register_rpm_package_signing_service(rpm_package_signing_service):
     assert "/api/v3/signing-services/" in service.pulp_href
 
 
+@dataclass
+class GPGMetadata:
+    pubkey: str
+    fingerprint: str
+    keyid: str
+
+
 @pytest.fixture
 def signing_gpg_extra(signing_gpg_metadata):
     """GPG instance with an extra gpg keypair registered."""
@@ -39,7 +47,10 @@ def signing_gpg_extra(signing_gpg_metadata):
 
     pubkey_a = gpg.export_keys(fingerprint_a)
     pubkey_b = gpg.export_keys(fingerprint_b)
-    return fingerprint_a, pubkey_a, fingerprint_b, pubkey_b
+    return (
+        GPGMetadata(pubkey_a, fingerprint_a, fingerprint_a[-8:]),
+        GPGMetadata(pubkey_b, fingerprint_b, fingerprint_b[-8:]),
+    )
 
 
 @pytest.mark.parallel
@@ -63,13 +74,14 @@ def test_sign_package_on_upload(
     This ensures different
     """
     # Setup RPM tool and package to upload
-    fingerprint_a, pubkey_a, fingerprint_b, pubkey_b = signing_gpg_extra
-    assert rpm_package_signing_service.pubkey_fingerprint == fingerprint_a
-    assert rpm_package_signing_service.pubkey_fingerprint != fingerprint_b
+    gpg_a, gpg_b = signing_gpg_extra
+    assert rpm_package_signing_service.pubkey_fingerprint == gpg_a.fingerprint
+    assert rpm_package_signing_service.pubkey_fingerprint != gpg_b.fingerprint
+    fingerprint_set = (gpg_a.fingerprint, gpg_b.fingerprint)
 
     rpm_tool = RpmTool(tmp_path)
-    rpm_tool.import_pubkey_string(pubkey_a)
-    rpm_tool.import_pubkey_string(pubkey_b)
+    rpm_tool.import_pubkey_string(gpg_a.pubkey)
+    rpm_tool.import_pubkey_string(gpg_b.pubkey)
 
     file_to_upload = tmp_path / RPM_PACKAGE_FILENAME
     file_to_upload.write_bytes(requests.get(RPM_UNSIGNED_URL).content)
@@ -78,7 +90,7 @@ def test_sign_package_on_upload(
 
     # Upload Package to Repository with signing-option on
     # The same file is uploaded, but signed with different keys each time
-    for fingerprint in (fingerprint_a, fingerprint_b):
+    for fingerprint in fingerprint_set:
         repository = rpm_repository_factory(
             package_signing_service=rpm_package_signing_service.pulp_href,
             package_signing_pubkey=fingerprint,
@@ -99,20 +111,3 @@ def test_sign_package_on_upload(
             download_content_unit(distribution.base_path, get_package_repo_path(pkg_location_href))
         )
         assert rpm_tool.verify_signature(downloaded_package)
-
-    # Can't upload same file with same key
-    with pytest.raises(InvalidSignatureError, match="The package is not signed: .*"):
-        rpm_tool.verify_signature(file_to_upload)
-    repository = rpm_repository_factory(
-        package_signing_service=rpm_package_signing_service.pulp_href,
-        package_signing_pubkey=fingerprint_a,
-    )
-    upload_response = rpm_package_api.create(
-        file=str(file_to_upload.absolute()),
-        repository=repository.pulp_href,
-        sign_package=True,
-    )
-    package_a_href = monitor_task(upload_response.task)
-    # import epdb;epdb.serve(port=12345)
-
-    # Cant use unreachable/invalid repository.signing_package_pubkey

pulp_rpm/tests/unit/test_rpm_tool.py

@@ -21,22 +21,18 @@ def get_fixture(tmp_path: Path, url: str):
     return file
 
 
-def test_get_empty_rpm_is_valid(tmp_path, monkeypatch):
-    """Can get a valid rpm."""
+def test_can_get_empty_rpm(tmp_path, monkeypatch):
+    """
+    Can get a valid rpm without hitting the internet.
+
+    This rpm can be used in production by the SigningService.validate() method, which
+    is used to validate a provided signing script knows how to sign an rpm blob.
+    """
     # Should't hit the internet
     # https://stackoverflow.com/a/18601897
     monkeypatch.setattr(socket, "socket", connection_guard)
-
-    # Assert is a valid rpm
     rpm_pkg = RpmTool.get_empty_rpm(tmp_path)
     assert rpm_pkg.exists()
-    with open(rpm_pkg, "rb") as pkg:
-        # https://rpm-software-management.github.io/rpm/manual/format_lead.html
-        rpm_magic_numbers = bytes([0xED, 0xAB, 0xEE, 0xDB])
-        pkg_lead = pkg.read(96)
-        rpm_major_version = pkg_lead[4]
-        assert pkg_lead[:4] == rpm_magic_numbers
-        assert rpm_major_version == 3
 
 
 def test_verify_signature_is_valid(tmp_path):

requirements.txt

@@ -6,4 +6,4 @@ productmd~=1.33.0
 pulpcore>=3.44.1,<3.55
 solv~=0.7.21
 aiohttp_xmlrpc~=1.5.0
-importlib-resources
+importlib-resources~=6.4.0