diff --git a/examples/task_result.py b/examples/task_result.py new file mode 100644 index 0000000..c24619b --- /dev/null +++ b/examples/task_result.py @@ -0,0 +1,35 @@ +import os + +from pytfe import TFEClient + + +def main(): + token = os.getenv("TFE_TOKEN") + task_result_id = os.getenv("TFE_TASK_RESULT_ID") + + if not token: + print("Set TFE_TOKEN") + return + + if not task_result_id: + print("Set TFE_TASK_RESULT_ID") + return + + client = TFEClient() + + try: + result = client.task_results.read(task_result_id) + + print("=== Task Result ===") + print(f"ID: {result.id}") + print(f"Status: {result.status}") + print(f"Message: {result.message}") + print(f"Task Name: {result.task_name}") + print(f"URL: {result.url}") + + except Exception as e: + print(f"Error: {e}") + + +if __name__ == "__main__": + main() diff --git a/src/pytfe/client.py b/src/pytfe/client.py index 4642d9a..8a55473 100644 --- a/src/pytfe/client.py +++ b/src/pytfe/client.py @@ -39,6 +39,7 @@ from .resources.stack_configuration import StackConfigurations from .resources.state_version_outputs import StateVersionOutputs from .resources.state_versions import StateVersions +from .resources.task_result import TaskResults from .resources.team import Teams from .resources.team_project_access import TeamProjectAccesses from .resources.user import Users @@ -80,6 +81,7 @@ def __init__(self, config: TFEConfig | None = None): self.organizations = Organizations(self._transport) self.organization_memberships = OrganizationMemberships(self._transport) self.users = Users(self._transport) + self.task_results = TaskResults(self._transport) self.organization_tokens = OrganizationTokens(self._transport) self.projects = Projects(self._transport) self.variables = Variables(self._transport) diff --git a/src/pytfe/models/task_result.py b/src/pytfe/models/task_result.py new file mode 100644 index 0000000..516da0a --- /dev/null +++ b/src/pytfe/models/task_result.py @@ -0,0 +1,68 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from __future__ import annotations + +from datetime import datetime +from enum import Enum + +from pydantic import BaseModel, ConfigDict, Field + +# Reuse, do NOT duplicate +from pytfe.models.task_stage import TaskStage + + +class TaskResultStatus(str, Enum): + passed = "passed" + failed = "failed" + pending = "pending" + running = "running" + unreachable = "unreachable" + errored = "errored" + + +class TaskEnforcementLevel(str, Enum): + advisory = "advisory" + mandatory = "mandatory" + + +class TaskResultStatusTimestamps(BaseModel): + model_config = ConfigDict(populate_by_name=True) + + errored_at: datetime | None = Field(None, alias="errored-at") + running_at: datetime | None = Field(None, alias="running-at") + canceled_at: datetime | None = Field(None, alias="canceled-at") + failed_at: datetime | None = Field(None, alias="failed-at") + passed_at: datetime | None = Field(None, alias="passed-at") + + +class TaskResult(BaseModel): + model_config = ConfigDict(populate_by_name=True) + + id: str + + status: TaskResultStatus | None = Field(None, alias="status") + message: str | None = Field(None, alias="message") + + status_timestamps: TaskResultStatusTimestamps | None = Field( + None, alias="status-timestamps" + ) + + url: str | None = Field(None, alias="url") + + created_at: datetime | None = Field(None, alias="created-at") + updated_at: datetime | None = Field(None, alias="updated-at") + + task_id: str | None = Field(None, alias="task-id") + task_name: str | None = Field(None, alias="task-name") + task_url: str | None = Field(None, alias="task-url") + + workspace_task_id: str | None = Field(None, alias="workspace-task-id") + workspace_task_enforcement_level: TaskEnforcementLevel | None = Field( + None, alias="workspace-task-enforcement-level" + ) + + agent_pool_id: str | None = Field(None, alias="agent-pool-id") + + # Relation (matches Go: *TaskStage) + task_stage: TaskStage | None = Field(None, alias="task-stage") diff --git a/src/pytfe/models/task_stage.py b/src/pytfe/models/task_stage.py index 54b9346..bae0766 100644 --- a/src/pytfe/models/task_stage.py +++ b/src/pytfe/models/task_stage.py @@ -8,7 +8,7 @@ # TaskStage represents a HCP Terraform or Terraform Enterprise run's stage where run tasks can occur class TaskStage(BaseModel): - model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + model_config = ConfigDict(populate_by_name=True) id: str # stage: Stage = Field(..., alias="stage") diff --git a/src/pytfe/resources/task_result.py b/src/pytfe/resources/task_result.py new file mode 100644 index 0000000..7640d01 --- /dev/null +++ b/src/pytfe/resources/task_result.py @@ -0,0 +1,32 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from typing import Any + +from pytfe.models.task_result import TaskResult +from pytfe.utils import valid_string_id + +from ._base import _Service + + +class TaskResults(_Service): + def read(self, task_result_id: str) -> TaskResult: + if not valid_string_id(task_result_id): + raise ValueError("Invalid task_result_id") + + path = f"/api/v2/task-results/{task_result_id}" + + response = self.t.request("GET", path) + data = response.json() + + if "data" not in data: + raise ValueError("Invalid response format") + + return self._parse_task_result(data["data"]) + + def _parse_task_result(self, data: dict[str, Any]) -> TaskResult: + attributes = data.get("attributes", {}) + + attributes["id"] = data.get("id") + + return TaskResult(**attributes) diff --git a/tests/units/test_task_results.py b/tests/units/test_task_results.py new file mode 100644 index 0000000..ff134fc --- /dev/null +++ b/tests/units/test_task_results.py @@ -0,0 +1,123 @@ +from unittest.mock import Mock + +import pytest + +from pytfe.models.task_result import TaskResult +from pytfe.resources.task_result import TaskResults + + +class TestTaskResults: + @pytest.fixture + def mock_transport(self): + return Mock() + + @pytest.fixture + def service(self, mock_transport): + return TaskResults(mock_transport) + + def test_read_success(self, service, mock_transport): + response = Mock() + response.json.return_value = { + "data": { + "id": "tr-123", + "attributes": { + "status": "passed", + "message": "ok", + "status-timestamps": {}, + "url": "url", + "created-at": "2024-01-01T00:00:00Z", + "updated-at": "2024-01-01T00:00:00Z", + "task-id": "t1", + "task-name": "name", + "task-url": "url", + "workspace-task-id": "wt1", + "workspace-task-enforcement-level": "advisory", + "agent-pool-id": None, + }, + } + } + mock_transport.request.return_value = response + + result = service.read("tr-123") + + assert isinstance(result, TaskResult) + assert result.id == "tr-123" + assert result.status == "passed" + + def test_invalid_id(self, service): + with pytest.raises(ValueError): + service.read("") + + def test_missing_data(self, service, mock_transport): + response = Mock() + response.json.return_value = {} + + mock_transport.request.return_value = response + + with pytest.raises(ValueError): + service.read("tr-123") + + def test_missing_attributes(self, service, mock_transport): + response = Mock() + response.json.return_value = {"data": {"id": "tr-123"}} + + mock_transport.request.return_value = response + + result = service.read("tr-123") + + assert result.id == "tr-123" + + def test_optional_fields(self, service, mock_transport): + response = Mock() + response.json.return_value = { + "data": { + "id": "tr-123", + "attributes": { + "status": "passed", + "message": None, + }, + } + } + + mock_transport.request.return_value = response + + result = service.read("tr-123") + + assert result.message is None + + def test_status_enum(self, service, mock_transport): + response = Mock() + response.json.return_value = { + "data": { + "id": "tr-123", + "attributes": { + "status": "failed", + "message": "fail", + }, + } + } + + mock_transport.request.return_value = response + + result = service.read("tr-123") + + assert result.status == "failed" + + def test_timestamps_parsing(self, service, mock_transport): + response = Mock() + response.json.return_value = { + "data": { + "id": "tr-123", + "attributes": { + "status": "passed", + "message": "ok", + "status-timestamps": {"passed-at": "2024-01-01T00:00:00Z"}, + }, + } + } + + mock_transport.request.return_value = response + + result = service.read("tr-123") + + assert result.status_timestamps is not None