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/client.py b/src/tfe/client.py index 93e1254c..7f871fe2 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 @@ -33,6 +35,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 b78b8065..18dff7bc 100644 --- a/src/tfe/errors.py +++ b/src/tfe/errors.py @@ -339,3 +339,19 @@ def __init__( 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.""" + + 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/models/apply.py b/src/tfe/models/apply.py index 16d04487..abfa02bb 100644 --- a/src/tfe/models/apply.py +++ b/src/tfe/models/apply.py @@ -6,35 +6,39 @@ 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 | None = Field(None, alias="log-read-url") - raiseesource_additions: int = Field(..., alias="resource-additions") - resource_changes: int = Field(..., alias="resource-changes") - resource_destructions: int = Field(..., alias="resource-destructions") - status: ApplyStatus = Field(..., alias="status") - status_timestamps: ApplyStatusTimestamps = Field(..., alias="status-timestamps") - - -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" + 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): 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") diff --git a/src/tfe/models/plan.py b/src/tfe/models/plan.py index a964efdb..2987a958 100644 --- a/src/tfe/models/plan.py +++ b/src/tfe/models/plan.py @@ -5,43 +5,49 @@ from pydantic import BaseModel, ConfigDict, Field +from ..models.plan_export import PlanExport + class PlanStatus(str, Enum): - 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" + """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") + 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") 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..c13a464b --- /dev/null +++ b/src/tfe/resources/apply.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +from ..errors import InvalidApplyIDError +from ..models.apply import ( + Apply, +) +from ..utils import valid_string_id, validate_log_url +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(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) + if not apply.log_read_url: + raise ValueError(f"Apply {apply_id} does not have a log URL") + + validate_log_url(apply.log_read_url) + + # 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.""" + 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 diff --git a/src/tfe/resources/plan.py b/src/tfe/resources/plan.py new file mode 100644 index 00000000..332f2c59 --- /dev/null +++ b/src/tfe/resources/plan.py @@ -0,0 +1,87 @@ +from __future__ import annotations + +from typing import Any + +from ..errors import InvalidPlanIDError +from ..models.plan import ( + Plan, + PlanStatus, +) +from ..utils import valid_string_id, validate_log_url +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(self, plan_id: str) -> str: + """Get logs for a specific plan. + + Args: + plan_id: Plan ID to get logs for + + Returns: + Log content as string (placeholder implementation) + """ + # Validate plan ID + if not valid_string_id(plan_id): + raise InvalidPlanIDError() + + # Get the plan and validate log URL + plan = self.read(plan_id) + if not plan.log_read_url: + raise ValueError(f"Plan {plan_id} does not have a log URL") + + validate_log_url(plan.log_read_url) + + # 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. + + 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.""" + 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 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 new file mode 100644 index 00000000..0d933cfd --- /dev/null +++ b/tests/units/test_plan.py @@ -0,0 +1,139 @@ +"""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): + 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.""" + + 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"]