18252174847

Committed 05 Oct 2025 01:36AM UTC coverage: 43.382% (-36.9%) from 80.261%

Build # 18252174847

Build Type

push

github

Committed by

web-flow

Commit Message

run tests on mac arm (#22717)

Just doing the minimal to pull forward the x86_64 pattern.

ref #20993

Run Details

25776 of 59416 relevant lines covered (43.38%)

1.3 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

63.41

/src/python/pants/engine/download_file.py

# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from dataclasses import dataclass
from fnmatch import fnmatch
from typing import ClassVar
from urllib.parse import urlparse

from pants.engine.fs import Digest, DownloadFile, NativeDownloadFile
from pants.engine.internals.native_engine import FileDigest
from pants.engine.intrinsics import download_file as native_download_file
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.unions import UnionMembership, union
from pants.option.global_options import GlobalOptions
from pants.util.strutil import bullet_list, softwrap


@union
@dataclass(frozen=True)
class URLDownloadHandler:
    """Union base for custom URL handler.

    To register a custom URL handler:
    - Subclass this class and declare one or both of the ClassVars.
    - Declare a rule that takes in your class type and returns a `Digest`.
    - Register your union member in your `rules()`: `UnionRule(URLDownloadHandler, YourClass)`.

    Example:

        class S3DownloadHandler(URLDownloadHandler):
            match_scheme = "s3"

        @rule
        async def download_s3_file(request: S3DownloadHandler) -> Digest:
            # Lookup auth tokens, etc...
            # Ideally, download the file using `NativeDownloadFile()`
            return digest

        def rules():
            return [
                *collect_rules(),
                UnionRule(URLDownloadHandler, S3DownloadHandler),
            ]
    """

    match_scheme: ClassVar[str | None] = None
    """The scheme to match (e.g. 'ftp' or 's3') or `None` to match all schemes.

    The scheme is matched using `fnmatch`, see https://docs.python.org/3/library/fnmatch.html for more
    information.
    """

    match_authority: ClassVar[str | None] = None
    """The authority to match (e.g. 'pantsbuild.org' or 's3.amazonaws.com') or `None` to match all authorities.

    The authority is matched using `fnmatch`, see https://docs.python.org/3/library/fnmatch.html for more
    information.

    Note that the authority matches userinfo (e.g. 'me@pantsbuild.org' or 'me:password@pantsbuild.org')
    as well as port (e.g. 'pantsbuild.org:80').
    """

    url: str
    expected_digest: FileDigest


@rule(polymorphic=True)
async def download_file_using_handler(handler: URLDownloadHandler) -> Digest:
    raise NotImplementedError()


@rule
async def download_file(
    request: DownloadFile,
    union_membership: UnionMembership,
    global_options: GlobalOptions,
) -> Digest:
    parsed_url = urlparse(request.url)

    handlers = union_membership.get(URLDownloadHandler)
    matched_handlers = []
    for handler in handlers:
        matches_scheme = handler.match_scheme is None or fnmatch(
            parsed_url.scheme, handler.match_scheme
        )
        matches_authority = handler.match_authority is None or fnmatch(
            parsed_url.netloc, handler.match_authority
        )
        if matches_scheme and matches_authority:
            matched_handlers.append(handler)

    if len(matched_handlers) > 1:
        raise Exception(
            softwrap(
                f"""
                Too many registered URL handlers match the URL '{request.url}'.

                Matched handlers:
                {bullet_list(map(str, handlers))}
                """
            )
        )
    if len(matched_handlers) == 1:
        handler = matched_handlers[0]
        return await download_file_using_handler(
            **implicitly({handler(request.url, request.expected_digest): URLDownloadHandler})
        )

    return await native_download_file(
        NativeDownloadFile(
            request.url,
            request.expected_digest,
            retry_delay_duration=global_options.file_downloads_retry_delay,
            max_attempts=global_options.file_downloads_max_attempts,
        ),
    )


def rules():
    return collect_rules()

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from dataclasses import dataclass	3✔
5	from fnmatch import fnmatch	3✔
6	from typing import ClassVar	3✔
7	from urllib.parse import urlparse	3✔
8
9	from pants.engine.fs import Digest, DownloadFile, NativeDownloadFile	3✔
10	from pants.engine.internals.native_engine import FileDigest	3✔
11	from pants.engine.intrinsics import download_file as native_download_file	3✔
12	from pants.engine.rules import collect_rules, implicitly, rule	3✔
13	from pants.engine.unions import UnionMembership, union	3✔
14	from pants.option.global_options import GlobalOptions	3✔
15	from pants.util.strutil import bullet_list, softwrap	3✔
16
17
18	@union	3✔
19	@dataclass(frozen=True)	3✔
20	class URLDownloadHandler:	3✔
21	"""Union base for custom URL handler.
22
23	To register a custom URL handler:
24	- Subclass this class and declare one or both of the ClassVars.
25	- Declare a rule that takes in your class type and returns a `Digest`.
26	- Register your union member in your `rules()`: `UnionRule(URLDownloadHandler, YourClass)`.
27
28	Example:
29
30	class S3DownloadHandler(URLDownloadHandler):
31	match_scheme = "s3"
32
33	@rule
34	async def download_s3_file(request: S3DownloadHandler) -> Digest:
35	# Lookup auth tokens, etc...
36	# Ideally, download the file using `NativeDownloadFile()`
37	return digest
38
39	def rules():
40	return [
41	*collect_rules(),
42	UnionRule(URLDownloadHandler, S3DownloadHandler),
43	]
44	"""
45
46	match_scheme: ClassVar[str \| None] = None	3✔
47	"""The scheme to match (e.g. 'ftp' or 's3') or `None` to match all schemes.	3✔
48
49	The scheme is matched using `fnmatch`, see https://docs.python.org/3/library/fnmatch.html for more
50	information.
51	"""
52
53	match_authority: ClassVar[str \| None] = None	3✔
54	"""The authority to match (e.g. 'pantsbuild.org' or 's3.amazonaws.com') or `None` to match all authorities.	3✔
55
56	The authority is matched using `fnmatch`, see https://docs.python.org/3/library/fnmatch.html for more
57	information.
58
59	Note that the authority matches userinfo (e.g. 'me@pantsbuild.org' or 'me:password@pantsbuild.org')
60	as well as port (e.g. 'pantsbuild.org:80').
61	"""
62
63	url: str	3✔
64	expected_digest: FileDigest	3✔
65
66
67	@rule(polymorphic=True)	3✔
68	async def download_file_using_handler(handler: URLDownloadHandler) -> Digest:	3✔
69	raise NotImplementedError()	×
70
71
72	@rule	3✔
73	async def download_file(	3✔
74	request: DownloadFile,
75	union_membership: UnionMembership,
76	global_options: GlobalOptions,
77	) -> Digest:
78	parsed_url = urlparse(request.url)	×
79
80	handlers = union_membership.get(URLDownloadHandler)	×
81	matched_handlers = []	×
82	for handler in handlers:	×
83	matches_scheme = handler.match_scheme is None or fnmatch(	×
84	parsed_url.scheme, handler.match_scheme
85	)
86	matches_authority = handler.match_authority is None or fnmatch(	×
87	parsed_url.netloc, handler.match_authority
88	)
89	if matches_scheme and matches_authority:	×
90	matched_handlers.append(handler)	×
91
92	if len(matched_handlers) > 1:	×
93	raise Exception(	×
94	softwrap(
95	f"""
96	Too many registered URL handlers match the URL '{request.url}'.
97
98	Matched handlers:
99	{bullet_list(map(str, handlers))}
100	"""
101	)
102	)
103	if len(matched_handlers) == 1:	×
104	handler = matched_handlers[0]	×
105	return await download_file_using_handler(	×
106	**implicitly({handler(request.url, request.expected_digest): URLDownloadHandler})
107	)
108
109	return await native_download_file(	×
110	NativeDownloadFile(
111	request.url,
112	request.expected_digest,
113	retry_delay_duration=global_options.file_downloads_retry_delay,
114	max_attempts=global_options.file_downloads_max_attempts,
115	),
116	)
117
118
119	def rules():	3✔
120	return collect_rules()	3✔

pantsbuild / pants / 18252174847

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous