22285099215

Committed 22 Feb 2026 08:52PM UTC coverage: 75.854% (-17.1%) from 92.936%

Build # 22285099215

Build Type

Pull #23121

github

Committed by

web-flow

Commit Message

Merge c7299df9c into ba8359840

Pull Request Pull Request #23121: fix issue with optional fields in dependency validator

Run Details

28 of 29 new or added lines in 2 files covered. (96.55%)

11174 existing lines in 400 files now uncovered.

53694 of 70786 relevant lines covered (75.85%)

1.88 hits per line

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

54.67

/src/python/pants/bsp/spec/task.py

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

from dataclasses import dataclass
from enum import Enum
from typing import Any

from pants.bsp.spec.base import StatusCode, TaskId
from pants.bsp.spec.compile import CompileReport, CompileTask
from pants.bsp.spec.notification import BSPNotification

# -----------------------------------------------------------------------------------------------
# Task Notifications
# See https://build-server-protocol.github.io/docs/specification.html#compile-request
# -----------------------------------------------------------------------------------------------


class TaskDataKind(Enum):
    # `data` field must contain a CompileTask object.
    COMPILE_TASK = "compile-task"

    #  `data` field must contain a CompileReport object.
    COMPILE_REPORT = "compile-report"

    # `data` field must contain a TestTask object.
    TEST_TASK = "test-task"

    # `data` field must contain a TestReport object.
    TEST_REPORT = "test-report"

    # `data` field must contain a TestStart object.
    TEST_START = "test-start"

    # `data` field must contain a TestFinish object.
    TEST_FINISH = "test-finish"


@dataclass(frozen=True)
class TaskStartParams(BSPNotification):
    notification_name = "build/taskStart"

    # Unique id of the task with optional reference to parent task id
    task_id: TaskId

    # Timestamp of when the event started in milliseconds since Epoch.
    event_time: int | None = None

    # Message describing the task.
    message: str | None = None

    # Task-specific data.
    # Note: This field is serialized as two fields: `dataKind` (for type name) and `data`.
    data: CompileTask | None = None

    def to_json_dict(self) -> dict[str, Any]:
        result: dict[str, Any] = {"taskId": self.task_id.to_json_dict()}
        if self.event_time is not None:
            result["eventTime"] = self.event_time
        if self.message is not None:
            result["message"] = self.message
        if self.data is not None:
            if isinstance(self.data, CompileTask):
                result["dataKind"] = TaskDataKind.COMPILE_TASK.value
            else:
                raise AssertionError(
                    f"TaskStartParams contained an unexpected instance: {self.data}"
                )
            result["data"] = self.data.to_json_dict()
        return result


@dataclass(frozen=True)
class TaskProgressParams(BSPNotification):
    notification_name = "build/taskProgress"

    # Unique id of the task with optional reference to parent task id
    task_id: TaskId

    # Timestamp of when the progress event was generated in milliseconds since Epoch.
    event_time: int | None = None

    # Message describing the task progress.
    # Information about the state of the task at the time the event is sent.
    message: str | None = None

    # If known, total amount of work units in this task.
    total: int | None = None

    # If known, completed amount of work units in this task.
    progress: int | None = None

    # Name of a work unit. For example, "files" or "tests". May be empty.
    unit: str | None = None

    # TODO: `data` field is not currently represented. Once we know what types will be sent, then it can be bound.

    def to_json_dict(self) -> dict[str, Any]:
        result: dict[str, Any] = {"taskId": self.task_id.to_json_dict()}
        if self.event_time is not None:
            result["eventTime"] = self.event_time
        if self.message is not None:
            result["message"] = self.message
        if self.total is not None:
            result["total"] = self.total
        if self.progress is not None:
            result["progress"] = self.progress
        if self.unit is not None:
            result["unit"] = self.unit
        return result


@dataclass(frozen=True)
class TaskFinishParams(BSPNotification):
    notification_name = "build/taskFinish"

    # Unique id of the task with optional reference to parent task id
    task_id: TaskId

    # Timestamp of the event in milliseconds.
    event_time: int | None = None

    # Message describing the finish event.
    message: str | None = None

    # Task completion status.
    status: StatusCode = StatusCode.OK

    # Task-specific data.
    # Note: This field is serialized as two fields: `dataKind` (for type name) and `data`.
    data: CompileReport | None = None

    def to_json_dict(self) -> dict[str, Any]:
        result: dict[str, Any] = {
            "taskId": self.task_id.to_json_dict(),
            "status": self.status.value,
        }
        if self.event_time is not None:
            result["eventTime"] = self.event_time
        if self.message is not None:
            result["message"] = self.message
        if self.data is not None:
            if isinstance(self.data, CompileReport):
                result["dataKind"] = TaskDataKind.COMPILE_REPORT.value
            else:
                raise AssertionError(
                    f"TaskFinishParams contained an unexpected instance: {self.data}"
                )
            result["data"] = self.data.to_json_dict()
        return result

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	from __future__ import annotations	1✔
4
5	from dataclasses import dataclass	1✔
6	from enum import Enum	1✔
7	from typing import Any	1✔
8
9	from pants.bsp.spec.base import StatusCode, TaskId	1✔
10	from pants.bsp.spec.compile import CompileReport, CompileTask	1✔
11	from pants.bsp.spec.notification import BSPNotification	1✔
12
13	# -----------------------------------------------------------------------------------------------
14	# Task Notifications
15	# See https://build-server-protocol.github.io/docs/specification.html#compile-request
16	# -----------------------------------------------------------------------------------------------
17
18
19	class TaskDataKind(Enum):	1✔
20	# `data` field must contain a CompileTask object.
21	COMPILE_TASK = "compile-task"	1✔
22
23	# `data` field must contain a CompileReport object.
24	COMPILE_REPORT = "compile-report"	1✔
25
26	# `data` field must contain a TestTask object.
27	TEST_TASK = "test-task"	1✔
28
29	# `data` field must contain a TestReport object.
30	TEST_REPORT = "test-report"	1✔
31
32	# `data` field must contain a TestStart object.
33	TEST_START = "test-start"	1✔
34
35	# `data` field must contain a TestFinish object.
36	TEST_FINISH = "test-finish"	1✔
37
38
39	@dataclass(frozen=True)	1✔
40	class TaskStartParams(BSPNotification):	1✔
41	notification_name = "build/taskStart"	1✔
42
43	# Unique id of the task with optional reference to parent task id
44	task_id: TaskId	1✔
45
46	# Timestamp of when the event started in milliseconds since Epoch.
47	event_time: int \| None = None	1✔
48
49	# Message describing the task.
50	message: str \| None = None	1✔
51
52	# Task-specific data.
53	# Note: This field is serialized as two fields: `dataKind` (for type name) and `data`.
54	data: CompileTask \| None = None	1✔
55
56	def to_json_dict(self) -> dict[str, Any]:	1✔
UNCOV 57	result: dict[str, Any] = {"taskId": self.task_id.to_json_dict()}	×
UNCOV 58	if self.event_time is not None:	×
UNCOV 59	result["eventTime"] = self.event_time	×
UNCOV 60	if self.message is not None:	×
UNCOV 61	result["message"] = self.message	×
UNCOV 62	if self.data is not None:	×
UNCOV 63	if isinstance(self.data, CompileTask):	×
UNCOV 64	result["dataKind"] = TaskDataKind.COMPILE_TASK.value	×
65	else:
66	raise AssertionError(	×
67	f"TaskStartParams contained an unexpected instance: {self.data}"
68	)
UNCOV 69	result["data"] = self.data.to_json_dict()	×
UNCOV 70	return result	×
71
72
73	@dataclass(frozen=True)	1✔
74	class TaskProgressParams(BSPNotification):	1✔
75	notification_name = "build/taskProgress"	1✔
76
77	# Unique id of the task with optional reference to parent task id
78	task_id: TaskId	1✔
79
80	# Timestamp of when the progress event was generated in milliseconds since Epoch.
81	event_time: int \| None = None	1✔
82
83	# Message describing the task progress.
84	# Information about the state of the task at the time the event is sent.
85	message: str \| None = None	1✔
86
87	# If known, total amount of work units in this task.
88	total: int \| None = None	1✔
89
90	# If known, completed amount of work units in this task.
91	progress: int \| None = None	1✔
92
93	# Name of a work unit. For example, "files" or "tests". May be empty.
94	unit: str \| None = None	1✔
95
96	# TODO: `data` field is not currently represented. Once we know what types will be sent, then it can be bound.
97
98	def to_json_dict(self) -> dict[str, Any]:	1✔
UNCOV 99	result: dict[str, Any] = {"taskId": self.task_id.to_json_dict()}	×
UNCOV 100	if self.event_time is not None:	×
UNCOV 101	result["eventTime"] = self.event_time	×
UNCOV 102	if self.message is not None:	×
UNCOV 103	result["message"] = self.message	×
UNCOV 104	if self.total is not None:	×
105	result["total"] = self.total	×
UNCOV 106	if self.progress is not None:	×
107	result["progress"] = self.progress	×
UNCOV 108	if self.unit is not None:	×
109	result["unit"] = self.unit	×
UNCOV 110	return result	×
111
112
113	@dataclass(frozen=True)	1✔
114	class TaskFinishParams(BSPNotification):	1✔
115	notification_name = "build/taskFinish"	1✔
116
117	# Unique id of the task with optional reference to parent task id
118	task_id: TaskId	1✔
119
120	# Timestamp of the event in milliseconds.
121	event_time: int \| None = None	1✔
122
123	# Message describing the finish event.
124	message: str \| None = None	1✔
125
126	# Task completion status.
127	status: StatusCode = StatusCode.OK	1✔
128
129	# Task-specific data.
130	# Note: This field is serialized as two fields: `dataKind` (for type name) and `data`.
131	data: CompileReport \| None = None	1✔
132
133	def to_json_dict(self) -> dict[str, Any]:	1✔
UNCOV 134	result: dict[str, Any] = {	×
135	"taskId": self.task_id.to_json_dict(),
136	"status": self.status.value,
137	}
UNCOV 138	if self.event_time is not None:	×
UNCOV 139	result["eventTime"] = self.event_time	×
UNCOV 140	if self.message is not None:	×
UNCOV 141	result["message"] = self.message	×
UNCOV 142	if self.data is not None:	×
UNCOV 143	if isinstance(self.data, CompileReport):	×
UNCOV 144	result["dataKind"] = TaskDataKind.COMPILE_REPORT.value	×
145	else:
146	raise AssertionError(	×
147	f"TaskFinishParams contained an unexpected instance: {self.data}"
148	)
UNCOV 149	result["data"] = self.data.to_json_dict()	×
UNCOV 150	return result	×

pantsbuild / pants / 22285099215

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