diff --git a/CHANGELOG.md b/CHANGELOG.md index 4db63b0d..55e5164c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,20 @@ # Unreleased +# Released +# v0.1.5 + +* `pytfe.__version__` added in src/pytfe/init.py via importlib.metadata.version("pytfe"). This will resolve to the version from pyproject.toml. +* Updated comments, sshkey, stateversion and cost-estimate models to have id as mandatory attribute by @isivaselvan [#137](https://github.com/hashicorp/python-tfe/pull/137) +* Updated workspace resource to include additional relationship models include AgentPool, Configuration-version, Run, Variables and State-version by @isivaselvan [#138](https://github.com/hashicorp/python-tfe/pull/138) + +## Bug Fixes +* Run.read / Run.create fail with pydantic ValidationError when response has a `cost-estimate` and `comments` relationship. + +# v0.1.4 + +## Enhancements +* Standardize Notification Configuration option models on Pydantic [#132](https://github.com/hashicorp/python-tfe/pull/132) + # v0.1.3 ## Enhancements diff --git a/examples/notification_configuration.py b/examples/notification_configuration.py index 07e1b857..3e333220 100644 --- a/examples/notification_configuration.py +++ b/examples/notification_configuration.py @@ -10,13 +10,9 @@ """ import os -import sys - -# Add the src directory to the Python path so we can import the tfe module -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) from pytfe.client import TFEClient -from pytfe.models.notification_configuration import ( +from pytfe.models import ( NotificationConfigurationCreateOptions, NotificationConfigurationListOptions, NotificationConfigurationSubscribableChoice, diff --git a/examples/organization_token.py b/examples/organization_token.py new file mode 100644 index 00000000..202a2554 --- /dev/null +++ b/examples/organization_token.py @@ -0,0 +1,212 @@ +#!/usr/bin/env python3 +""" +Organization Token Operations Example + +Demonstrates usage of all 6 organization token operations: +1. create() - Create a new organization token, replacing any existing token +2. create_with_options() - Create with options like expiration date and token type +3. read() - Read the organization token +4. read_with_options() - Read with options like token type +5. delete() - Delete the organization token +6. delete_with_options() - Delete with options like token type + +Usage: +- Modify organization names as needed for your environment +- Ensure you have proper TFE credentials and organization access +- Organization tokens are used for organization-level API access + +Prerequisites: +- Set TFE_TOKEN and TFE_ADDRESS environment variables +- You need an existing organization or admin permissions to create one +- Appropriate permissions to manage organization tokens +""" + +from datetime import datetime, timedelta + +# Add the src directory to the path +##sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) +from pytfe import TFEClient, TFEConfig +from pytfe.models import ( + OrganizationTokenCreateOptions, + OrganizationTokenDeleteOptions, + OrganizationTokenReadOptions, + TokenType, +) + + +def redact_token(token_value: str | None) -> str: + """Redact token value for safe display.""" + if not token_value: + return "None" + if len(token_value) <= 8: + return f"{'*' * len(token_value)}" + # Show first 3 and last 3 characters + return f"{token_value[:3]}...{token_value[-3:]}".replace( + token_value[3:-3], "*" * (len(token_value) - 6) + ) + + +def redact_id(id_value: str | None) -> str: + """Redact ID for safe display.""" + if not id_value: + return "None" + if len(id_value) <= 6: + return f"{'*' * len(id_value)}" + # Show first 3 and last 3 characters + return f"{id_value[:3]}...{id_value[-3:]}" + + +def main(): + """Execute organization token operation examples.""" + + print("=" * 80) + print("ORGANIZATION TOKEN OPERATIONS") + print("=" * 80) + + # Initialize the TFE client + client = TFEClient(TFEConfig.from_env()) + organization_name = "prab-sandbox01" + # ===================================================== + # 1. CREATE ORGANIZATION TOKEN (BASIC) + # ===================================================== + print("\n1. create() - Create a new organization token:") + print("-" * 40) + try: + print(f"Creating token for organization: {organization_name}") + token = client.organization_tokens.create(organization_name) + + print("Token created successfully!") + print(f" Token ID: {redact_id(token.id)}") + print(f" Created At: {token.created_at}") + print(f" Description: {token.description}") + print(f" Token Value: {redact_token(token.token)}") + if token.expired_at: + print(f" Expires At: {token.expired_at}") + print() + + except Exception as e: + print(f" Error: {e}") + print() + + # ===================================================== + # 2. CREATE WITH OPTIONS (WITH EXPIRATION) + # ===================================================== + print("2. create_with_options() - Create token with expiration date:") + print("-" * 40) + try: + # Create a token that expires in 30 days + expiry_date = datetime.utcnow() + timedelta(days=30) + options = OrganizationTokenCreateOptions(expired_at=expiry_date) + + print(f"Creating organization token with expiration date: {expiry_date}") + token = client.organization_tokens.create_with_options( + organization_name, options + ) + + print("Token created with options successfully!") + print(f" Token ID: {redact_id(token.id)}") + print(f" Created At: {token.created_at}") + if token.expired_at: + print(f" Expires At: {token.expired_at}") + print() + + except Exception as e: + print(f" Error: {e}") + print() + + # ===================================================== + print("3. create_with_options() - Create audit-trails token:") + print("-" * 40) + try: + options = OrganizationTokenCreateOptions(token_type=TokenType.AUDIT_TRAILS) + + print(f"Creating audit-trails token for organization: {organization_name}") + token = client.organization_tokens.create_with_options( + organization_name, options + ) + + print(" Audit-trails token created successfully!") + print(f" Token ID: {redact_id(token.id)}") + print(f" Token Value: {redact_token(token.token)}") + print() + + except Exception as e: + print(f"Error: {e}") + print() + + # ===================================================== + print("4. read() - Read the organization token:") + print("-" * 40) + try: + print(f"Reading organization token for organization: {organization_name}") + token = client.organization_tokens.read(organization_name) + + print("Token read successfully!") + print(f" Token ID: {redact_id(token.id)}") + print(f" Created At: {token.created_at}") + print(f" Description: {token.description}") + if token.last_used_at: + print(f" Last Used At: {token.last_used_at}") + if token.expired_at: + print(f" Expires At: {token.expired_at}") + print() + + except Exception as e: + print(f" Error: {e}") + print() + + # ===================================================== + print("5. read_with_options() - Read audit-trails token:") + print("-" * 40) + try: + options = OrganizationTokenReadOptions(token_type=TokenType.AUDIT_TRAILS) + + print(f"Reading audit-trails token for organization: {organization_name}") + token = client.organization_tokens.read_with_options(organization_name, options) + + print(" Audit-trails token read successfully!") + print(f" Token ID: {redact_id(token.id)}") + print(f" Token Value: {redact_token(token.token)}") + print() + + except Exception as e: + print(f" Error: {e}") + print() + + # ===================================================== + print("6. delete() - Delete the organization token:") + print("-" * 40) + try: + print(f"Deleting organization token for organization: {organization_name}") + client.organization_tokens.delete(organization_name) + + print(" Token deleted successfully!") + print() + + except Exception as e: + print(f" Error: {e}") + print() + + # ===================================================== + print("7. delete_with_options() - Delete audit-trails token:") + print("-" * 40) + try: + options = OrganizationTokenDeleteOptions(token_type=TokenType.AUDIT_TRAILS) + + print(f"Deleting audit-trails token for organization: {organization_name}") + client.organization_tokens.delete_with_options(organization_name, options) + + print(" Audit-trails token deleted successfully!") + print() + + except Exception as e: + print(f"Error: {e}") + print() + + print("=" * 80) + print("ORGANIZATION TOKEN OPERATIONS COMPLETED") + print("=" * 80) + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml index c8986fbd..4f311fc4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "pytfe" -version = "0.1.3" +version = "0.1.5" description = "Official Python SDK for HashiCorp Terraform Cloud / Terraform Enterprise (TFE) API v2" readme = "README.md" license = { text = "MPL-2.0" } diff --git a/src/pytfe/__init__.py b/src/pytfe/__init__.py index 68cea566..9d518c48 100644 --- a/src/pytfe/__init__.py +++ b/src/pytfe/__init__.py @@ -1,8 +1,16 @@ # Copyright IBM Corp. 2025, 2026 # SPDX-License-Identifier: MPL-2.0 +from importlib.metadata import PackageNotFoundError +from importlib.metadata import version as _pkg_version + from . import errors, models from .client import TFEClient from .config import TFEConfig -__all__ = ["TFEConfig", "TFEClient", "errors", "models"] +try: + __version__ = _pkg_version("pytfe") +except PackageNotFoundError: # running from a source checkout without install + __version__ = "0.0.0+unknown" + +__all__ = ["TFEConfig", "TFEClient", "errors", "models", "__version__"] diff --git a/src/pytfe/client.py b/src/pytfe/client.py index 30b506b9..dc1972ce 100644 --- a/src/pytfe/client.py +++ b/src/pytfe/client.py @@ -13,6 +13,7 @@ from .resources.oauth_client import OAuthClients from .resources.oauth_token import OAuthTokens from .resources.organization_membership import OrganizationMemberships +from .resources.organization_token import OrganizationTokens from .resources.organizations import Organizations from .resources.plan import Plans from .resources.policy import Policies @@ -72,7 +73,7 @@ def __init__(self, config: TFEConfig | None = None): self.plans = Plans(self._transport) self.organizations = Organizations(self._transport) self.organization_memberships = OrganizationMemberships(self._transport) - + self.organization_tokens = OrganizationTokens(self._transport) self.projects = Projects(self._transport) self.variables = Variables(self._transport) self.variable_sets = VariableSets(self._transport) diff --git a/src/pytfe/models/__init__.py b/src/pytfe/models/__init__.py index 0f1435d8..ea2bbec3 100644 --- a/src/pytfe/models/__init__.py +++ b/src/pytfe/models/__init__.py @@ -59,6 +59,19 @@ DataRetentionPolicySetOptions, ) +# ── Notification Configurations ─────────────────────────────────────────────── +from .notification_configuration import ( + DeliveryResponse, + NotificationConfiguration, + NotificationConfigurationCreateOptions, + NotificationConfigurationList, + NotificationConfigurationListOptions, + NotificationConfigurationSubscribableChoice, + NotificationConfigurationUpdateOptions, + NotificationDestinationType, + NotificationTriggerType, +) + # ── OAuth ───────────────────────────────────────────────────────────────────── from .oauth_client import ( OAuthClient, @@ -96,6 +109,15 @@ OrganizationMembershipStatus, OrgMembershipIncludeOpt, ) + +# ── Organization Token ──────────────────────────────────────────────────────── +from .organization_token import ( + OrganizationToken, + OrganizationTokenCreateOptions, + OrganizationTokenDeleteOptions, + OrganizationTokenReadOptions, + TokenType, +) from .policy import ( Policy, PolicyCreateOptions, @@ -376,6 +398,16 @@ # ── Public surface ──────────────────────────────────────────────────────────── __all__ = [ + # Notification configurations + "DeliveryResponse", + "NotificationConfiguration", + "NotificationConfigurationCreateOptions", + "NotificationConfigurationList", + "NotificationConfigurationListOptions", + "NotificationConfigurationSubscribableChoice", + "NotificationConfigurationUpdateOptions", + "NotificationDestinationType", + "NotificationTriggerType", # OAuth "OAuthClient", "OAuthClientAddProjectsOptions", @@ -497,6 +529,12 @@ "OrganizationMembershipReadOptions", "OrganizationMembershipStatus", "OrgMembershipIncludeOpt", + # Organization tokens + "OrganizationToken", + "OrganizationTokenCreateOptions", + "OrganizationTokenDeleteOptions", + "OrganizationTokenReadOptions", + "TokenType", "OrganizationAccess", "Team", "TeamPermissions", diff --git a/src/pytfe/models/comment.py b/src/pytfe/models/comment.py index da2cd213..19cc25ca 100644 --- a/src/pytfe/models/comment.py +++ b/src/pytfe/models/comment.py @@ -10,4 +10,4 @@ class Comment(BaseModel): model_config = ConfigDict(populate_by_name=True, validate_by_name=True) id: str - body: str = Field(..., alias="body") + body: str = Field(default="", alias="body") diff --git a/src/pytfe/models/cost_estimate.py b/src/pytfe/models/cost_estimate.py index d1b6ff6b..4ae1c614 100644 --- a/src/pytfe/models/cost_estimate.py +++ b/src/pytfe/models/cost_estimate.py @@ -13,17 +13,17 @@ class CostEstimate(BaseModel): model_config = ConfigDict(populate_by_name=True, validate_by_name=True) id: str - delta_monthly_cost: str = Field(..., alias="delta-monthly-cost") - error_message: str = Field(..., alias="error-message") - matched_resources_count: int = Field(..., alias="matched-resources-count") - prior_monthly_cost: str = Field(..., alias="prior-monthly-cost") - proposed_monthly_cost: str = Field(..., alias="proposed-monthly-cost") - resources_count: int = Field(..., alias="resources-count") - status: CostEstimateStatus = Field(..., alias="status") - status_timestamps: CostEstimateStatusTimestamps = Field( - ..., alias="status-timestamps" + delta_monthly_cost: str = Field(default="", alias="delta-monthly-cost") + error_message: str = Field(default="", alias="error-message") + matched_resources_count: int = Field(default=0, alias="matched-resources-count") + prior_monthly_cost: str = Field(default="", alias="prior-monthly-cost") + proposed_monthly_cost: str = Field(default="", alias="proposed-monthly-cost") + resources_count: int = Field(default=0, alias="resources-count") + status: CostEstimateStatus | None = Field(default=None, alias="status") + status_timestamps: CostEstimateStatusTimestamps | None = Field( + default=None, alias="status-timestamps" ) - unmatched_resources_count: int = Field(..., alias="unmatched-resources-count") + unmatched_resources_count: int = Field(default=0, alias="unmatched-resources-count") class CostEstimateStatus(str, Enum): diff --git a/src/pytfe/models/notification_configuration.py b/src/pytfe/models/notification_configuration.py index e1af877a..c9b2934a 100644 --- a/src/pytfe/models/notification_configuration.py +++ b/src/pytfe/models/notification_configuration.py @@ -9,10 +9,13 @@ from __future__ import annotations +from collections.abc import Iterator from datetime import datetime from enum import Enum from typing import Any +from pydantic import BaseModel, ConfigDict, Field, field_validator + class NotificationTriggerType(Enum): """Represents the different TFE notifications that can be sent as a run's progress transitions between different states.""" @@ -47,209 +50,143 @@ class NotificationDestinationType(Enum): MICROSOFT_TEAMS = "microsoft-teams" -class DeliveryResponse: +class DeliveryResponse(BaseModel): """Represents a notification configuration delivery response.""" - # Type annotations for instance attributes - body: str - code: str - headers: dict[str, Any] - sent_at: datetime | None - successful: str - url: str - - def __init__(self, data: dict[str, Any]): - self.body = data.get("body", "") - self.code = data.get("code", "") - self.headers = data.get("headers", {}) - self.sent_at = self._parse_datetime(data.get("sent-at")) - self.successful = data.get("successful", "") - self.url = data.get("url", "") - - def _parse_datetime(self, date_str: str | None) -> datetime | None: - """Parse ISO 8601 datetime string.""" - if not date_str: - return None - try: - return datetime.fromisoformat(date_str.replace("Z", "+00:00")) - except (ValueError, AttributeError): - return None + model_config = ConfigDict(populate_by_name=True) - def __repr__(self) -> str: - return f"DeliveryResponse(url='{self.url}', code='{self.code}', successful='{self.successful}')" + body: str | None = None + code: str | None = None + headers: dict[str, Any] | None = Field(default_factory=dict) + sent_at: datetime | None = Field(default=None, alias="sent-at") + successful: str | None = None + url: str | None = None + def __init__(self, data: dict[str, Any] | None = None, /, **kwargs: Any) -> None: + if data is not None: + super().__init__(**{**data, **kwargs}) + else: + super().__init__(**kwargs) -class NotificationConfigurationSubscribableChoice: - """Choice type struct that represents the possible values within a polymorphic relation.""" - # Type annotations for instance attributes - team: Any | None - workspace: Any | None +class NotificationConfigurationSubscribableChoice(BaseModel): + """Choice type struct that represents the possible values within a polymorphic relation.""" - def __init__(self, team: Any | None = None, workspace: Any | None = None): - self.team = team - self.workspace = workspace + model_config = ConfigDict(arbitrary_types_allowed=True) - def __repr__(self) -> str: - if self.team: - return f"NotificationConfigurationSubscribableChoice(team={self.team})" - elif self.workspace: - return f"NotificationConfigurationSubscribableChoice(workspace={self.workspace})" - return "NotificationConfigurationSubscribableChoice()" + team: Any | None = None + workspace: Any | None = None -class NotificationConfiguration: +class NotificationConfiguration(BaseModel): """Represents a Notification Configuration.""" - # Type annotations for instance attributes - id: str | None - created_at: datetime | None - updated_at: datetime | None - destination_type: str | None - enabled: bool - name: str - token: str - url: str - triggers: list[NotificationTriggerType] - delivery_responses: list[Any] - email_addresses: list[str] - email_users: list[Any] - subscribable: Any - subscribable_choice: Any | None - - def __init__(self, data: dict[str, Any]): - self.id = data.get("id") - self.created_at = self._parse_datetime(data.get("created-at")) - self.updated_at = self._parse_datetime(data.get("updated-at")) - - # Core attributes - self.destination_type = data.get("destination-type") - self.enabled = data.get("enabled", False) - self.name = data.get("name", "") - self.token = data.get("token", "") - self.url = data.get("url", "") - - # Triggers - convert from strings to enum values - self.triggers = self._parse_triggers(data.get("triggers", [])) - - # Delivery responses - delivery_responses_data = data.get("delivery-responses", []) - self.delivery_responses = [ - DeliveryResponse(dr) for dr in delivery_responses_data - ] - - # Email configuration - self.email_addresses = data.get("email-addresses", []) - self.email_users = data.get("email-users", []) - - # Relationships - using polymorphic relation pattern - self.subscribable = data.get( - "subscribable" - ) # Deprecated but maintained for compatibility - self.subscribable_choice = self._parse_subscribable_choice( - data.get("subscribable-choice") - ) - - def _parse_datetime(self, date_str: str | None) -> datetime | None: - """Parse ISO 8601 datetime string.""" - if not date_str: - return None - try: - return datetime.fromisoformat(date_str.replace("Z", "+00:00")) - except (ValueError, AttributeError): - return None - - def _parse_triggers(self, triggers: list[str]) -> list[NotificationTriggerType]: - """Parse trigger strings to enum values.""" - parsed_triggers = [] - for trigger in triggers: + model_config = ConfigDict(populate_by_name=True, arbitrary_types_allowed=True) + + id: str | None = None + created_at: datetime | None = Field(default=None, alias="created-at") + updated_at: datetime | None = Field(default=None, alias="updated-at") + destination_type: str | None = Field(default=None, alias="destination-type") + enabled: bool = False + name: str | None = None + token: str | None = None + url: str | None = None + triggers: list[NotificationTriggerType] = Field(default_factory=list) + delivery_responses: list[DeliveryResponse] = Field( + default_factory=list, alias="delivery-responses" + ) + email_addresses: list[str] = Field(default_factory=list, alias="email-addresses") + email_users: list[Any] = Field(default_factory=list, alias="email-users") + subscribable: Any = None + subscribable_choice: NotificationConfigurationSubscribableChoice | None = Field( + default=None, alias="subscribable-choice" + ) + + @field_validator( + "delivery_responses", + "email_addresses", + "email_users", + mode="before", + ) + @classmethod + def _none_to_empty_list(cls, value: Any) -> Any: + return [] if value is None else value + + @field_validator("triggers", mode="before") + @classmethod + def _coerce_triggers(cls, value: Any) -> list[NotificationTriggerType]: + if not value: + return [] + parsed: list[NotificationTriggerType] = [] + for trigger in value: + if isinstance(trigger, NotificationTriggerType): + parsed.append(trigger) + continue try: - parsed_triggers.append(NotificationTriggerType(trigger)) - except ValueError: - # If trigger is not in enum, keep as string for backwards compatibility + parsed.append(NotificationTriggerType(trigger)) + except (ValueError, TypeError): + # Silently drop unknown triggers for backwards compatibility pass - return parsed_triggers - - def _parse_subscribable_choice( - self, choice_data: dict[str, Any] | None - ) -> NotificationConfigurationSubscribableChoice | None: - """Parse subscribable choice data.""" - if not choice_data: - return None - - team = choice_data.get("team") - workspace = choice_data.get("workspace") - return NotificationConfigurationSubscribableChoice( - team=team, workspace=workspace - ) - - def __repr__(self) -> str: - return f"NotificationConfiguration(id='{self.id}', name='{self.name}', enabled={self.enabled})" + return parsed + + def __init__(self, data: dict[str, Any] | None = None, /, **kwargs: Any) -> None: + if data is not None: + super().__init__(**{**data, **kwargs}) + else: + super().__init__(**kwargs) + + +def _serialize_triggers( + triggers: list[NotificationTriggerType | str], +) -> list[str]: + """Serialize trigger enums or raw strings to their wire value.""" + return [t.value if isinstance(t, NotificationTriggerType) else t for t in triggers] + + +def _validate_triggers( + triggers: list[NotificationTriggerType | str], +) -> list[str]: + """Collect errors for any non-enum, non-known-string trigger entries.""" + errors: list[str] = [] + for trigger in triggers: + if isinstance(trigger, NotificationTriggerType): + continue + try: + NotificationTriggerType(trigger) + except ValueError: + errors.append(f"Invalid trigger type: {trigger}") + return errors -class NotificationConfigurationListOptions: +class NotificationConfigurationListOptions(BaseModel): """Represents the options for listing notification configurations.""" - # Type annotations for instance attributes - page_size: int | None - subscribable_choice: NotificationConfigurationSubscribableChoice | None + model_config = ConfigDict(populate_by_name=True, arbitrary_types_allowed=True) - def __init__( - self, - page_size: int | None = None, - subscribable_choice: NotificationConfigurationSubscribableChoice | None = None, - ): - self.page_size = page_size - self.subscribable_choice = subscribable_choice + page_size: int | None = Field(default=None, alias="page[size]") + subscribable_choice: NotificationConfigurationSubscribableChoice | None = Field( + default=None, exclude=True + ) def to_dict(self) -> dict[str, Any]: """Convert to dictionary for API requests.""" - params = {} - - if self.page_size is not None: - params["page[size]"] = self.page_size + return self.model_dump(by_alias=True, exclude_none=True) - return params - -class NotificationConfigurationCreateOptions: +class NotificationConfigurationCreateOptions(BaseModel): """Represents the options for creating a new notification configuration.""" - # Type annotations for instance attributes + model_config = ConfigDict(populate_by_name=True, arbitrary_types_allowed=True) + destination_type: NotificationDestinationType enabled: bool name: str - token: str | None - triggers: list[NotificationTriggerType] - url: str | None - email_addresses: list[str] - email_users: list[Any] - subscribable_choice: NotificationConfigurationSubscribableChoice | None - - def __init__( - self, - destination_type: NotificationDestinationType, - enabled: bool, - name: str, - token: str | None = None, - triggers: list[NotificationTriggerType] | None = None, - url: str | None = None, - email_addresses: list[str] | None = None, - email_users: list[Any] | None = None, - subscribable_choice: NotificationConfigurationSubscribableChoice | None = None, - ): - # Required fields - self.destination_type = destination_type - self.enabled = enabled - self.name = name - - # Optional fields - self.token = token - self.triggers = triggers or [] - self.url = url - self.email_addresses = email_addresses or [] - self.email_users = email_users or [] - self.subscribable_choice = subscribable_choice + token: str | None = None + triggers: list[NotificationTriggerType | str] = Field(default_factory=list) + url: str | None = None + email_addresses: list[str] = Field(default_factory=list) + email_users: list[Any] = Field(default_factory=list) + subscribable_choice: NotificationConfigurationSubscribableChoice | None = None def to_dict(self) -> dict[str, Any]: """Convert to dictionary for API requests.""" @@ -262,14 +199,11 @@ def to_dict(self) -> dict[str, Any]: }, } - # Add optional attributes if self.token is not None: data["attributes"]["token"] = self.token if self.triggers: - data["attributes"]["triggers"] = [ - trigger.value for trigger in self.triggers - ] + data["attributes"]["triggers"] = _serialize_triggers(self.triggers) if self.url is not None: data["attributes"]["url"] = self.url @@ -277,84 +211,58 @@ def to_dict(self) -> dict[str, Any]: if self.email_addresses: data["attributes"]["email-addresses"] = self.email_addresses - # Handle relationships if self.email_users: - data["relationships"] = data.get("relationships", {}) - data["relationships"]["users"] = { - "data": [ - { - "type": "users", - "id": user.id if hasattr(user, "id") else str(user), - } - for user in self.email_users - ] + data["relationships"] = { + "users": { + "data": [ + { + "type": "users", + "id": user.id if hasattr(user, "id") else str(user), + } + for user in self.email_users + ] + } } return data - def validate(self) -> list[str]: + def validate(self) -> list[str]: # type: ignore[override] """Validate the create options and return any errors.""" - errors = [] + errors: list[str] = [] - # Required field validation if not self.name or not self.name.strip(): errors.append("Name is required") - if not isinstance(self.enabled, bool): - errors.append("Enabled must be a boolean") # type: ignore[unreachable] - - # URL validation for certain destination types - if self.destination_type in [ + if self.destination_type in ( NotificationDestinationType.GENERIC, NotificationDestinationType.SLACK, NotificationDestinationType.MICROSOFT_TEAMS, - ]: + ): if not self.url: errors.append("URL is required for this destination type") - # Trigger validation - for trigger in self.triggers: - if not isinstance(trigger, NotificationTriggerType): - errors.append(f"Invalid trigger type: {trigger}") # type: ignore[unreachable] + errors.extend(_validate_triggers(self.triggers)) return errors -class NotificationConfigurationUpdateOptions: +class NotificationConfigurationUpdateOptions(BaseModel): """Represents the options for updating an existing notification configuration.""" - # Type annotations for instance attributes - enabled: bool | None - name: str | None - token: str | None - triggers: list[NotificationTriggerType] | None - url: str | None - email_addresses: list[str] | None - email_users: list[Any] | None - - def __init__( - self, - enabled: bool | None = None, - name: str | None = None, - token: str | None = None, - triggers: list[NotificationTriggerType] | None = None, - url: str | None = None, - email_addresses: list[str] | None = None, - email_users: list[Any] | None = None, - ): - self.enabled = enabled - self.name = name - self.token = token - self.triggers = triggers - self.url = url - self.email_addresses = email_addresses - self.email_users = email_users + model_config = ConfigDict(populate_by_name=True, arbitrary_types_allowed=True) + + enabled: bool | None = None + name: str | None = None + token: str | None = None + triggers: list[NotificationTriggerType | str] | None = None + url: str | None = None + email_addresses: list[str] | None = None + email_users: list[Any] | None = None def to_dict(self) -> dict[str, Any]: """Convert to dictionary for API requests.""" data: dict[str, Any] = {"type": "notification-configurations", "attributes": {}} - # Add only specified attributes if self.enabled is not None: data["attributes"]["enabled"] = self.enabled @@ -365,9 +273,7 @@ def to_dict(self) -> dict[str, Any]: data["attributes"]["token"] = self.token if self.triggers is not None: - data["attributes"]["triggers"] = [ - trigger.value for trigger in self.triggers - ] + data["attributes"]["triggers"] = _serialize_triggers(self.triggers) if self.url is not None: data["attributes"]["url"] = self.url @@ -375,75 +281,71 @@ def to_dict(self) -> dict[str, Any]: if self.email_addresses is not None: data["attributes"]["email-addresses"] = self.email_addresses - # Handle relationships if self.email_users is not None: - data["relationships"] = data.get("relationships", {}) - data["relationships"]["users"] = { - "data": [ - { - "type": "users", - "id": user.id if hasattr(user, "id") else str(user), - } - for user in self.email_users - ] + data["relationships"] = { + "users": { + "data": [ + { + "type": "users", + "id": user.id if hasattr(user, "id") else str(user), + } + for user in self.email_users + ] + } } return data - def validate(self) -> list[str]: + def validate(self) -> list[str]: # type: ignore[override] """Validate the update options and return any errors.""" - errors = [] + errors: list[str] = [] - # Name validation (if provided) if self.name is not None and (not self.name or not self.name.strip()): errors.append("Name cannot be empty") - # Trigger validation (if provided) if self.triggers is not None: - for trigger in self.triggers: - if not isinstance(trigger, NotificationTriggerType): - errors.append(f"Invalid trigger type: {trigger}") # type: ignore[unreachable] + errors.extend(_validate_triggers(self.triggers)) return errors -class NotificationConfigurationList: +class NotificationConfigurationList(BaseModel): """Represents a list of notification configurations with pagination.""" - # Type annotations for instance attributes - items: list[NotificationConfiguration] - current_page: int - page_size: int - prev_page: int | None - next_page: int | None - total_pages: int - total_count: int - - def __init__(self, data: dict[str, Any]): - self.items = [ - NotificationConfiguration(item.get("attributes", {})) - for item in data.get("data", []) - ] - - # Pagination metadata - meta = data.get("meta", {}) - pagination = meta.get("pagination", {}) - - self.current_page = pagination.get("current-page", 0) - self.page_size = pagination.get("page-size", 20) - self.prev_page = pagination.get("prev-page") - self.next_page = pagination.get("next-page") - self.total_pages = pagination.get("total-pages", 0) - self.total_count = pagination.get("total-count", 0) + model_config = ConfigDict(populate_by_name=True) + + items: list[NotificationConfiguration] = Field(default_factory=list) + current_page: int = 0 + page_size: int = 20 + prev_page: int | None = None + next_page: int | None = None + total_pages: int = 0 + total_count: int = 0 + + def __init__(self, data: dict[str, Any] | None = None, /, **kwargs: Any) -> None: + if data is None: + super().__init__(**kwargs) + return + + items_data = [item.get("attributes", {}) for item in data.get("data") or []] + pagination = (data.get("meta") or {}).get("pagination") or {} + parsed: dict[str, Any] = { + "items": items_data, + "current_page": pagination.get("current-page", 0), + "page_size": pagination.get("page-size", 20), + "prev_page": pagination.get("prev-page"), + "next_page": pagination.get("next-page"), + "total_pages": pagination.get("total-pages", 0), + "total_count": pagination.get("total-count", 0), + } + parsed.update(kwargs) + super().__init__(**parsed) def __len__(self) -> int: return len(self.items) - def __iter__(self) -> Any: + def __iter__(self) -> Iterator[NotificationConfiguration]: # type: ignore[override] return iter(self.items) def __getitem__(self, index: int) -> NotificationConfiguration: return self.items[index] - - def __repr__(self) -> str: - return f"NotificationConfigurationList(count={len(self.items)}, page={self.current_page}, total={self.total_count})" diff --git a/src/pytfe/models/organization_token.py b/src/pytfe/models/organization_token.py new file mode 100644 index 00000000..24f1cb0c --- /dev/null +++ b/src/pytfe/models/organization_token.py @@ -0,0 +1,72 @@ +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: + pass + + +class TokenType(str, Enum): + """Token type enumeration.""" + + AUDIT_TRAILS = "audit-trails" + + +class OrganizationToken(BaseModel): + """Organization token represents a Terraform Enterprise organization token.""" + + model_config = ConfigDict(extra="forbid") + + id: str = Field(..., description="Organization token ID") + created_at: datetime = Field(..., description="Creation timestamp") + description: str | None = Field(None, description="Token description") + last_used_at: datetime | None = Field(None, description="Last usage timestamp") + token: str | None = Field(None, description="The actual token value") + expired_at: datetime | None = Field(None, description="Token expiration timestamp") + created_by: Any | None = Field( + None, description="The entity that created this token" + ) + + +class OrganizationTokenCreateOptions(BaseModel): + """Options for creating an organization token.""" + + model_config = ConfigDict(extra="forbid", populate_by_name=True) + + expired_at: datetime | None = Field( + None, + description="The token's expiration date. Available in TFE release v202305-1 and later", + ) + token_type: TokenType | None = Field( + None, + alias="token", + description="What type of token to create. Only applicable to HCP Terraform", + ) + + +class OrganizationTokenReadOptions(BaseModel): + """Options for reading an organization token.""" + + model_config = ConfigDict(extra="forbid", populate_by_name=True) + + token_type: TokenType | None = Field( + None, + alias="token", + description="What type of token to read. Only applicable to HCP Terraform", + ) + + +class OrganizationTokenDeleteOptions(BaseModel): + """Options for deleting an organization token.""" + + model_config = ConfigDict(extra="forbid", populate_by_name=True) + + token_type: TokenType | None = Field( + None, + alias="token", + description="What type of token to delete. Only applicable to HCP Terraform", + ) diff --git a/src/pytfe/models/ssh_key.py b/src/pytfe/models/ssh_key.py index 4df3d0a0..52423fb8 100644 --- a/src/pytfe/models/ssh_key.py +++ b/src/pytfe/models/ssh_key.py @@ -13,7 +13,7 @@ class SSHKey(BaseModel): id: str = Field(..., description="The unique identifier for this SSH key") type: str = Field(default="ssh-keys", description="The type of this resource") - name: str = Field(..., description="A name to identify the SSH key") + name: str = Field(default="", description="A name to identify the SSH key") class SSHKeyCreateOptions(BaseModel): diff --git a/src/pytfe/models/state_version.py b/src/pytfe/models/state_version.py index 4d8607dd..dab42619 100644 --- a/src/pytfe/models/state_version.py +++ b/src/pytfe/models/state_version.py @@ -32,7 +32,7 @@ class StateVersion(BaseModel): model_config = ConfigDict(populate_by_name=True, validate_by_name=True) id: str = Field(..., alias="id") - created_at: datetime = Field(..., alias="created-at") + created_at: datetime | None = Field(None, alias="created-at") hosted_state_download_url: str | None = Field( None, alias="hosted-state-download-url" ) diff --git a/src/pytfe/models/workspace.py b/src/pytfe/models/workspace.py index d24ab35d..e0be77a7 100644 --- a/src/pytfe/models/workspace.py +++ b/src/pytfe/models/workspace.py @@ -23,9 +23,13 @@ from ..utils import has_tags_regex_defined, is_valid_workspace_name, valid_string from .agent import AgentPool from .common import EffectiveTagBinding, Tag, TagBinding +from .configuration_version import ConfigurationVersion from .data_retention_policy import DataRetentionPolicyChoice from .organization import ExecutionMode, Organization from .project import Project +from .ssh_key import SSHKey +from .state_version import StateVersion +from .variable import Variable if TYPE_CHECKING: from .run import Run @@ -168,15 +172,17 @@ class Workspace(BaseModel): # Relations agent_pool: AgentPool | None = None # AgentPool object current_run: Run | None = None # Run object - current_state_version: Any | None = None # StateVersion object + current_state_version: StateVersion | None = None # StateVersion object organization: Organization | None = None project: Project | None = None - ssh_key: Any | None = None # SSHKey object + ssh_key: SSHKey | None = None # SSHKey object outputs: list[WorkspaceOutputs] = Field(default_factory=list) tags: list[Tag] = Field(default_factory=list) - current_configuration_version: Any | None = None # ConfigurationVersion object + current_configuration_version: ConfigurationVersion | None = ( + None # ConfigurationVersion object + ) locked_by: LockedByChoice | None = None - variables: list[Any] = Field(default_factory=list) # Variable objects + variables: list[Variable] = Field(default_factory=list) # Variable objects tag_bindings: list[TagBinding] = Field(default_factory=list) effective_tag_bindings: list[EffectiveTagBinding] = Field(default_factory=list) diff --git a/src/pytfe/resources/organization_token.py b/src/pytfe/resources/organization_token.py new file mode 100644 index 00000000..dcbcfb28 --- /dev/null +++ b/src/pytfe/resources/organization_token.py @@ -0,0 +1,220 @@ +from __future__ import annotations + +from datetime import datetime +from typing import Any +from urllib.parse import quote + +from ..errors import ERR_INVALID_ORG +from ..models.organization_token import ( + OrganizationToken, + OrganizationTokenCreateOptions, + OrganizationTokenDeleteOptions, + OrganizationTokenReadOptions, +) +from ..utils import valid_string_id +from ._base import _Service + + +class OrganizationTokens(_Service): + """Organization tokens service for managing TFE organization tokens.""" + + def create(self, organization: str) -> OrganizationToken: + """Create a new organization token, replacing any existing token. + + Args: + organization: The organization name or ID + + Returns: + OrganizationToken: The created organization token + + Raises: + ValueError: If the organization name is invalid + """ + return self.create_with_options(organization) + + def create_with_options( + self, + organization: str, + options: OrganizationTokenCreateOptions | None = None, + ) -> OrganizationToken: + """Create a new organization token with options, replacing any existing token. + + Args: + organization: The organization name or ID + options: Options for creating the token + + Returns: + OrganizationToken: The created organization token + + Raises: + ValueError: If the organization name is invalid + """ + if not valid_string_id(organization): + raise ValueError(ERR_INVALID_ORG) + + path = f"/api/v2/organizations/{quote(organization)}/authentication-token" + + # Build request body + body: dict[str, Any] = { + "data": { + "type": "authentication-token", + "attributes": {}, + } + } + + # Add optional attributes + if options and options.expired_at is not None: + body["data"]["attributes"]["expired-at"] = options.expired_at.isoformat() + + # Add query parameters for token type if specified + params = {} + if options and options.token_type is not None: + params["token"] = options.token_type.value + + if params: + response = self.t.request("POST", path, json_body=body, params=params) + else: + response = self.t.request("POST", path, json_body=body) + + data = response.json() + + if "data" in data: + return self._parse_organization_token(data["data"]) + + raise ValueError("Invalid response format") + + def read(self, organization: str) -> OrganizationToken: + """Read an organization token. + + Args: + organization: The organization name or ID + + Returns: + OrganizationToken: The organization token + + Raises: + ValueError: If the organization name is invalid + """ + return self.read_with_options(organization, None) + + def read_with_options( + self, + organization: str, + options: OrganizationTokenReadOptions | None = None, + ) -> OrganizationToken: + """Read an organization token with options. + + Args: + organization: The organization name or ID + options: Options for reading the token + + Returns: + OrganizationToken: The organization token + + Raises: + ValueError: If the organization name is invalid + """ + if not valid_string_id(organization): + raise ValueError(ERR_INVALID_ORG) + + path = f"/api/v2/organizations/{quote(organization)}/authentication-token" + + # Add query parameters for token type if specified + params = {} + if options and options.token_type is not None: + params["token"] = options.token_type.value + + response = self.t.request("GET", path, params=params if params else None) + data = response.json() + + if "data" in data: + return self._parse_organization_token(data["data"]) + + raise ValueError("Invalid response format") + + def delete(self, organization: str) -> None: + """Delete an organization token. + + Args: + organization: The organization name or ID + + Raises: + ValueError: If the organization name is invalid + """ + return self.delete_with_options(organization, None) + + def delete_with_options( + self, + organization: str, + options: OrganizationTokenDeleteOptions | None = None, + ) -> None: + """Delete an organization token with options. + + Args: + organization: The organization name or ID + options: Options for deleting the token + + Raises: + ValueError: If the organization name is invalid + """ + if not valid_string_id(organization): + raise ValueError(ERR_INVALID_ORG) + + path = f"/api/v2/organizations/{quote(organization)}/authentication-token" + + # Add query parameters for token type if specified + params = {} + if options and options.token_type is not None: + params["token"] = options.token_type.value + + if params: + self.t.request("DELETE", path, params=params) + else: + self.t.request("DELETE", path) + + def _parse_organization_token(self, data: dict[str, Any]) -> OrganizationToken: + """Parse organization token data from API response. + + Args: + data: The token data from the API response + + Returns: + OrganizationToken: The parsed organization token + """ + attributes = data.get("attributes", {}) + + # Parse timestamps + created_at_str = attributes.get("created-at") + created_at = ( + datetime.fromisoformat(created_at_str.replace("Z", "+00:00")) + if created_at_str + else datetime.now() + ) + + last_used_at_str = attributes.get("last-used-at") + last_used_at = ( + datetime.fromisoformat(last_used_at_str.replace("Z", "+00:00")) + if last_used_at_str + else None + ) + + expired_at_str = attributes.get("expired-at") + expired_at = ( + datetime.fromisoformat(expired_at_str.replace("Z", "+00:00")) + if expired_at_str + else None + ) + + # Parse created-by relationship + created_by = None + # For now, just set to None since it's mainly for display + + return OrganizationToken( + id=data.get("id", ""), + created_at=created_at, + description=attributes.get("description", ""), + last_used_at=last_used_at, + token=attributes.get("token", ""), + expired_at=expired_at, + created_by=created_by, + ) diff --git a/src/pytfe/resources/workspaces.py b/src/pytfe/resources/workspaces.py index 2bd2d9b3..1a6ac6cb 100644 --- a/src/pytfe/resources/workspaces.py +++ b/src/pytfe/resources/workspaces.py @@ -7,6 +7,8 @@ from collections.abc import Iterator from typing import Any +from pytfe.models.ssh_key import SSHKey + from ..errors import ( InvalidOrgError, InvalidSSHKeyIDError, @@ -19,11 +21,13 @@ WorkspaceMinimumLimitError, WorkspaceRequiredError, ) +from ..models.agent import AgentPool from ..models.common import ( EffectiveTagBinding, Tag, TagBinding, ) +from ..models.configuration_version import ConfigurationVersion from ..models.data_retention_policy import ( DataRetentionPolicy, DataRetentionPolicyChoice, @@ -34,6 +38,9 @@ ) from ..models.organization import Organization from ..models.project import Project +from ..models.run import Run +from ..models.state_version import StateVersion +from ..models.variable import Variable from ..models.workspace import ( ExecutionMode, LockedByChoice, @@ -181,7 +188,32 @@ def _ws_from(d: dict[str, Any]) -> Workspace: {"id": relationships["project"]["data"].get("id")} ) if relationships.get("ssh-key", {}).get("data"): - attr["ssh_key"] = relationships["ssh-key"]["data"].get("id") + attr["ssh_key"] = SSHKey.model_validate( + {"id": relationships["ssh-key"]["data"].get("id")} + ) + if relationships.get("agent-pool", {}).get("data"): + attr["agent_pools"] = AgentPool.model_validate( + {"id": relationships["agent-pool"]["data"].get("id")} + ) + if relationships.get("current-run", {}).get("data"): + attr["current_run"] = Run.model_validate( + {"id": relationships["current-run"]["data"].get("id")} + ) + if relationships.get("current-configuration-version", {}).get("data"): + attr["current_configuration_version"] = ConfigurationVersion.model_validate( + {"id": relationships["current-configuration-version"]["data"].get("id")} + ) + if relationships.get("vars", {}).get("data"): + attr["variables"] = [ + Variable.model_validate({"id": item.get("id")}) + for item in relationships["vars"]["data"] + if item.get("id") + ] + if relationships.get("current-state-version", {}).get("data"): + attr["current_state_version"] = StateVersion.model_validate( + {"id": relationships["current-state-version"]["data"].get("id")} + ) + attr["outputs"] = outputs attr["locked_by"] = locked_by attr["data_retention_policy_choice"] = data_retention_policy_choice diff --git a/tests/units/test_organization_token.py b/tests/units/test_organization_token.py new file mode 100644 index 00000000..826f2239 --- /dev/null +++ b/tests/units/test_organization_token.py @@ -0,0 +1,313 @@ +"""Unit tests for the organization token module.""" + +from datetime import datetime +from unittest.mock import Mock, patch + +import pytest + +from pytfe._http import HTTPTransport +from pytfe.errors import ERR_INVALID_ORG +from pytfe.models.organization_token import ( + OrganizationToken, + OrganizationTokenCreateOptions, + OrganizationTokenDeleteOptions, + OrganizationTokenReadOptions, + TokenType, +) +from pytfe.resources.organization_token import OrganizationTokens + + +class TestOrganizationTokens: + """Test the OrganizationTokens service class.""" + + @pytest.fixture + def mock_transport(self): + """Create a mock HTTPTransport.""" + return Mock(spec=HTTPTransport) + + @pytest.fixture + def org_tokens_service(self, mock_transport): + """Create an OrganizationTokens service with mocked transport.""" + return OrganizationTokens(mock_transport) + + def test_create_success(self, org_tokens_service): + """Test successful create operation.""" + mock_response_data = { + "data": { + "id": "at-test123", + "attributes": { + "created-at": "2023-01-01T00:00:00Z", + "description": "Test token", + "token": "test-token-value", + }, + } + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + + with patch.object(org_tokens_service, "t") as mock_t: + mock_t.request.return_value = mock_response + + result = org_tokens_service.create("test-org") + + mock_t.request.assert_called_once() + call_args = mock_t.request.call_args + + assert call_args[0][0] == "POST" + assert ( + call_args[0][1] == "/api/v2/organizations/test-org/authentication-token" + ) + assert "json_body" in call_args[1] + assert "data" in call_args[1]["json_body"] + assert "attributes" in call_args[1]["json_body"]["data"] + assert isinstance(result, OrganizationToken) + assert result.id == "at-test123" + assert result.description == "Test token" + + def test_create_validation_errors(self, org_tokens_service): + """Test create with invalid organization name.""" + with pytest.raises(ValueError, match=ERR_INVALID_ORG): + org_tokens_service.create("") + + with pytest.raises(ValueError, match=ERR_INVALID_ORG): + org_tokens_service.create(None) + + def test_create_with_options_expiration_success(self, org_tokens_service): + """Test create with options including expiration date.""" + mock_response_data = { + "data": { + "id": "at-exp-123", + "attributes": { + "created-at": "2023-01-01T00:00:00Z", + "token": "token-value", + "expired-at": "2024-01-01T00:00:00Z", + }, + } + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + + with patch.object(org_tokens_service, "t") as mock_t: + mock_t.request.return_value = mock_response + + expiry = datetime(2024, 1, 1, 0, 0, 0) + options = OrganizationTokenCreateOptions(expired_at=expiry) + + result = org_tokens_service.create_with_options("test-org", options) + + assert isinstance(result, OrganizationToken) + assert result.expired_at is not None + + call_args = mock_t.request.call_args + assert call_args[0][0] == "POST" + assert ( + call_args[0][1] == "/api/v2/organizations/test-org/authentication-token" + ) + body = call_args[1]["json_body"] + assert "expired-at" in body["data"]["attributes"] + assert body["data"]["attributes"]["expired-at"] == "2024-01-01T00:00:00" + + def test_create_with_options_token_type_success(self, org_tokens_service): + """Test create with options including token type.""" + mock_response_data = { + "data": { + "id": "at-audit-123", + "attributes": { + "created-at": "2023-01-01T00:00:00Z", + "token": "audit-token-value", + }, + } + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + + with patch.object(org_tokens_service, "t") as mock_t: + mock_t.request.return_value = mock_response + + options = OrganizationTokenCreateOptions(token_type=TokenType.AUDIT_TRAILS) + result = org_tokens_service.create_with_options("test-org", options) + + assert isinstance(result, OrganizationToken) + call_args = mock_t.request.call_args + assert call_args[0][0] == "POST" + assert ( + call_args[0][1] == "/api/v2/organizations/test-org/authentication-token" + ) + assert "params" in call_args[1] + assert call_args[1]["params"]["token"] == "audit-trails" + assert "json_body" in call_args[1] + + def test_read_success(self, org_tokens_service): + """Test successful read operation.""" + mock_response_data = { + "data": { + "id": "at-read-123", + "attributes": { + "created-at": "2023-01-01T00:00:00Z", + "description": "Read token", + "token": "read-token-value", + }, + } + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + + with patch.object(org_tokens_service, "t") as mock_t: + mock_t.request.return_value = mock_response + + result = org_tokens_service.read("test-org") + + assert isinstance(result, OrganizationToken) + assert result.id == "at-read-123" + + call_args = mock_t.request.call_args + assert call_args[0][0] == "GET" + assert ( + call_args[0][1] == "/api/v2/organizations/test-org/authentication-token" + ) + + def test_read_validation_errors(self, org_tokens_service): + """Test read with invalid organization name.""" + with pytest.raises(ValueError, match=ERR_INVALID_ORG): + org_tokens_service.read("") + + with pytest.raises(ValueError, match=ERR_INVALID_ORG): + org_tokens_service.read(None) + + def test_read_with_options_token_type_success(self, org_tokens_service): + """Test read with options including token type.""" + mock_response_data = { + "data": { + "id": "at-audit-read-123", + "attributes": { + "created-at": "2023-01-01T00:00:00Z", + "token": "audit-read-value", + }, + } + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + + with patch.object(org_tokens_service, "t") as mock_t: + mock_t.request.return_value = mock_response + + options = OrganizationTokenReadOptions(token_type=TokenType.AUDIT_TRAILS) + result = org_tokens_service.read_with_options("test-org", options) + + assert isinstance(result, OrganizationToken) + call_args = mock_t.request.call_args + assert call_args[0][0] == "GET" + assert ( + call_args[0][1] == "/api/v2/organizations/test-org/authentication-token" + ) + assert call_args[1]["params"]["token"] == "audit-trails" + + def test_delete_success(self, org_tokens_service): + """Test successful delete operation.""" + with patch.object(org_tokens_service, "t") as mock_t: + mock_t.request.return_value = Mock() + + result = org_tokens_service.delete("test-org") + + assert result is None + call_args = mock_t.request.call_args + assert call_args[0][0] == "DELETE" + assert ( + call_args[0][1] == "/api/v2/organizations/test-org/authentication-token" + ) + + def test_delete_validation_errors(self, org_tokens_service): + """Test delete with invalid organization name.""" + with pytest.raises(ValueError, match=ERR_INVALID_ORG): + org_tokens_service.delete("") + + with pytest.raises(ValueError, match=ERR_INVALID_ORG): + org_tokens_service.delete(None) + + def test_delete_with_options_token_type_success(self, org_tokens_service): + """Test delete with options including token type.""" + with patch.object(org_tokens_service, "t") as mock_t: + mock_t.request.return_value = Mock() + + options = OrganizationTokenDeleteOptions(token_type=TokenType.AUDIT_TRAILS) + result = org_tokens_service.delete_with_options("test-org", options) + + assert result is None + call_args = mock_t.request.call_args + assert call_args[0][0] == "DELETE" + assert ( + call_args[0][1] == "/api/v2/organizations/test-org/authentication-token" + ) + assert call_args[1]["params"]["token"] == "audit-trails" + + def test_parse_token_minimal(self, org_tokens_service): + """Test parsing token with minimal data.""" + data = { + "id": "at-minimal-123", + "attributes": { + "created-at": "2023-01-01T00:00:00Z", + "description": "Minimal token", + "token": "minimal-value", + }, + "relationships": {}, + } + + result = org_tokens_service._parse_organization_token(data) + + assert result.id == "at-minimal-123" + assert isinstance(result.created_at, datetime) + assert result.description == "Minimal token" + assert result.token == "minimal-value" + assert result.last_used_at is None + assert result.expired_at is None + + def test_parse_token_all_fields(self, org_tokens_service): + """Test parsing token with all fields populated.""" + data = { + "id": "at-full-123", + "attributes": { + "created-at": "2023-01-01T00:00:00Z", + "description": "Full token", + "token": "full-value", + "last-used-at": "2023-01-15T12:30:00Z", + "expired-at": "2024-01-01T00:00:00Z", + }, + "relationships": {}, + } + + result = org_tokens_service._parse_organization_token(data) + + assert result.id == "at-full-123" + assert result.description == "Full token" + assert result.token == "full-value" + assert result.last_used_at is not None + assert result.expired_at is not None + assert isinstance(result.last_used_at, datetime) + assert isinstance(result.expired_at, datetime) + + def test_invalid_response_format_on_create(self, org_tokens_service): + """Test handling of invalid response format when creating.""" + mock_response = Mock() + mock_response.json.return_value = {"error": "Invalid"} + + with patch.object(org_tokens_service, "t") as mock_t: + mock_t.request.return_value = mock_response + + with pytest.raises(ValueError, match="Invalid response format"): + org_tokens_service.create("test-org") + + def test_invalid_response_format_on_read(self, org_tokens_service): + """Test handling of invalid response format when reading.""" + mock_response = Mock() + mock_response.json.return_value = {"error": "Invalid"} + + with patch.object(org_tokens_service, "t") as mock_t: + mock_t.request.return_value = mock_response + + with pytest.raises(ValueError, match="Invalid response format"): + org_tokens_service.read("test-org")