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

0.0

/src/python/pants/backend/openapi/goals/tailor.py

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

from __future__ import annotations

import itertools
from collections.abc import Iterable
from dataclasses import dataclass

from pants.backend.openapi.dependency_inference import ParseOpenApiSources, parse_openapi_sources
from pants.backend.openapi.subsystems.openapi import OpenApiSubsystem
from pants.backend.openapi.target_types import (
    OPENAPI_FILE_EXTENSIONS,
    OpenApiDocumentGeneratorTarget,
    OpenApiSourceGeneratorTarget,
)
from pants.core.goals.tailor import (
    AllOwnedSources,
    PutativeTarget,
    PutativeTargets,
    PutativeTargetsRequest,
)
from pants.engine.fs import PathGlobs
from pants.engine.intrinsics import path_globs_to_digest, path_globs_to_paths
from pants.engine.rules import Rule, collect_rules, rule
from pants.engine.unions import UnionRule
from pants.util.dirutil import group_by_dir
from pants.util.logging import LogLevel


@dataclass(frozen=True)
class PutativeOpenApiTargetsRequest(PutativeTargetsRequest):
    pass


@rule(level=LogLevel.DEBUG, desc="Determine candidate OpenAPI targets to create")
async def find_putative_targets(
    req: PutativeOpenApiTargetsRequest,
    all_owned_sources: AllOwnedSources,
    openapi_subsystem: OpenApiSubsystem,
) -> PutativeTargets:
    if not openapi_subsystem.tailor_targets:
        return PutativeTargets()

    all_openapi_files = await path_globs_to_paths(
        req.path_globs(*(f"**/openapi{ext}" for ext in OPENAPI_FILE_EXTENSIONS))
    )
    targets = {*all_openapi_files.files}
    paths = tuple(targets)

    while paths:
        digest = await path_globs_to_digest(PathGlobs(paths))  # noqa: PNT30: this is inherently sequential
        result = await parse_openapi_sources(
            ParseOpenApiSources(
                sources_digest=digest,
                paths=tuple(paths),
            )
        )

        targets.update(result.dependencies.keys())
        paths = tuple(v for v in itertools.chain(*result.dependencies.values()) if v not in targets)

    unowned_openapi_documents = set(all_openapi_files.files) - set(all_owned_sources)
    unowned_openapi_sources = targets - set(all_owned_sources)
    classified_unowned_openapi_files = {
        OpenApiDocumentGeneratorTarget: unowned_openapi_documents,
        OpenApiSourceGeneratorTarget: unowned_openapi_sources,
    }

    putative_targets = []
    for tgt_type, filepaths in classified_unowned_openapi_files.items():
        for dirname, filenames in group_by_dir(filepaths).items():
            name = "openapi" if tgt_type == OpenApiDocumentGeneratorTarget else None
            putative_targets.append(
                PutativeTarget.for_target_type(
                    tgt_type, path=dirname, name=name, triggering_sources=sorted(filenames)
                )
            )

    return PutativeTargets(putative_targets)


def rules() -> Iterable[Rule | UnionRule]:
    return (
        *collect_rules(),
        UnionRule(PutativeTargetsRequest, PutativeOpenApiTargetsRequest),
    )

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	×
5
6	import itertools	×
7	from collections.abc import Iterable	×
8	from dataclasses import dataclass	×
9
10	from pants.backend.openapi.dependency_inference import ParseOpenApiSources, parse_openapi_sources	×
11	from pants.backend.openapi.subsystems.openapi import OpenApiSubsystem	×
12	from pants.backend.openapi.target_types import (	×
13	OPENAPI_FILE_EXTENSIONS,
14	OpenApiDocumentGeneratorTarget,
15	OpenApiSourceGeneratorTarget,
16	)
17	from pants.core.goals.tailor import (	×
18	AllOwnedSources,
19	PutativeTarget,
20	PutativeTargets,
21	PutativeTargetsRequest,
22	)
23	from pants.engine.fs import PathGlobs	×
24	from pants.engine.intrinsics import path_globs_to_digest, path_globs_to_paths	×
25	from pants.engine.rules import Rule, collect_rules, rule	×
26	from pants.engine.unions import UnionRule	×
27	from pants.util.dirutil import group_by_dir	×
28	from pants.util.logging import LogLevel	×
29
30
31	@dataclass(frozen=True)	×
32	class PutativeOpenApiTargetsRequest(PutativeTargetsRequest):	×
33	pass	×
34
35
36	@rule(level=LogLevel.DEBUG, desc="Determine candidate OpenAPI targets to create")	×
37	async def find_putative_targets(	×
38	req: PutativeOpenApiTargetsRequest,
39	all_owned_sources: AllOwnedSources,
40	openapi_subsystem: OpenApiSubsystem,
41	) -> PutativeTargets:
42	if not openapi_subsystem.tailor_targets:	×
43	return PutativeTargets()	×
44
45	all_openapi_files = await path_globs_to_paths(	×
46	req.path_globs((f"*/openapi{ext}" for ext in OPENAPI_FILE_EXTENSIONS))
47	)
48	targets = {*all_openapi_files.files}	×
49	paths = tuple(targets)	×
50
51	while paths:	×
52	digest = await path_globs_to_digest(PathGlobs(paths)) # noqa: PNT30: this is inherently sequential	×
53	result = await parse_openapi_sources(	×
54	ParseOpenApiSources(
55	sources_digest=digest,
56	paths=tuple(paths),
57	)
58	)
59
60	targets.update(result.dependencies.keys())	×
61	paths = tuple(v for v in itertools.chain(*result.dependencies.values()) if v not in targets)	×
62
63	unowned_openapi_documents = set(all_openapi_files.files) - set(all_owned_sources)	×
64	unowned_openapi_sources = targets - set(all_owned_sources)	×
65	classified_unowned_openapi_files = {	×
66	OpenApiDocumentGeneratorTarget: unowned_openapi_documents,
67	OpenApiSourceGeneratorTarget: unowned_openapi_sources,
68	}
69
70	putative_targets = []	×
71	for tgt_type, filepaths in classified_unowned_openapi_files.items():	×
72	for dirname, filenames in group_by_dir(filepaths).items():	×
73	name = "openapi" if tgt_type == OpenApiDocumentGeneratorTarget else None	×
74	putative_targets.append(	×
75	PutativeTarget.for_target_type(
76	tgt_type, path=dirname, name=name, triggering_sources=sorted(filenames)
77	)
78	)
79
80	return PutativeTargets(putative_targets)	×
81
82
83	def rules() -> Iterable[Rule \| UnionRule]:	×
84	return (	×
85	*collect_rules(),
86	UnionRule(PutativeTargetsRequest, PutativeOpenApiTargetsRequest),
87	)

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