Add tests and type hints for add_to_pydotorg

Author: hugovk

.coveragerc

@@ -4,5 +4,5 @@
 # Regexes for lines to exclude from consideration
 exclude_also =
     # Don't complain if non-runnable code isn't run:
-    if __name__ == .__main__.:
+    if __name__ == .__main__.
     def main

.gitignore

@@ -1,6 +1,5 @@
 #
 .idea/
-venv/
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

add-to-pydotorg.py renamed to add_to_pydotorg.py

@@ -15,7 +15,7 @@
 
 * Put an AUTH_INFO variable containing "username:api_key" in your environment.
 
-* Call this script as "python add-to-pydotorg.py RELEASE".
+* Call this script as "python add_to_pydotorg.py RELEASE".
 
   Each call will remove all previous file objects, so you can call the script
   multiple times.
@@ -30,20 +30,23 @@
 import subprocess
 import sys
 from os import path
+from typing import Any, Generator
 
 import requests
 
 
 # Copied from release.py
-def error(*msgs):
+def error(*msgs: Any) -> None:
     print("**ERROR**", file=sys.stderr)
     for msg in msgs:
         print(msg, file=sys.stderr)
     sys.exit(1)
 
 
 # Copied from release.py
-def run_cmd(cmd, silent=False, shell=True, **kwargs):
+def run_cmd(
+    cmd: list[str] | str, silent: bool = False, shell: bool = False, **kwargs: Any
+) -> None:
     if shell:
         cmd = " ".join(cmd)
     if not silent:
@@ -89,7 +92,9 @@ def run_cmd(cmd, silent=False, shell=True, **kwargs):
 }
 
 
-def get_file_descriptions(release):
+def get_file_descriptions(
+    release: str,
+) -> list[tuple[re.Pattern[str], tuple[str, int, bool, str]]]:
     v = minor_version_tuple(release)
     rx = re.compile
     # value is (file "name", OS id, download button, file "description").
@@ -157,54 +162,59 @@ def get_file_descriptions(release):
     ]
 
 
-def changelog_for(release):
-    new_url = f"http://docs.python.org/release/{release}/whatsnew/changelog.html"
-    if requests.head(new_url).status_code != 200:
-        return f"http://hg.python.org/cpython/file/v{release}/Misc/NEWS"
-
-
-def slug_for(release):
+def slug_for(release: str) -> str:
     return base_version(release).replace(".", "") + (
         "-" + release[len(base_version(release)) :]
         if release[len(base_version(release)) :]
         else ""
     )
 
 
-def sigfile_for(release, rfile):
+def sigfile_for(release: str, rfile: str) -> str:
     return download_root + f"{release}/{rfile}.asc"
 
 
-def md5sum_for(release, rfile):
+def md5sum_for(release: str, rfile: str) -> str:
     return hashlib.md5(
         open(ftp_root + base_version(release) + "/" + rfile, "rb").read()
     ).hexdigest()
 
 
-def filesize_for(release, rfile):
+def filesize_for(release: str, rfile: str) -> int:
     return path.getsize(ftp_root + base_version(release) + "/" + rfile)
 
 
-def make_slug(text):
+def make_slug(text: str) -> str:
     return re.sub("[^a-zA-Z0-9_-]", "", text.replace(" ", "-"))
 
 
-def base_version(release):
+def base_version(release: str) -> str:
     m = tag_cre.match(release)
+    assert m is not None, f"Invalid release: {release}"
     return ".".join(m.groups()[:3])
 
 
-def minor_version(release):
+def minor_version(release: str) -> str:
     m = tag_cre.match(release)
+    assert m is not None, f"Invalid release: {release}"
     return ".".join(m.groups()[:2])
 
 
-def minor_version_tuple(release):
+def minor_version_tuple(release: str) -> tuple[int, int]:
     m = tag_cre.match(release)
