diff --git a/tests/test_commands.py b/tests/test_commands.py index 94cd631a..c5aa31a2 100644 --- a/tests/test_commands.py +++ b/tests/test_commands.py @@ -2,6 +2,7 @@ import pytest +from tests import helpers from twine import commands from twine import exceptions @@ -49,3 +50,43 @@ def test_find_dists_handles_real_files(): ] files = commands._find_dists(expected) assert expected == files + + +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 = commands._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: [], + } diff --git a/tests/test_upload.py b/tests/test_upload.py index 0bce83f2..a2dc5135 100644 --- a/tests/test_upload.py +++ b/tests/test_upload.py @@ -116,46 +116,6 @@ def test_make_package_attestations_flagged_but_missing(upload_settings): upload._make_package(helpers.NEW_WHEEL_FIXTURE, {}, [], upload_settings) -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} diff --git a/twine/commands/__init__.py b/twine/commands/__init__.py index acd80f2e..91b8d39e 100644 --- a/twine/commands/__init__.py +++ b/twine/commands/__init__.py @@ -17,9 +17,10 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import fnmatch import glob import os.path -from typing import List +from typing import Dict, List, NamedTuple from twine import exceptions @@ -52,3 +53,41 @@ def _find_dists(dists: List[str]) -> List[str]: # Otherwise, files will be filenames that exist uploads.extend(files) return _group_wheel_files_first(uploads) + + +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) diff --git a/twine/commands/check.py b/twine/commands/check.py index 4101cc11..6b0fe5c7 100644 --- a/twine/commands/check.py +++ b/twine/commands/check.py @@ -127,7 +127,7 @@ def check( :return: ``True`` if there are rendering errors, otherwise ``False``. """ - uploads = [i for i in commands._find_dists(dists) if not i.endswith(".asc")] + uploads, _, _ = commands._split_inputs(dists) if not uploads: # Return early, if there are no files to check. logger.error("No files to check.") return False diff --git a/twine/commands/upload.py b/twine/commands/upload.py index 0f4f3633..939e1dfa 100644 --- a/twine/commands/upload.py +++ b/twine/commands/upload.py @@ -14,10 +14,8 @@ # 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, NamedTuple, cast +from typing import Dict, List, cast import requests from rich import print @@ -110,44 +108,6 @@ 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. @@ -187,7 +147,7 @@ def upload(upload_settings: settings.Settings, dists: List[str]) -> None: dists = commands._find_dists(dists) # Determine if the user has passed in pre-signed distributions or any attestations. - uploads, signatures, attestations_by_dist = _split_inputs(dists) + uploads, signatures, attestations_by_dist = commands._split_inputs(dists) print(f"Uploading distributions to {utils.sanitize_url(repository_url)}")