diff --git a/examples/task_result.py b/examples/task_result.py new file mode 100644 index 00000000..ca229213 --- /dev/null +++ b/examples/task_result.py @@ -0,0 +1,36 @@ +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}") + print(f"Task Stage: {result.task_stage.id if result.task_stage else None}") + + 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 7b5bd993..409a36c3 100644 --- a/src/pytfe/client.py +++ b/src/pytfe/client.py @@ -44,6 +44,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.team_token import TeamTokens @@ -93,6 +94,7 @@ def __init__(self, config: TFEConfig | None = None): ) # org Explorer queries and saved views self.users = Users(self._transport) + self.task_results = TaskResults(self._transport) self.organization_tags = OrganizationTags(self._transport) self.organization_tokens = OrganizationTokens(self._transport) self.projects = Projects(self._transport) diff --git a/src/pytfe/models/__init__.py b/src/pytfe/models/__init__.py index 2e2cb323..f7cda576 100644 --- a/src/pytfe/models/__init__.py +++ b/src/pytfe/models/__init__.py @@ -366,6 +366,17 @@ StateVersionOutput, StateVersionOutputsListOptions, ) + +# ── Task Result ─────────────────────────────────────────────────────────────── +from .task_result import ( + TaskEnforcementLevel as TaskResultEnforcementLevel, +) +from .task_result import ( + TaskResult, + TaskResultStatus, + TaskResultStatusTimestamps, +) +from .task_stage import TaskStage from .team import ( OrganizationAccess, Team, @@ -717,6 +728,12 @@ "RunTaskCreateOptions", "RunTaskUpdateOptions", "RunTaskReadOptions", + # Task Result + "TaskResult", + "TaskResultEnforcementLevel", + "TaskResultStatus", + "TaskResultStatusTimestamps", + "TaskStage", # Run task integration (callback) "TaskResultCallbackRequestOptions", "TaskResultCallbackStatus", @@ -809,3 +826,14 @@ RegistryProvider.model_rebuild() RegistryProviderVersion.model_rebuild() RegistryProviderPlatform.model_rebuild() + +# Rebuild TaskResult to resolve Run, Workspace, PolicyEvaluation, TaskStage refs +TaskResult.model_rebuild( + raise_errors=False, + _types_namespace={ + "PolicyEvaluation": PolicyEvaluation, + "Run": Run, + "TaskStage": TaskStage, + "Workspace": Workspace, + }, +) diff --git a/src/pytfe/models/task_result.py b/src/pytfe/models/task_result.py new file mode 100644 index 00000000..588dc9b7 --- /dev/null +++ b/src/pytfe/models/task_result.py @@ -0,0 +1,117 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from __future__ import annotations + +from datetime import datetime +from enum import Enum +from typing import TYPE_CHECKING, Any + +from pydantic import BaseModel, ConfigDict, Field + +if TYPE_CHECKING: + # Imported only for type checking to avoid circular imports. + from pytfe.models.policy_evaluation import PolicyEvaluation + from pytfe.models.run import Run + from pytfe.models.task_stage import TaskStage + from pytfe.models.workspace import Workspace + + +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") + + # Relationships + # Forward-referenced to avoid circular imports; resolved lazily below. + task_stage: TaskStage | None = Field(None, alias="task-stage") + run: Run | None = Field(None, alias="run") + workspace: Workspace | None = Field(None, alias="workspace") + policy_evaluations: list[PolicyEvaluation] | None = Field( + None, + alias="policy-evaluations", + ) + + @classmethod + def model_validate(cls, *args: Any, **kwargs: Any) -> TaskResult: + # Ensure the TaskStage forward reference is resolved before validating. + # The import-time rebuild may run while task_stage.py is still + # partially loaded (circular import), in which case we retry here. + if not getattr(cls, "__pydantic_complete__", True): + _rebuild_task_result_model() + return super().model_validate(*args, **kwargs) + + +def _rebuild_task_result_model() -> None: + # Resolve all forward references once all modules are loaded. + try: + from pytfe.models.policy_evaluation import PolicyEvaluation + from pytfe.models.run import Run + from pytfe.models.task_stage import TaskStage + from pytfe.models.workspace import Workspace + + TaskResult.model_rebuild( + raise_errors=False, + _types_namespace={ + "PolicyEvaluation": PolicyEvaluation, + "Run": Run, + "TaskStage": TaskStage, + "Workspace": Workspace, + }, + ) + except Exception: + # One or more models not yet importable during partial init; safe to skip. + pass + + +_rebuild_task_result_model() diff --git a/src/pytfe/models/task_stage.py b/src/pytfe/models/task_stage.py index 54b9346b..bae0766b 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 00000000..ed6a7b61 --- /dev/null +++ b/src/pytfe/resources/task_result.py @@ -0,0 +1,77 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from typing import Any + +from pytfe.models.policy_evaluation import PolicyEvaluation +from pytfe.models.run import Run +from pytfe.models.task_result import TaskResult +from pytfe.models.task_stage import TaskStage +from pytfe.models.workspace import Workspace +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: + # Ensure forward references in TaskResult are resolved before use. + TaskResult.model_rebuild( + raise_errors=False, + _types_namespace={ + "PolicyEvaluation": PolicyEvaluation, + "Run": Run, + "TaskStage": TaskStage, + "Workspace": Workspace, + }, + ) + + attributes = data.get("attributes", {}) + attributes["id"] = data.get("id") + + relationships = data.get("relationships", {}) + + # Map task-stage relationship into the TaskStage SDK model. + task_stage_data = relationships.get("task-stage", {}).get("data") + if task_stage_data: + attributes["task-stage"] = TaskStage.model_validate(task_stage_data) + else: + attributes["task-stage"] = None + + # Map run relationship into the Run SDK model. + run_data = relationships.get("run", {}).get("data") + if run_data: + attributes["run"] = Run.model_validate(run_data) + else: + attributes["run"] = None + + # Map workspace relationship into the Workspace SDK model. + workspace_data = relationships.get("workspace", {}).get("data") + if workspace_data: + attributes["workspace"] = Workspace.model_validate(workspace_data) + else: + attributes["workspace"] = None + + # Map policy-evaluations relationship into a list of PolicyEvaluation models. + policy_evaluations_data = relationships.get("policy-evaluations", {}).get( + "data", [] + ) + attributes["policy-evaluations"] = [ + PolicyEvaluation.model_validate(pe) for pe in policy_evaluations_data + ] + + return TaskResult.model_validate(attributes) diff --git a/tests/units/test_task_results.py b/tests/units/test_task_results.py new file mode 100644 index 00000000..13d0ba98 --- /dev/null +++ b/tests/units/test_task_results.py @@ -0,0 +1,285 @@ +from unittest.mock import Mock + +import pytest + +from pytfe.models.policy_evaluation import PolicyEvaluation +from pytfe.models.run import Run +from pytfe.models.task_result import TaskResult +from pytfe.models.task_stage import TaskStage +from pytfe.models.workspace import Workspace +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 + + # ── Relationship mapping tests ───────────────────────────────────────────── + + def test_task_stage_relationship_mapped(self, service, mock_transport): + response = Mock() + response.json.return_value = { + "data": { + "id": "tr-123", + "attributes": {"status": "passed"}, + "relationships": { + "task-stage": {"data": {"id": "ts-456", "type": "task-stages"}} + }, + } + } + mock_transport.request.return_value = response + + result = service.read("tr-123") + + assert isinstance(result.task_stage, TaskStage) + assert result.task_stage.id == "ts-456" + + def test_task_stage_relationship_null(self, service, mock_transport): + response = Mock() + response.json.return_value = { + "data": { + "id": "tr-123", + "attributes": {"status": "passed"}, + "relationships": {"task-stage": {"data": None}}, + } + } + mock_transport.request.return_value = response + + result = service.read("tr-123") + + assert result.task_stage is None + + def test_run_relationship_mapped(self, service, mock_transport): + response = Mock() + response.json.return_value = { + "data": { + "id": "tr-123", + "attributes": {"status": "passed"}, + "relationships": {"run": {"data": {"id": "run-789", "type": "runs"}}}, + } + } + mock_transport.request.return_value = response + + result = service.read("tr-123") + + assert isinstance(result.run, Run) + assert result.run.id == "run-789" + + def test_run_relationship_null(self, service, mock_transport): + response = Mock() + response.json.return_value = { + "data": { + "id": "tr-123", + "attributes": {"status": "passed"}, + "relationships": {"run": {"data": None}}, + } + } + mock_transport.request.return_value = response + + result = service.read("tr-123") + + assert result.run is None + + def test_workspace_relationship_mapped(self, service, mock_transport): + response = Mock() + response.json.return_value = { + "data": { + "id": "tr-123", + "attributes": {"status": "passed"}, + "relationships": { + "workspace": {"data": {"id": "ws-abc", "type": "workspaces"}} + }, + } + } + mock_transport.request.return_value = response + + result = service.read("tr-123") + + assert isinstance(result.workspace, Workspace) + assert result.workspace.id == "ws-abc" + + def test_workspace_relationship_null(self, service, mock_transport): + response = Mock() + response.json.return_value = { + "data": { + "id": "tr-123", + "attributes": {"status": "passed"}, + "relationships": {"workspace": {"data": None}}, + } + } + mock_transport.request.return_value = response + + result = service.read("tr-123") + + assert result.workspace is None + + def test_policy_evaluations_relationship_mapped(self, service, mock_transport): + response = Mock() + response.json.return_value = { + "data": { + "id": "tr-123", + "attributes": {"status": "passed"}, + "relationships": { + "policy-evaluations": { + "data": [ + {"id": "pe-001", "type": "policy-evaluations"}, + {"id": "pe-002", "type": "policy-evaluations"}, + ] + } + }, + } + } + mock_transport.request.return_value = response + + result = service.read("tr-123") + + assert isinstance(result.policy_evaluations, list) + assert len(result.policy_evaluations) == 2 + assert all(isinstance(pe, PolicyEvaluation) for pe in result.policy_evaluations) + assert result.policy_evaluations[0].id == "pe-001" + assert result.policy_evaluations[1].id == "pe-002" + + def test_policy_evaluations_relationship_empty(self, service, mock_transport): + response = Mock() + response.json.return_value = { + "data": { + "id": "tr-123", + "attributes": {"status": "passed"}, + "relationships": {"policy-evaluations": {"data": []}}, + } + } + mock_transport.request.return_value = response + + result = service.read("tr-123") + + assert result.policy_evaluations == [] + + def test_no_relationships_key(self, service, mock_transport): + """When 'relationships' is absent, all relationship fields stay None.""" + response = Mock() + response.json.return_value = { + "data": { + "id": "tr-123", + "attributes": {"status": "passed"}, + } + } + mock_transport.request.return_value = response + + result = service.read("tr-123") + + assert result.task_stage is None + assert result.run is None + assert result.workspace is None + assert result.policy_evaluations == []