19015773527

Committed 02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%

Build # 19015773527

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a12d75757 into 6c024e162

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

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

26.97

/src/python/pants/option/options_fingerprinter.py

# Copyright 2015 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
import hashlib
import json
import os
from collections import defaultdict
from enum import Enum
from hashlib import sha1
from typing import Any, cast

from pants.base.build_environment import get_buildroot
from pants.option.custom_types import UnsetBool, dict_with_files_option, dir_option, file_option
from pants.option.options import Options
from pants.util.strutil import softwrap


class OptionEncoder(json.JSONEncoder):
    def default(self, o):
        if o is UnsetBool:
            return "_UNSET_BOOL_ENCODING"
        if isinstance(o, Enum):
            return o.value
        if isinstance(o, dict):
            # Sort by key to ensure that we don't invalidate if the insertion order changes.
            return {k: self.default(v) for k, v in sorted(o.items())}
        return super().default(o)


def stable_option_fingerprint(obj):
    json_str = json.dumps(
        obj, ensure_ascii=True, allow_nan=False, sort_keys=True, cls=OptionEncoder
    )
    digest = hashlib.sha1()
    digest.update(json_str.encode("utf8"))
    return digest.hexdigest()


class OptionsFingerprinter:
    """Handles fingerprinting options under a given build_graph.

    :API: public
    """

    @classmethod
    def options_map_for_scope(cls, scope: str, options: Options) -> dict[str, Any]:
        """Given options and a scope, create a JSON-serializable map of all fingerprintable options
        in scope to their values."""
        options_map = {}
        option_items = options.get_fingerprintable_for_scope(scope, daemon_only=False)
        fingerprinter = cls()

        for option_name, option_type, option_value in option_items:
            value = option_value
            if option_type in (dir_option, file_option, dict_with_files_option):
                fingerprint = fingerprinter.fingerprint(option_type, option_value)
                if fingerprint is None:
                    # This isn't necessarily a good value to be using here, but it preserves behavior from
                    # before the commit which added it. I suspect that using the empty string would be
                    # reasonable too, but haven't done any archaeology to check.
                    fingerprint = "None"
                value = f"{option_value} (hash: {fingerprint})"
            options_map[option_name] = value

        serializable_map = cast(
            dict[str, Any], json.loads(json.dumps(options_map, cls=OptionEncoder))
        )

        return serializable_map

    def fingerprint(self, option_type, option_val):
        """Returns a hash of the given option_val based on the option_type.

        :API: public

        Returns None if option_val is None.
        """
        if option_val is None:
            return None

        # Wrapping all other values in a list here allows us to easily handle single-valued and
        # list-valued options uniformly. For non-list-valued options, this will be a singleton list
        # (with the exception of dict, which is not modified). This dict exception works because we do
        # not currently have any "list of dict" type, so there is no ambiguity.
        if not isinstance(option_val, (list, tuple, dict)):
            option_val = [option_val]

        if option_type == dir_option:
            return self._fingerprint_dirs(option_val)
        elif option_type == file_option:
            return self._fingerprint_files(option_val)
        elif option_type == dict_with_files_option:
            return self._fingerprint_dict_with_files(option_val)
        else:
            return self._fingerprint_primitives(option_val)

    def _assert_in_buildroot(self, filepath):
        """Raises an error if the given filepath isn't in the buildroot.

        Returns the normalized, absolute form of the path.
        """
        filepath = os.path.normpath(filepath)
        root = get_buildroot()
        if not os.path.abspath(filepath) == filepath:
            # If not absolute, assume relative to the build root.
            return os.path.join(root, filepath)
        else:
            if ".." in os.path.relpath(filepath, root).split(os.path.sep):
                # The path wasn't in the buildroot. This is an error because it violates pants being
                # hermetic.
                raise ValueError(
                    softwrap(
                        f"""
                        Received a file_option that was not inside the build root:

                            file_option: {filepath}
                            build_root:  {root}
                        """
                    )
                )
            return filepath

    def _fingerprint_dirs(self, dirpaths, topdown=True, onerror=None, followlinks=False):
        """Returns a fingerprint of the given file directories and all their sub contents.

        This assumes that the file directories are of reasonable size to cause memory or performance
        issues.
        """
        # Note that we don't sort the dirpaths, as their order may have meaning.
        filepaths = []
        for dirpath in dirpaths:
            dirs = os.walk(dirpath, topdown=topdown, onerror=onerror, followlinks=followlinks)
            sorted_dirs = sorted(dirs, key=lambda d: d[0])
            filepaths.extend(
                [
                    os.path.join(dirpath, filename)
                    for dirpath, dirnames, filenames in sorted_dirs
                    for filename in sorted(filenames)
                ]
            )
        return self._fingerprint_files(filepaths)

    def _fingerprint_files(self, filepaths):
        """Returns a fingerprint of the given filepaths and their contents.

        This assumes the files are small enough to be read into memory.
        """
        hasher = sha1()
        # Note that we don't sort the filepaths, as their order may have meaning.
        for filepath in filepaths:
            filepath = self._assert_in_buildroot(filepath)
            hasher.update(os.path.relpath(filepath, get_buildroot()).encode())
            with open(filepath, "rb") as f:
                hasher.update(f.read())
        return hasher.hexdigest()

    def _fingerprint_primitives(self, val):
        return stable_option_fingerprint(val)

    @staticmethod
    def _fingerprint_dict_with_files(option_val):
        """Returns a fingerprint of the given dictionary containing file paths.

        Any value which is a file path which exists on disk will be fingerprinted by that file's
        contents rather than by its path.

        This assumes the files are small enough to be read into memory.

        NB: The keys of the dict are assumed to be strings -- if they are not, the dict should be
        converted to encode its keys with `stable_option_fingerprint()`, as is done in the `fingerprint()`
        method.
        """
        final = defaultdict(list)
        for k, v in option_val.items():
            for sub_value in sorted(v.split(",")):
                if os.path.isfile(sub_value):
                    with open(sub_value) as f:
                        final[k].append(f.read())
                else:
                    final[k].append(sub_value)
        fingerprint = stable_option_fingerprint(final)
        return fingerprint

