diff --git a/examples/registry_provider_platform.py b/examples/registry_provider_platform.py new file mode 100644 index 00000000..a6cf01ac --- /dev/null +++ b/examples/registry_provider_platform.py @@ -0,0 +1,236 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from __future__ import annotations + +import argparse +import os + +from pytfe import TFEClient, TFEConfig +from pytfe.models import ( + RegistryProviderPlatformCreateOptions, + RegistryProviderPlatformID, + RegistryProviderPlatformListOptions, + RegistryProviderVersionID, +) + + +def _print_header(title: str): + print("\n" + "=" * 80) + print(title) + print("=" * 80) + + +def main(): + parser = argparse.ArgumentParser( + description="Registry Provider Platforms 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("--organization", required=True, help="Organization name") + parser.add_argument( + "--registry-name", + default="private", + help="Registry name (default: private)", + ) + parser.add_argument("--namespace", required=True, help="Provider namespace") + parser.add_argument("--name", required=True, help="Provider name") + parser.add_argument( + "--version", required=True, help="Provider version (e.g., 1.0.0)" + ) + parser.add_argument( + "--page-size", + type=int, + default=100, + help="Page size for listing platforms", + ) + parser.add_argument("--create", action="store_true", help="Create a platform") + parser.add_argument("--read", action="store_true", help="Read a specific platform") + parser.add_argument( + "--delete", action="store_true", help="Delete a specific platform" + ) + parser.add_argument( + "--os", dest="os", help="Operating system (e.g., linux, darwin)" + ) + parser.add_argument("--arch", help="Architecture (e.g., amd64, arm64)") + parser.add_argument("--shasum", help="SHA256 checksum of the provider binary") + parser.add_argument("--filename", help="Filename of the provider binary zip") + args = parser.parse_args() + + cfg = TFEConfig(address=args.address, token=args.token) + client = TFEClient(cfg) + + version_id = RegistryProviderVersionID( + organization_name=args.organization, + registry_name=args.registry_name, + namespace=args.namespace, + name=args.name, + version=args.version, + ) + + # 1) List all platforms for the provider version + _print_header( + f"Listing platforms for {args.registry_name}/{args.namespace}/{args.name} @ {args.version}" + ) + + list_options = RegistryProviderPlatformListOptions(page_size=args.page_size) + + platform_count = 0 + for platform in client.registry_provider_platforms.list( + version_id=version_id, + options=list_options, + ): + platform_count += 1 + print(f"- Platform {platform.os}/{platform.arch} (ID: {platform.id})") + print(f" Filename: {platform.filename}") + print(f" Shasum: {platform.shasum}") + print(f" Provider Binary Uploaded: {platform.provider_binary_uploaded}") + if platform.permissions: + print(" Permissions:") + print(f" Can Delete: {platform.permissions.can_delete}") + print(f" Can Upload Asset: {platform.permissions.can_upload_asset}") + if platform.links: + print(" Links:") + for key, value in platform.links.items(): + print(f" {key}: {value}") + print() + + if platform_count == 0: + print("No platforms found.") + else: + print(f"Total: {platform_count} platforms") + + # 2) Create a new platform (if --create flag is provided) + if args.create: + if not args.os: + print("Error: --os is required for create operation") + return + if not args.arch: + print("Error: --arch is required for create operation") + return + if not args.shasum: + print("Error: --shasum is required for create operation") + return + if not args.filename: + print("Error: --filename is required for create operation") + return + + _print_header(f"Creating platform: {args.os}/{args.arch}") + + create_options = RegistryProviderPlatformCreateOptions( + os=args.os, + arch=args.arch, + shasum=args.shasum, + filename=args.filename, + ) + + new_platform = client.registry_provider_platforms.create( + version_id=version_id, + options=create_options, + ) + + print(f"Created platform: {new_platform.id}") + print(f" OS: {new_platform.os}") + print(f" Arch: {new_platform.arch}") + print(f" Filename: {new_platform.filename}") + print(f" Shasum: {new_platform.shasum}") + print(f" Provider Binary Uploaded: {new_platform.provider_binary_uploaded}") + + if new_platform.links: + print("\n Upload URLs:") + if "provider-binary-upload" in new_platform.links: + print( + f" Provider Binary: {new_platform.links['provider-binary-upload']}" + ) + + # 3) Read a specific platform (if --read flag is provided) + if args.read: + if not args.os: + print("Error: --os is required for read operation") + return + if not args.arch: + print("Error: --arch is required for read operation") + return + + _print_header(f"Reading platform: {args.os}/{args.arch}") + + platform_id = RegistryProviderPlatformID( + organization_name=args.organization, + registry_name=args.registry_name, + namespace=args.namespace, + name=args.name, + version=args.version, + os=args.os, + arch=args.arch, + ) + + platform = client.registry_provider_platforms.read(platform_id) + + print(f"Platform ID: {platform.id}") + print(f" OS: {platform.os}") + print(f" Arch: {platform.arch}") + print(f" Filename: {platform.filename}") + print(f" Shasum: {platform.shasum}") + print(f" Provider Binary Uploaded: {platform.provider_binary_uploaded}") + + if platform.permissions: + print(" Permissions:") + print(f" Can Delete: {platform.permissions.can_delete}") + print(f" Can Upload Asset: {platform.permissions.can_upload_asset}") + + if platform.links: + print(" Links:") + for key, value in platform.links.items(): + print(f" {key}: {value}") + + # 4) Delete a platform (if --delete flag is provided) + if args.delete: + if not args.os: + print("Error: --os is required for delete operation") + return + if not args.arch: + print("Error: --arch is required for delete operation") + return + + _print_header(f"Deleting platform: {args.os}/{args.arch}") + + platform_id = RegistryProviderPlatformID( + organization_name=args.organization, + registry_name=args.registry_name, + namespace=args.namespace, + name=args.name, + version=args.version, + os=args.os, + arch=args.arch, + ) + + try: + platform_to_delete = client.registry_provider_platforms.read(platform_id) + print("Platform to delete:") + print(f" ID: {platform_to_delete.id}") + print(f" OS/Arch: {platform_to_delete.os}/{platform_to_delete.arch}") + print(f" Filename: {platform_to_delete.filename}") + except Exception as e: + print(f"Error reading platform: {e}") + return + + client.registry_provider_platforms.delete(platform_id) + print(f"\n Successfully deleted platform: {args.os}/{args.arch}") + + # List remaining platforms + _print_header("Listing platforms after deletion") + remaining_count = 0 + for platform in client.registry_provider_platforms.list(version_id=version_id): + remaining_count += 1 + print(f"- {platform.os}/{platform.arch} (ID: {platform.id})") + + if remaining_count == 0: + print("No platforms remaining.") + else: + print(f"Total remaining: {remaining_count} platforms") + + +if __name__ == "__main__": + main() diff --git a/src/pytfe/client.py b/src/pytfe/client.py index dc1972ce..d60221b4 100644 --- a/src/pytfe/client.py +++ b/src/pytfe/client.py @@ -27,6 +27,7 @@ from .resources.query_run import QueryRuns from .resources.registry_module import RegistryModules from .resources.registry_provider import RegistryProviders +from .resources.registry_provider_platform import RegistryProviderPlatforms from .resources.registry_provider_version import RegistryProviderVersions from .resources.reserved_tag_key import ReservedTagKeys from .resources.run import Runs @@ -83,6 +84,7 @@ def __init__(self, config: TFEConfig | None = None): self.registry_modules = RegistryModules(self._transport) self.registry_providers = RegistryProviders(self._transport) self.registry_provider_versions = RegistryProviderVersions(self._transport) + self.registry_provider_platforms = RegistryProviderPlatforms(self._transport) # State and execution resources self.state_versions = StateVersions(self._transport) diff --git a/src/pytfe/errors.py b/src/pytfe/errors.py index 4d616a17..f2340af3 100644 --- a/src/pytfe/errors.py +++ b/src/pytfe/errors.py @@ -567,3 +567,63 @@ class InvalidTeamProjectAccessIDError(InvalidValues): def __init__(self, message: str = "invalid value for team project access ID"): super().__init__(message) + + +# Registry Provider Platform errors +class RequiredOSError(RequiredFieldMissing): + """Raised when a required OS field is missing.""" + + def __init__(self, message: str = "os is required"): + super().__init__(message) + + +class RequiredArchError(RequiredFieldMissing): + """Raised when a required architecture field is missing.""" + + def __init__(self, message: str = "arch is required"): + super().__init__(message) + + +class RequiredShasumError(RequiredFieldMissing): + """Raised when a required shasum field is missing.""" + + def __init__(self, message: str = "shasum is required"): + super().__init__(message) + + +class RequiredFilenameError(RequiredFieldMissing): + """Raised when a required filename field is missing.""" + + def __init__(self, message: str = "filename is required"): + super().__init__(message) + + +class InvalidOSError(InvalidValues): + """Raised when an invalid OS field is provided.""" + + def __init__(self, message: str = "invalid value for os"): + super().__init__(message) + + +class InvalidArchError(InvalidValues): + """Raised when an invalid architecture field is provided.""" + + def __init__(self, message: str = "invalid value for arch"): + super().__init__(message) + + +class InvalidNamespaceError(InvalidValues): + """Raised when an invalid namespace field is provided.""" + + def __init__(self, message: str = "invalid value for namespace"): + super().__init__(message) + + +class InvalidRegistryNameError(InvalidValues): + """Raised when an invalid registry name field is provided.""" + + def __init__( + self, + message: str = "invalid value for registry-name. It must be either private or public", + ): + super().__init__(message) diff --git a/src/pytfe/models/__init__.py b/src/pytfe/models/__init__.py index f332fbbf..c0d6c773 100644 --- a/src/pytfe/models/__init__.py +++ b/src/pytfe/models/__init__.py @@ -230,6 +230,13 @@ RegistryProviderPermissions, RegistryProviderReadOptions, ) +from .registry_provider_platform import ( + RegistryProviderPlatform, + RegistryProviderPlatformCreateOptions, + RegistryProviderPlatformID, + RegistryProviderPlatformListOptions, + RegistryProviderPlatformPermissions, +) from .registry_provider_version import ( RegistryProviderVersion, RegistryProviderVersionCreateOptions, @@ -500,6 +507,12 @@ "RegistryProviderVersionID", "RegistryProviderVersionListOptions", "RegistryProviderVersionPermissions", + # Registry provider platforms + "RegistryProviderPlatform", + "RegistryProviderPlatformCreateOptions", + "RegistryProviderPlatformID", + "RegistryProviderPlatformListOptions", + "RegistryProviderPlatformPermissions", # Query runs "QueryRun", "QueryRunActions", @@ -706,3 +719,6 @@ # Rebuild models with forward references after all models are loaded PolicyCheck.model_rebuild() +RegistryProvider.model_rebuild() +RegistryProviderVersion.model_rebuild() +RegistryProviderPlatform.model_rebuild() diff --git a/src/pytfe/models/registry_provider.py b/src/pytfe/models/registry_provider.py index 2861acac..5cd57414 100644 --- a/src/pytfe/models/registry_provider.py +++ b/src/pytfe/models/registry_provider.py @@ -7,7 +7,15 @@ from enum import Enum from typing import Any -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, model_validator + +from ..errors import ( + InvalidNameError, + InvalidNamespaceError, + InvalidOrgError, + InvalidValues, +) +from ..utils import valid_string_id class RegistryName(Enum): @@ -35,12 +43,14 @@ class RegistryProvider(BaseModel): """Registry provider model.""" id: str - name: str - namespace: str - created_at: datetime = Field(alias="created-at") - updated_at: datetime = Field(alias="updated-at") - registry_name: RegistryName = Field(alias="registry-name") - permissions: RegistryProviderPermissions + name: str = Field(alias="name", default="") + namespace: str = Field(alias="namespace", default="") + created_at: datetime | None = Field(alias="created-at", default=None) + updated_at: datetime | None = Field(alias="updated-at", default=None) + registry_name: RegistryName | None = Field(alias="registry-name", default=None) + permissions: RegistryProviderPermissions | None = Field( + alias="permissions", default=None + ) # Relations organization: dict[str, Any] | None = None @@ -62,6 +72,19 @@ class RegistryProviderID(BaseModel): namespace: str name: str + @model_validator(mode="after") + def valid(self) -> RegistryProviderID: + """Validate the registry provider ID.""" + if not valid_string_id(self.organization_name): + raise InvalidOrgError() + if not valid_string_id(self.name): + raise InvalidNameError() + if not valid_string_id(self.namespace): + raise InvalidNamespaceError() + if not valid_string_id(self.registry_name.value): + raise InvalidValues("invalid value for registry name") + return self + class RegistryProviderCreateOptions(BaseModel): """Options for creating a registry provider.""" @@ -72,6 +95,15 @@ class RegistryProviderCreateOptions(BaseModel): model_config = {"populate_by_name": True} + @model_validator(mode="after") + def valid(self) -> RegistryProviderCreateOptions: + """Validate the create options.""" + if not valid_string_id(self.name): + raise InvalidNameError() + if not valid_string_id(self.namespace): + raise InvalidNamespaceError() + return self + class RegistryProviderReadOptions(BaseModel): """Options for reading a registry provider.""" diff --git a/src/pytfe/models/registry_provider_platform.py b/src/pytfe/models/registry_provider_platform.py new file mode 100644 index 00000000..716ac6e4 --- /dev/null +++ b/src/pytfe/models/registry_provider_platform.py @@ -0,0 +1,105 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from __future__ import annotations + +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from ..errors import ( + InvalidArchError, + InvalidOSError, + RequiredArchError, + RequiredFilenameError, + RequiredOSError, + RequiredShasumError, +) +from ..utils import valid_string, valid_string_id +from .registry_provider_version import ( + RegistryProviderVersion, + RegistryProviderVersionID, +) + + +class RegistryProviderPlatformPermissions(BaseModel): + """Registry provider platform permissions.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + can_delete: bool = Field(alias="can-delete") + can_upload_asset: bool = Field(alias="can-upload-asset") + + +class RegistryProviderPlatform(BaseModel): + """Registry provider platform model.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + id: str + os: str = Field(alias="os", default="") + arch: str = Field(alias="arch", default="") + filename: str = Field(alias="filename", default="") + shasum: str = Field(alias="shasum", default="") + provider_binary_uploaded: bool | None = Field( + alias="provider-binary-uploaded", default=None + ) + permissions: RegistryProviderPlatformPermissions | None = None + + # Relations + registry_provider_version: RegistryProviderVersion | None = Field( + alias="registry-provider-version", default=None + ) + + # Links + links: dict[str, Any] | None = None + + +class RegistryProviderPlatformID(RegistryProviderVersionID): + """Registry provider platform identifier. + + Extends RegistryProviderVersionID with OS and arch to uniquely + identify a specific platform of a provider version. + """ + + os: str + arch: str + + @model_validator(mode="after") + def valid_platform_id(self) -> RegistryProviderPlatformID: + if not valid_string_id(self.os): + raise InvalidOSError() + if not valid_string_id(self.arch): + raise InvalidArchError() + return self + + +class RegistryProviderPlatformCreateOptions(BaseModel): + """Options for creating a registry provider platform.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + os: str = Field(alias="os") + arch: str = Field(alias="arch") + shasum: str = Field(alias="shasum") + filename: str = Field(alias="filename") + + @model_validator(mode="after") + def valid(self) -> RegistryProviderPlatformCreateOptions: + if not valid_string(self.os): + raise RequiredOSError() + if not valid_string(self.arch): + raise RequiredArchError() + if not valid_string_id(self.shasum): + raise RequiredShasumError() + if not valid_string_id(self.filename): + raise RequiredFilenameError() + return self + + +class RegistryProviderPlatformListOptions(BaseModel): + """Options for listing registry provider platforms.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + page_size: int | None = Field(alias="page[size]", default=None) diff --git a/src/pytfe/models/registry_provider_version.py b/src/pytfe/models/registry_provider_version.py index e5875051..8ec0d442 100644 --- a/src/pytfe/models/registry_provider_version.py +++ b/src/pytfe/models/registry_provider_version.py @@ -4,7 +4,7 @@ from __future__ import annotations from datetime import datetime -from typing import Any +from typing import TYPE_CHECKING, Any from pydantic import BaseModel, ConfigDict, Field, model_validator @@ -16,9 +16,13 @@ from ..utils import valid_string_id from .registry_provider import ( RegistryName, + RegistryProvider, RegistryProviderID, ) +if TYPE_CHECKING: + from .registry_provider_platform import RegistryProviderPlatform + class RegistryProviderVersionPermissions(BaseModel): """Registry provider version permissions.""" @@ -35,20 +39,24 @@ class RegistryProviderVersion(BaseModel): model_config = ConfigDict(populate_by_name=True, validate_by_name=True) id: str - version: str - created_at: datetime = Field(alias="created-at") - updated_at: datetime = Field(alias="updated-at") - key_id: str = Field(alias="key-id") - protocols: list[str] - permissions: RegistryProviderVersionPermissions - shasums_uploaded: bool = Field(alias="shasums-uploaded") - shasums_sig_uploaded: bool = Field(alias="shasums-sig-uploaded") + version: str = Field(alias="version", default="") + created_at: datetime | None = Field(alias="created-at", default=None) + updated_at: datetime | None = Field(alias="updated-at", default=None) + key_id: str = Field(alias="key-id", default="") + protocols: list[str] = Field(alias="protocols", default_factory=list) + permissions: RegistryProviderVersionPermissions | None = Field( + alias="permissions", default=None + ) + shasums_uploaded: bool | None = Field(alias="shasums-uploaded", default=None) + shasums_sig_uploaded: bool | None = Field( + alias="shasums-sig-uploaded", default=None + ) # Relations - registry_provider: dict[str, Any] | None = Field( + registry_provider: RegistryProvider | None = Field( alias="registry-provider", default=None ) - registry_provider_platforms: list[dict[str, Any]] | None = Field( + registry_provider_platforms: list[RegistryProviderPlatform] | None = Field( alias="platforms", default=None ) @@ -142,7 +150,7 @@ class RegistryProviderVersionID(RegistryProviderID): version: str @model_validator(mode="after") - def valid(self) -> RegistryProviderVersionID: + def valid_version_id(self) -> RegistryProviderVersionID: if not valid_string_id(self.version): raise InvalidVersionError() if self.registry_name != RegistryName.PRIVATE: diff --git a/src/pytfe/resources/registry_provider.py b/src/pytfe/resources/registry_provider.py index d4ae122b..e9f2b48d 100644 --- a/src/pytfe/resources/registry_provider.py +++ b/src/pytfe/resources/registry_provider.py @@ -61,9 +61,6 @@ def create( if not valid_string_id(organization): raise ValueError(ERR_INVALID_ORG) - if not self._validate_create_options(options): - raise ValueError("Invalid create options") - path = f"/api/v2/organizations/{organization}/registry-providers" # Prepare the data payload @@ -88,9 +85,6 @@ def read( options: RegistryProviderReadOptions | None = None, ) -> RegistryProvider: """Read a specific registry provider.""" - if not self._validate_provider_id(provider_id): - raise ValueError("Invalid provider ID") - path = ( f"/api/v2/organizations/{provider_id.organization_name}/" f"registry-providers/{provider_id.registry_name.value}/" @@ -107,9 +101,6 @@ def read( def delete(self, provider_id: RegistryProviderID) -> None: """Delete a registry provider.""" - if not self._validate_provider_id(provider_id): - raise ValueError("Invalid provider ID") - path = ( f"/api/v2/organizations/{provider_id.organization_name}/" f"registry-providers/{provider_id.registry_name.value}/" @@ -118,28 +109,6 @@ def delete(self, provider_id: RegistryProviderID) -> None: self.t.request("DELETE", path) - def _validate_provider_id(self, provider_id: RegistryProviderID) -> bool: - """Validate a registry provider ID.""" - if not valid_string_id(provider_id.organization_name): - return False - if not valid_string_id(provider_id.name): - return False - if not valid_string_id(provider_id.namespace): - return False - if provider_id.registry_name not in [RegistryName.PRIVATE, RegistryName.PUBLIC]: - return False - return True - - def _validate_create_options(self, options: RegistryProviderCreateOptions) -> bool: - """Validate create options.""" - if not valid_string_id(options.name): - return False - if not valid_string_id(options.namespace): - return False - if options.registry_name not in [RegistryName.PRIVATE, RegistryName.PUBLIC]: - return False - return True - def _parse_registry_provider(self, data: dict[str, Any]) -> RegistryProvider: """Parse a registry provider from API response data.""" if data is None: diff --git a/src/pytfe/resources/registry_provider_platform.py b/src/pytfe/resources/registry_provider_platform.py new file mode 100644 index 00000000..a25c8e17 --- /dev/null +++ b/src/pytfe/resources/registry_provider_platform.py @@ -0,0 +1,106 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +from ..models.registry_provider_platform import ( + RegistryProviderPlatform, + RegistryProviderPlatformCreateOptions, + RegistryProviderPlatformID, + RegistryProviderPlatformListOptions, +) +from ..models.registry_provider_version import ( + RegistryProviderVersion, + RegistryProviderVersionID, +) +from ._base import _Service + + +class RegistryProviderPlatforms(_Service): + """Service for managing Terraform registry provider platforms.""" + + def create( + self, + version_id: RegistryProviderVersionID, + options: RegistryProviderPlatformCreateOptions, + ) -> RegistryProviderPlatform: + """Create a registry provider platform""" + path = f"/api/v2/organizations/{version_id.organization_name}/registry-providers/{version_id.registry_name.value}/{version_id.namespace}/{version_id.name}/versions/{version_id.version}/platforms" + attributes = options.model_dump(by_alias=True, exclude_none=True) + payload = { + "data": { + "type": "registry-provider-platforms", + "attributes": attributes, + } + } + r = self.t.request("POST", path=path, json_body=payload) + data = r.json().get("data", {}) + return self._registry_provider_platform_from(data) + + def list( + self, + version_id: RegistryProviderVersionID, + options: RegistryProviderPlatformListOptions | None = None, + ) -> Iterator[RegistryProviderPlatform]: + """List registry provider platforms for a specific version""" + path = ( + f"/api/v2/organizations/{version_id.organization_name}" + f"/registry-providers/{version_id.registry_name.value}" + f"/{version_id.namespace}/{version_id.name}" + f"/versions/{version_id.version}/platforms" + ) + params = options.model_dump(by_alias=True) if options else {} + for item in self._list(path=path, params=params): + yield self._registry_provider_platform_from(item) + + def read(self, platform_id: RegistryProviderPlatformID) -> RegistryProviderPlatform: + """Read a specific registry provider platform""" + path = ( + f"/api/v2/organizations/{platform_id.organization_name}" + f"/registry-providers/{platform_id.registry_name.value}" + f"/{platform_id.namespace}/{platform_id.name}" + f"/versions/{platform_id.version}" + f"/platforms/{platform_id.os}/{platform_id.arch}" + ) + r = self.t.request("GET", path=path) + data = r.json().get("data", {}) + return self._registry_provider_platform_from(data) + + def delete(self, platform_id: RegistryProviderPlatformID) -> None: + """Delete a specific registry provider platform""" + path = ( + f"/api/v2/organizations/{platform_id.organization_name}" + f"/registry-providers/{platform_id.registry_name.value}" + f"/{platform_id.namespace}/{platform_id.name}" + f"/versions/{platform_id.version}" + f"/platforms/{platform_id.os}/{platform_id.arch}" + ) + self.t.request("DELETE", path=path) + return None + + def _registry_provider_platform_from( + self, data: dict[str, Any] + ) -> RegistryProviderPlatform: + """Parse a registry provider platform from API response data.""" + attrs = data.get("attributes", {}) + relationships = data.get("relationships", {}) + attrs["id"] = data.get("id") + + if ( + "registry-provider-version" in relationships + and "data" in relationships["registry-provider-version"] + and relationships["registry-provider-version"]["data"] is not None + ): + attrs["registry-provider-version"] = ( + RegistryProviderVersion.model_construct( + id=relationships["registry-provider-version"]["data"].get("id") + ) + ) + + if "links" in data: + attrs["links"] = data["links"] + + return RegistryProviderPlatform.model_validate(attrs) diff --git a/src/pytfe/resources/registry_provider_version.py b/src/pytfe/resources/registry_provider_version.py index 08735c48..03156afe 100644 --- a/src/pytfe/resources/registry_provider_version.py +++ b/src/pytfe/resources/registry_provider_version.py @@ -11,15 +11,16 @@ ) from ..models.registry_provider import ( RegistryName, + RegistryProvider, RegistryProviderID, ) +from ..models.registry_provider_platform import RegistryProviderPlatform from ..models.registry_provider_version import ( RegistryProviderVersion, RegistryProviderVersionCreateOptions, RegistryProviderVersionID, RegistryProviderVersionListOptions, ) -from ..utils import valid_string_id from ._base import _Service @@ -32,9 +33,6 @@ def create( options: RegistryProviderVersionCreateOptions, ) -> RegistryProviderVersion: """Create a registry provider version""" - if not self._validate_provider_id(provider_id): - raise ValueError("Invalid provider ID") - if provider_id.registry_name != RegistryName.PRIVATE: raise RequiredPrivateRegistryError() path = f"/api/v2/organizations/{provider_id.organization_name}/registry-providers/{provider_id.registry_name.value}/{provider_id.namespace}/{provider_id.name}/versions" @@ -53,18 +51,6 @@ def create( data = r.json().get("data", {}) return self._registry_provider_version_from(data) - def _validate_provider_id(self, provider_id: RegistryProviderID) -> bool: - """Validate a registry provider ID.""" - if not valid_string_id(provider_id.organization_name): - return False - if not valid_string_id(provider_id.name): - return False - if not valid_string_id(provider_id.namespace): - return False - if provider_id.registry_name not in [RegistryName.PRIVATE, RegistryName.PUBLIC]: - return False - return True - def _registry_provider_version_from( self, data: dict[str, Any] ) -> RegistryProviderVersion: @@ -74,16 +60,22 @@ def _registry_provider_version_from( relationships = data.get("relationships", {}) attrs["id"] = data.get("id") - # Parse relationships + # Parse relationships as typed stubs if "registry-provider" in relationships: - attrs["registry_provider"] = relationships["registry-provider"].get( - "data", {} - ) + rp_data = relationships["registry-provider"].get("data") + if rp_data and rp_data.get("id"): + attrs["registry_provider"] = RegistryProvider.model_construct( + id=rp_data["id"] + ) if "platforms" in relationships: - attrs["registry_provider_platforms"] = relationships["platforms"].get( - "data", [] - ) + platforms_data = relationships["platforms"].get("data", []) + if platforms_data: + attrs["registry_provider_platforms"] = [ + RegistryProviderPlatform.model_construct(id=p["id"]) + for p in platforms_data + if p.get("id") + ] return RegistryProviderVersion.model_validate(attrs) @@ -93,9 +85,6 @@ def list( options: RegistryProviderVersionListOptions | None = None, ) -> Iterator[RegistryProviderVersion]: """List registry provider versions""" - if not self._validate_provider_id(provider_id): - raise ValueError("Invalid provider ID") - path = f"/api/v2/organizations/{provider_id.organization_name}/registry-providers/{provider_id.registry_name.value}/{provider_id.namespace}/{provider_id.name}/versions" params = options.model_dump(by_alias=True) if options else {} for item in self._list(path=path, params=params): @@ -103,9 +92,6 @@ def list( def read(self, version_id: RegistryProviderVersionID) -> RegistryProviderVersion: """Read a specific registry provider version""" - if not self._validate_provider_id(version_id): - raise ValueError("Invalid provider ID") - path = f"/api/v2/organizations/{version_id.organization_name}/registry-providers/{version_id.registry_name.value}/{version_id.namespace}/{version_id.name}/versions/{version_id.version}" r = self.t.request( "GET", @@ -116,9 +102,6 @@ def read(self, version_id: RegistryProviderVersionID) -> RegistryProviderVersion def delete(self, version_id: RegistryProviderVersionID) -> None: """Delete a specific registry provider version""" - if not self._validate_provider_id(version_id): - raise ValueError("Invalid provider ID") - path = f"/api/v2/organizations/{version_id.organization_name}/registry-providers/{version_id.registry_name.value}/{version_id.namespace}/{version_id.name}/versions/{version_id.version}" self.t.request( "DELETE", diff --git a/tests/units/test_registry_provider_platform.py b/tests/units/test_registry_provider_platform.py new file mode 100644 index 00000000..817157aa --- /dev/null +++ b/tests/units/test_registry_provider_platform.py @@ -0,0 +1,392 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +"""Unit tests for the registry_provider_platform module.""" + +from unittest.mock import Mock, patch + +import pytest + +from pytfe._http import HTTPTransport +from pytfe.errors import ( + InvalidArchError, + InvalidNameError, + InvalidNamespaceError, + InvalidOrgError, + InvalidOSError, + InvalidVersionError, + RequiredArchError, + RequiredFilenameError, + RequiredOSError, + RequiredPrivateRegistryError, + RequiredShasumError, +) +from pytfe.models.registry_provider import RegistryName +from pytfe.models.registry_provider_platform import ( + RegistryProviderPlatform, + RegistryProviderPlatformCreateOptions, + RegistryProviderPlatformID, + RegistryProviderPlatformListOptions, +) +from pytfe.models.registry_provider_version import ( + RegistryProviderVersion, + RegistryProviderVersionID, +) +from pytfe.resources.registry_provider_platform import RegistryProviderPlatforms + + +class TestRegistryProviderPlatforms: + """Test the RegistryProviderPlatforms service class.""" + + @pytest.fixture + def mock_transport(self): + """Create a mock HTTPTransport.""" + return Mock(spec=HTTPTransport) + + @pytest.fixture + def platforms_service(self, mock_transport): + """Create a RegistryProviderPlatforms service with mocked transport.""" + return RegistryProviderPlatforms(mock_transport) + + @pytest.fixture + def valid_version_id(self): + """Create a valid version ID.""" + return RegistryProviderVersionID( + organization_name="test-org", + registry_name=RegistryName.PRIVATE, + namespace="test-namespace", + name="test-provider", + version="1.0.0", + ) + + @pytest.fixture + def valid_platform_id(self): + """Create a valid platform ID.""" + return RegistryProviderPlatformID( + organization_name="test-org", + registry_name=RegistryName.PRIVATE, + namespace="test-namespace", + name="test-provider", + version="1.0.0", + os="linux", + arch="amd64", + ) + + @pytest.fixture + def platform_api_data(self): + """Typical API response data for a single platform.""" + return { + "id": "provpltfrm-123", + "type": "registry-provider-platforms", + "attributes": { + "os": "linux", + "arch": "amd64", + "filename": "terraform-provider-test_1.0.0_linux_amd64.zip", + "shasum": "abc123def456", + "provider-binary-uploaded": False, + "permissions": { + "can-delete": True, + "can-upload-asset": True, + }, + }, + "relationships": { + "registry-provider-version": { + "data": { + "id": "provver-456", + "type": "registry-provider-versions", + } + } + }, + "links": { + "provider-binary-upload": "https://example.com/upload", + }, + } + + # ------------------------------------------------------------------------- + # ID validation tests + # ------------------------------------------------------------------------- + + def test_invalid_platform_id_fields(self): + """Test RegistryProviderPlatformID raises correct error for each invalid field.""" + base = { + "organization_name": "test-org", + "registry_name": RegistryName.PRIVATE, + "namespace": "test-namespace", + "name": "test-provider", + "version": "1.0.0", + "os": "linux", + "arch": "amd64", + } + with pytest.raises(InvalidOrgError): + RegistryProviderPlatformID(**{**base, "organization_name": ""}) + with pytest.raises(InvalidOrgError): + RegistryProviderPlatformID(**{**base, "organization_name": " "}) + with pytest.raises(InvalidNameError): + RegistryProviderPlatformID(**{**base, "name": ""}) + with pytest.raises(InvalidNamespaceError): + RegistryProviderPlatformID(**{**base, "namespace": ""}) + with pytest.raises(InvalidVersionError): + RegistryProviderPlatformID(**{**base, "version": ""}) + with pytest.raises(RequiredPrivateRegistryError): + RegistryProviderPlatformID(**{**base, "registry_name": RegistryName.PUBLIC}) + with pytest.raises(InvalidOSError): + RegistryProviderPlatformID(**{**base, "os": ""}) + with pytest.raises(InvalidArchError): + RegistryProviderPlatformID(**{**base, "arch": ""}) + + def test_valid_platform_id(self, valid_platform_id): + """Test RegistryProviderPlatformID with valid data.""" + assert valid_platform_id.organization_name == "test-org" + assert valid_platform_id.registry_name == RegistryName.PRIVATE + assert valid_platform_id.namespace == "test-namespace" + assert valid_platform_id.name == "test-provider" + assert valid_platform_id.version == "1.0.0" + assert valid_platform_id.os == "linux" + assert valid_platform_id.arch == "amd64" + + # ------------------------------------------------------------------------- + # CreateOptions validation tests + # ------------------------------------------------------------------------- + + def test_create_options_invalid_fields(self): + """Test RegistryProviderPlatformCreateOptions raises correct error for each invalid field.""" + base = { + "os": "linux", + "arch": "amd64", + "shasum": "abc123", + "filename": "provider.zip", + } + with pytest.raises(RequiredOSError): + RegistryProviderPlatformCreateOptions(**{**base, "os": ""}) + with pytest.raises(RequiredArchError): + RegistryProviderPlatformCreateOptions(**{**base, "arch": ""}) + with pytest.raises(RequiredShasumError): + RegistryProviderPlatformCreateOptions(**{**base, "shasum": ""}) + with pytest.raises(RequiredFilenameError): + RegistryProviderPlatformCreateOptions(**{**base, "filename": ""}) + + def test_create_options_valid(self): + """Test RegistryProviderPlatformCreateOptions with valid data.""" + options = RegistryProviderPlatformCreateOptions( + os="linux", + arch="amd64", + shasum="abc123def456", + filename="terraform-provider-test_1.0.0_linux_amd64.zip", + ) + assert options.os == "linux" + assert options.arch == "amd64" + assert options.shasum == "abc123def456" + assert options.filename == "terraform-provider-test_1.0.0_linux_amd64.zip" + + # ------------------------------------------------------------------------- + # create() + # ------------------------------------------------------------------------- + + def test_create_platform_success( + self, platforms_service, valid_version_id, mock_transport, platform_api_data + ): + """Test successful create operation.""" + mock_response = Mock() + mock_response.json.return_value = {"data": platform_api_data} + mock_transport.request.return_value = mock_response + + options = RegistryProviderPlatformCreateOptions( + os="linux", + arch="amd64", + shasum="abc123def456", + filename="terraform-provider-test_1.0.0_linux_amd64.zip", + ) + + result = platforms_service.create(valid_version_id, options) + + mock_transport.request.assert_called_once_with( + "POST", + path="/api/v2/organizations/test-org/registry-providers/private/test-namespace/test-provider/versions/1.0.0/platforms", + json_body={ + "data": { + "type": "registry-provider-platforms", + "attributes": { + "os": "linux", + "arch": "amd64", + "shasum": "abc123def456", + "filename": "terraform-provider-test_1.0.0_linux_amd64.zip", + }, + } + }, + ) + + assert isinstance(result, RegistryProviderPlatform) + assert result.id == "provpltfrm-123" + assert result.os == "linux" + assert result.arch == "amd64" + assert result.shasum == "abc123def456" + assert result.provider_binary_uploaded is False + assert result.permissions.can_delete is True + assert result.permissions.can_upload_asset is True + + # ------------------------------------------------------------------------- + # list() + # ------------------------------------------------------------------------- + + def test_list_platforms_success( + self, platforms_service, valid_version_id, platform_api_data + ): + """Test successful list operation.""" + second = {**platform_api_data, "id": "provpltfrm-456"} + second["attributes"] = { + **platform_api_data["attributes"], + "os": "darwin", + "arch": "arm64", + } + + with patch.object( + platforms_service, "_list", return_value=[platform_api_data, second] + ): + result = list(platforms_service.list(valid_version_id)) + + assert len(result) == 2 + assert result[0].id == "provpltfrm-123" + assert result[0].os == "linux" + assert result[0].arch == "amd64" + assert result[1].id == "provpltfrm-456" + assert result[1].os == "darwin" + assert result[1].arch == "arm64" + + def test_list_platforms_with_options( + self, platforms_service, valid_version_id, mock_transport, platform_api_data + ): + """Test list operation passes page_size param.""" + mock_response = Mock() + mock_response.json.return_value = {"data": [platform_api_data]} + mock_transport.request.return_value = mock_response + + options = RegistryProviderPlatformListOptions(page_size=10) + + with patch.object( + platforms_service, "_list", return_value=[platform_api_data] + ) as mock_list: + result = list(platforms_service.list(valid_version_id, options)) + mock_list.assert_called_once_with( + path="/api/v2/organizations/test-org/registry-providers/private/test-namespace/test-provider/versions/1.0.0/platforms", + params={"page[size]": 10}, + ) + + assert len(result) == 1 + + def test_list_platforms_empty(self, platforms_service, valid_version_id): + """Test list operation returns empty iterator when no platforms exist.""" + with patch.object(platforms_service, "_list", return_value=[]): + result = list(platforms_service.list(valid_version_id)) + + assert result == [] + + # ------------------------------------------------------------------------- + # read() + # ------------------------------------------------------------------------- + + def test_read_platform_success( + self, platforms_service, valid_platform_id, mock_transport, platform_api_data + ): + """Test successful read operation.""" + mock_response = Mock() + mock_response.json.return_value = {"data": platform_api_data} + mock_transport.request.return_value = mock_response + + result = platforms_service.read(valid_platform_id) + + mock_transport.request.assert_called_once_with( + "GET", + path="/api/v2/organizations/test-org/registry-providers/private/test-namespace/test-provider/versions/1.0.0/platforms/linux/amd64", + ) + + assert isinstance(result, RegistryProviderPlatform) + assert result.id == "provpltfrm-123" + assert result.os == "linux" + assert result.arch == "amd64" + + # ------------------------------------------------------------------------- + # delete() + # ------------------------------------------------------------------------- + + def test_delete_platform_success( + self, platforms_service, valid_platform_id, mock_transport + ): + """Test successful delete operation.""" + result = platforms_service.delete(valid_platform_id) + + mock_transport.request.assert_called_once_with( + "DELETE", + path="/api/v2/organizations/test-org/registry-providers/private/test-namespace/test-provider/versions/1.0.0/platforms/linux/amd64", + ) + + assert result is None + + # ------------------------------------------------------------------------- + # _registry_provider_platform_from() + # ------------------------------------------------------------------------- + + def test_platform_from_full_data(self, platforms_service, platform_api_data): + """Test _registry_provider_platform_from with full API response including relationships and links.""" + result = platforms_service._registry_provider_platform_from(platform_api_data) + + assert isinstance(result, RegistryProviderPlatform) + assert result.id == "provpltfrm-123" + assert result.os == "linux" + assert result.arch == "amd64" + assert result.filename == "terraform-provider-test_1.0.0_linux_amd64.zip" + assert result.shasum == "abc123def456" + assert result.provider_binary_uploaded is False + assert result.permissions.can_delete is True + assert result.permissions.can_upload_asset is True + # registry-provider-version relation parsed as typed stub + assert isinstance(result.registry_provider_version, RegistryProviderVersion) + assert result.registry_provider_version.id == "provver-456" + # links preserved + assert result.links is not None + assert "provider-binary-upload" in result.links + + def test_platform_from_no_relationships(self, platforms_service): + """Test _registry_provider_platform_from when relationships are absent.""" + data = { + "id": "provpltfrm-789", + "type": "registry-provider-platforms", + "attributes": { + "os": "windows", + "arch": "amd64", + "filename": "terraform-provider-test_1.0.0_windows_amd64.zip", + "shasum": "deadbeef", + "provider-binary-uploaded": True, + "permissions": { + "can-delete": False, + "can-upload-asset": False, + }, + }, + } + + result = platforms_service._registry_provider_platform_from(data) + + assert result.id == "provpltfrm-789" + assert result.os == "windows" + assert result.arch == "amd64" + assert result.registry_provider_version is None + assert result.links is None + + def test_platform_from_null_version_relationship(self, platforms_service): + """Test _registry_provider_platform_from when registry-provider-version data is null.""" + data = { + "id": "provpltfrm-abc", + "type": "registry-provider-platforms", + "attributes": { + "os": "linux", + "arch": "arm64", + "filename": "provider.zip", + "shasum": "abc123", + "provider-binary-uploaded": False, + "permissions": {"can-delete": True, "can-upload-asset": True}, + }, + "relationships": {"registry-provider-version": {"data": None}}, + } + + result = platforms_service._registry_provider_platform_from(data) + + assert result.registry_provider_version is None diff --git a/tests/units/test_registry_provider_version.py b/tests/units/test_registry_provider_version.py index e291e602..a1239fc0 100644 --- a/tests/units/test_registry_provider_version.py +++ b/tests/units/test_registry_provider_version.py @@ -10,11 +10,15 @@ from pytfe._http import HTTPTransport from pytfe.errors import ( InvalidKeyIDError, + InvalidNameError, + InvalidNamespaceError, + InvalidOrgError, InvalidVersionError, RequiredPrivateRegistryError, ) from pytfe.models.registry_provider import ( RegistryName, + RegistryProvider, RegistryProviderID, ) from pytfe.models.registry_provider_version import ( @@ -59,34 +63,52 @@ def valid_version_id(self): version="1.0.0", ) - def test_validate_provider_id_success(self, versions_service, valid_provider_id): - """Test _validate_provider_id with valid provider ID.""" - result = versions_service._validate_provider_id(valid_provider_id) - assert result is True - - def test_validate_provider_id_invalid_organization( - self, versions_service, valid_provider_id - ): - """Test _validate_provider_id with invalid organization name.""" - valid_provider_id.organization_name = "" - result = versions_service._validate_provider_id(valid_provider_id) - assert result is False - def test_create_version_validations(self, versions_service): - """Test create method validations.""" - # Test with invalid provider ID - invalid_provider_id = RegistryProviderID( - organization_name="", - registry_name=RegistryName.PRIVATE, - namespace="test-namespace", - name="test-provider", - ) - options = RegistryProviderVersionCreateOptions( - version="1.0.0", **{"key-id": "test-key-id"}, protocols=["5.0"] - ) + """Test create method raises error when constructing invalid provider ID.""" + with pytest.raises(InvalidOrgError): + RegistryProviderID( + organization_name="", + registry_name=RegistryName.PRIVATE, + namespace="test-namespace", + name="test-provider", + ) - with pytest.raises(ValueError, match="Invalid provider ID"): - versions_service.create(invalid_provider_id, options) + def test_invalid_provider_id_fields(self): + """Test RegistryProviderID raises correct error for each invalid field.""" + base = { + "organization_name": "test-org", + "registry_name": RegistryName.PRIVATE, + "namespace": "test-namespace", + "name": "test-provider", + } + with pytest.raises(InvalidOrgError): + RegistryProviderID(**{**base, "organization_name": ""}) + with pytest.raises(InvalidOrgError): + RegistryProviderID(**{**base, "organization_name": " "}) + with pytest.raises(InvalidNameError): + RegistryProviderID(**{**base, "name": ""}) + with pytest.raises(InvalidNamespaceError): + RegistryProviderID(**{**base, "namespace": ""}) + + def test_invalid_version_id_fields(self): + """Test RegistryProviderVersionID raises correct error for each invalid field.""" + base = { + "organization_name": "test-org", + "registry_name": RegistryName.PRIVATE, + "namespace": "test-namespace", + "name": "test-provider", + "version": "1.0.0", + } + with pytest.raises(InvalidOrgError): + RegistryProviderVersionID(**{**base, "organization_name": ""}) + with pytest.raises(InvalidNameError): + RegistryProviderVersionID(**{**base, "name": ""}) + with pytest.raises(InvalidNamespaceError): + RegistryProviderVersionID(**{**base, "namespace": ""}) + with pytest.raises(InvalidVersionError): + RegistryProviderVersionID(**{**base, "version": ""}) + with pytest.raises(RequiredPrivateRegistryError): + RegistryProviderVersionID(**{**base, "registry_name": RegistryName.PUBLIC}) def test_create_version_requires_private_registry( self, versions_service, mock_transport @@ -240,17 +262,15 @@ def test_list_versions_success_without_options( assert result[1].shasums_uploaded is True def test_read_version_validations(self, versions_service): - """Test read method with invalid version ID.""" - invalid_version_id = RegistryProviderVersionID( - organization_name="", - registry_name=RegistryName.PRIVATE, - namespace="test-namespace", - name="test-provider", - version="1.0.0", - ) - - with pytest.raises(ValueError, match="Invalid provider ID"): - versions_service.read(invalid_version_id) + """Test read method raises error when constructing invalid version ID.""" + with pytest.raises(InvalidOrgError): + RegistryProviderVersionID( + organization_name="", + registry_name=RegistryName.PRIVATE, + namespace="test-namespace", + name="test-provider", + version="1.0.0", + ) def test_read_version_success( self, versions_service, valid_version_id, mock_transport @@ -359,10 +379,8 @@ def test_registry_provider_version_from_success(self, versions_service): assert result.id == "provver-123" assert result.version == "1.0.0" assert result.key_id == "test-key-id" - assert result.registry_provider == { - "id": "prov-123", - "type": "registry-providers", - } + assert isinstance(result.registry_provider, RegistryProvider) + assert result.registry_provider.id == "prov-123" assert result.registry_provider_platforms is not None assert len(result.registry_provider_platforms) == 2