Repository: pypa/twine
Pull request #1095 (2024-04-29 15:47:34)

twine/upload: attestations scaffolding
Initial work towards #1094. 

Summary:

* Adds an `--attestations` flag (default `False`) and propagates its value into `Settings`
* Refactors the current input splitting logic into a separate `_split_inputs` helper, which returns dists, signatures, and attestations as separate data structures
* Unit tests for all of the above

I've tried to keep this change small (~50 lines without the tests), so `--attestations` is currently a no-op. But if you'd prefer it do something substantive, I can add the "fail if the user passes `--attestations` but one or more files are missing attestations" behavior to this changeset 🙂 


Patch:
diff --git a/twine/commands/upload.py b/twine/commands/upload.py
index b289d6b..5bc84c2 100644
--- a/twine/commands/upload.py
+++ b/twine/commands/upload.py
@@ -14,9 +14,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import argparse
+import fnmatch
 import logging
 import os.path
-from typing import Dict, List, cast
+from typing import Dict, List, NamedTuple, cast
 
 import requests
 from rich import print
@@ -91,6 +92,44 @@ def _make_package(
     return package
 
 
+class Inputs(NamedTuple):
+    """Represents structured user inputs."""
+
+    dists: List[str]
+    signatures: Dict[str, str]
+    attestations_by_dist: Dict[str, List[str]]
+
+
+def _split_inputs(
+    inputs: List[str],
+) -> Inputs:
+    """
+    Split the unstructured list of input files provided by the user into groups.
+
+    Three groups are returned: upload files (i.e. dists), signatures, and attestations.
+
+    Upload files are returned as a linear list, signatures are returned as a
+    dict of ``basename -> path``, and attestations are returned as a dict of
+    ``dist-path -> [attestation-path]``.
+    """
+    signatures = {os.path.basename(i): i for i in fnmatch.filter(inputs, "*.asc")}
+    attestations = fnmatch.filter(inputs, "*.*.attestation")
+    dists = [
+        dist
+        for dist in inputs
+        if dist not in (set(signatures.values()) | set(attestations))
+    ]
+
+    attestations_by_dist = {}
+    for dist in dists:
+        dist_basename = os.path.basename(dist)
+        attestations_by_dist[dist] = [
+            a for a in attestations if os.path.basename(a).startswith(dist_basename)
+        ]
+
+    return Inputs(dists, signatures, attestations_by_dist)
+
+
 def upload(upload_settings: settings.Settings, dists: List[str]) -> None:
     """Upload one or more distributions to a repository, and display the progress.
 
@@ -105,7 +144,8 @@ def upload(upload_settings: settings.Settings, dists: List[str]) -> None:
         The configured options related to uploading to a repository.
     :param dists:
         The distribution files to upload to the repository. This can also include
-        ``.asc`` files; the GPG signatures will be added to the corresponding uploads.
+        ``.asc`` and ``.attestation`` files, which will be added to their respective
+        file uploads.
 
     :raises twine.exceptions.TwineException:
         The upload failed due to a configuration error.
@@ -113,9 +153,8 @@ def upload(upload_settings: settings.Settings, dists: List[str]) -> None:
         The repository responded with an error.
     """
     dists = commands._find_dists(dists)
-    # Determine if the user has passed in pre-signed distributions
-    signatures = {os.path.basename(d): d for d in dists if d.endswith(".asc")}
-    uploads = [i for i in dists if not i.endswith(".asc")]
+    # Determine if the user has passed in pre-signed distributions or any attestations.
+    uploads, signatures, _ = _split_inputs(dists)
 
     upload_settings.check_repository_url()
     repository_url = cast(str, upload_settings.repository_config["repository"])
diff --git a/twine/settings.py b/twine/settings.py
index 78d382e..dc73adc 100644
--- a/twine/settings.py
+++ b/twine/settings.py
@@ -45,6 +45,7 @@ class Settings:
     def __init__(
         self,
         *,
+        attestations: bool = False,
         sign: bool = False,
         sign_with: str = "gpg",
         identity: Optional[str] = None,
@@ -64,6 +65,8 @@ class Settings:
     ) -> None:
         """Initialize our settings instance.
 
+        :param attestations:
+            Whether the package file should be uploaded with attestations.
         :param sign:
             Configure whether the package file should be signed.
         :param sign_with:
@@ -114,6 +117,7 @@ class Settings:
             repository_name=repository_name,
             repository_url=repository_url,
         )
+        self.attestations = attestations
         self._handle_package_signing(
             sign=sign,
             sign_with=sign_with,
@@ -175,6 +179,12 @@ class Settings:
             " This overrides --repository. "
             "(Can also be set via %(env)s environment variable.)",
         )
+        parser.add_argument(
+            "--attestations",
+            action="store_true",
+            default=False,
+            help="Upload each file's associated attestations.",
+        )
         parser.add_argument(
             "-s",
             "--sign",


Test patch:
diff --git a/tests/test_settings.py b/tests/test_settings.py
index ea31182..39944d2 100644
--- a/tests/test_settings.py
+++ b/tests/test_settings.py
@@ -164,3 +164,7 @@ class TestArgumentParsing:
         monkeypatch.setenv("TWINE_NON_INTERACTIVE", "0")
         args = self.parse_args([])
         assert not args.non_interactive
+
+    def test_attestations_flag(self):
+        args = self.parse_args(["--attestations"])
+        assert args.attestations
diff --git a/tests/test_upload.py b/tests/test_upload.py
index 291f070..b6477b6 100644
--- a/tests/test_upload.py
+++ b/tests/test_upload.py
@@ -105,6 +105,46 @@ def test_make_package_unsigned_dist(upload_settings, monkeypatch, caplog):
     ]
 
 
+def test_split_inputs():
+    """Split inputs into dists, signatures, and attestations."""
+    inputs = [
+        helpers.WHEEL_FIXTURE,
+        helpers.WHEEL_FIXTURE + ".asc",
+        helpers.WHEEL_FIXTURE + ".build.attestation",
+        helpers.WHEEL_FIXTURE + ".publish.attestation",
+        helpers.SDIST_FIXTURE,
+        helpers.SDIST_FIXTURE + ".asc",
+        helpers.NEW_WHEEL_FIXTURE,
+        helpers.NEW_WHEEL_FIXTURE + ".frob.attestation",
+        helpers.NEW_SDIST_FIXTURE,
+    ]
+
+    inputs = upload._split_inputs(inputs)
+
+    assert inputs.dists == [
+        helpers.WHEEL_FIXTURE,
+        helpers.SDIST_FIXTURE,
+        helpers.NEW_WHEEL_FIXTURE,
+        helpers.NEW_SDIST_FIXTURE,
+    ]
+
+    expected_signatures = {
+        os.path.basename(dist) + ".asc": dist + ".asc"
+        for dist in [helpers.WHEEL_FIXTURE, helpers.SDIST_FIXTURE]
+    }
+    assert inputs.signatures == expected_signatures
+
+    assert inputs.attestations_by_dist == {
+        helpers.WHEEL_FIXTURE: [
+            helpers.WHEEL_FIXTURE + ".build.attestation",
+            helpers.WHEEL_FIXTURE + ".publish.attestation",
+        ],
+        helpers.SDIST_FIXTURE: [],
+        helpers.NEW_WHEEL_FIXTURE: [helpers.NEW_WHEEL_FIXTURE + ".frob.attestation"],
+        helpers.NEW_SDIST_FIXTURE: [],
+    }
+
+
 def test_successs_prints_release_urls(upload_settings, stub_repository, capsys):
     """Print PyPI release URLS for each uploaded package."""
     stub_repository.release_urls = lambda packages: {RELEASE_URL, NEW_RELEASE_URL}