TST: Extending typing to tests; cover generic and scripts folder files

.github/workflows/github-ci.yaml

@@ -262,7 +262,7 @@ jobs:
         ruff check .
     - name: Test with mypy
       run : |
-        mypy pypdf
+        mypy .
     - name: Install docs requirements
       run: |
         pip install -r requirements/docs.txt

make_release.py

@@ -198,12 +198,12 @@ def get_formatted_changes(git_tag: str) -> tuple[str, str]:
     commits = get_git_commits_since_tag(git_tag)
 
     # Group by prefix
-    grouped = {}
+    grouped: dict[str, list[Change]] = {}
     for commit in commits:
         if commit.prefix not in grouped:
             grouped[commit.prefix] = []
         grouped[commit.prefix].append(
-            {"msg": commit.message, "author": commit.author_login}
+            commit
         )
 
     # Order prefixes
@@ -246,18 +246,18 @@ def get_formatted_changes(git_tag: str) -> tuple[str, str]:
         output += tmp
         output_with_user += tmp
         for commit in grouped[prefix]:
-            output += f"- {commit['msg']}\n"
-            output_with_user += f"- {commit['msg']} by @{commit['author']}\n"
+            output += f"- {commit.message}\n"
+            output_with_user += f"- {commit.message} by @{commit.author_login}\n"
         del grouped[prefix]
 
     if grouped:
         output += "\n### Other\n"
         output_with_user += "\n### Other\n"
         for prefix, commits in grouped.items():
             for commit in commits:
-                output += f"- {prefix}: {commit['msg']}\n"
+                output += f"- {prefix}: {commit.message}\n"
                 output_with_user += (
-                    f"- {prefix}: {commit['msg']} by @{commit['author']}\n"
+                    f"- {prefix}: {commit.message} by @{commit.author_login}\n"
                 )
 
     return output, output_with_user

pypdf/generic/_image_xobject.py

@@ -181,7 +181,7 @@ def _handle_flate(
     size: tuple[int, int],
     data: bytes,
     mode: mode_str_type,
-    color_space: str,
+    color_space: Union[str, ArrayObject],
     colors: int,
     obj_as_text: str,
 ) -> tuple[Image.Image, str, str, bool]:

pyproject.toml

@@ -248,4 +248,17 @@ disallow_incomplete_defs = true
 warn_redundant_casts = true
 warn_unused_ignores = true
 warn_unused_configs = true
-exclude = ['venv', '.venv', 'tests', 'make_release.py']
+exclude = ['venv', '.venv']
+
+
+[[tool.mypy.overrides]]
+module = "tests.*"
+ignore_errors = true
+
+[[tool.mypy.overrides]]
+module = "tests.generic.*"
+ignore_errors = false
+
+[[tool.mypy.overrides]]
+module = "tests.scripts.*"
+ignore_errors = false

requirements/ci-3.11.txt

@@ -30,6 +30,8 @@ mypy-extensions==1.0.0
     # via mypy
 packaging==24.1
     # via pytest
+pathspec==1.0.4
+    # via mypy
 pillow==12.1.1
     # via
     #   -r requirements/ci.in
@@ -69,6 +71,8 @@ tomli==2.0.2
     #   pytest
 typeguard==4.3.0
     # via -r requirements/ci.in
+types-pyyaml==6.0.12.20250915
+    # via -r requirements/ci.in
 typing-extensions==4.12.2
     # via
     #   mypy

requirements/ci.in

@@ -12,3 +12,4 @@ pytest-cov
 # ruff  # only take this for 3.11
 typeguard
 pyyaml
+types-pyyaml

requirements/ci.txt

@@ -63,6 +63,8 @@ tomli==2.0.2
     #   pytest
 typeguard==4.3.0
     # via -r requirements/ci.in
+types-pyyaml==6.0.12.20250915
+    # via -r requirements/ci.in
 typing-extensions==4.13.2
     # via
     #   mypy

tests/__init__.py

@@ -4,7 +4,8 @@
 import sys
 import urllib.request
 from pathlib import Path
-from typing import Optional
+from types import TracebackType
+from typing import Callable, Optional, cast
 from urllib.error import HTTPError
 
 if sys.version_info >= (3, 11):
@@ -21,7 +22,10 @@
 
 
 def _get_data_from_url(url: str) -> bytes:
-    ssl._create_default_https_context = ssl._create_unverified_context
+    ssl._create_default_https_context = cast(
+        Callable[..., ssl.SSLContext],
+        ssl._create_unverified_context,
+    )
     attempts = 0
     while attempts < 3:
         try:
