diff --git a/examples/registry_provider_version.py b/examples/registry_provider_version.py new file mode 100644 index 0000000..4da1c83 --- /dev/null +++ b/examples/registry_provider_version.py @@ -0,0 +1,244 @@ +from __future__ import annotations + +import argparse +import os + +from pytfe import TFEClient, TFEConfig +from pytfe.models import ( + RegistryProviderID, + RegistryProviderVersionCreateOptions, + RegistryProviderVersionID, + RegistryProviderVersionListOptions, +) + + +def _print_header(title: str): + print("\n" + "=" * 80) + print(title) + print("=" * 80) + + +def main(): + parser = argparse.ArgumentParser( + description="Registry Provider Versions 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( + "--page-size", + type=int, + default=100, + help="Page size for fetching versions", + ) + parser.add_argument("--create", action="store_true", help="Create a test version") + parser.add_argument("--read", action="store_true", help="Read a specific version") + parser.add_argument( + "--delete", action="store_true", help="Delete a specific version" + ) + parser.add_argument("--version", help="Version number (e.g., 1.0.0)") + parser.add_argument("--key-id", help="GPG key ID for version signing") + parser.add_argument( + "--protocols", + nargs="+", + help="Supported protocols (e.g., 5.0 6.0)", + ) + args = parser.parse_args() + + cfg = TFEConfig(address=args.address, token=args.token) + client = TFEClient(cfg) + + # 1) List all versions for the registry provider + _print_header( + f"Listing versions for {args.registry_name}/{args.namespace}/{args.name}" + ) + provider_id = RegistryProviderID( + organization_name=args.organization, + registry_name=args.registry_name, + namespace=args.namespace, + name=args.name, + ) + + options = RegistryProviderVersionListOptions( + page_size=args.page_size, + ) + + version_count = 0 + for version in client.registry_provider_versions.list( + provider_id=provider_id, + options=options, + ): + version_count += 1 + print(f"- Version {version.version} (ID: {version.id})") + print(f" Created: {version.created_at}") + print(f" Updated: {version.updated_at}") + print(f" Key ID: {version.key_id}") + print(f" Protocols: {', '.join(version.protocols)}") + print(f" Shasums Uploaded: {version.shasums_uploaded}") + print(f" Shasums Signature Uploaded: {version.shasums_sig_uploaded}") + if version.permissions: + print(" Permissions:") + print(f" Can Delete: {version.permissions.can_delete}") + print(f" Can Upload Asset: {version.permissions.can_upload_asset}") + print() + + if version_count == 0: + print("No versions found.") + else: + print(f"Total: {version_count} versions") + + # 2) Create a new version (if --create flag is provided) + if args.create: + if not args.version: + print("Error: --version is required for create operation") + return + if not args.key_id: + print("Error: --key-id is required for create operation") + return + if not args.protocols: + print("Error: --protocols is required for create operation") + return + + _print_header(f"Creating new version: {args.version}") + + create_options = RegistryProviderVersionCreateOptions( + version=args.version, + key_id=args.key_id, + protocols=args.protocols, + ) + + new_version = client.registry_provider_versions.create( + provider_id=provider_id, + options=create_options, + ) + + print(f"Created version: {new_version.id}") + print(f" Version: {new_version.version}") + print(f" Created: {new_version.created_at}") + print(f" Key ID: {new_version.key_id}") + print(f" Protocols: {', '.join(new_version.protocols)}") + print(f" Shasums Uploaded: {new_version.shasums_uploaded}") + print(f" Shasums Signature Uploaded: {new_version.shasums_sig_uploaded}") + + # Show upload URLs if available in links + if new_version.links: + print("\n Upload URLs:") + if "shasums-upload" in new_version.links: + print(f" Shasums: {new_version.links['shasums-upload']}") + if "shasums-sig-upload" in new_version.links: + print( + f" Shasums Signature: {new_version.links['shasums-sig-upload']}" + ) + + # 3) Read a specific version (if --read flag is provided) + if args.read: + if not args.version: + print("Error: --version is required for read operation") + return + + _print_header(f"Reading version: {args.version}") + + version_id = RegistryProviderVersionID( + organization_name=args.organization, + registry_name=args.registry_name, + namespace=args.namespace, + name=args.name, + version=args.version, + ) + + version = client.registry_provider_versions.read(version_id) + + print(f"Version ID: {version.id}") + print(f" Version: {version.version}") + print(f" Created: {version.created_at}") + print(f" Updated: {version.updated_at}") + print(f" Key ID: {version.key_id}") + print(f" Protocols: {', '.join(version.protocols)}") + print(f" Shasums Uploaded: {version.shasums_uploaded}") + print(f" Shasums Signature Uploaded: {version.shasums_sig_uploaded}") + + if version.permissions: + print(" Permissions:") + print(f" Can Delete: {version.permissions.can_delete}") + print(f" Can Upload Asset: {version.permissions.can_upload_asset}") + + # Show links if available + if version.links: + print(" Links:") + for key, value in version.links.items(): + print(f" {key}: {value}") + + # 4) Delete a version (if --delete flag is provided) + if args.delete: + if not args.version: + print("Error: --version is required for delete operation") + return + + _print_header(f"Deleting version: {args.version}") + + version_id = RegistryProviderVersionID( + organization_name=args.organization, + registry_name=args.registry_name, + namespace=args.namespace, + name=args.name, + version=args.version, + ) + + # First read the version to show what's being deleted + try: + version_to_delete = client.registry_provider_versions.read(version_id) + print("Version to delete:") + print(f" ID: {version_to_delete.id}") + print(f" Version: {version_to_delete.version}") + print(f" Protocols: {', '.join(version_to_delete.protocols)}") + print(f" Key ID: {version_to_delete.key_id}") + except Exception as e: + print(f"Error reading version: {e}") + return + + # Delete the version + client.registry_provider_versions.delete(version_id) + print(f"\n Successfully deleted version: {args.version}") + + # List remaining versions + _print_header("Listing versions after deletion") + provider_id = RegistryProviderID( + organization_name=args.organization, + registry_name=args.registry_name, + namespace=args.namespace, + name=args.name, + ) + + options = RegistryProviderVersionListOptions( + page_size=args.page_size, + ) + print("Remaining versions:") + remaining_count = 0 + for version in client.registry_provider_versions.list( + provider_id=provider_id, + options=options, + ): + remaining_count += 1 + print( + f"- Version {version.version}: " + f" protocols={', '.join(version.protocols)}, " + f" shasums_uploaded={version.shasums_uploaded}" + ) + + if remaining_count == 0: + print("No versions remaining.") + else: + print(f"\nTotal: {remaining_count} versions") + + +if __name__ == "__main__": + main() diff --git a/src/pytfe/client.py b/src/pytfe/client.py index 3894886..816df1e 100644 --- a/src/pytfe/client.py +++ b/src/pytfe/client.py @@ -21,6 +21,7 @@ from .resources.query_run import QueryRuns from .resources.registry_module import RegistryModules from .resources.registry_provider import RegistryProviders +from .resources.registry_provider_version import RegistryProviderVersions from .resources.reserved_tag_key import ReservedTagKey from .resources.run import Runs from .resources.run_event import RunEvents @@ -71,6 +72,7 @@ def __init__(self, config: TFEConfig | None = None): self.workspaces = Workspaces(self._transport) self.registry_modules = RegistryModules(self._transport) self.registry_providers = RegistryProviders(self._transport) + self.registry_provider_versions = RegistryProviderVersions(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 3eac2be..5ff6149 100644 --- a/src/pytfe/errors.py +++ b/src/pytfe/errors.py @@ -460,3 +460,25 @@ class InvalidPolicyEvaluationIDError(InvalidValues): def __init__(self, message: str = "invalid value for policy evaluation ID"): super().__init__(message) + + +# Registry Provider Version errors +class RequiredPrivateRegistryError(RequiredFieldMissing): + """Raised when a required private registry field is missing.""" + + def __init__(self, message: str = "only private registry is allowed"): + super().__init__(message) + + +class InvalidVersionError(InvalidValues): + """Raised when an invalid version is provided.""" + + def __init__(self, message: str = "invalid value for version"): + super().__init__(message) + + +class InvalidKeyIDError(InvalidValues): + """Raised when an invalid key ID is provided.""" + + def __init__(self, message: str = "invalid value for key-id"): + super().__init__(message) diff --git a/src/pytfe/models/__init__.py b/src/pytfe/models/__init__.py index f3eb33b..520c22b 100644 --- a/src/pytfe/models/__init__.py +++ b/src/pytfe/models/__init__.py @@ -203,6 +203,13 @@ RegistryProviderPermissions, RegistryProviderReadOptions, ) +from .registry_provider_version import ( + RegistryProviderVersion, + RegistryProviderVersionCreateOptions, + RegistryProviderVersionID, + RegistryProviderVersionListOptions, + RegistryProviderVersionPermissions, +) # ── Reserved Tag Keys ───────────────────────────────────────────────────────── from .reserved_tag_key import ( @@ -430,6 +437,12 @@ "RegistryProviderListOptions", "RegistryProviderPermissions", "RegistryProviderReadOptions", + # Registry provider versions + "RegistryProviderVersion", + "RegistryProviderVersionCreateOptions", + "RegistryProviderVersionID", + "RegistryProviderVersionListOptions", + "RegistryProviderVersionPermissions", # Query runs "QueryRun", "QueryRunCancelOptions", diff --git a/src/pytfe/models/registry_provider_version.py b/src/pytfe/models/registry_provider_version.py new file mode 100644 index 0000000..397e385 --- /dev/null +++ b/src/pytfe/models/registry_provider_version.py @@ -0,0 +1,175 @@ +from __future__ import annotations + +from datetime import datetime +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from ..errors import ( + InvalidKeyIDError, + InvalidVersionError, + RequiredPrivateRegistryError, +) +from ..utils import valid_string_id +from .registry_provider import ( + RegistryName, + RegistryProviderID, +) + + +class RegistryProviderVersionPermissions(BaseModel): + """Registry provider version 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 RegistryProviderVersion(BaseModel): + """Registry provider version model.""" + + 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") + + # Relations + registry_provider: dict[str, Any] | None = Field( + alias="registry-provider", default=None + ) + registry_provider_platforms: list[dict[str, Any]] | None = Field( + alias="platforms", default=None + ) + + # Links + links: dict[str, Any] | None = None + + def shasums_upload_url(self) -> str: + """ShasumsUploadURL returns the upload URL to upload shasums if one is available""" + if self.links is None: + raise ValueError( + "The registry provider version does not contain a shasums upload link" + ) + upload_url = str(self.links.get("shasums-upload")) + if not upload_url: + raise ValueError( + "The registry provider version does not contain a shasums upload link" + ) + + if upload_url == "": + raise ValueError( + "The registry provider version shasums upload URL is empty" + ) + + return upload_url + + def shasums_sig_upload_url(self) -> str: + """ShasumsSigUploadURL returns the URL to upload a shasums sig""" + if self.links is None: + raise ValueError( + "The registry provider version does not contain a shasums sig upload link" + ) + upload_url = str(self.links.get("shasums-sig-upload")) + if not upload_url: + raise ValueError( + "The registry provider version does not contain a shasums sig upload link" + ) + + if upload_url == "": + raise ValueError( + "The registry provider version shasums sig upload URL is empty" + ) + + return upload_url + + def shasums_download_url(self) -> str: + """ShasumsDownloadURL returns the URL to download the shasums for the registry version""" + if self.links is None: + raise ValueError( + "The registry provider version does not contain a shasums download link" + ) + download_url = str(self.links.get("shasums-download")) + if not download_url: + raise ValueError( + "The registry provider version does not contain a shasums download link" + ) + + if download_url == "": + raise ValueError( + "The registry provider version shasums download URL is empty" + ) + + return download_url + + def shasums_sig_download_url(self) -> str: + """ShasumsSigDownloadURL returns the URL to download the shasums sig for the registry version""" + if self.links is None: + raise ValueError( + "The registry provider version does not contain a shasums sig download link" + ) + download_url = str(self.links.get("shasums-sig-download")) + if not download_url: + raise ValueError( + "The registry provider version does not contain a shasums sig download link" + ) + + if download_url == "": + raise ValueError( + "The registry provider version shasums sig download URL is empty" + ) + + return download_url + + +class RegistryProviderVersionID(RegistryProviderID): + """Registry provider version identifier. + + This extends RegistryProviderID with a version field to uniquely + identify a specific version of a provider. + """ + + version: str + + @model_validator(mode="after") + def valid(self) -> RegistryProviderVersionID: + if not valid_string_id(self.version): + raise InvalidVersionError() + if self.registry_name != RegistryName.PRIVATE: + raise RequiredPrivateRegistryError() + return self + + +class RegistryProviderVersionCreateOptions(BaseModel): + """Options for creating a registry provider version.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + version: str + key_id: str = Field(alias="key-id") + protocols: list[str] + + # validation method for version and key_id + @model_validator(mode="after") + def valid(self) -> RegistryProviderVersionCreateOptions: + if not valid_string_id(self.version): + raise InvalidVersionError() + if not valid_string_id(self.key_id): + raise InvalidKeyIDError() + return self + + +class RegistryProviderVersionListOptions(BaseModel): + """Options for listing registry provider versions.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + page_number: int | None = Field(alias="page[number]", default=None) + page_size: int | None = Field(alias="page[size]", default=None) diff --git a/src/pytfe/resources/registry_provider_version.py b/src/pytfe/resources/registry_provider_version.py new file mode 100644 index 0000000..f2d4fb3 --- /dev/null +++ b/src/pytfe/resources/registry_provider_version.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +from ..errors import ( + RequiredPrivateRegistryError, +) +from ..models.registry_provider import ( + RegistryName, + RegistryProviderID, +) +from ..models.registry_provider_version import ( + RegistryProviderVersion, + RegistryProviderVersionCreateOptions, + RegistryProviderVersionID, + RegistryProviderVersionListOptions, +) +from ..utils import valid_string_id +from ._base import _Service + + +class RegistryProviderVersions(_Service): + """Registry providers service for managing Terraform registry providers.""" + + def create( + self, + provider_id: RegistryProviderID, + 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" + attributes = options.model_dump(by_alias=True, exclude_none=True) + payload = { + "data": { + "type": "registry-provider-versions", + "attributes": attributes, + } + } + r = self.t.request( + "POST", + path=path, + json_body=payload, + ) + 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: + """Parse a registry provider version from API response data.""" + + attrs = data.get("attributes", {}) + relationships = data.get("relationships", {}) + attrs["id"] = data.get("id") + + # Parse relationships + if "registry-provider" in relationships: + attrs["registry_provider"] = relationships["registry-provider"].get( + "data", {} + ) + + if "platforms" in relationships: + attrs["registry_provider_platforms"] = relationships["platforms"].get( + "data", [] + ) + + return RegistryProviderVersion.model_validate(attrs) + + def list( + self, + provider_id: RegistryProviderID, + 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): + yield self._registry_provider_version_from(item) + + 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", + path=path, + ) + data = r.json().get("data", {}) + return self._registry_provider_version_from(data) + + 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", + path=path, + ) + return None diff --git a/src/pytfe/utils.py b/src/pytfe/utils.py index d6e9b38..02c43cc 100644 --- a/src/pytfe/utils.py +++ b/src/pytfe/utils.py @@ -37,7 +37,7 @@ WorkspaceUpdateOptions, ) -_STRING_ID_PATTERN = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_-]{2,}$") +_STRING_ID_PATTERN = re.compile(r"^[^/\s]+$") _WS_ID_RE = re.compile(r"^ws-[A-Za-z0-9]+$") _VERSION_PATTERN = re.compile( r"^\d+\.\d+\.\d+(?:-[a-zA-Z0-9]+(?:\.[a-zA-Z0-9]+)*)?(?:\+[a-zA-Z0-9]+(?:\.[a-zA-Z0-9]+)*)?$" diff --git a/tests/units/test_apply.py b/tests/units/test_apply.py index 458c87b..62f7509 100644 --- a/tests/units/test_apply.py +++ b/tests/units/test_apply.py @@ -25,7 +25,7 @@ def test_read_apply_validation_errors(self): self.applies.read("") with self.assertRaises(InvalidApplyIDError): - self.applies.read("a") + self.applies.read("! / nope") # Contains spaces and slashes def test_read_apply_success(self): """Test successful apply read.""" diff --git a/tests/units/test_project.py b/tests/units/test_project.py index 262787a..801a29f 100644 --- a/tests/units/test_project.py +++ b/tests/units/test_project.py @@ -345,7 +345,9 @@ def test_list_tag_bindings_invalid_project_id(self): with pytest.raises( ValueError, match="Project ID is required and must be valid" ): - self.projects_service.list_tag_bindings("x") # Too short + self.projects_service.list_tag_bindings( + "! / nope" + ) # Contains spaces and slashes def test_list_effective_tag_bindings_success(self): """Test successful listing of effective tag bindings""" @@ -541,5 +543,5 @@ def test_delete_tag_bindings_invalid_project_id(self): ValueError, match="Project ID is required and must be valid" ): self.projects_service.delete_tag_bindings( - "ab" - ) # Too short (needs at least 3 chars) + "bad/id" + ) # Contains forward slash diff --git a/tests/units/test_registry_provider_version.py b/tests/units/test_registry_provider_version.py new file mode 100644 index 0000000..46b76bd --- /dev/null +++ b/tests/units/test_registry_provider_version.py @@ -0,0 +1,402 @@ +"""Unit tests for the registry_provider_version module.""" + +from unittest.mock import Mock, patch + +import pytest + +from pytfe._http import HTTPTransport +from pytfe.errors import ( + InvalidKeyIDError, + InvalidVersionError, + RequiredPrivateRegistryError, +) +from pytfe.models.registry_provider import ( + RegistryName, + RegistryProviderID, +) +from pytfe.models.registry_provider_version import ( + RegistryProviderVersion, + RegistryProviderVersionCreateOptions, + RegistryProviderVersionID, +) +from pytfe.resources.registry_provider_version import RegistryProviderVersions + + +class TestRegistryProviderVersions: + """Test the RegistryProviderVersions service class.""" + + @pytest.fixture + def mock_transport(self): + """Create a mock HTTPTransport.""" + return Mock(spec=HTTPTransport) + + @pytest.fixture + def versions_service(self, mock_transport): + """Create a RegistryProviderVersions service with mocked transport.""" + return RegistryProviderVersions(mock_transport) + + @pytest.fixture + def valid_provider_id(self): + """Create a valid provider ID.""" + return RegistryProviderID( + organization_name="test-org", + registry_name=RegistryName.PRIVATE, + namespace="test-namespace", + name="test-provider", + ) + + @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", + ) + + 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"] + ) + + with pytest.raises(ValueError, match="Invalid provider ID"): + versions_service.create(invalid_provider_id, options) + + def test_create_version_requires_private_registry( + self, versions_service, mock_transport + ): + """Test create method requires private registry.""" + public_provider_id = RegistryProviderID( + organization_name="test-org", + registry_name=RegistryName.PUBLIC, + namespace="hashicorp", + name="aws", + ) + options = RegistryProviderVersionCreateOptions( + version="1.0.0", **{"key-id": "test-key-id"}, protocols=["5.0"] + ) + + with pytest.raises(RequiredPrivateRegistryError): + versions_service.create(public_provider_id, options) + + def test_create_version_success( + self, versions_service, valid_provider_id, mock_transport + ): + """Test successful create operation.""" + mock_response_data = { + "data": { + "id": "provver-123", + "type": "registry-provider-versions", + "attributes": { + "version": "1.0.0", + "created-at": "2023-01-01T12:00:00Z", + "updated-at": "2023-01-01T12:00:00Z", + "key-id": "test-key-id", + "protocols": ["5.0"], + "shasums-uploaded": False, + "shasums-sig-uploaded": False, + "permissions": { + "can-delete": True, + "can-upload-asset": True, + }, + }, + "relationships": { + "registry-provider": { + "data": {"id": "prov-123", "type": "registry-providers"} + } + }, + "links": { + "shasums-upload": "https://example.com/upload", + "shasums-sig-upload": "https://example.com/sig-upload", + }, + } + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + mock_transport.request.return_value = mock_response + + options = RegistryProviderVersionCreateOptions( + version="1.0.0", **{"key-id": "test-key-id"}, protocols=["5.0"] + ) + + result = versions_service.create(valid_provider_id, options) + + mock_transport.request.assert_called_once_with( + "POST", + path="/api/v2/organizations/test-org/registry-providers/private/test-namespace/test-provider/versions", + json_body={ + "data": { + "type": "registry-provider-versions", + "attributes": { + "version": "1.0.0", + "key-id": "test-key-id", + "protocols": ["5.0"], + }, + } + }, + ) + + assert isinstance(result, RegistryProviderVersion) + assert result.id == "provver-123" + assert result.version == "1.0.0" + assert result.key_id == "test-key-id" + assert result.protocols == ["5.0"] + assert result.permissions.can_delete is True + + def test_list_versions_success_without_options( + self, versions_service, valid_provider_id, mock_transport + ): + """Test successful list operation without options.""" + mock_response_data = { + "data": [ + { + "id": "provver-123", + "type": "registry-provider-versions", + "attributes": { + "version": "1.0.0", + "created-at": "2023-01-01T12:00:00Z", + "updated-at": "2023-01-01T12:00:00Z", + "key-id": "test-key-id", + "protocols": ["5.0"], + "shasums-uploaded": False, + "shasums-sig-uploaded": False, + "permissions": { + "can-delete": True, + "can-upload-asset": True, + }, + }, + }, + { + "id": "provver-456", + "type": "registry-provider-versions", + "attributes": { + "version": "1.1.0", + "created-at": "2023-02-01T12:00:00Z", + "updated-at": "2023-02-01T12:00:00Z", + "key-id": "test-key-id-2", + "protocols": ["5.0", "6.0"], + "shasums-uploaded": True, + "shasums-sig-uploaded": True, + "permissions": { + "can-delete": True, + "can-upload-asset": False, + }, + }, + }, + ], + "meta": { + "pagination": { + "current-page": 1, + "total-pages": 1, + "prev-page": None, + "next-page": None, + "total-count": 2, + } + }, + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + mock_transport.request.return_value = mock_response + + with patch.object( + versions_service, "_list", return_value=mock_response_data["data"] + ): + result = list(versions_service.list(valid_provider_id)) + + assert len(result) == 2 + assert result[0].id == "provver-123" + assert result[0].version == "1.0.0" + assert result[0].shasums_uploaded is False + assert result[1].id == "provver-456" + assert result[1].version == "1.1.0" + 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) + + def test_read_version_success( + self, versions_service, valid_version_id, mock_transport + ): + """Test successful read operation.""" + mock_response_data = { + "data": { + "id": "provver-789", + "type": "registry-provider-versions", + "attributes": { + "version": "1.0.0", + "created-at": "2023-01-01T12:00:00Z", + "updated-at": "2023-01-01T12:00:00Z", + "key-id": "test-key-id", + "protocols": ["5.0", "6.0"], + "shasums-uploaded": True, + "shasums-sig-uploaded": True, + "permissions": { + "can-delete": True, + "can-upload-asset": False, + }, + }, + "relationships": { + "registry-provider": { + "data": {"id": "prov-123", "type": "registry-providers"} + }, + "platforms": { + "data": [ + {"id": "plat-123", "type": "registry-provider-platforms"} + ] + }, + }, + "links": { + "shasums-download": "https://example.com/download", + "shasums-sig-download": "https://example.com/sig-download", + }, + } + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + mock_transport.request.return_value = mock_response + + result = versions_service.read(valid_version_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", + ) + + assert isinstance(result, RegistryProviderVersion) + assert result.id == "provver-789" + assert result.version == "1.0.0" + assert result.key_id == "test-key-id" + assert result.protocols == ["5.0", "6.0"] + assert result.shasums_uploaded is True + assert result.shasums_sig_uploaded is True + + def test_delete_version_success( + self, versions_service, valid_version_id, mock_transport + ): + """Test successful delete operation.""" + result = versions_service.delete(valid_version_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", + ) + + assert result is None + + def test_registry_provider_version_from_success(self, versions_service): + """Test _registry_provider_version_from with valid data.""" + data = { + "id": "provver-123", + "type": "registry-provider-versions", + "attributes": { + "version": "1.0.0", + "created-at": "2023-01-01T12:00:00Z", + "updated-at": "2023-01-01T12:00:00Z", + "key-id": "test-key-id", + "protocols": ["5.0"], + "shasums-uploaded": False, + "shasums-sig-uploaded": False, + "permissions": { + "can-delete": True, + "can-upload-asset": True, + }, + }, + "relationships": { + "registry-provider": { + "data": {"id": "prov-123", "type": "registry-providers"} + }, + "platforms": { + "data": [ + {"id": "plat-123", "type": "registry-provider-platforms"}, + {"id": "plat-456", "type": "registry-provider-platforms"}, + ] + }, + }, + } + + result = versions_service._registry_provider_version_from(data) + + assert isinstance(result, RegistryProviderVersion) + 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 result.registry_provider_platforms is not None + assert len(result.registry_provider_platforms) == 2 + + def test_create_options_validation_invalid_version(self): + """Test RegistryProviderVersionCreateOptions with invalid version.""" + with pytest.raises(InvalidVersionError): + RegistryProviderVersionCreateOptions( + version="", **{"key-id": "test-key-id"}, protocols=["5.0"] + ) + + def test_create_options_validation_invalid_key_id(self): + """Test RegistryProviderVersionCreateOptions with invalid key_id.""" + with pytest.raises(InvalidKeyIDError): + RegistryProviderVersionCreateOptions( + version="1.0.0", **{"key-id": ""}, protocols=["5.0"] + ) + + def test_create_options_validation_success(self): + """Test RegistryProviderVersionCreateOptions with valid data.""" + options = RegistryProviderVersionCreateOptions( + version="1.0.0", **{"key-id": "test-key-id"}, protocols=["5.0", "6.0"] + ) + assert options.version == "1.0.0" + assert options.key_id == "test-key-id" + assert options.protocols == ["5.0", "6.0"] + + def test_version_id_validation_success(self): + """Test RegistryProviderVersionID with valid data.""" + version_id = RegistryProviderVersionID( + organization_name="test-org", + registry_name=RegistryName.PRIVATE, + namespace="test-namespace", + name="test-provider", + version="1.0.0", + ) + assert version_id.organization_name == "test-org" + assert version_id.registry_name == RegistryName.PRIVATE + assert version_id.namespace == "test-namespace" + assert version_id.name == "test-provider" + assert version_id.version == "1.0.0"