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/core/goals/generate_snapshots.py

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

from __future__ import annotations

import logging
from abc import ABCMeta
from dataclasses import dataclass

from pants.core.environments.rules import EnvironmentNameRequest, resolve_environment_name
from pants.engine.environment import EnvironmentName
from pants.engine.fs import MergeDigests, Snapshot, Workspace
from pants.engine.goal import Goal, GoalSubsystem
from pants.engine.internals.specs_rules import find_valid_field_sets_for_target_roots
from pants.engine.intrinsics import digest_to_snapshot
from pants.engine.rules import collect_rules, concurrently, goal_rule, implicitly, rule
from pants.engine.target import FieldSet, NoApplicableTargetsBehavior, TargetRootsToFieldSetsRequest
from pants.engine.unions import UnionMembership, union

logger = logging.getLogger(__name__)


@union(in_scope_types=[EnvironmentName])
class GenerateSnapshotsFieldSet(FieldSet, metaclass=ABCMeta):
    """The fields necessary to generate snapshots from a target."""


@dataclass(frozen=True)
class GenerateSnapshotsResult:
    snapshot: Snapshot


@rule(polymorphic=True)
async def generate_snapshots(
    field_set: GenerateSnapshotsFieldSet, environment_name: EnvironmentName
) -> GenerateSnapshotsResult:
    raise NotImplementedError()


@dataclass(frozen=True)
class EnvironmentAwareGenerateSnapshotsRequest:
    """Request class to request a `GenerateSnapshotsResult` in an environment-aware fashion."""

    field_set: GenerateSnapshotsFieldSet


@rule
async def environment_aware_generate_snapshots(
    request: EnvironmentAwareGenerateSnapshotsRequest,
) -> GenerateSnapshotsResult:
    environment_name = await resolve_environment_name(
        EnvironmentNameRequest.from_field_set(request.field_set),
        **implicitly(),
    )
    result = await generate_snapshots(
        **implicitly(
            {request.field_set: GenerateSnapshotsFieldSet, environment_name: EnvironmentName}
        )
    )
    return result


class GenerateSnapshotsSubsystem(GoalSubsystem):
    name = "generate-snapshots"
    help = "Generate test snapshots."

    @classmethod
    def activated(cls, union_membership: UnionMembership) -> bool:
        return GenerateSnapshotsFieldSet in union_membership


class GenerateSnapshots(Goal):
    subsystem_cls = GenerateSnapshotsSubsystem
    environment_behavior = Goal.EnvironmentBehavior.USES_ENVIRONMENTS


@goal_rule
async def generate_snapshots_goal(workspace: Workspace) -> GenerateSnapshots:
    target_roots_to_field_sets = await find_valid_field_sets_for_target_roots(
        TargetRootsToFieldSetsRequest(
            GenerateSnapshotsFieldSet,
            goal_description=f"the `{GenerateSnapshotsSubsystem.name}` goal",
            no_applicable_targets_behavior=NoApplicableTargetsBehavior.error,
        ),
        **implicitly(),
    )

    if not target_roots_to_field_sets.field_sets:
        return GenerateSnapshots(exit_code=0)

    snapshot_results = await concurrently(
        environment_aware_generate_snapshots(EnvironmentAwareGenerateSnapshotsRequest(field_set))
        for field_set in target_roots_to_field_sets.field_sets
    )

    all_snapshots = await digest_to_snapshot(
        **implicitly(MergeDigests([result.snapshot.digest for result in snapshot_results]))
    )
    workspace.write_digest(all_snapshots.digest)
    for file in all_snapshots.files:
        logger.info(f"Generated {file}")
    return GenerateSnapshots(exit_code=0)


def rules():
    return collect_rules()

1	# Copyright 2023 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 logging	×
7	from abc import ABCMeta	×
8	from dataclasses import dataclass	×
9
10	from pants.core.environments.rules import EnvironmentNameRequest, resolve_environment_name	×
11	from pants.engine.environment import EnvironmentName	×
12	from pants.engine.fs import MergeDigests, Snapshot, Workspace	×
13	from pants.engine.goal import Goal, GoalSubsystem	×
14	from pants.engine.internals.specs_rules import find_valid_field_sets_for_target_roots	×
15	from pants.engine.intrinsics import digest_to_snapshot	×
16	from pants.engine.rules import collect_rules, concurrently, goal_rule, implicitly, rule	×
17	from pants.engine.target import FieldSet, NoApplicableTargetsBehavior, TargetRootsToFieldSetsRequest	×
18	from pants.engine.unions import UnionMembership, union	×
19
20	logger = logging.getLogger(__name__)	×
21
22
23	@union(in_scope_types=[EnvironmentName])	×
24	class GenerateSnapshotsFieldSet(FieldSet, metaclass=ABCMeta):	×
25	"""The fields necessary to generate snapshots from a target."""
26
27
28	@dataclass(frozen=True)	×
29	class GenerateSnapshotsResult:	×
30	snapshot: Snapshot	×
31
32
33	@rule(polymorphic=True)	×
34	async def generate_snapshots(	×
35	field_set: GenerateSnapshotsFieldSet, environment_name: EnvironmentName
36	) -> GenerateSnapshotsResult:
37	raise NotImplementedError()	×
38
39
40	@dataclass(frozen=True)	×
41	class EnvironmentAwareGenerateSnapshotsRequest:	×
42	"""Request class to request a `GenerateSnapshotsResult` in an environment-aware fashion."""
43
44	field_set: GenerateSnapshotsFieldSet	×
45
46
47	@rule	×
48	async def environment_aware_generate_snapshots(	×
49	request: EnvironmentAwareGenerateSnapshotsRequest,
50	) -> GenerateSnapshotsResult:
51	environment_name = await resolve_environment_name(	×
52	EnvironmentNameRequest.from_field_set(request.field_set),
53	**implicitly(),
54	)
55	result = await generate_snapshots(	×
56	**implicitly(
57	{request.field_set: GenerateSnapshotsFieldSet, environment_name: EnvironmentName}
58	)
59	)
60	return result	×
61
62
63	class GenerateSnapshotsSubsystem(GoalSubsystem):	×
64	name = "generate-snapshots"	×
65	help = "Generate test snapshots."	×
66
67	@classmethod	×
68	def activated(cls, union_membership: UnionMembership) -> bool:	×
69	return GenerateSnapshotsFieldSet in union_membership	×
70
71
72	class GenerateSnapshots(Goal):	×
73	subsystem_cls = GenerateSnapshotsSubsystem	×
74	environment_behavior = Goal.EnvironmentBehavior.USES_ENVIRONMENTS	×
75
76
77	@goal_rule	×
78	async def generate_snapshots_goal(workspace: Workspace) -> GenerateSnapshots:	×
79	target_roots_to_field_sets = await find_valid_field_sets_for_target_roots(	×
80	TargetRootsToFieldSetsRequest(
81	GenerateSnapshotsFieldSet,
82	goal_description=f"the `{GenerateSnapshotsSubsystem.name}` goal",
83	no_applicable_targets_behavior=NoApplicableTargetsBehavior.error,
84	),
85	**implicitly(),
86	)
87
88	if not target_roots_to_field_sets.field_sets:	×
89	return GenerateSnapshots(exit_code=0)	×
90
91	snapshot_results = await concurrently(	×
92	environment_aware_generate_snapshots(EnvironmentAwareGenerateSnapshotsRequest(field_set))
93	for field_set in target_roots_to_field_sets.field_sets
94	)
95
96	all_snapshots = await digest_to_snapshot(	×
97	**implicitly(MergeDigests([result.snapshot.digest for result in snapshot_results]))
98	)
99	workspace.write_digest(all_snapshots.digest)	×
100	for file in all_snapshots.files:	×
101	logger.info(f"Generated {file}")	×
102	return GenerateSnapshots(exit_code=0)	×
103
104
105	def rules():	×
106	return collect_rules()	×

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