• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

pantsbuild / pants / 19015773527

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

Pull #22816

github

web-flow
Merge a12d75757 into 6c024e162
Pull Request #22816: Update Pants internal Python to 3.14

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

0.0
/src/python/pants/backend/tools/preamble/rules.py
1
# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2
# Licensed under the Apache License, Version 2.0 (see LICENSE).
3

UNCOV
4
from __future__ import annotations
×
5

UNCOV
6
import dataclasses
×
UNCOV
7
import datetime
×
UNCOV
8
import re
×
UNCOV
9
import string
×
UNCOV
10
from typing import Any
×
11

UNCOV
12
from pants.backend.tools.preamble.subsystem import PreambleSubsystem
×
UNCOV
13
from pants.core.goals.fmt import FmtFilesRequest, FmtResult, Partitions
×
UNCOV
14
from pants.engine.fs import CreateDigest
×
UNCOV
15
from pants.engine.intrinsics import digest_to_snapshot, get_digest_contents
×
UNCOV
16
from pants.engine.rules import collect_rules, implicitly, rule
×
UNCOV
17
from pants.util.logging import LogLevel
×
UNCOV
18
from pants.util.memo import memoized
×
19

20

UNCOV
21
class PreambleRequest(FmtFilesRequest):
×
UNCOV
22
    tool_subsystem = PreambleSubsystem  # type: ignore[assignment]
×
23

24

UNCOV
25
@memoized
×
UNCOV
26
def _template_checker_regex(template: str) -> re.Pattern:
×
27
    shebang = r"(?P<shebang>#!.*\n)"
×
28
    if re.match(shebang, template):
×
29
        # If the template already contains a shebang, don't attempt to match one.
30
        shebang = r"(?P<shebang>)"
×
31
    maybe_shebang = f"{shebang}?"
×
32

33
    subbed = string.Template(template).safe_substitute(year=r"====YEAR====")
×
34
    raw_regex = re.escape(subbed).replace("====YEAR====", r"\d{4}")
×
35
    maybe_template = f"(?P<template>{raw_regex})?"
×
36

37
    # Wrap in `?s` to enable matching newlines with `.`.
38
    body = "(?s:(?P<body>.*))"
×
39

40
    return re.compile(maybe_shebang + maybe_template + body)
×
41

42

UNCOV
43
@memoized
×
UNCOV
44
def _substituted_template(template: str) -> str:
×
45
    return string.Template(template).safe_substitute(year=datetime.date.today().year)
×
46

47

UNCOV
48
@rule
×
UNCOV
49
async def partition_inputs(
×
50
    request: PreambleRequest.PartitionRequest, preamble_subsystem: PreambleSubsystem
51
) -> Partitions[Any]:
52
    if preamble_subsystem.skip:
×
53
        return Partitions()
×
54

55
    return Partitions.single_partition(
×
56
        sorted(preamble_subsystem.get_template_by_path(request.files))
57
    )
58

59

UNCOV
60
@rule(desc="Add preambles", level=LogLevel.DEBUG)
×
UNCOV
61
async def preamble_fmt(
×
62
    request: PreambleRequest.Batch, preamble_subsystem: PreambleSubsystem
63
) -> FmtResult:
64
    template_by_path = preamble_subsystem.get_template_by_path(request.snapshot.files)
×
65
    digest_contents = await get_digest_contents(request.snapshot.digest)
×
66
    contents_by_path = {file_content.path: file_content for file_content in digest_contents}
×
67

68
    for path, template in template_by_path.items():
×
69
        regex = _template_checker_regex(template)
×
70
        file_content = contents_by_path[path].content
×
71
        matched = regex.fullmatch(file_content.decode("utf-8"))
×
72
        # The regex produced by `_template_checker_regex` is infallible.
73
        assert matched
×
74
        if not matched.group("template"):
×
75
            shebang = matched.group("shebang")
×
76
            body = matched.group("body")
×
77
            new_content = (shebang if shebang else "") + _substituted_template(template) + body
×
78
            contents_by_path[path] = dataclasses.replace(
×
79
                contents_by_path[path], content=new_content.encode("utf-8")
80
            )
81

82
    output_snapshot = await digest_to_snapshot(
×
83
        **implicitly(CreateDigest(contents_by_path.values()))
84
    )
85

86
    return FmtResult(
×
87
        input=request.snapshot,
88
        output=output_snapshot,
89
        stdout="",
90
        stderr="",
91
        tool_name=request.tool_name,
92
    )
93

94

UNCOV
95
def rules():
×
UNCOV
96
    return (
×
97
        *collect_rules(),
98
        *PreambleRequest.rules(),
99
    )
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc