24145945949

Committed 08 Apr 2026 04:14PM UTC coverage: 82.077% (-10.8%) from 92.91%

Build # 24145945949

Build Type

Pull #23233

github

Committed by

web-flow

Commit Message

Merge 089d98e3c into 9036734c9

Pull Request Pull Request #23233: Introduce a LockfileFormat enum.

Coverage Stats

8 of 11 new or added lines in 4 files covered. (72.73%)

7635 existing lines in 306 files now uncovered.

63732 of 77649 relevant lines covered (82.08%)

2.96 hits per line

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

73.08

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

pantsbuild / pants / 24145945949

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