-    return (int(m.groups()[0]), int(m.groups()[1]))
-
-
-def build_file_dict(release, rfile, rel_pk, file_desc, os_pk, add_download, add_desc):
+    assert m is not None, f"Invalid release: {release}"
+    return int(m.groups()[0]), int(m.groups()[1])
+
+
+def build_file_dict(
+    release: str,
+    rfile: str,
+    rel_pk: int,
+    file_desc: str,
+    os_pk: int,
+    add_download: bool,
+    add_desc: str,
+) -> dict[str, Any]:
     """Return a dictionary with all needed fields for a ReleaseFile object."""
     d = {
         "name": file_desc,
@@ -243,7 +253,7 @@ def build_file_dict(release, rfile, rel_pk, file_desc, os_pk, add_download, add_
     return d
 
 
-def list_files(release):
+def list_files(release: str) -> Generator[tuple[str, str, int, bool, str], None, None]:
     """List all of the release's download files."""
     reldir = base_version(release)
     for rfile in os.listdir(path.join(ftp_root, reldir)):
@@ -279,7 +289,7 @@ def list_files(release):
             continue
 
 
-def query_object(objtype, **params):
+def query_object(objtype: str, **params: Any) -> int:
     """Find an API object by query parameters."""
     uri = base_url + f"downloads/{objtype}/"
     uri += "?" + "&".join(f"{k}={v}" for k, v in params.items())
@@ -290,7 +300,7 @@ def query_object(objtype, **params):
     return int(obj["resource_uri"].strip("/").split("/")[-1])
 
 
-def post_object(objtype, datadict):
+def post_object(objtype: str, datadict: dict[str, Any]) -> int:
     """Create a new API object."""
     resp = requests.post(
         base_url + "downloads/" + objtype + "/",
@@ -311,13 +321,15 @@ def post_object(objtype, datadict):
     return pk
 
 
-def sign_release_files_with_sigstore(release, release_files):
+def sign_release_files_with_sigstore(
+    release: str, release_files: list[tuple[str, str, int, bool, str]]
+) -> None:
     filenames = [
         ftp_root + f"{base_version(release)}/{rfile}"
         for rfile, file_desc, os_pk, add_download, add_desc in release_files
     ]
 
-    def has_sigstore_signature(filename):
+    def has_sigstore_signature(filename: str) -> bool:
         return os.path.exists(filename + ".sigstore") or (
             os.path.exists(filename + ".sig") and os.path.exists(filename + ".crt")
         )
@@ -378,7 +390,7 @@ def has_sigstore_signature(filename):
             )
 
 
-def main():
+def main() -> None:
     rel = sys.argv[1]
     print("Querying python.org for release", rel)
     rel_pk = query_object("release", name="Python+" + rel)
@@ -412,5 +424,5 @@ def main():
     print(f"Done - {n} files added")
 
 
-if not sys.flags.interactive:
+if __name__ == "__main__" and not sys.flags.interactive:
     main()

mypy-requirements.txt

@@ -3,3 +3,4 @@ mypy==1.10
 pytest
 pytest-mock
 sigstore==1.1.2
+types-requests

pyproject.toml

@@ -9,7 +9,6 @@ extra_checks = true
 warn_unreachable = true
 
 exclude = [
-  "^add-to-pydotorg.py$",
   "^buildbotapi.py$",
   "^run_release.py$",
   "^sbom.py$",

run_release.py

@@ -8,7 +8,6 @@
 
 import argparse
 import asyncio
-import builtins
 import contextlib
 import functools
 import getpass
@@ -24,7 +23,7 @@
 import urllib.request
 from dataclasses import dataclass
 from shelve import DbfilenameShelf
-from typing import Any, Callable, Generator, Iterator
+from typing import Any, Callable, Iterator
 
 import aiohttp
 import gnupg
@@ -282,14 +281,6 @@ def cd(path: str) -> Iterator[None]:
     os.chdir(current_path)
 
 
-@contextlib.contextmanager
-def supress_print() -> Generator[None, None, None]:
-    print_func = builtins.print
-    builtins.print = lambda *args, **kwargs: None
-    yield
-    builtins.print = print_func
-
-
 def check_tool(db: DbfilenameShelf, tool: str) -> None:
     if shutil.which(tool) is None:
         raise ReleaseException(f"{tool} is not available")
@@ -533,8 +524,8 @@ def sign_source_artifacts(db: DbfilenameShelf) -> None:
         uid = input("Please enter key ID to use for signing: ")
 
     tarballs_path = pathlib.Path(db["git_repo"] / str(db["release"]) / "src")
-    tgz = str(tarballs_path / (f"Python-{db['release']}.tgz"))
-    xz = str(tarballs_path / (f"Python-{db['release']}.tar.xz"))
+    tgz = str(tarballs_path / f"Python-{db['release']}.tgz")
+    xz = str(tarballs_path / f"Python-{db['release']}.tar.xz")
 
     subprocess.check_call(["gpg", "-bas", "-u", uid, tgz])
     subprocess.check_call(["gpg", "-bas", "-u", uid, xz])
@@ -867,8 +858,8 @@ def run_add_to_python_dot_org(db: DbfilenameShelf) -> None:
     client.connect(DOWNLOADS_SERVER, port=22, username=db["ssh_user"])
 
     # Ensure the file is there
-    source = pathlib.Path(__file__).parent / "add-to-pydotorg.py"
-    destination = pathlib.Path(f"/home/psf-users/{db['ssh_user']}/add-to-pydotorg.py")
+    source = pathlib.Path(__file__).parent / "add_to_pydotorg.py"
+    destination = pathlib.Path(f"/home/psf-users/{db['ssh_user']}/add_to_pydotorg.py")
     ftp_client = MySFTPClient.from_transport(client.get_transport())
     ftp_client.put(str(source), str(destination))
     ftp_client.close()
@@ -881,7 +872,7 @@ def run_add_to_python_dot_org(db: DbfilenameShelf) -> None:
     identity_token = issuer.identity_token()
 
     stdin, stdout, stderr = client.exec_command(
-        f"AUTH_INFO={auth_info} SIGSTORE_IDENTITY_TOKEN={identity_token} python3 add-to-pydotorg.py {db['release']}"
+        f"AUTH_INFO={auth_info} SIGSTORE_IDENTITY_TOKEN={identity_token} python3 add_to_pydotorg.py {db['release']}"
     )
     stderr_text = stderr.read().decode()
     if stderr_text:

tests/test_add_to_pydotorg.py

@@ -0,0 +1,81 @@
+import os
+
+import pytest
+
+os.environ["AUTH_INFO"] = "test_username:test_api_key"
+
+import add_to_pydotorg  # noqa: E402
+
+RELEASE = "3.14.0a0"
+
+
+@pytest.mark.parametrize(
+    ["release", "expected"],
+    [
+        ("3.9.0a0", "390-a0"),
+        ("3.10.0b3", "3100-b3"),
+        ("3.11.0rc2", "3110-rc2"),
+        ("3.12.15", "31215"),
+    ],
+)
+def test_slug_for(release: str, expected: str) -> None:
+    assert add_to_pydotorg.slug_for(release) == expected
+
+
+def test_sigfile_for() -> None:
+    assert (
+        add_to_pydotorg.sigfile_for("3.14.0", "Python-3.13.0.tgz")
+        == "https://www.python.org/ftp/python/3.14.0/Python-3.13.0.tgz.asc"
+    )
+
+
+@pytest.mark.parametrize(
+    ["release", "expected"],
+    [
+        ("3.9.0a0", "390a0"),
+        ("3.10.0b3", "3100b3"),
+        ("3.11.0rc2", "3110rc2"),
+        ("3.12.15", "31215"),
+    ],
+)
+def test_make_slug(release: str, expected: str) -> None:
+    assert add_to_pydotorg.make_slug(release) == expected
+
+
+@pytest.mark.parametrize(
+    ["release", "expected"],
+    [
+        ("3.9.0a0", "3.9.0"),
+        ("3.10.0b3", "3.10.0"),
+        ("3.11.0rc2", "3.11.0"),
+        ("3.12.15", "3.12.15"),
+    ],
+)
+def test_base_version(release: str, expected: str) -> None:
+    assert add_to_pydotorg.base_version(release) == expected
+
+
+@pytest.mark.parametrize(
+    ["release", "expected"],
+    [
+        ("3.9.0a0", "3.9"),
+        ("3.10.0b3", "3.10"),
+        ("3.11.0rc2", "3.11"),
+        ("3.12.15", "3.12"),
+    ],
+)
+def test_minor_version(release: str, expected: str) -> None:
+    assert add_to_pydotorg.minor_version(release) == expected
+
+
+@pytest.mark.parametrize(
+    ["release", "expected"],
+    [
+        ("3.9.0a0", (3, 9)),
+        ("3.10.0b3", (3, 10)),
+        ("3.11.0rc2", (3, 11)),
+        ("3.12.15", (3, 12)),
+    ],
+)
+def test_minor_version_tuple(release: str, expected: tuple[int, int]) -> None:
+    assert add_to_pydotorg.minor_version_tuple(release) == expected