1	# Copyright 2015 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	import hashlib	1✔
4	import json	1✔
5	import os	1✔
6	from collections import defaultdict	1✔
7	from enum import Enum	1✔
8	from hashlib import sha1	1✔
9	from typing import Any, cast	1✔
10
11	from pants.base.build_environment import get_buildroot	1✔
12	from pants.option.custom_types import UnsetBool, dict_with_files_option, dir_option, file_option	1✔
13	from pants.option.options import Options	1✔
14	from pants.util.strutil import softwrap	1✔
15
16
17	class OptionEncoder(json.JSONEncoder):	1✔
18	def default(self, o):	1✔
UNCOV 19	if o is UnsetBool:	×
UNCOV 20	return "_UNSET_BOOL_ENCODING"	×
UNCOV 21	if isinstance(o, Enum):	×
UNCOV 22	return o.value	×
23	if isinstance(o, dict):	×
24	# Sort by key to ensure that we don't invalidate if the insertion order changes.
25	return {k: self.default(v) for k, v in sorted(o.items())}	×
26	return super().default(o)	×
27
28
29	def stable_option_fingerprint(obj):	1✔
UNCOV 30	json_str = json.dumps(	×
31	obj, ensure_ascii=True, allow_nan=False, sort_keys=True, cls=OptionEncoder
32	)
UNCOV 33	digest = hashlib.sha1()	×
UNCOV 34	digest.update(json_str.encode("utf8"))	×
UNCOV 35	return digest.hexdigest()	×
36
37
38	class OptionsFingerprinter:	1✔
39	"""Handles fingerprinting options under a given build_graph.
40
41	:API: public
42	"""
43
44	@classmethod	1✔
45	def options_map_for_scope(cls, scope: str, options: Options) -> dict[str, Any]:	1✔
46	"""Given options and a scope, create a JSON-serializable map of all fingerprintable options
47	in scope to their values."""
UNCOV 48	options_map = {}	×
UNCOV 49	option_items = options.get_fingerprintable_for_scope(scope, daemon_only=False)	×
UNCOV 50	fingerprinter = cls()	×
51
UNCOV 52	for option_name, option_type, option_value in option_items:	×
UNCOV 53	value = option_value	×
UNCOV 54	if option_type in (dir_option, file_option, dict_with_files_option):	×
UNCOV 55	fingerprint = fingerprinter.fingerprint(option_type, option_value)	×
UNCOV 56	if fingerprint is None:	×
57	# This isn't necessarily a good value to be using here, but it preserves behavior from
58	# before the commit which added it. I suspect that using the empty string would be
59	# reasonable too, but haven't done any archaeology to check.
UNCOV 60	fingerprint = "None"	×
UNCOV 61	value = f"{option_value} (hash: {fingerprint})"	×
UNCOV 62	options_map[option_name] = value	×
63
UNCOV 64	serializable_map = cast(	×
65	dict[str, Any], json.loads(json.dumps(options_map, cls=OptionEncoder))
66	)
67
UNCOV 68	return serializable_map	×
69
70	def fingerprint(self, option_type, option_val):	1✔
71	"""Returns a hash of the given option_val based on the option_type.
72
73	:API: public
74
75	Returns None if option_val is None.
76	"""
UNCOV 77	if option_val is None:	×
UNCOV 78	return None	×
79
80	# Wrapping all other values in a list here allows us to easily handle single-valued and
81	# list-valued options uniformly. For non-list-valued options, this will be a singleton list
82	# (with the exception of dict, which is not modified). This dict exception works because we do
83	# not currently have any "list of dict" type, so there is no ambiguity.
UNCOV 84	if not isinstance(option_val, (list, tuple, dict)):	×
UNCOV 85	option_val = [option_val]	×
86
UNCOV 87	if option_type == dir_option:	×
UNCOV 88	return self._fingerprint_dirs(option_val)	×
UNCOV 89	elif option_type == file_option:	×
UNCOV 90	return self._fingerprint_files(option_val)	×
UNCOV 91	elif option_type == dict_with_files_option:	×
UNCOV 92	return self._fingerprint_dict_with_files(option_val)	×
93	else:
UNCOV 94	return self._fingerprint_primitives(option_val)	×
95
96	def _assert_in_buildroot(self, filepath):	1✔
97	"""Raises an error if the given filepath isn't in the buildroot.
98
99	Returns the normalized, absolute form of the path.
100	"""
UNCOV 101	filepath = os.path.normpath(filepath)	×
UNCOV 102	root = get_buildroot()	×
UNCOV 103	if not os.path.abspath(filepath) == filepath:	×
104	# If not absolute, assume relative to the build root.
105	return os.path.join(root, filepath)	×
106	else:
UNCOV 107	if ".." in os.path.relpath(filepath, root).split(os.path.sep):	×
108	# The path wasn't in the buildroot. This is an error because it violates pants being
109	# hermetic.
UNCOV 110	raise ValueError(	×
111	softwrap(
112	f"""
113	Received a file_option that was not inside the build root:
114
115	file_option: {filepath}
116	build_root: {root}
117	"""
118	)
119	)
UNCOV 120	return filepath	×
121
122	def _fingerprint_dirs(self, dirpaths, topdown=True, onerror=None, followlinks=False):	1✔
123	"""Returns a fingerprint of the given file directories and all their sub contents.
124
125	This assumes that the file directories are of reasonable size to cause memory or performance
126	issues.
127	"""
128	# Note that we don't sort the dirpaths, as their order may have meaning.
UNCOV 129	filepaths = []	×
UNCOV 130	for dirpath in dirpaths:	×
UNCOV 131	dirs = os.walk(dirpath, topdown=topdown, onerror=onerror, followlinks=followlinks)	×
UNCOV 132	sorted_dirs = sorted(dirs, key=lambda d: d[0])	×
UNCOV 133	filepaths.extend(	×
134	[
135	os.path.join(dirpath, filename)
136	for dirpath, dirnames, filenames in sorted_dirs
137	for filename in sorted(filenames)
138	]
139	)
UNCOV 140	return self._fingerprint_files(filepaths)	×
141
142	def _fingerprint_files(self, filepaths):	1✔
143	"""Returns a fingerprint of the given filepaths and their contents.
144
145	This assumes the files are small enough to be read into memory.
146	"""
UNCOV 147	hasher = sha1()	×
148	# Note that we don't sort the filepaths, as their order may have meaning.
UNCOV 149	for filepath in filepaths:	×
UNCOV 150	filepath = self._assert_in_buildroot(filepath)	×
UNCOV 151	hasher.update(os.path.relpath(filepath, get_buildroot()).encode())	×
UNCOV 152	with open(filepath, "rb") as f:	×
UNCOV 153	hasher.update(f.read())	×
UNCOV 154	return hasher.hexdigest()	×
155
156	def _fingerprint_primitives(self, val):	1✔
UNCOV 157	return stable_option_fingerprint(val)	×
158
159	@staticmethod	1✔
160	def _fingerprint_dict_with_files(option_val):	1✔
161	"""Returns a fingerprint of the given dictionary containing file paths.
162
163	Any value which is a file path which exists on disk will be fingerprinted by that file's
164	contents rather than by its path.
165
166	This assumes the files are small enough to be read into memory.
167
168	NB: The keys of the dict are assumed to be strings -- if they are not, the dict should be
169	converted to encode its keys with `stable_option_fingerprint()`, as is done in the `fingerprint()`
170	method.
171	"""
UNCOV 172	final = defaultdict(list)	×
UNCOV 173	for k, v in option_val.items():	×
UNCOV 174	for sub_value in sorted(v.split(",")):	×
UNCOV 175	if os.path.isfile(sub_value):	×
UNCOV 176	with open(sub_value) as f:	×
UNCOV 177	final[k].append(f.read())	×
178	else:
179	final[k].append(sub_value)	×
UNCOV 180	fingerprint = stable_option_fingerprint(final)	×
UNCOV 181	return fingerprint	×

pantsbuild / pants / 19015773527

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