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/codegen/python/package_mapper.py

# Copyright 2024 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from __future__ import annotations

import logging
from collections import defaultdict
from pathlib import PurePath
from typing import DefaultDict

from pants.backend.openapi.codegen.python.generate import (
    GeneratePythonFromOpenAPIRequest,
    generate_python_from_openapi,
)
from pants.backend.openapi.target_types import AllOpenApiDocumentTargets, OpenApiDocumentField
from pants.backend.python.dependency_inference.module_mapper import (
    FirstPartyPythonMappingImpl,
    FirstPartyPythonMappingImplMarker,
    ModuleProvider,
    ModuleProviderType,
    ResolveName,
    module_from_stripped_path,
)
from pants.backend.python.subsystems.setup import PythonSetup
from pants.backend.python.target_types import PythonResolveField
from pants.core.util_rules.stripped_source_files import StrippedFileNameRequest, strip_file_name
from pants.engine.internals.graph import hydrate_sources
from pants.engine.internals.selectors import concurrently
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.target import HydrateSourcesRequest
from pants.engine.unions import UnionRule

logger = logging.getLogger(__name__)


class PythonOpenApiMappingMarker(FirstPartyPythonMappingImplMarker):
    pass


@rule
async def map_openapi_documents_to_python_modules(
    all_openapi_document_targets: AllOpenApiDocumentTargets,
    python_setup: PythonSetup,
    _: PythonOpenApiMappingMarker,
) -> FirstPartyPythonMappingImpl:
    hydrated_sources = await concurrently(
        hydrate_sources(HydrateSourcesRequest(target[OpenApiDocumentField]), **implicitly())
        for target in all_openapi_document_targets
    )
    generated_sources = await concurrently(
        generate_python_from_openapi(GeneratePythonFromOpenAPIRequest(sources.snapshot, target))
        for (target, sources) in zip(all_openapi_document_targets, hydrated_sources)
    )
    stripped_file_per_target_sources = await concurrently(
        concurrently(
            strip_file_name(StrippedFileNameRequest(file)) for file in sources.snapshot.files
        )
        for sources in generated_sources
    )

    resolves_to_modules_to_providers: DefaultDict[
        ResolveName, DefaultDict[str, list[ModuleProvider]]
    ] = defaultdict(lambda: defaultdict(list))

    for target, files in zip(all_openapi_document_targets, stripped_file_per_target_sources):
        resolve_name = target[PythonResolveField].normalized_value(python_setup)
        provider = ModuleProvider(target.address, ModuleProviderType.IMPL)
        for stripped_file in files:
            stripped_f = PurePath(stripped_file.value)
            module = module_from_stripped_path(stripped_f)
            resolves_to_modules_to_providers[resolve_name][module].append(provider)

    return FirstPartyPythonMappingImpl.create(resolves_to_modules_to_providers)


def rules():
    return [
        *collect_rules(),
        UnionRule(FirstPartyPythonMappingImplMarker, PythonOpenApiMappingMarker),
    ]

1	# Copyright 2024 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	from __future__ import annotations	×
4
5	import logging	×
6	from collections import defaultdict	×
7	from pathlib import PurePath	×
8	from typing import DefaultDict	×
9
10	from pants.backend.openapi.codegen.python.generate import (	×
11	GeneratePythonFromOpenAPIRequest,
12	generate_python_from_openapi,
13	)
14	from pants.backend.openapi.target_types import AllOpenApiDocumentTargets, OpenApiDocumentField	×
15	from pants.backend.python.dependency_inference.module_mapper import (	×
16	FirstPartyPythonMappingImpl,
17	FirstPartyPythonMappingImplMarker,
18	ModuleProvider,
19	ModuleProviderType,
20	ResolveName,
21	module_from_stripped_path,
22	)
23	from pants.backend.python.subsystems.setup import PythonSetup	×
24	from pants.backend.python.target_types import PythonResolveField	×
25	from pants.core.util_rules.stripped_source_files import StrippedFileNameRequest, strip_file_name	×
26	from pants.engine.internals.graph import hydrate_sources	×
27	from pants.engine.internals.selectors import concurrently	×
28	from pants.engine.rules import collect_rules, implicitly, rule	×
29	from pants.engine.target import HydrateSourcesRequest	×
30	from pants.engine.unions import UnionRule	×
31
32	logger = logging.getLogger(__name__)	×
33
34
35	class PythonOpenApiMappingMarker(FirstPartyPythonMappingImplMarker):	×
36	pass	×
37
38
39	@rule	×
40	async def map_openapi_documents_to_python_modules(	×
41	all_openapi_document_targets: AllOpenApiDocumentTargets,
42	python_setup: PythonSetup,
43	_: PythonOpenApiMappingMarker,
44	) -> FirstPartyPythonMappingImpl:
45	hydrated_sources = await concurrently(	×
46	hydrate_sources(HydrateSourcesRequest(target[OpenApiDocumentField]), **implicitly())
47	for target in all_openapi_document_targets
48	)
49	generated_sources = await concurrently(	×
50	generate_python_from_openapi(GeneratePythonFromOpenAPIRequest(sources.snapshot, target))
51	for (target, sources) in zip(all_openapi_document_targets, hydrated_sources)
52	)
53	stripped_file_per_target_sources = await concurrently(	×
54	concurrently(
55	strip_file_name(StrippedFileNameRequest(file)) for file in sources.snapshot.files
56	)
57	for sources in generated_sources
58	)
59
60	resolves_to_modules_to_providers: DefaultDict[	×
61	ResolveName, DefaultDict[str, list[ModuleProvider]]
62	] = defaultdict(lambda: defaultdict(list))
63
64	for target, files in zip(all_openapi_document_targets, stripped_file_per_target_sources):	×
65	resolve_name = target[PythonResolveField].normalized_value(python_setup)	×
66	provider = ModuleProvider(target.address, ModuleProviderType.IMPL)	×
67	for stripped_file in files:	×
68	stripped_f = PurePath(stripped_file.value)	×
69	module = module_from_stripped_path(stripped_f)	×
70	resolves_to_modules_to_providers[resolve_name][module].append(provider)	×
71
72	return FirstPartyPythonMappingImpl.create(resolves_to_modules_to_providers)	×
73
74
75	def rules():	×
76	return [	×
77	*collect_rules(),
78	UnionRule(FirstPartyPythonMappingImplMarker, PythonOpenApiMappingMarker),
79	]

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