From 4c0a04070d36a0b4fa36685fa550f7805b353932 Mon Sep 17 00:00:00 2001 From: Sivaselvan32 Date: Wed, 24 Sep 2025 17:36:05 +0530 Subject: [PATCH 1/3] Features providing Plan and Apply API Specs --- src/tfe/client.py | 4 + src/tfe/errors.py | 16 +++ src/tfe/log_reader.py | 246 ++++++++++++++++++++++++++++++++++ src/tfe/models/apply.py | 42 ++++++ src/tfe/models/plan.py | 51 +++++++ src/tfe/models/plan_export.py | 9 ++ src/tfe/resources/apply.py | 134 ++++++++++++++++++ src/tfe/resources/plan.py | 161 ++++++++++++++++++++++ tests/units/test_plan.py | 163 ++++++++++++++++++++++ 9 files changed, 826 insertions(+) create mode 100644 src/tfe/log_reader.py create mode 100644 src/tfe/models/apply.py create mode 100644 src/tfe/models/plan.py create mode 100644 src/tfe/models/plan_export.py create mode 100644 src/tfe/resources/apply.py create mode 100644 src/tfe/resources/plan.py create mode 100644 tests/units/test_plan.py diff --git a/src/tfe/client.py b/src/tfe/client.py index 38fc5e57..5448117b 100644 --- a/src/tfe/client.py +++ b/src/tfe/client.py @@ -2,7 +2,9 @@ from ._http import HTTPTransport from .config import TFEConfig +from .resources.apply import Applies from .resources.organizations import Organizations +from .resources.plan import Plans from .resources.projects import Projects from .resources.registry_module import RegistryModules from .resources.registry_provider import RegistryProviders @@ -32,6 +34,8 @@ def __init__(self, config: TFEConfig | None = None): proxies=cfg.proxies, ca_bundle=cfg.ca_bundle, ) + self.applies = Applies(self._transport) + self.plans = Plans(self._transport) self.organizations = Organizations(self._transport) self.projects = Projects(self._transport) self.variables = Variables(self._transport) diff --git a/src/tfe/errors.py b/src/tfe/errors.py index 84eaf3c0..9b9b118c 100644 --- a/src/tfe/errors.py +++ b/src/tfe/errors.py @@ -314,3 +314,19 @@ class InvalidRunTriggerIDError(InvalidValues): def __init__(self, message: str = "invalid value for run trigger ID"): super().__init__(message) + + +# Plan errors +class InvalidPlanIDError(InvalidValues): + """Raised when an invalid plan ID is provided.""" + + def __init__(self, message: str = "invalid value for plan ID"): + super().__init__(message) + + +# Apply errors +class InvalidApplyIDError(InvalidValues): + """Raised when an invalid apply ID is provided.""" + + def __init__(self, message: str = "invalid value for apply ID"): + super().__init__(message) diff --git a/src/tfe/log_reader.py b/src/tfe/log_reader.py new file mode 100644 index 00000000..8d4dfc20 --- /dev/null +++ b/src/tfe/log_reader.py @@ -0,0 +1,246 @@ +"""LogReader implementation for streaming TFE plan/apply logs.""" + +from __future__ import annotations + +import asyncio +import math +import time +from collections.abc import Callable +from typing import Any +from urllib.parse import urlparse, urlunparse + +import httpx + +from .models.plan import PlanStatus + + +class LogReader: + """ + LogReader implements io.Reader for streaming logs. + + This class exactly mirrors the Go LogReader implementation: + - Implements Read() method that works with bytes (like io.Reader) + - Handles context cancellation with select-like behavior + - STX/ETX control character handling at byte level + - Exponential backoff with exact same algorithm as Go + - Proper HTTP error handling via checkResponseCode equivalent + + Usage: + # For streaming logs byte by byte (like Go's io.Reader) + log_reader = LogReader(transport, log_url, done_func, context) + buffer = bytearray(4096) + while True: + n, err = await log_reader.read(buffer) + if n > 0: + print(buffer[:n].decode('utf-8', errors='ignore'), end='') + if err: + break + + # For reading all logs at once + all_logs = await log_reader.read_all() + """ + + def __init__( + self, + transport: Any, + log_url: str, + done_func: Callable[[], tuple[bool, Exception | None]], + context: Any = None, + ) -> None: + """ + Initialize LogReader. + + Args: + transport: HTTP transport for internal requests + log_url: URL to fetch logs from + done_func: Function that returns (done, error) tuple + context: Optional context for cancellation + """ + self.transport = transport + self.done_func = done_func + self.context = context + + # State tracking (exactly like Go implementation) + self.offset = 0 + self.reads = 0 + self.start_of_text = False + self.end_of_text = False + + # Parse URL for validation (like Go url.Parse) + self.parsed_url = urlparse(log_url) + if not self.parsed_url.scheme or not self.parsed_url.netloc: + raise ValueError(f"Invalid log URL: {log_url}") + + async def read(self, buffer: bytearray | bytes) -> tuple[int, Exception | None]: + """ + Read data into the provided buffer (io.Reader equivalent). + + This method exactly mirrors the Go LogReader.Read() behavior: + - Returns (bytes_read, error) tuple like Go + - Handles context cancellation with select-like behavior + - Implements exponential backoff + - Processes STX/ETX control characters at byte level + + Args: + buffer: Buffer to read data into + + Returns: + Tuple of (bytes_read, error). Returns (0, EOFError) when done, + (0, NoProgressError) for no progress, or (n, None) for n bytes read. + """ + # First attempt to read (like Go: if written, err := r.read(l)) + written, err = await self._read(buffer) + if err is not None and not isinstance(err, NoProgressError): + return written, err + + # Loop until we get data, context is cancelled, or run is finished + # This exactly mirrors the Go implementation's for loop + self.reads = 1 + while True: + try: + # Context cancellation check (equivalent to Go's select case <-r.ctx.Done()) + if self.context and hasattr(self.context, 'cancelled') and self.context.cancelled(): + return 0, self.context.exception() + + # Wait with backoff (equivalent to Go's case <-time.After(backoff(...))) + await asyncio.sleep(self._backoff(500, 2000, self.reads) / 1000.0) + + written, err = await self._read(buffer) + if err is not None and not isinstance(err, NoProgressError): + return written, err + + self.reads += 1 + except asyncio.CancelledError as e: + return 0, e + + async def _read(self, buffer: bytearray | bytes) -> tuple[int, Exception | None]: + """ + Internal read method that handles HTTP requests and data processing. + + This method exactly mirrors the Go LogReader.read() method. + + Args: + buffer: Buffer to read data into + + Returns: + Tuple of (bytes_read, error) + """ + # Update the query string (exactly like Go: r.logURL.RawQuery = fmt.Sprintf(...)) + url_parts = list(self.parsed_url) + query = f"limit={len(buffer)}&offset={self.offset}" + url_parts[4] = query # query component + chunk_url = urlunparse(url_parts) + + try: + # Create a new request (like Go: req, err := http.NewRequest("GET", ...)) + # Use the transport to make the request (like Go client.http.HTTPClient.Do) + response = await self.transport.arequest("GET", chunk_url) + + # Read the response body as bytes (like Go: written, err := resp.Body.Read(l)) + chunk_data = response.content + + except Exception as e: + return 0, e + + if not chunk_data: + return 0, NoProgressError() + + written = len(chunk_data) + + # Handle STX/ETX control characters at byte level (exactly like Go) + if written > 0: + # Check for STX (Start of Text) ASCII control marker + if not self.start_of_text and chunk_data[0] == 2: + self.start_of_text = True + + # Remove the STX marker from the received chunk (like Go copy operation) + chunk_data = chunk_data[1:] + self.offset += 1 + written -= 1 + + # Return early if we only received the STX marker + if written == 0: + return 0, NoProgressError() + + # If we found an STX ASCII control character, start looking for ETX + if self.start_of_text and chunk_data[-1] == 3: + self.end_of_text = True + + # Remove the ETX marker from the received chunk + chunk_data = chunk_data[:-1] + self.offset += 1 + written -= 1 + + # Copy data to buffer + if written > 0: + buffer[:written] = chunk_data[:written] + + # Check if we need to continue the loop (exactly like Go logic) + if written != 0: + # Update the offset for the next read + self.offset += written + return written, None + + # Check completion conditions (exactly like Go implementation) + if ( + (self.start_of_text and self.end_of_text) or # The logstream finished without issues + (self.start_of_text and self.reads % 10 == 0) or # The logstream terminated unexpectedly + (not self.start_of_text and self.reads > 1) # The logstream doesn't support STX/ETX + ): + # Check if operation is done (like Go: done, err := r.done()) + try: + done, err = self.done_func() + if err: + return 0, err + if done: + return 0, EOFError("End of log stream") + except Exception as e: + return 0, e + + return 0, NoProgressError() + + async def read_all(self, chunk_size: int = 4096) -> str: + """ + Read all available logs as a single string. + + Args: + chunk_size: Size of each chunk to read + + Returns: + Complete log content as string + """ + buffer = bytearray(chunk_size) + result = bytearray() + + while True: + n, err = await self.read(buffer) + if n > 0: + result.extend(buffer[:n]) + if err: + if isinstance(err, EOFError): + break + raise err + + return result.decode('utf-8', errors='ignore') + + def _backoff(self, minimum: float, maximum: float, iter: int) -> float: + """ + Calculate exponential backoff duration (exactly like Go implementation). + + Args: + minimum: Minimum backoff in milliseconds + maximum: Maximum backoff in milliseconds + iter: Current iteration number + + Returns: + Backoff duration in milliseconds + """ + backoff = math.pow(2, iter / 5) * minimum + if backoff > maximum: + backoff = maximum + return backoff + + +class NoProgressError(Exception): + """Error indicating no progress was made (equivalent to Go's io.ErrNoProgress).""" + pass \ No newline at end of file diff --git a/src/tfe/models/apply.py b/src/tfe/models/apply.py new file mode 100644 index 00000000..3de7e37b --- /dev/null +++ b/src/tfe/models/apply.py @@ -0,0 +1,42 @@ +from __future__ import annotations + +from datetime import datetime +from enum import Enum + +from pydantic import BaseModel, ConfigDict, Field + + +class ApplyStatus(str, Enum): + APPLY_CANCELED = "canceled" + APPLY_CREATED = "created" + APPLY_ERRORED = "errored" + APPLY_FINISHED = "finished" + APPLY_MFA_WAITING = "mfa_waiting" + APPLY_PENDING = "pending" + APPLY_QUEUED = "queued" + APPLY_RUNNING = "running" + APPLY_UNREACHABLE = "unreachable" + + +class Apply(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + id: str + log_read_url: str = Field(..., alias="log-read-url") + resource_additions: int = Field(..., alias="resource-additions") + resource_changes: int = Field(..., alias="resource-changes") + resource_destructions: int = Field(..., alias="resource-destructions") + resource_imports: int = Field(..., alias="resource-imports") + status: ApplyStatus = Field(..., alias="status") + status_timestamps: ApplyStatusTimestamps = Field(..., alias="status-timestamps") + + +class ApplyStatusTimestamps(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + canceled_at: datetime | None = Field(None, alias="canceled-at") + errored_at: datetime | None = Field(None, alias="errored-at") + finished_at: datetime | None = Field(None, alias="finished-at") + force_canceled_at: datetime | None = Field(None, alias="force-canceled-at") + queued_at: datetime | None = Field(None, alias="queued-at") + started_at: datetime | None = Field(None, alias="started-at") diff --git a/src/tfe/models/plan.py b/src/tfe/models/plan.py new file mode 100644 index 00000000..ee8ac898 --- /dev/null +++ b/src/tfe/models/plan.py @@ -0,0 +1,51 @@ +from __future__ import annotations + +from datetime import datetime +from enum import Enum + +from pydantic import BaseModel, ConfigDict, Field + +from ..models.plan_export import PlanExport + + +class PlanStatus(str, Enum): + """The status of a plan.""" + + PLAN_CANCELED = "canceled" + PLAN_CREATED = "created" + PLAN_ERRORED = "errored" + PLAN_FINISHED = "finished" + PLAN_MFA_WAITING = "mfa_waiting" + PLAN_PENDING = "pending" + PLAN_QUEUED = "queued" + PLAN_RUNNING = "running" + PLAN_UNREACHABLE = "unreachable" + + +class Plan(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + id: str + has_changes: bool = Field(..., alias="has-changes") + generated_configuration: bool = Field(..., alias="generated-configuration") + log_read_url: str = Field(..., alias="log-read-url") + resource_additions: int = Field(..., alias="resource-additions") + resource_changes: int = Field(..., alias="resource-changes") + resource_destructions: int = Field(..., alias="resource-destructions") + resource_imports: int = Field(..., alias="resource-imports") + status: PlanStatus = Field(..., alias="status") + status_timestamps: PlanStatusTimestamps = Field(..., alias="status-timestamps") + + # Relations + exports: list[PlanExport] = Field(..., alias="exports") + + +class PlanStatusTimestamps(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + canceled_at: datetime = Field(..., alias="canceled-at") + errored_at: datetime = Field(..., alias="errored-at") + finished_at: datetime = Field(..., alias="finished-at") + force_canceled_at: datetime = Field(..., alias="force-canceled-at") + queued_at: datetime = Field(..., alias="queued-at") + started_at: datetime = Field(..., alias="started-at") diff --git a/src/tfe/models/plan_export.py b/src/tfe/models/plan_export.py new file mode 100644 index 00000000..0d5e45a9 --- /dev/null +++ b/src/tfe/models/plan_export.py @@ -0,0 +1,9 @@ +from __future__ import annotations + +from pydantic import BaseModel, ConfigDict + + +class PlanExport(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + id: str diff --git a/src/tfe/resources/apply.py b/src/tfe/resources/apply.py new file mode 100644 index 00000000..c87c9ea9 --- /dev/null +++ b/src/tfe/resources/apply.py @@ -0,0 +1,134 @@ +from __future__ import annotations + +from collections.abc import Callable + +from ..errors import InvalidApplyIDError +from ..log_reader import LogReader +from ..models.apply import ( + Apply, +) +from ..utils import valid_string_id +from ._base import _Service + + +class Applies(_Service): + def read(self, apply_id: str) -> Apply: + """Read a specific apply by its ID.""" + if not valid_string_id(apply_id): + raise InvalidApplyIDError() + + r = self.t.request( + "GET", + f"/api/v2/applies/{apply_id}", + ) + d = r.json()["data"] + attr = d.get("attributes", {}) or {} + return Apply( + id=d.get("id"), + **{k.replace("-", "_"): v for k, v in attr.items()}, + ) + + def logs_reader(self, apply_id: str) -> LogReader: + """Get a LogReader for streaming logs from a specific apply. + + This method follows the Go LogReader pattern, providing: + - Chunked reading with offset/limit parameters + - Status checking to determine when logs are complete + - STX/ETX control character handling + - Exponential backoff for retries + + Returns: + LogReader instance for streaming logs + + Usage: + # Stream logs chunk by chunk (async) + log_reader = applies.logs_reader("apply-123") + buffer = bytearray(4096) + while True: + n, err = await log_reader.read(buffer) + if n > 0: + print(buffer[:n].decode('utf-8', errors='ignore'), end='') + if err: + break + + # Or get all logs at once (async) + all_logs = await log_reader.read_all() + + # Or use the convenience logs() method for synchronous access + all_logs = applies.logs("apply-123") + """ + # Validate apply ID + if not valid_string_id(apply_id): + raise InvalidApplyIDError() + + # Get the apply and validate log URL + apply = self.read(apply_id) + self._validate_log_url(apply.log_read_url, apply_id) + + # Create done function for status checking + done_func = lambda: self._done(apply_id) + + # Return LogReader configured with transport, URL, and done function + return LogReader( + transport=self.t, + log_url=apply.log_read_url, + done_func=done_func, + ) + + def logs(self, apply_id: str) -> str: + """Get all logs for a specific apply as a string. + + This is a convenience method that uses logs_reader() internally + to fetch all logs at once. For streaming logs, use logs_reader() instead. + + Args: + apply_id: Apply ID to get logs for + + Returns: + Complete log content as string + """ + import asyncio + + log_reader = self.logs_reader(apply_id) + return asyncio.run(log_reader.read_all()) + + def _done(self, apply_id: str) -> tuple[bool, Exception | None]: + """ + Check if an apply is in a terminal state. + + Args: + apply_id: Apply ID to check + + Returns: + Tuple of (is_complete, error) + """ + try: + apply_obj = self.read(apply_id) + terminal_states = {"canceled", "errored", "finished", "unreachable"} + is_complete = apply_obj.status in terminal_states + return is_complete, None + except Exception as e: + return False, e + + def _validate_log_url(self, log_url: str, resource_id: str) -> None: + """ + Validate that a log URL exists and has the correct format. + + Args: + log_url: The log URL to validate + resource_id: The resource ID for error messages + + Raises: + ValueError: If the log URL is invalid or empty + """ + if not log_url: + raise ValueError(f"Apply {resource_id} does not have a log URL") + + from urllib.parse import urlparse + + try: + parsed_url = urlparse(log_url) + if not parsed_url.scheme or not parsed_url.netloc: + raise ValueError(f"Invalid log URL format: {log_url}") + except Exception as e: + raise ValueError(f"Invalid log URL: {log_url}") from e diff --git a/src/tfe/resources/plan.py b/src/tfe/resources/plan.py new file mode 100644 index 00000000..948d7bf4 --- /dev/null +++ b/src/tfe/resources/plan.py @@ -0,0 +1,161 @@ +from __future__ import annotations + +from collections.abc import Callable +from typing import Any + +from ..errors import InvalidPlanIDError +from ..log_reader import LogReader +from ..models.plan import ( + Plan, + PlanStatus, +) +from ..utils import valid_string_id +from ._base import _Service + + +class Plans(_Service): + def read(self, plan_id: str) -> Plan: + """Read a specific plan by its ID.""" + if not valid_string_id(plan_id): + raise InvalidPlanIDError() + + r = self.t.request( + "GET", + f"/api/v2/plans/{plan_id}", + ) + d = r.json()["data"] + attr = d.get("attributes", {}) or {} + return Plan( + id=d.get("id"), + **{k.replace("-", "_"): v for k, v in attr.items()}, + ) + + def logs_reader(self, plan_id: str) -> LogReader: + """Get a LogReader for streaming logs from a specific plan. + + This method follows the Go LogReader pattern, providing: + - Chunked reading with offset/limit parameters + - Status checking to determine when logs are complete + - STX/ETX control character handling + - Exponential backoff for retries + + Returns: + LogReader instance for streaming logs + + Usage: + # Stream logs chunk by chunk (async) + log_reader = plans.logs_reader("plan-123") + buffer = bytearray(4096) + while True: + n, err = await log_reader.read(buffer) + if n > 0: + print(buffer[:n].decode('utf-8', errors='ignore'), end='') + if err: + break + + # Or get all logs at once (async) + all_logs = await log_reader.read_all() + + # Or use the convenience logs() method for synchronous access + all_logs = plans.logs("plan-123") + """ + # Validate plan ID + if not valid_string_id(plan_id): + raise InvalidPlanIDError() + + # Get the plan and validate log URL + plan = self.read(plan_id) + self._validate_log_url(plan.log_read_url, plan_id) + + # Create done function for status checking + done_func = lambda: self._done(plan_id) + + # Return LogReader configured with transport, URL, and done function + return LogReader( + transport=self.t, + log_url=plan.log_read_url, + done_func=done_func, + ) + + def logs(self, plan_id: str) -> str: + """Get all logs for a specific plan as a string. + + This is a convenience method that uses logs_reader() internally + to fetch all logs at once. For streaming logs, use logs_reader() instead. + + Args: + plan_id: Plan ID to get logs for + + Returns: + Complete log content as string + """ + import asyncio + + log_reader = self.logs_reader(plan_id) + return asyncio.run(log_reader.read_all()) + + def read_json_output(self, plan_id: str) -> dict[str, Any]: + """Get the JSON execution plan for a specific plan by its ID. + + Returns the JSON representation of the Terraform execution plan, + which includes detailed information about planned changes. + """ + if not valid_string_id(plan_id): + raise InvalidPlanIDError() + + r = self.t.request( + "GET", + f"/api/v2/plans/{plan_id}/json-output", + ) + + # Return the raw JSON data - this endpoint returns JSON directly + # not wrapped in a JSON:API format + json_data = r.json() + # Ensure we return a dictionary, not Any + if isinstance(json_data, dict): + return json_data + else: + # If somehow the response isn't a dict, wrap it + return {"data": json_data} + + def _done(self, plan_id: str) -> bool: + """ + Create a done function for plan log reading. + + Args: + plan_id: Plan ID to check + + Returns: + Function that returns boolean + """ + plan = self.read(plan_id) + terminal_states = { + PlanStatus.PLAN_CANCELED, + PlanStatus.PLAN_ERRORED, + PlanStatus.PLAN_FINISHED, + PlanStatus.PLAN_UNREACHABLE, + } + return plan.status in terminal_states + + def _validate_log_url(self, log_url: str, resource_id: str) -> None: + """ + Validate that a log URL exists and has the correct format. + + Args: + log_url: The log URL to validate + resource_id: The resource ID for error messages + + Raises: + ValueError: If the log URL is invalid or empty + """ + if not log_url: + raise ValueError(f"Plan {resource_id} does not have a log URL") + + from urllib.parse import urlparse + + try: + parsed_url = urlparse(log_url) + if not parsed_url.scheme or not parsed_url.netloc: + raise ValueError(f"Invalid log URL format: {log_url}") + except Exception as e: + raise ValueError(f"Invalid log URL: {log_url}") from e diff --git a/tests/units/test_plan.py b/tests/units/test_plan.py new file mode 100644 index 00000000..62666eb7 --- /dev/null +++ b/tests/units/test_plan.py @@ -0,0 +1,163 @@ +"""Unit tests for the plan module.""" + +from unittest.mock import Mock, patch + +import pytest + +from tfe.errors import InvalidPlanIDError +from tfe.resources.plan import Plans + + +class TestPlans: + @pytest.fixture + def plans_service(self): + """Create a Plans service for testing.""" + mock_transport = Mock() + return Plans(mock_transport) + + def test_plans_service_init(self, plans_service): + """Test Plans service initialization.""" + assert plans_service.t is not None + + def test_read_plan_validation_errors(self, plans_service): + """Test read method with invalid plan ID.""" + + # Test empty plan ID + with pytest.raises(InvalidPlanIDError): + plans_service.read("") + + # Test None plan ID + with pytest.raises(InvalidPlanIDError): + plans_service.read(None) + + def test_read_plan_success(self, plans_service): + """Test successful read operation.""" + + mock_response_data = { + "data": { + "id": "plan-123", + "attributes": { + "has-changes": True, + "generated-configuration": False, + "log-read-url": "https://example.com/logs/plan-123", + "resource-additions": 3, + "resource-changes": 1, + "resource-destructions": 0, + "resource-imports": 0, + "status": "finished", + "status-timestamps": { + "canceled-at": "2023-01-01T00:00:00Z", + "errored-at": "2023-01-01T00:00:00Z", + "finished-at": "2023-01-01T10:00:00Z", + "force-canceled-at": "2023-01-01T00:00:00Z", + "queued-at": "2023-01-01T09:00:00Z", + "started-at": "2023-01-01T09:30:00Z", + }, + "exports": [], + }, + } + } + + with patch.object(plans_service, "t") as mock_transport: + mock_response = Mock() + mock_response.json.return_value = mock_response_data + mock_transport.request.return_value = mock_response + + result = plans_service.read("plan-123") + + # Verify request was made correctly + mock_transport.request.assert_called_once_with( + "GET", "/api/v2/plans/plan-123" + ) + + # Verify plan object + assert result.id == "plan-123" + assert result.has_changes is True + assert result.generated_configuration is False + assert result.log_read_url == "https://example.com/logs/plan-123" + assert result.resource_additions == 3 + assert result.resource_changes == 1 + assert result.resource_destructions == 0 + assert result.resource_imports == 0 + assert result.status.value == "finished" + + def test_logs_success(self, plans_service): + """Test successful logs operation.""" + + # Mock the read method to return a plan with log URL + mock_plan = Mock() + mock_plan.log_read_url = "https://example.com/logs/plan-123" + + with patch.object(plans_service, "read", return_value=mock_plan): + # Mock httpx.Client and its response + with patch("httpx.Client") as mock_httpx_client: + mock_context_manager = Mock() + mock_httpx_client.return_value.__enter__ = Mock( + return_value=mock_context_manager + ) + mock_httpx_client.return_value.__exit__ = Mock(return_value=None) + + mock_response = Mock() + mock_response.text = "Terraform will perform the following actions:\n\n + resource will be created" + mock_response.raise_for_status = Mock() # Don't raise any exceptions + + mock_context_manager.get.return_value = mock_response + + result = plans_service.logs("plan-123") + + # Verify read was called first + plans_service.read.assert_called_once_with("plan-123") + + # Verify httpx client was used correctly + mock_httpx_client.assert_called_once_with(timeout=30.0) + mock_context_manager.get.assert_called_once_with( + "https://example.com/logs/plan-123" + ) + mock_response.raise_for_status.assert_called_once() + + # Verify log content + assert ( + result + == "Terraform will perform the following actions:\n\n + resource will be created" + ) + + def test_read_json_output_success(self, plans_service): + """Test successful read_json_output operation.""" + + mock_json_data = { + "format_version": "1.1", + "terraform_version": "1.5.0", + "planned_values": {"root_module": {"resources": []}}, + "resource_changes": [ + { + "address": "resource.example", + "mode": "managed", + "type": "resource", + "name": "example", + "change": { + "actions": ["create"], + "before": None, + "after": {"name": "example"}, + }, + } + ], + } + + with patch.object(plans_service, "t") as mock_transport: + mock_response = Mock() + mock_response.json.return_value = mock_json_data + mock_transport.request.return_value = mock_response + + result = plans_service.read_json_output("plan-123") + + # Verify request was made correctly + mock_transport.request.assert_called_once_with( + "GET", "/api/v2/plans/plan-123/json-output" + ) + + # Verify JSON data is returned + assert result == mock_json_data + assert result["format_version"] == "1.1" + assert result["terraform_version"] == "1.5.0" + assert len(result["resource_changes"]) == 1 + assert result["resource_changes"][0]["change"]["actions"] == ["create"] From 816e5ce27e60ad07184341d4ca74de269dd31178 Mon Sep 17 00:00:00 2001 From: Sivaselvan32 Date: Thu, 25 Sep 2025 11:10:48 +0530 Subject: [PATCH 2/3] Plan and Apply API Specs --- examples/apply.py | 54 ++++++++ examples/plan.py | 88 +++++++++++++ src/tfe/log_reader.py | 246 ------------------------------------- src/tfe/resources/apply.py | 100 ++------------- src/tfe/resources/plan.py | 98 ++------------- src/tfe/utils.py | 11 ++ tests/units/test_apply.py | 96 +++++++++++++++ tests/units/test_plan.py | 38 ++---- 8 files changed, 277 insertions(+), 454 deletions(-) create mode 100644 examples/apply.py create mode 100644 examples/plan.py delete mode 100644 src/tfe/log_reader.py create mode 100644 tests/units/test_apply.py diff --git a/examples/apply.py b/examples/apply.py new file mode 100644 index 00000000..218697d1 --- /dev/null +++ b/examples/apply.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +import argparse +import os + +from tfe import TFEClient, TFEConfig + + +def _print_header(title: str): + print("\n" + "=" * 80) + print(title) + print("=" * 80) + + +def main(): + parser = argparse.ArgumentParser(description="Applies demo for python-tfe SDK") + parser.add_argument( + "--address", default=os.getenv("TFE_ADDRESS", "https://app.terraform.io") + ) + parser.add_argument("--token", default=os.getenv("TFE_TOKEN", "")) + parser.add_argument("--apply-id", required=True, help="Apply ID to work with") + args = parser.parse_args() + + cfg = TFEConfig(address=args.address, token=args.token) + client = TFEClient(cfg) + + # 1) Read the apply details + _print_header("Reading Apply Details") + try: + apply = client.applies.read(args.apply_id) + print(f"Apply ID: {apply.id}") + print(f"Status: {apply.status}") + print(f"Resource Additions: {apply.resource_additions}") + print(f"Resource Changes: {apply.resource_changes}") + print(f"Resource Destructions: {apply.resource_destructions}") + print(f"Resource Imports: {apply.resource_imports}") + print(f"Created At: {apply.created_at}") + print(f"Status Timestamps: {apply.status_timestamps}") + print(f"Log Read URL: {apply.log_read_url}") + print( + f"Execution Details ID: {apply.execution_details.id if apply.execution_details else 'None'}" + ) + except Exception as e: + print(f"Error reading apply: {e}") + return 1 + + print("\n" + "=" * 80) + print("Apply demo completed successfully!") + print("=" * 80) + return 0 + + +if __name__ == "__main__": + exit(main()) diff --git a/examples/plan.py b/examples/plan.py new file mode 100644 index 00000000..ec36269a --- /dev/null +++ b/examples/plan.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +import argparse +import json +import os + +from tfe import TFEClient, TFEConfig + + +def _print_header(title: str): + print("\n" + "=" * 80) + print(title) + print("=" * 80) + + +def main(): + parser = argparse.ArgumentParser(description="Plans demo for python-tfe SDK") + parser.add_argument( + "--address", default=os.getenv("TFE_ADDRESS", "https://app.terraform.io") + ) + parser.add_argument("--token", default=os.getenv("TFE_TOKEN", "")) + parser.add_argument("--plan-id", required=True, help="Plan ID to work with") + parser.add_argument("--save-json", help="Path to save JSON output") + args = parser.parse_args() + + cfg = TFEConfig(address=args.address, token=args.token) + client = TFEClient(cfg) + + # 1) Read the plan details + _print_header("Reading Plan Details") + try: + plan = client.plans.read(args.plan_id) + print(f"Plan ID: {plan.id}") + print(f"Status: {plan.status}") + print(f"Has Changes: {plan.has_changes}") + print(f"Resource Additions: {plan.resource_additions}") + print(f"Resource Changes: {plan.resource_changes}") + print(f"Resource Destructions: {plan.resource_destructions}") + print(f"Resource Imports: {plan.resource_imports}") + print(f"Status Timestamps: {plan.status_timestamps}") + print(f"Log Read URL: {plan.log_read_url}") + except Exception as e: + print(f"Error reading plan: {e}") + return 1 + + # 2) Get JSON output if the plan has it + _print_header("Reading JSON Output") + try: + json_output = client.plans.read_json_output(args.plan_id) + print( + f"JSON Output Keys: {list(json_output.keys()) if isinstance(json_output, dict) else 'Not a dict'}" + ) + + if isinstance(json_output, dict): + # Print some key information from the JSON output + if "format_version" in json_output: + print(f"Format Version: {json_output['format_version']}") + if "terraform_version" in json_output: + print(f"Terraform Version: {json_output['terraform_version']}") + if "resource_changes" in json_output: + changes = json_output["resource_changes"] + print(f"Number of Resource Changes: {len(changes) if changes else 0}") + + # Show first few resource changes + if changes: + print("\nFirst few resource changes:") + for i, change in enumerate(changes[:3]): + action = change.get("change", {}).get("actions", []) + address = change.get("address", "unknown") + print(f" {i + 1}. {address}: {action}") + + # Save JSON output if requested + if args.save_json: + with open(args.save_json, "w") as f: + json.dump(json_output, f, indent=2, default=str) + print(f"\nJSON output saved to: {args.save_json}") + + except Exception as e: + print(f"Error reading JSON output: {e}") + + print("\n" + "=" * 80) + print("Plan demo completed successfully!") + print("=" * 80) + return 0 + + +if __name__ == "__main__": + exit(main()) diff --git a/src/tfe/log_reader.py b/src/tfe/log_reader.py deleted file mode 100644 index 8d4dfc20..00000000 --- a/src/tfe/log_reader.py +++ /dev/null @@ -1,246 +0,0 @@ -"""LogReader implementation for streaming TFE plan/apply logs.""" - -from __future__ import annotations - -import asyncio -import math -import time -from collections.abc import Callable -from typing import Any -from urllib.parse import urlparse, urlunparse - -import httpx - -from .models.plan import PlanStatus - - -class LogReader: - """ - LogReader implements io.Reader for streaming logs. - - This class exactly mirrors the Go LogReader implementation: - - Implements Read() method that works with bytes (like io.Reader) - - Handles context cancellation with select-like behavior - - STX/ETX control character handling at byte level - - Exponential backoff with exact same algorithm as Go - - Proper HTTP error handling via checkResponseCode equivalent - - Usage: - # For streaming logs byte by byte (like Go's io.Reader) - log_reader = LogReader(transport, log_url, done_func, context) - buffer = bytearray(4096) - while True: - n, err = await log_reader.read(buffer) - if n > 0: - print(buffer[:n].decode('utf-8', errors='ignore'), end='') - if err: - break - - # For reading all logs at once - all_logs = await log_reader.read_all() - """ - - def __init__( - self, - transport: Any, - log_url: str, - done_func: Callable[[], tuple[bool, Exception | None]], - context: Any = None, - ) -> None: - """ - Initialize LogReader. - - Args: - transport: HTTP transport for internal requests - log_url: URL to fetch logs from - done_func: Function that returns (done, error) tuple - context: Optional context for cancellation - """ - self.transport = transport - self.done_func = done_func - self.context = context - - # State tracking (exactly like Go implementation) - self.offset = 0 - self.reads = 0 - self.start_of_text = False - self.end_of_text = False - - # Parse URL for validation (like Go url.Parse) - self.parsed_url = urlparse(log_url) - if not self.parsed_url.scheme or not self.parsed_url.netloc: - raise ValueError(f"Invalid log URL: {log_url}") - - async def read(self, buffer: bytearray | bytes) -> tuple[int, Exception | None]: - """ - Read data into the provided buffer (io.Reader equivalent). - - This method exactly mirrors the Go LogReader.Read() behavior: - - Returns (bytes_read, error) tuple like Go - - Handles context cancellation with select-like behavior - - Implements exponential backoff - - Processes STX/ETX control characters at byte level - - Args: - buffer: Buffer to read data into - - Returns: - Tuple of (bytes_read, error). Returns (0, EOFError) when done, - (0, NoProgressError) for no progress, or (n, None) for n bytes read. - """ - # First attempt to read (like Go: if written, err := r.read(l)) - written, err = await self._read(buffer) - if err is not None and not isinstance(err, NoProgressError): - return written, err - - # Loop until we get data, context is cancelled, or run is finished - # This exactly mirrors the Go implementation's for loop - self.reads = 1 - while True: - try: - # Context cancellation check (equivalent to Go's select case <-r.ctx.Done()) - if self.context and hasattr(self.context, 'cancelled') and self.context.cancelled(): - return 0, self.context.exception() - - # Wait with backoff (equivalent to Go's case <-time.After(backoff(...))) - await asyncio.sleep(self._backoff(500, 2000, self.reads) / 1000.0) - - written, err = await self._read(buffer) - if err is not None and not isinstance(err, NoProgressError): - return written, err - - self.reads += 1 - except asyncio.CancelledError as e: - return 0, e - - async def _read(self, buffer: bytearray | bytes) -> tuple[int, Exception | None]: - """ - Internal read method that handles HTTP requests and data processing. - - This method exactly mirrors the Go LogReader.read() method. - - Args: - buffer: Buffer to read data into - - Returns: - Tuple of (bytes_read, error) - """ - # Update the query string (exactly like Go: r.logURL.RawQuery = fmt.Sprintf(...)) - url_parts = list(self.parsed_url) - query = f"limit={len(buffer)}&offset={self.offset}" - url_parts[4] = query # query component - chunk_url = urlunparse(url_parts) - - try: - # Create a new request (like Go: req, err := http.NewRequest("GET", ...)) - # Use the transport to make the request (like Go client.http.HTTPClient.Do) - response = await self.transport.arequest("GET", chunk_url) - - # Read the response body as bytes (like Go: written, err := resp.Body.Read(l)) - chunk_data = response.content - - except Exception as e: - return 0, e - - if not chunk_data: - return 0, NoProgressError() - - written = len(chunk_data) - - # Handle STX/ETX control characters at byte level (exactly like Go) - if written > 0: - # Check for STX (Start of Text) ASCII control marker - if not self.start_of_text and chunk_data[0] == 2: - self.start_of_text = True - - # Remove the STX marker from the received chunk (like Go copy operation) - chunk_data = chunk_data[1:] - self.offset += 1 - written -= 1 - - # Return early if we only received the STX marker - if written == 0: - return 0, NoProgressError() - - # If we found an STX ASCII control character, start looking for ETX - if self.start_of_text and chunk_data[-1] == 3: - self.end_of_text = True - - # Remove the ETX marker from the received chunk - chunk_data = chunk_data[:-1] - self.offset += 1 - written -= 1 - - # Copy data to buffer - if written > 0: - buffer[:written] = chunk_data[:written] - - # Check if we need to continue the loop (exactly like Go logic) - if written != 0: - # Update the offset for the next read - self.offset += written - return written, None - - # Check completion conditions (exactly like Go implementation) - if ( - (self.start_of_text and self.end_of_text) or # The logstream finished without issues - (self.start_of_text and self.reads % 10 == 0) or # The logstream terminated unexpectedly - (not self.start_of_text and self.reads > 1) # The logstream doesn't support STX/ETX - ): - # Check if operation is done (like Go: done, err := r.done()) - try: - done, err = self.done_func() - if err: - return 0, err - if done: - return 0, EOFError("End of log stream") - except Exception as e: - return 0, e - - return 0, NoProgressError() - - async def read_all(self, chunk_size: int = 4096) -> str: - """ - Read all available logs as a single string. - - Args: - chunk_size: Size of each chunk to read - - Returns: - Complete log content as string - """ - buffer = bytearray(chunk_size) - result = bytearray() - - while True: - n, err = await self.read(buffer) - if n > 0: - result.extend(buffer[:n]) - if err: - if isinstance(err, EOFError): - break - raise err - - return result.decode('utf-8', errors='ignore') - - def _backoff(self, minimum: float, maximum: float, iter: int) -> float: - """ - Calculate exponential backoff duration (exactly like Go implementation). - - Args: - minimum: Minimum backoff in milliseconds - maximum: Maximum backoff in milliseconds - iter: Current iteration number - - Returns: - Backoff duration in milliseconds - """ - backoff = math.pow(2, iter / 5) * minimum - if backoff > maximum: - backoff = maximum - return backoff - - -class NoProgressError(Exception): - """Error indicating no progress was made (equivalent to Go's io.ErrNoProgress).""" - pass \ No newline at end of file diff --git a/src/tfe/resources/apply.py b/src/tfe/resources/apply.py index c87c9ea9..c13a464b 100644 --- a/src/tfe/resources/apply.py +++ b/src/tfe/resources/apply.py @@ -1,13 +1,10 @@ from __future__ import annotations -from collections.abc import Callable - from ..errors import InvalidApplyIDError -from ..log_reader import LogReader from ..models.apply import ( Apply, ) -from ..utils import valid_string_id +from ..utils import valid_string_id, validate_log_url from ._base import _Service @@ -28,80 +25,24 @@ def read(self, apply_id: str) -> Apply: **{k.replace("-", "_"): v for k, v in attr.items()}, ) - def logs_reader(self, apply_id: str) -> LogReader: - """Get a LogReader for streaming logs from a specific apply. - - This method follows the Go LogReader pattern, providing: - - Chunked reading with offset/limit parameters - - Status checking to determine when logs are complete - - STX/ETX control character handling - - Exponential backoff for retries - - Returns: - LogReader instance for streaming logs - - Usage: - # Stream logs chunk by chunk (async) - log_reader = applies.logs_reader("apply-123") - buffer = bytearray(4096) - while True: - n, err = await log_reader.read(buffer) - if n > 0: - print(buffer[:n].decode('utf-8', errors='ignore'), end='') - if err: - break - - # Or get all logs at once (async) - all_logs = await log_reader.read_all() - - # Or use the convenience logs() method for synchronous access - all_logs = applies.logs("apply-123") - """ + def logs(self, apply_id: str) -> str: + """Get logs for a specific apply""" # Validate apply ID if not valid_string_id(apply_id): raise InvalidApplyIDError() # Get the apply and validate log URL apply = self.read(apply_id) - self._validate_log_url(apply.log_read_url, apply_id) - - # Create done function for status checking - done_func = lambda: self._done(apply_id) - - # Return LogReader configured with transport, URL, and done function - return LogReader( - transport=self.t, - log_url=apply.log_read_url, - done_func=done_func, - ) - - def logs(self, apply_id: str) -> str: - """Get all logs for a specific apply as a string. - - This is a convenience method that uses logs_reader() internally - to fetch all logs at once. For streaming logs, use logs_reader() instead. + if not apply.log_read_url: + raise ValueError(f"Apply {apply_id} does not have a log URL") - Args: - apply_id: Apply ID to get logs for + validate_log_url(apply.log_read_url) - Returns: - Complete log content as string - """ - import asyncio - - log_reader = self.logs_reader(apply_id) - return asyncio.run(log_reader.read_all()) + # Placeholder implementation - in future this would stream logs + return "" def _done(self, apply_id: str) -> tuple[bool, Exception | None]: - """ - Check if an apply is in a terminal state. - - Args: - apply_id: Apply ID to check - - Returns: - Tuple of (is_complete, error) - """ + """Check if an apply is in a terminal state.""" try: apply_obj = self.read(apply_id) terminal_states = {"canceled", "errored", "finished", "unreachable"} @@ -109,26 +50,3 @@ def _done(self, apply_id: str) -> tuple[bool, Exception | None]: return is_complete, None except Exception as e: return False, e - - def _validate_log_url(self, log_url: str, resource_id: str) -> None: - """ - Validate that a log URL exists and has the correct format. - - Args: - log_url: The log URL to validate - resource_id: The resource ID for error messages - - Raises: - ValueError: If the log URL is invalid or empty - """ - if not log_url: - raise ValueError(f"Apply {resource_id} does not have a log URL") - - from urllib.parse import urlparse - - try: - parsed_url = urlparse(log_url) - if not parsed_url.scheme or not parsed_url.netloc: - raise ValueError(f"Invalid log URL format: {log_url}") - except Exception as e: - raise ValueError(f"Invalid log URL: {log_url}") from e diff --git a/src/tfe/resources/plan.py b/src/tfe/resources/plan.py index 948d7bf4..332f2c59 100644 --- a/src/tfe/resources/plan.py +++ b/src/tfe/resources/plan.py @@ -1,15 +1,13 @@ from __future__ import annotations -from collections.abc import Callable from typing import Any from ..errors import InvalidPlanIDError -from ..log_reader import LogReader from ..models.plan import ( Plan, PlanStatus, ) -from ..utils import valid_string_id +from ..utils import valid_string_id, validate_log_url from ._base import _Service @@ -30,34 +28,14 @@ def read(self, plan_id: str) -> Plan: **{k.replace("-", "_"): v for k, v in attr.items()}, ) - def logs_reader(self, plan_id: str) -> LogReader: - """Get a LogReader for streaming logs from a specific plan. + def logs(self, plan_id: str) -> str: + """Get logs for a specific plan. - This method follows the Go LogReader pattern, providing: - - Chunked reading with offset/limit parameters - - Status checking to determine when logs are complete - - STX/ETX control character handling - - Exponential backoff for retries + Args: + plan_id: Plan ID to get logs for Returns: - LogReader instance for streaming logs - - Usage: - # Stream logs chunk by chunk (async) - log_reader = plans.logs_reader("plan-123") - buffer = bytearray(4096) - while True: - n, err = await log_reader.read(buffer) - if n > 0: - print(buffer[:n].decode('utf-8', errors='ignore'), end='') - if err: - break - - # Or get all logs at once (async) - all_logs = await log_reader.read_all() - - # Or use the convenience logs() method for synchronous access - all_logs = plans.logs("plan-123") + Log content as string (placeholder implementation) """ # Validate plan ID if not valid_string_id(plan_id): @@ -65,34 +43,13 @@ def logs_reader(self, plan_id: str) -> LogReader: # Get the plan and validate log URL plan = self.read(plan_id) - self._validate_log_url(plan.log_read_url, plan_id) - - # Create done function for status checking - done_func = lambda: self._done(plan_id) - - # Return LogReader configured with transport, URL, and done function - return LogReader( - transport=self.t, - log_url=plan.log_read_url, - done_func=done_func, - ) - - def logs(self, plan_id: str) -> str: - """Get all logs for a specific plan as a string. + if not plan.log_read_url: + raise ValueError(f"Plan {plan_id} does not have a log URL") - This is a convenience method that uses logs_reader() internally - to fetch all logs at once. For streaming logs, use logs_reader() instead. + validate_log_url(plan.log_read_url) - Args: - plan_id: Plan ID to get logs for - - Returns: - Complete log content as string - """ - import asyncio - - log_reader = self.logs_reader(plan_id) - return asyncio.run(log_reader.read_all()) + # Placeholder implementation - in future this would stream logs + return "" def read_json_output(self, plan_id: str) -> dict[str, Any]: """Get the JSON execution plan for a specific plan by its ID. @@ -119,15 +76,7 @@ def read_json_output(self, plan_id: str) -> dict[str, Any]: return {"data": json_data} def _done(self, plan_id: str) -> bool: - """ - Create a done function for plan log reading. - - Args: - plan_id: Plan ID to check - - Returns: - Function that returns boolean - """ + """Create a done function for plan log reading.""" plan = self.read(plan_id) terminal_states = { PlanStatus.PLAN_CANCELED, @@ -136,26 +85,3 @@ def _done(self, plan_id: str) -> bool: PlanStatus.PLAN_UNREACHABLE, } return plan.status in terminal_states - - def _validate_log_url(self, log_url: str, resource_id: str) -> None: - """ - Validate that a log URL exists and has the correct format. - - Args: - log_url: The log URL to validate - resource_id: The resource ID for error messages - - Raises: - ValueError: If the log URL is invalid or empty - """ - if not log_url: - raise ValueError(f"Plan {resource_id} does not have a log URL") - - from urllib.parse import urlparse - - try: - parsed_url = urlparse(log_url) - if not parsed_url.scheme or not parsed_url.netloc: - raise ValueError(f"Invalid log URL format: {log_url}") - except Exception as e: - raise ValueError(f"Invalid log URL: {log_url}") from e diff --git a/src/tfe/utils.py b/src/tfe/utils.py index d8be76d8..875408a4 100644 --- a/src/tfe/utils.py +++ b/src/tfe/utils.py @@ -4,6 +4,7 @@ import time from collections.abc import Callable, Mapping from typing import Any +from urllib.parse import urlparse from .errors import ( InvalidNameError, @@ -197,3 +198,13 @@ def validate_workspace_update_options(options: WorkspaceUpdateOptions) -> None: if options.file_triggers_enabled is not None and options.file_triggers_enabled: raise UnsupportedBothTagsRegexAndFileTriggersEnabledError() + + +def validate_log_url(log_url: str) -> None: + """Validate a log URL for Terraform resources.""" + try: + parsed_url = urlparse(log_url) + if not parsed_url.scheme or not parsed_url.netloc: + raise ValueError(f"Invalid log URL format: {log_url}") + except Exception as e: + raise ValueError(f"Invalid log URL: {log_url}") from e diff --git a/tests/units/test_apply.py b/tests/units/test_apply.py new file mode 100644 index 00000000..7553873d --- /dev/null +++ b/tests/units/test_apply.py @@ -0,0 +1,96 @@ +"""Test cases for Apply resources.""" + +from __future__ import annotations + +import unittest +from unittest.mock import MagicMock, patch + +from tfe.errors import InvalidApplyIDError +from tfe.models.apply import Apply +from tfe.resources.apply import Applies + + +class TestApplies(unittest.TestCase): + def setUp(self): + self.mock_transport = MagicMock() + self.applies = Applies(self.mock_transport) + + def test_applies_service_init(self): + """Test that the applies service initializes correctly.""" + assert self.applies.t == self.mock_transport + + def test_read_apply_validation_errors(self): + """Test apply read with invalid IDs.""" + with self.assertRaises(InvalidApplyIDError): + self.applies.read("") + + with self.assertRaises(InvalidApplyIDError): + self.applies.read("a") + + def test_read_apply_success(self): + """Test successful apply read.""" + # Mock the transport response + mock_response = MagicMock() + mock_response.json.return_value = { + "data": { + "id": "apply-123", + "attributes": { + "status": "finished", + "resource-additions": 2, + "resource-changes": 1, + "resource-destructions": 0, + "resource-imports": 0, + "created-at": "2023-01-01T00:00:00Z", + "log-read-url": "https://app.terraform.io/api/v2/applies/apply-123/logs", + "status-timestamps": {}, + }, + } + } + self.mock_transport.request.return_value = mock_response + + # Call the method + result = self.applies.read("apply-123") + + # Verify the request was made correctly + self.mock_transport.request.assert_called_once_with( + "GET", "/api/v2/applies/apply-123" + ) + + # Verify the result + assert isinstance(result, Apply) + assert result.id == "apply-123" + assert result.status == "finished" + assert result.resource_additions == 2 + assert result.resource_changes == 1 + assert result.resource_destructions == 0 + assert result.resource_imports == 0 + + @patch("tfe.resources.apply.Applies.read") + def test_logs_success(self, mock_read): + """Test successful logs retrieval.""" + # Mock the apply object + mock_apply = MagicMock() + mock_apply.log_read_url = ( + "https://app.terraform.io/api/v2/applies/apply-123/logs" + ) + mock_read.return_value = mock_apply + + # Call the method + result = self.applies.logs("apply-123") + + # Verify it returns empty string (placeholder implementation) + assert result == "" + + @patch("tfe.resources.apply.Applies.read") + def test_logs_no_url_error(self, mock_read): + """Test logs method when apply has no log URL.""" + # Mock apply with no log URL + mock_apply = MagicMock() + mock_apply.log_read_url = None + mock_read.return_value = mock_apply + + # Call the method and expect error + with self.assertRaises(ValueError) as cm: + self.applies.logs("apply-123") + + assert "Apply apply-123 does not have a log URL" in str(cm.exception) diff --git a/tests/units/test_plan.py b/tests/units/test_plan.py index 62666eb7..0d933cfd 100644 --- a/tests/units/test_plan.py +++ b/tests/units/test_plan.py @@ -89,37 +89,13 @@ def test_logs_success(self, plans_service): mock_plan.log_read_url = "https://example.com/logs/plan-123" with patch.object(plans_service, "read", return_value=mock_plan): - # Mock httpx.Client and its response - with patch("httpx.Client") as mock_httpx_client: - mock_context_manager = Mock() - mock_httpx_client.return_value.__enter__ = Mock( - return_value=mock_context_manager - ) - mock_httpx_client.return_value.__exit__ = Mock(return_value=None) - - mock_response = Mock() - mock_response.text = "Terraform will perform the following actions:\n\n + resource will be created" - mock_response.raise_for_status = Mock() # Don't raise any exceptions - - mock_context_manager.get.return_value = mock_response - - result = plans_service.logs("plan-123") - - # Verify read was called first - plans_service.read.assert_called_once_with("plan-123") - - # Verify httpx client was used correctly - mock_httpx_client.assert_called_once_with(timeout=30.0) - mock_context_manager.get.assert_called_once_with( - "https://example.com/logs/plan-123" - ) - mock_response.raise_for_status.assert_called_once() - - # Verify log content - assert ( - result - == "Terraform will perform the following actions:\n\n + resource will be created" - ) + result = plans_service.logs("plan-123") + + # Verify read was called first + plans_service.read.assert_called_once_with("plan-123") + + # The current implementation returns empty string as placeholder + assert result == "" def test_read_json_output_success(self, plans_service): """Test successful read_json_output operation.""" From 1985e6e42bd7e5b7f47fa36ebbae4fd04c9360b1 Mon Sep 17 00:00:00 2001 From: Sivaselvan32 Date: Thu, 25 Sep 2025 12:18:54 +0530 Subject: [PATCH 3/3] Resolved Conflicts --- src/tfe/errors.py | 25 +++++++++++++++++++++++++ src/tfe/models/apply.py | 16 +++++++++------- src/tfe/models/plan.py | 34 ++++++++++++++++++---------------- 3 files changed, 52 insertions(+), 23 deletions(-) diff --git a/src/tfe/errors.py b/src/tfe/errors.py index 9b9b118c..18dff7bc 100644 --- a/src/tfe/errors.py +++ b/src/tfe/errors.py @@ -246,6 +246,13 @@ def __init__(self, message: str = "name is required"): super().__init__(message) +class RequiredWorkspaceError(RequiredFieldMissing): + """Raised when a required workspace field is missing.""" + + def __init__(self, message: str = "workspace is required"): + super().__init__(message) + + # Run Task errors class InvalidRunTaskIDError(InvalidValues): """Raised when an invalid run task ID is provided.""" @@ -316,6 +323,24 @@ def __init__(self, message: str = "invalid value for run trigger ID"): super().__init__(message) +# Run errors +class InvalidRunIDError(InvalidValues): + """Raised when an invalid run ID is provided.""" + + def __init__(self, message: str = "invalid value for run ID"): + super().__init__(message) + + +class TerraformVersionValidForPlanOnlyError(ValidationError): + """Raised when terraform_version is set without plan_only being true.""" + + def __init__( + self, + message: str = "setting terraform-version is only valid when plan-only is set to true", + ): + super().__init__(message) + + # Plan errors class InvalidPlanIDError(InvalidValues): """Raised when an invalid plan ID is provided.""" diff --git a/src/tfe/models/apply.py b/src/tfe/models/apply.py index 3de7e37b..abfa02bb 100644 --- a/src/tfe/models/apply.py +++ b/src/tfe/models/apply.py @@ -22,13 +22,15 @@ class Apply(BaseModel): model_config = ConfigDict(populate_by_name=True, validate_by_name=True) id: str - log_read_url: str = Field(..., alias="log-read-url") - resource_additions: int = Field(..., alias="resource-additions") - resource_changes: int = Field(..., alias="resource-changes") - resource_destructions: int = Field(..., alias="resource-destructions") - resource_imports: int = Field(..., alias="resource-imports") - status: ApplyStatus = Field(..., alias="status") - status_timestamps: ApplyStatusTimestamps = Field(..., alias="status-timestamps") + log_read_url: str | None = Field(None, alias="log-read-url") + resource_additions: int | None = Field(None, alias="resource-additions") + resource_changes: int | None = Field(None, alias="resource-changes") + resource_destructions: int | None = Field(None, alias="resource-destructions") + resource_imports: int | None = Field(None, alias="resource-imports") + status: ApplyStatus | None = Field(None, alias="status") + status_timestamps: ApplyStatusTimestamps | None = Field( + None, alias="status-timestamps" + ) class ApplyStatusTimestamps(BaseModel): diff --git a/src/tfe/models/plan.py b/src/tfe/models/plan.py index ee8ac898..2987a958 100644 --- a/src/tfe/models/plan.py +++ b/src/tfe/models/plan.py @@ -26,26 +26,28 @@ class Plan(BaseModel): model_config = ConfigDict(populate_by_name=True, validate_by_name=True) id: str - has_changes: bool = Field(..., alias="has-changes") - generated_configuration: bool = Field(..., alias="generated-configuration") - log_read_url: str = Field(..., alias="log-read-url") - resource_additions: int = Field(..., alias="resource-additions") - resource_changes: int = Field(..., alias="resource-changes") - resource_destructions: int = Field(..., alias="resource-destructions") - resource_imports: int = Field(..., alias="resource-imports") - status: PlanStatus = Field(..., alias="status") - status_timestamps: PlanStatusTimestamps = Field(..., alias="status-timestamps") + has_changes: bool | None = Field(None, alias="has-changes") + generated_configuration: bool | None = Field(None, alias="generated-configuration") + log_read_url: str | None = Field(None, alias="log-read-url") + resource_additions: int | None = Field(None, alias="resource-additions") + resource_changes: int | None = Field(None, alias="resource-changes") + resource_destructions: int | None = Field(None, alias="resource-destructions") + resource_imports: int | None = Field(None, alias="resource-imports") + status: PlanStatus | None = Field(None, alias="status") + status_timestamps: PlanStatusTimestamps | None = Field( + None, alias="status-timestamps" + ) # Relations - exports: list[PlanExport] = Field(..., alias="exports") + exports: list[PlanExport] | None = Field(None, alias="exports") class PlanStatusTimestamps(BaseModel): model_config = ConfigDict(populate_by_name=True, validate_by_name=True) - canceled_at: datetime = Field(..., alias="canceled-at") - errored_at: datetime = Field(..., alias="errored-at") - finished_at: datetime = Field(..., alias="finished-at") - force_canceled_at: datetime = Field(..., alias="force-canceled-at") - queued_at: datetime = Field(..., alias="queued-at") - started_at: datetime = Field(..., alias="started-at") + canceled_at: datetime | None = Field(None, alias="canceled-at") + errored_at: datetime | None = Field(None, alias="errored-at") + finished_at: datetime | None = Field(None, alias="finished-at") + force_canceled_at: datetime | None = Field(None, alias="force-canceled-at") + queued_at: datetime | None = Field(None, alias="queued-at") + started_at: datetime | None = Field(None, alias="started-at")