@@ -99,7 +103,7 @@ def normalize_warnings(caplog_text: str) -> list[str]:
     return [_strip_position(line) for line in caplog_text.strip().split("\n")]
 
 
-def is_sublist(child_list, parent_list):
+def is_sublist(child_list: list[int], parent_list: list[int]) -> bool:
     """
     Check if child_list is a sublist of parent_list, with respect to
     * elements order
@@ -121,7 +125,7 @@ def read_yaml_to_list_of_dicts(yaml_file: Path) -> list[dict[str, str]]:
         return yaml.safe_load(yaml_input)
 
 
-def download_test_pdfs():
+def download_test_pdfs() -> None:
     """
     Run this before the tests are executed to ensure you have everything locally.
 
@@ -146,11 +150,17 @@ def __init__(self) -> None:
     def __enter__(self) -> Self:
         # Allow loading incomplete images.
         from PIL import ImageFile  # noqa: PLC0415
+
         self._saved_load_truncated_images = ImageFile.LOAD_TRUNCATED_IMAGES
         ImageFile.LOAD_TRUNCATED_IMAGES = True
         return self
 
-    def __exit__(self, type_, value, traceback) -> Optional[bool]:
+    def __exit__(
+        self,
+        type_: Optional[type[BaseException]],
+        value: Optional[BaseException],
+        traceback: Optional[TracebackType]
+    ) -> Optional[bool]:
         from PIL import ImageFile  # noqa: PLC0415
         ImageFile.LOAD_TRUNCATED_IMAGES = self._saved_load_truncated_images
         if type_:

tests/generic/test_base.py

@@ -22,7 +22,7 @@ def test_text_string_object__looks_like_bom(source: bytes, expected: str) -> Non
 
 
 @pytest.mark.enable_socket
-def test_text_string_object__wrongly_detected_bom():
+def test_text_string_object__wrongly_detected_bom() -> None:
     url = "https://github.com/user-attachments/files/24401507/minimal.pdf"
     name = "issue3587.pdf"
     reader = PdfReader(BytesIO(get_data_from_url(url, name=name)))

tests/generic/test_data_structures.py

@@ -25,10 +25,10 @@
 try:
     import resource
 except ImportError:
-    resource = None
+    resource = None  # type: ignore[assignment]
 
 
-def test_dictionary_object__get_next_object_position():
+def test_dictionary_object__get_next_object_position() -> None:
     reader = PdfReader(RESOURCE_ROOT / "crazyones.pdf")
 
     # reader.xref = {0: {7: 15, 9: 10245, 12: 939, 14: 2999, 16: 4982, 18: 9949, 22: 11160}}
@@ -49,12 +49,13 @@ def test_dictionary_object__get_next_object_position():
     ) == 15
 
 
-def test_tree_object__cyclic_reference(caplog):
+def test_tree_object__cyclic_reference(caplog: pytest.LogCaptureFixture) -> None:
     writer = PdfWriter()
-    child1 = writer._add_object(DictionaryObject())
+    child1_object = DictionaryObject()
+    child1 = writer._add_object(child1_object)
     child2 = writer._add_object(DictionaryObject({NameObject("/Next"): child1}))
     child3 = writer._add_object(DictionaryObject({NameObject("/Next"): child2}))
-    child1.get_object()[NameObject("/Next")] = child3
+    child1_object[NameObject("/Next")] = child3
     tree = TreeObject()
     tree[NameObject("/First")] = child2
     tree[NameObject("/Last")] = writer._add_object(DictionaryObject())
@@ -64,7 +65,7 @@ def test_tree_object__cyclic_reference(caplog):
 
 
 @pytest.mark.enable_socket
-def test_array_object__clone_same_object_multiple_times(caplog):
+def test_array_object__clone_same_object_multiple_times(caplog: pytest.LogCaptureFixture) -> None:
     url = "https://github.com/user-attachments/files/25412858/Draft_OSMF_financial_statement_2013.pdf"
     name = "issue2991.pdf"
     reader = PdfReader(BytesIO(get_data_from_url(url=url, name=name)))
@@ -76,7 +77,7 @@ def test_array_object__clone_same_object_multiple_times(caplog):
     assert caplog.messages == []
 
 
-def test_array_object__clone_same_stream_multiple_times():
+def test_array_object__clone_same_stream_multiple_times() -> None:
     writer = PdfWriter()
 
     # Unique streams.
@@ -87,7 +88,8 @@ def test_array_object__clone_same_stream_multiple_times():
 
     # Shared streams.
     shared_streams = [StreamObject() for _ in range(3)]
-    [shared_stream.set_data(f"Shared stream {index}".encode()) for index, shared_stream in enumerate(shared_streams)]
+    for index, shared_stream in enumerate(shared_streams):
+        shared_stream.set_data(f"Shared stream {index}".encode())
 
     # Add to writer.
     writer._add_object(stream1)
@@ -121,7 +123,7 @@ def test_array_object__clone_same_stream_multiple_times():
 
 
 @pytest.mark.enable_socket
-def test_dictionary_object__read_from_stream__limit():
+def test_dictionary_object__read_from_stream__limit() -> None:
     name = "read_from_stream__length_2gb.pdf"
     url = "https://github.com/user-attachments/files/25842437/read_from_stream__length_2gb.pdf"
 
@@ -164,7 +166,7 @@ def limit_virtual_memory() -> None:
 @pytest.mark.enable_socket
 @pytest.mark.skipif(condition=resource is None, reason="Does not have 'resource' module.")
 @pytest.mark.skipif(sys.platform == "darwin", reason="RLIMIT_AS is unreliable.")
-def test_dictionary_object__read_from_stream__no_limit(tmp_path):
+def test_dictionary_object__read_from_stream__no_limit(tmp_path: Path) -> None:
     pdf_path_str, env, limit_virtual_memory = _prepare_test_dictionary_object__read_from_stream__no_limit(tmp_path)
 
     source_file = tmp_path / "script.py"
@@ -196,7 +198,7 @@ def test_dictionary_object__read_from_stream__no_limit(tmp_path):
 @pytest.mark.enable_socket
 @pytest.mark.skipif(condition=resource is None, reason="Does not have 'resource' module.")
 @pytest.mark.skipif(sys.platform == "darwin", reason="RLIMIT_AS is unreliable.")
-def test_dictionary_object__read_from_stream__no_limit__path(tmp_path):
+def test_dictionary_object__read_from_stream__no_limit__path(tmp_path: Path) -> None:
     pdf_path_str, env, limit_virtual_memory = _prepare_test_dictionary_object__read_from_stream__no_limit(tmp_path)
 
     source_file = tmp_path / "script.py"
@@ -230,7 +232,8 @@ def _get_array_based_buffer(stream_count: int, chunk_bytes: int) -> BytesIO:
 
     streams = [ContentStream(stream=None, pdf=writer) for _ in range(stream_count)]
     chunk = b"q\n" + (b"A" * chunk_bytes) + b"\nQ\n"
-    [stream.set_data(chunk) for stream in streams]
+    for stream in streams:
+        stream.set_data(chunk)
     contents = ArrayObject([writer._add_object(stream) for stream in streams])
     page[NameObject("/Contents")] = contents
 
@@ -241,13 +244,13 @@ def _get_array_based_buffer(stream_count: int, chunk_bytes: int) -> BytesIO:
 
 
 @pytest.mark.timeout(10)
-def test_content_stream__array_based__performance():
+def test_content_stream__array_based__performance() -> None:
     buffer = _get_array_based_buffer(stream_count=10_000, chunk_bytes=7000)
     reader = PdfReader(buffer)
     _ = reader.pages[0].get_contents()
 
 
-def test_content_stream__array_based__length():
+def test_content_stream__array_based__length() -> None:
     buffer = _get_array_based_buffer(stream_count=11_000, chunk_bytes=1)
     reader = PdfReader(buffer)
     with pytest.raises(
@@ -257,7 +260,7 @@ def test_content_stream__array_based__length():
 
 
 @pytest.mark.timeout(10)
-def test_content_stream__array_based__output_length():
+def test_content_stream__array_based__output_length() -> None:
     buffer = _get_array_based_buffer(stream_count=10_000, chunk_bytes=8192)
     reader = PdfReader(buffer)
     with pytest.raises(
@@ -268,7 +271,7 @@ def test_content_stream__array_based__output_length():
 
 
 @pytest.mark.timeout(5)
-def test_dictionary_object__read_from_stream__infinite_loop(caplog):
+def test_dictionary_object__read_from_stream__infinite_loop(caplog: pytest.LogCaptureFixture) -> None:
     data = b"""1 0 obj
 <<
 <</Length 1 0 R>>