diff --git a/.release-please-manifest.json b/.release-please-manifest.json index d2d60a3d..a36746b8 100644 --- a/.release-please-manifest.json +++ b/.release-please-manifest.json @@ -1,3 +1,3 @@ { - ".": "0.24.0" + ".": "0.25.0" } \ No newline at end of file diff --git a/.stats.yml b/.stats.yml index 1d92b3c5..99de8068 100644 --- a/.stats.yml +++ b/.stats.yml @@ -1,4 +1,4 @@ configured_endpoints: 640 -openapi_spec_url: https://storage.googleapis.com/stainless-sdk-openapi-specs/gcore%2Fgcore-4079df5b80ccb3ccbdfbfab9543db244c7e236675b3ef9fd4f9b9f21e77cbb19.yml -openapi_spec_hash: 2ab81d6ee1696810acf27cfbfd559700 -config_hash: 12d04b6067da5a07e6f7f104987c0360 +openapi_spec_url: https://storage.googleapis.com/stainless-sdk-openapi-specs/gcore%2Fgcore-739b0d84ccefaa883a2334ad16738cc87ec625ded28ee0e2e11781f17ba36580.yml +openapi_spec_hash: deb6a4c127edd808cfd8c8eb9eb9610b +config_hash: 1607041fe51cd832db709af848355624 diff --git a/CHANGELOG.md b/CHANGELOG.md index fb1287ee..d954d7ed 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,18 @@ # Changelog +## 0.25.0 (2025-12-12) + +Full Changelog: [v0.24.0...v0.25.0](https://github.com/G-Core/gcore-python/compare/v0.24.0...v0.25.0) + +### ⚠ BREAKING CHANGES + +* **cloud:** streamline vip connected and candidate ports + +### Bug Fixes + +* **cloud:** fix vip examples ([f4f6c46](https://github.com/G-Core/gcore-python/commit/f4f6c46f6352fb55eae9d91a2060f24289ed4dda)) +* **cloud:** streamline vip connected and candidate ports ([958b2a7](https://github.com/G-Core/gcore-python/commit/958b2a735265407c29cc2ccd4b06566f9151dd15)) + ## 0.24.0 (2025-12-10) Full Changelog: [v0.23.0...v0.24.0](https://github.com/G-Core/gcore-python/compare/v0.23.0...v0.24.0) diff --git a/api.md b/api.md index 8ae67ebb..06fef89f 100644 --- a/api.md +++ b/api.md @@ -311,22 +311,38 @@ Methods: Types: ```python -from gcore.types.cloud.reserved_fixed_ips import ( - CandidatePort, - CandidatePortList, - ConnectedPort, - ConnectedPortList, - IPWithSubnet, -) +from gcore.types.cloud.reserved_fixed_ips import IPWithSubnet +``` + +Methods: + +- client.cloud.reserved_fixed_ips.vip.toggle(port_id, \*, project_id, region_id, \*\*params) -> ReservedFixedIP + +#### CandidatePorts + +Types: + +```python +from gcore.types.cloud.reserved_fixed_ips.vip import CandidatePort, CandidatePortList +``` + +Methods: + +- client.cloud.reserved_fixed_ips.vip.candidate_ports.list(port_id, \*, project_id, region_id) -> CandidatePortList + +#### ConnectedPorts + +Types: + +```python +from gcore.types.cloud.reserved_fixed_ips.vip import ConnectedPort, ConnectedPortList ``` Methods: -- client.cloud.reserved_fixed_ips.vip.list_candidate_ports(port_id, \*, project_id, region_id) -> CandidatePortList -- client.cloud.reserved_fixed_ips.vip.list_connected_ports(port_id, \*, project_id, region_id) -> ConnectedPortList -- client.cloud.reserved_fixed_ips.vip.replace_connected_ports(port_id, \*, project_id, region_id, \*\*params) -> ConnectedPortList -- client.cloud.reserved_fixed_ips.vip.toggle(port_id, \*, project_id, region_id, \*\*params) -> ReservedFixedIP -- client.cloud.reserved_fixed_ips.vip.update_connected_ports(port_id, \*, project_id, region_id, \*\*params) -> ConnectedPortList +- client.cloud.reserved_fixed_ips.vip.connected_ports.list(port_id, \*, project_id, region_id) -> ConnectedPortList +- client.cloud.reserved_fixed_ips.vip.connected_ports.add(port_id, \*, project_id, region_id, \*\*params) -> ConnectedPortList +- client.cloud.reserved_fixed_ips.vip.connected_ports.replace(port_id, \*, project_id, region_id, \*\*params) -> ConnectedPortList ## Networks diff --git a/examples/cloud/reserved_fixed_ips.py b/examples/cloud/reserved_fixed_ips.py index 313c6069..7ff0d28c 100644 --- a/examples/cloud/reserved_fixed_ips.py +++ b/examples/cloud/reserved_fixed_ips.py @@ -73,7 +73,7 @@ def toggle_reserved_fixed_ip_vip(*, client: Gcore, port_id: str, is_vip: bool) - def list_candidate_ports(*, client: Gcore, port_id: str) -> None: print("\n=== LIST CANDIDATE PORTS ===") - candidate_ports = client.cloud.reserved_fixed_ips.vip.list_candidate_ports(port_id) + candidate_ports = client.cloud.reserved_fixed_ips.vip.candidate_ports.list(port_id) for count, port in enumerate(candidate_ports.results, 1): print(f"{count}. Candidate port: ID={port.port_id}, instance name={port.instance_name}") print("========================") @@ -81,7 +81,7 @@ def list_candidate_ports(*, client: Gcore, port_id: str) -> None: def list_connected_ports(*, client: Gcore, port_id: str) -> None: print("\n=== LIST CONNECTED PORTS ===") - connected_ports = client.cloud.reserved_fixed_ips.vip.list_connected_ports(port_id) + connected_ports = client.cloud.reserved_fixed_ips.vip.connected_ports.list(port_id) for count, port in enumerate(connected_ports.results, 1): print(f"{count}. Connected port: ID={port.port_id}, instance name={port.instance_name}") print("========================") diff --git a/examples/cloud/reserved_fixed_ips_async.py b/examples/cloud/reserved_fixed_ips_async.py index 2a20d763..13de7071 100644 --- a/examples/cloud/reserved_fixed_ips_async.py +++ b/examples/cloud/reserved_fixed_ips_async.py @@ -77,7 +77,7 @@ async def toggle_reserved_fixed_ip_vip(*, client: AsyncGcore, port_id: str, is_v async def list_candidate_ports(*, client: AsyncGcore, port_id: str) -> None: print("\n=== LIST CANDIDATE PORTS ===") - candidate_ports = await client.cloud.reserved_fixed_ips.vip.list_candidate_ports(port_id) + candidate_ports = await client.cloud.reserved_fixed_ips.vip.candidate_ports.list(port_id) for count, port in enumerate(candidate_ports.results, 1): print(f"{count}. Candidate port: ID={port.port_id}, instance name={port.instance_name}") print("========================") @@ -85,7 +85,7 @@ async def list_candidate_ports(*, client: AsyncGcore, port_id: str) -> None: async def list_connected_ports(*, client: AsyncGcore, port_id: str) -> None: print("\n=== LIST CONNECTED PORTS ===") - connected_ports = await client.cloud.reserved_fixed_ips.vip.list_connected_ports(port_id) + connected_ports = await client.cloud.reserved_fixed_ips.vip.connected_ports.list(port_id) for count, port in enumerate(connected_ports.results, 1): print(f"{count}. Connected port: ID={port.port_id}, instance name={port.instance_name}") print("========================") diff --git a/pyproject.toml b/pyproject.toml index f9e9af97..c810ecea 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "gcore" -version = "0.24.0" +version = "0.25.0" description = "The official Python library for the gcore API" dynamic = ["readme"] license = "Apache-2.0" diff --git a/src/gcore/_version.py b/src/gcore/_version.py index 08a76896..7771dae8 100644 --- a/src/gcore/_version.py +++ b/src/gcore/_version.py @@ -1,4 +1,4 @@ # File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. __title__ = "gcore" -__version__ = "0.24.0" # x-release-please-version +__version__ = "0.25.0" # x-release-please-version diff --git a/src/gcore/resources/cloud/reserved_fixed_ips/reserved_fixed_ips.py b/src/gcore/resources/cloud/reserved_fixed_ips/reserved_fixed_ips.py index 7651922c..c41f3adc 100644 --- a/src/gcore/resources/cloud/reserved_fixed_ips/reserved_fixed_ips.py +++ b/src/gcore/resources/cloud/reserved_fixed_ips/reserved_fixed_ips.py @@ -7,7 +7,7 @@ import httpx -from .vip import ( +from .vip.vip import ( VipResource, AsyncVipResource, VipResourceWithRawResponse, diff --git a/src/gcore/resources/cloud/reserved_fixed_ips/vip/__init__.py b/src/gcore/resources/cloud/reserved_fixed_ips/vip/__init__.py new file mode 100644 index 00000000..81fcb02c --- /dev/null +++ b/src/gcore/resources/cloud/reserved_fixed_ips/vip/__init__.py @@ -0,0 +1,47 @@ +# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. + +from .vip import ( + VipResource, + AsyncVipResource, + VipResourceWithRawResponse, + AsyncVipResourceWithRawResponse, + VipResourceWithStreamingResponse, + AsyncVipResourceWithStreamingResponse, +) +from .candidate_ports import ( + CandidatePortsResource, + AsyncCandidatePortsResource, + CandidatePortsResourceWithRawResponse, + AsyncCandidatePortsResourceWithRawResponse, + CandidatePortsResourceWithStreamingResponse, + AsyncCandidatePortsResourceWithStreamingResponse, +) +from .connected_ports import ( + ConnectedPortsResource, + AsyncConnectedPortsResource, + ConnectedPortsResourceWithRawResponse, + AsyncConnectedPortsResourceWithRawResponse, + ConnectedPortsResourceWithStreamingResponse, + AsyncConnectedPortsResourceWithStreamingResponse, +) + +__all__ = [ + "CandidatePortsResource", + "AsyncCandidatePortsResource", + "CandidatePortsResourceWithRawResponse", + "AsyncCandidatePortsResourceWithRawResponse", + "CandidatePortsResourceWithStreamingResponse", + "AsyncCandidatePortsResourceWithStreamingResponse", + "ConnectedPortsResource", + "AsyncConnectedPortsResource", + "ConnectedPortsResourceWithRawResponse", + "AsyncConnectedPortsResourceWithRawResponse", + "ConnectedPortsResourceWithStreamingResponse", + "AsyncConnectedPortsResourceWithStreamingResponse", + "VipResource", + "AsyncVipResource", + "VipResourceWithRawResponse", + "AsyncVipResourceWithRawResponse", + "VipResourceWithStreamingResponse", + "AsyncVipResourceWithStreamingResponse", +] diff --git a/src/gcore/resources/cloud/reserved_fixed_ips/vip/candidate_ports.py b/src/gcore/resources/cloud/reserved_fixed_ips/vip/candidate_ports.py new file mode 100644 index 00000000..fdc29228 --- /dev/null +++ b/src/gcore/resources/cloud/reserved_fixed_ips/vip/candidate_ports.py @@ -0,0 +1,175 @@ +# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. + +from __future__ import annotations + +import httpx + +from ....._types import Body, Query, Headers, NotGiven, not_given +from ....._compat import cached_property +from ....._resource import SyncAPIResource, AsyncAPIResource +from ....._response import ( + to_raw_response_wrapper, + to_streamed_response_wrapper, + async_to_raw_response_wrapper, + async_to_streamed_response_wrapper, +) +from ....._base_client import make_request_options +from .....types.cloud.reserved_fixed_ips.vip.candidate_port_list import CandidatePortList + +__all__ = ["CandidatePortsResource", "AsyncCandidatePortsResource"] + + +class CandidatePortsResource(SyncAPIResource): + @cached_property + def with_raw_response(self) -> CandidatePortsResourceWithRawResponse: + """ + This property can be used as a prefix for any HTTP method call to return + the raw response object instead of the parsed content. + + For more information, see https://www.github.com/G-Core/gcore-python#accessing-raw-response-data-eg-headers + """ + return CandidatePortsResourceWithRawResponse(self) + + @cached_property + def with_streaming_response(self) -> CandidatePortsResourceWithStreamingResponse: + """ + An alternative to `.with_raw_response` that doesn't eagerly read the response body. + + For more information, see https://www.github.com/G-Core/gcore-python#with_streaming_response + """ + return CandidatePortsResourceWithStreamingResponse(self) + + def list( + self, + port_id: str, + *, + project_id: int | None = None, + region_id: int | None = None, + # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. + # The extra values given here take precedence over values defined on the client or passed to this method. + extra_headers: Headers | None = None, + extra_query: Query | None = None, + extra_body: Body | None = None, + timeout: float | httpx.Timeout | None | NotGiven = not_given, + ) -> CandidatePortList: + """ + List all instance ports that are available for connecting to a VIP. + + Args: + extra_headers: Send extra headers + + extra_query: Add additional query parameters to the request + + extra_body: Add additional JSON properties to the request + + timeout: Override the client-level default timeout for this request, in seconds + """ + if project_id is None: + project_id = self._client._get_cloud_project_id_path_param() + if region_id is None: + region_id = self._client._get_cloud_region_id_path_param() + if not port_id: + raise ValueError(f"Expected a non-empty value for `port_id` but received {port_id!r}") + return self._get( + f"/cloud/v1/reserved_fixed_ips/{project_id}/{region_id}/{port_id}/available_devices", + options=make_request_options( + extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout + ), + cast_to=CandidatePortList, + ) + + +class AsyncCandidatePortsResource(AsyncAPIResource): + @cached_property + def with_raw_response(self) -> AsyncCandidatePortsResourceWithRawResponse: + """ + This property can be used as a prefix for any HTTP method call to return + the raw response object instead of the parsed content. + + For more information, see https://www.github.com/G-Core/gcore-python#accessing-raw-response-data-eg-headers + """ + return AsyncCandidatePortsResourceWithRawResponse(self) + + @cached_property + def with_streaming_response(self) -> AsyncCandidatePortsResourceWithStreamingResponse: + """ + An alternative to `.with_raw_response` that doesn't eagerly read the response body. + + For more information, see https://www.github.com/G-Core/gcore-python#with_streaming_response + """ + return AsyncCandidatePortsResourceWithStreamingResponse(self) + + async def list( + self, + port_id: str, + *, + project_id: int | None = None, + region_id: int | None = None, + # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. + # The extra values given here take precedence over values defined on the client or passed to this method. + extra_headers: Headers | None = None, + extra_query: Query | None = None, + extra_body: Body | None = None, + timeout: float | httpx.Timeout | None | NotGiven = not_given, + ) -> CandidatePortList: + """ + List all instance ports that are available for connecting to a VIP. + + Args: + extra_headers: Send extra headers + + extra_query: Add additional query parameters to the request + + extra_body: Add additional JSON properties to the request + + timeout: Override the client-level default timeout for this request, in seconds + """ + if project_id is None: + project_id = self._client._get_cloud_project_id_path_param() + if region_id is None: + region_id = self._client._get_cloud_region_id_path_param() + if not port_id: + raise ValueError(f"Expected a non-empty value for `port_id` but received {port_id!r}") + return await self._get( + f"/cloud/v1/reserved_fixed_ips/{project_id}/{region_id}/{port_id}/available_devices", + options=make_request_options( + extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout + ), + cast_to=CandidatePortList, + ) + + +class CandidatePortsResourceWithRawResponse: + def __init__(self, candidate_ports: CandidatePortsResource) -> None: + self._candidate_ports = candidate_ports + + self.list = to_raw_response_wrapper( + candidate_ports.list, + ) + + +class AsyncCandidatePortsResourceWithRawResponse: + def __init__(self, candidate_ports: AsyncCandidatePortsResource) -> None: + self._candidate_ports = candidate_ports + + self.list = async_to_raw_response_wrapper( + candidate_ports.list, + ) + + +class CandidatePortsResourceWithStreamingResponse: + def __init__(self, candidate_ports: CandidatePortsResource) -> None: + self._candidate_ports = candidate_ports + + self.list = to_streamed_response_wrapper( + candidate_ports.list, + ) + + +class AsyncCandidatePortsResourceWithStreamingResponse: + def __init__(self, candidate_ports: AsyncCandidatePortsResource) -> None: + self._candidate_ports = candidate_ports + + self.list = async_to_streamed_response_wrapper( + candidate_ports.list, + ) diff --git a/src/gcore/resources/cloud/reserved_fixed_ips/vip.py b/src/gcore/resources/cloud/reserved_fixed_ips/vip/connected_ports.py similarity index 50% rename from src/gcore/resources/cloud/reserved_fixed_ips/vip.py rename to src/gcore/resources/cloud/reserved_fixed_ips/vip/connected_ports.py index 966a51e0..f98e4b8a 100644 --- a/src/gcore/resources/cloud/reserved_fixed_ips/vip.py +++ b/src/gcore/resources/cloud/reserved_fixed_ips/vip/connected_ports.py @@ -4,89 +4,44 @@ import httpx -from ...._types import Body, Omit, Query, Headers, NotGiven, SequenceNotStr, omit, not_given -from ...._utils import maybe_transform, async_maybe_transform -from ...._compat import cached_property -from ...._resource import SyncAPIResource, AsyncAPIResource -from ...._response import ( +from ....._types import Body, Omit, Query, Headers, NotGiven, SequenceNotStr, omit, not_given +from ....._utils import maybe_transform, async_maybe_transform +from ....._compat import cached_property +from ....._resource import SyncAPIResource, AsyncAPIResource +from ....._response import ( to_raw_response_wrapper, to_streamed_response_wrapper, async_to_raw_response_wrapper, async_to_streamed_response_wrapper, ) -from ...._base_client import make_request_options -from ....types.cloud.reserved_fixed_ip import ReservedFixedIP -from ....types.cloud.reserved_fixed_ips import ( - vip_toggle_params, - vip_update_connected_ports_params, - vip_replace_connected_ports_params, -) -from ....types.cloud.reserved_fixed_ips.candidate_port_list import CandidatePortList -from ....types.cloud.reserved_fixed_ips.connected_port_list import ConnectedPortList +from ....._base_client import make_request_options +from .....types.cloud.reserved_fixed_ips.vip import connected_port_add_params, connected_port_replace_params +from .....types.cloud.reserved_fixed_ips.vip.connected_port_list import ConnectedPortList -__all__ = ["VipResource", "AsyncVipResource"] +__all__ = ["ConnectedPortsResource", "AsyncConnectedPortsResource"] -class VipResource(SyncAPIResource): +class ConnectedPortsResource(SyncAPIResource): @cached_property - def with_raw_response(self) -> VipResourceWithRawResponse: + def with_raw_response(self) -> ConnectedPortsResourceWithRawResponse: """ This property can be used as a prefix for any HTTP method call to return the raw response object instead of the parsed content. For more information, see https://www.github.com/G-Core/gcore-python#accessing-raw-response-data-eg-headers """ - return VipResourceWithRawResponse(self) + return ConnectedPortsResourceWithRawResponse(self) @cached_property - def with_streaming_response(self) -> VipResourceWithStreamingResponse: + def with_streaming_response(self) -> ConnectedPortsResourceWithStreamingResponse: """ An alternative to `.with_raw_response` that doesn't eagerly read the response body. For more information, see https://www.github.com/G-Core/gcore-python#with_streaming_response """ - return VipResourceWithStreamingResponse(self) - - def list_candidate_ports( - self, - port_id: str, - *, - project_id: int | None = None, - region_id: int | None = None, - # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. - # The extra values given here take precedence over values defined on the client or passed to this method. - extra_headers: Headers | None = None, - extra_query: Query | None = None, - extra_body: Body | None = None, - timeout: float | httpx.Timeout | None | NotGiven = not_given, - ) -> CandidatePortList: - """ - List all instance ports that are available for connecting to a VIP. - - Args: - extra_headers: Send extra headers - - extra_query: Add additional query parameters to the request - - extra_body: Add additional JSON properties to the request - - timeout: Override the client-level default timeout for this request, in seconds - """ - if project_id is None: - project_id = self._client._get_cloud_project_id_path_param() - if region_id is None: - region_id = self._client._get_cloud_region_id_path_param() - if not port_id: - raise ValueError(f"Expected a non-empty value for `port_id` but received {port_id!r}") - return self._get( - f"/cloud/v1/reserved_fixed_ips/{project_id}/{region_id}/{port_id}/available_devices", - options=make_request_options( - extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout - ), - cast_to=CandidatePortList, - ) + return ConnectedPortsResourceWithStreamingResponse(self) - def list_connected_ports( + def list( self, port_id: str, *, @@ -125,7 +80,7 @@ def list_connected_ports( cast_to=ConnectedPortList, ) - def replace_connected_ports( + def add( self, port_id: str, *, @@ -140,7 +95,7 @@ def replace_connected_ports( timeout: float | httpx.Timeout | None | NotGiven = not_given, ) -> ConnectedPortList: """ - Replace the list of instance ports that share a VIP. + Add instance ports to share a VIP. Args: port_ids: List of port IDs that will share one VIP @@ -159,61 +114,16 @@ def replace_connected_ports( region_id = self._client._get_cloud_region_id_path_param() if not port_id: raise ValueError(f"Expected a non-empty value for `port_id` but received {port_id!r}") - return self._put( + return self._patch( f"/cloud/v1/reserved_fixed_ips/{project_id}/{region_id}/{port_id}/connected_devices", - body=maybe_transform( - {"port_ids": port_ids}, vip_replace_connected_ports_params.VipReplaceConnectedPortsParams - ), + body=maybe_transform({"port_ids": port_ids}, connected_port_add_params.ConnectedPortAddParams), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=ConnectedPortList, ) - def toggle( - self, - port_id: str, - *, - project_id: int | None = None, - region_id: int | None = None, - is_vip: bool, - # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. - # The extra values given here take precedence over values defined on the client or passed to this method. - extra_headers: Headers | None = None, - extra_query: Query | None = None, - extra_body: Body | None = None, - timeout: float | httpx.Timeout | None | NotGiven = not_given, - ) -> ReservedFixedIP: - """ - Update the VIP status of a reserved fixed IP. - - Args: - is_vip: If reserved fixed IP should be a VIP - - extra_headers: Send extra headers - - extra_query: Add additional query parameters to the request - - extra_body: Add additional JSON properties to the request - - timeout: Override the client-level default timeout for this request, in seconds - """ - if project_id is None: - project_id = self._client._get_cloud_project_id_path_param() - if region_id is None: - region_id = self._client._get_cloud_region_id_path_param() - if not port_id: - raise ValueError(f"Expected a non-empty value for `port_id` but received {port_id!r}") - return self._patch( - f"/cloud/v1/reserved_fixed_ips/{project_id}/{region_id}/{port_id}", - body=maybe_transform({"is_vip": is_vip}, vip_toggle_params.VipToggleParams), - options=make_request_options( - extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout - ), - cast_to=ReservedFixedIP, - ) - - def update_connected_ports( + def replace( self, port_id: str, *, @@ -228,7 +138,7 @@ def update_connected_ports( timeout: float | httpx.Timeout | None | NotGiven = not_given, ) -> ConnectedPortList: """ - Add instance ports to share a VIP. + Replace the list of instance ports that share a VIP. Args: port_ids: List of port IDs that will share one VIP @@ -247,11 +157,9 @@ def update_connected_ports( region_id = self._client._get_cloud_region_id_path_param() if not port_id: raise ValueError(f"Expected a non-empty value for `port_id` but received {port_id!r}") - return self._patch( + return self._put( f"/cloud/v1/reserved_fixed_ips/{project_id}/{region_id}/{port_id}/connected_devices", - body=maybe_transform( - {"port_ids": port_ids}, vip_update_connected_ports_params.VipUpdateConnectedPortsParams - ), + body=maybe_transform({"port_ids": port_ids}, connected_port_replace_params.ConnectedPortReplaceParams), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), @@ -259,66 +167,27 @@ def update_connected_ports( ) -class AsyncVipResource(AsyncAPIResource): +class AsyncConnectedPortsResource(AsyncAPIResource): @cached_property - def with_raw_response(self) -> AsyncVipResourceWithRawResponse: + def with_raw_response(self) -> AsyncConnectedPortsResourceWithRawResponse: """ This property can be used as a prefix for any HTTP method call to return the raw response object instead of the parsed content. For more information, see https://www.github.com/G-Core/gcore-python#accessing-raw-response-data-eg-headers """ - return AsyncVipResourceWithRawResponse(self) + return AsyncConnectedPortsResourceWithRawResponse(self) @cached_property - def with_streaming_response(self) -> AsyncVipResourceWithStreamingResponse: + def with_streaming_response(self) -> AsyncConnectedPortsResourceWithStreamingResponse: """ An alternative to `.with_raw_response` that doesn't eagerly read the response body. For more information, see https://www.github.com/G-Core/gcore-python#with_streaming_response """ - return AsyncVipResourceWithStreamingResponse(self) + return AsyncConnectedPortsResourceWithStreamingResponse(self) - async def list_candidate_ports( - self, - port_id: str, - *, - project_id: int | None = None, - region_id: int | None = None, - # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. - # The extra values given here take precedence over values defined on the client or passed to this method. - extra_headers: Headers | None = None, - extra_query: Query | None = None, - extra_body: Body | None = None, - timeout: float | httpx.Timeout | None | NotGiven = not_given, - ) -> CandidatePortList: - """ - List all instance ports that are available for connecting to a VIP. - - Args: - extra_headers: Send extra headers - - extra_query: Add additional query parameters to the request - - extra_body: Add additional JSON properties to the request - - timeout: Override the client-level default timeout for this request, in seconds - """ - if project_id is None: - project_id = self._client._get_cloud_project_id_path_param() - if region_id is None: - region_id = self._client._get_cloud_region_id_path_param() - if not port_id: - raise ValueError(f"Expected a non-empty value for `port_id` but received {port_id!r}") - return await self._get( - f"/cloud/v1/reserved_fixed_ips/{project_id}/{region_id}/{port_id}/available_devices", - options=make_request_options( - extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout - ), - cast_to=CandidatePortList, - ) - - async def list_connected_ports( + async def list( self, port_id: str, *, @@ -357,7 +226,7 @@ async def list_connected_ports( cast_to=ConnectedPortList, ) - async def replace_connected_ports( + async def add( self, port_id: str, *, @@ -372,7 +241,7 @@ async def replace_connected_ports( timeout: float | httpx.Timeout | None | NotGiven = not_given, ) -> ConnectedPortList: """ - Replace the list of instance ports that share a VIP. + Add instance ports to share a VIP. Args: port_ids: List of port IDs that will share one VIP @@ -391,61 +260,16 @@ async def replace_connected_ports( region_id = self._client._get_cloud_region_id_path_param() if not port_id: raise ValueError(f"Expected a non-empty value for `port_id` but received {port_id!r}") - return await self._put( + return await self._patch( f"/cloud/v1/reserved_fixed_ips/{project_id}/{region_id}/{port_id}/connected_devices", - body=await async_maybe_transform( - {"port_ids": port_ids}, vip_replace_connected_ports_params.VipReplaceConnectedPortsParams - ), + body=await async_maybe_transform({"port_ids": port_ids}, connected_port_add_params.ConnectedPortAddParams), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout ), cast_to=ConnectedPortList, ) - async def toggle( - self, - port_id: str, - *, - project_id: int | None = None, - region_id: int | None = None, - is_vip: bool, - # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. - # The extra values given here take precedence over values defined on the client or passed to this method. - extra_headers: Headers | None = None, - extra_query: Query | None = None, - extra_body: Body | None = None, - timeout: float | httpx.Timeout | None | NotGiven = not_given, - ) -> ReservedFixedIP: - """ - Update the VIP status of a reserved fixed IP. - - Args: - is_vip: If reserved fixed IP should be a VIP - - extra_headers: Send extra headers - - extra_query: Add additional query parameters to the request - - extra_body: Add additional JSON properties to the request - - timeout: Override the client-level default timeout for this request, in seconds - """ - if project_id is None: - project_id = self._client._get_cloud_project_id_path_param() - if region_id is None: - region_id = self._client._get_cloud_region_id_path_param() - if not port_id: - raise ValueError(f"Expected a non-empty value for `port_id` but received {port_id!r}") - return await self._patch( - f"/cloud/v1/reserved_fixed_ips/{project_id}/{region_id}/{port_id}", - body=await async_maybe_transform({"is_vip": is_vip}, vip_toggle_params.VipToggleParams), - options=make_request_options( - extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout - ), - cast_to=ReservedFixedIP, - ) - - async def update_connected_ports( + async def replace( self, port_id: str, *, @@ -460,7 +284,7 @@ async def update_connected_ports( timeout: float | httpx.Timeout | None | NotGiven = not_given, ) -> ConnectedPortList: """ - Add instance ports to share a VIP. + Replace the list of instance ports that share a VIP. Args: port_ids: List of port IDs that will share one VIP @@ -479,10 +303,10 @@ async def update_connected_ports( region_id = self._client._get_cloud_region_id_path_param() if not port_id: raise ValueError(f"Expected a non-empty value for `port_id` but received {port_id!r}") - return await self._patch( + return await self._put( f"/cloud/v1/reserved_fixed_ips/{project_id}/{region_id}/{port_id}/connected_devices", body=await async_maybe_transform( - {"port_ids": port_ids}, vip_update_connected_ports_params.VipUpdateConnectedPortsParams + {"port_ids": port_ids}, connected_port_replace_params.ConnectedPortReplaceParams ), options=make_request_options( extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout @@ -491,85 +315,61 @@ async def update_connected_ports( ) -class VipResourceWithRawResponse: - def __init__(self, vip: VipResource) -> None: - self._vip = vip +class ConnectedPortsResourceWithRawResponse: + def __init__(self, connected_ports: ConnectedPortsResource) -> None: + self._connected_ports = connected_ports - self.list_candidate_ports = to_raw_response_wrapper( - vip.list_candidate_ports, - ) - self.list_connected_ports = to_raw_response_wrapper( - vip.list_connected_ports, + self.list = to_raw_response_wrapper( + connected_ports.list, ) - self.replace_connected_ports = to_raw_response_wrapper( - vip.replace_connected_ports, + self.add = to_raw_response_wrapper( + connected_ports.add, ) - self.toggle = to_raw_response_wrapper( - vip.toggle, - ) - self.update_connected_ports = to_raw_response_wrapper( - vip.update_connected_ports, + self.replace = to_raw_response_wrapper( + connected_ports.replace, ) -class AsyncVipResourceWithRawResponse: - def __init__(self, vip: AsyncVipResource) -> None: - self._vip = vip +class AsyncConnectedPortsResourceWithRawResponse: + def __init__(self, connected_ports: AsyncConnectedPortsResource) -> None: + self._connected_ports = connected_ports - self.list_candidate_ports = async_to_raw_response_wrapper( - vip.list_candidate_ports, - ) - self.list_connected_ports = async_to_raw_response_wrapper( - vip.list_connected_ports, + self.list = async_to_raw_response_wrapper( + connected_ports.list, ) - self.replace_connected_ports = async_to_raw_response_wrapper( - vip.replace_connected_ports, + self.add = async_to_raw_response_wrapper( + connected_ports.add, ) - self.toggle = async_to_raw_response_wrapper( - vip.toggle, - ) - self.update_connected_ports = async_to_raw_response_wrapper( - vip.update_connected_ports, + self.replace = async_to_raw_response_wrapper( + connected_ports.replace, ) -class VipResourceWithStreamingResponse: - def __init__(self, vip: VipResource) -> None: - self._vip = vip +class ConnectedPortsResourceWithStreamingResponse: + def __init__(self, connected_ports: ConnectedPortsResource) -> None: + self._connected_ports = connected_ports - self.list_candidate_ports = to_streamed_response_wrapper( - vip.list_candidate_ports, - ) - self.list_connected_ports = to_streamed_response_wrapper( - vip.list_connected_ports, - ) - self.replace_connected_ports = to_streamed_response_wrapper( - vip.replace_connected_ports, + self.list = to_streamed_response_wrapper( + connected_ports.list, ) - self.toggle = to_streamed_response_wrapper( - vip.toggle, + self.add = to_streamed_response_wrapper( + connected_ports.add, ) - self.update_connected_ports = to_streamed_response_wrapper( - vip.update_connected_ports, + self.replace = to_streamed_response_wrapper( + connected_ports.replace, ) -class AsyncVipResourceWithStreamingResponse: - def __init__(self, vip: AsyncVipResource) -> None: - self._vip = vip +class AsyncConnectedPortsResourceWithStreamingResponse: + def __init__(self, connected_ports: AsyncConnectedPortsResource) -> None: + self._connected_ports = connected_ports - self.list_candidate_ports = async_to_streamed_response_wrapper( - vip.list_candidate_ports, - ) - self.list_connected_ports = async_to_streamed_response_wrapper( - vip.list_connected_ports, - ) - self.replace_connected_ports = async_to_streamed_response_wrapper( - vip.replace_connected_ports, + self.list = async_to_streamed_response_wrapper( + connected_ports.list, ) - self.toggle = async_to_streamed_response_wrapper( - vip.toggle, + self.add = async_to_streamed_response_wrapper( + connected_ports.add, ) - self.update_connected_ports = async_to_streamed_response_wrapper( - vip.update_connected_ports, + self.replace = async_to_streamed_response_wrapper( + connected_ports.replace, ) diff --git a/src/gcore/resources/cloud/reserved_fixed_ips/vip/vip.py b/src/gcore/resources/cloud/reserved_fixed_ips/vip/vip.py new file mode 100644 index 00000000..58ff9bd5 --- /dev/null +++ b/src/gcore/resources/cloud/reserved_fixed_ips/vip/vip.py @@ -0,0 +1,249 @@ +# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. + +from __future__ import annotations + +import httpx + +from ....._types import Body, Query, Headers, NotGiven, not_given +from ....._utils import maybe_transform, async_maybe_transform +from ....._compat import cached_property +from ....._resource import SyncAPIResource, AsyncAPIResource +from ....._response import ( + to_raw_response_wrapper, + to_streamed_response_wrapper, + async_to_raw_response_wrapper, + async_to_streamed_response_wrapper, +) +from .candidate_ports import ( + CandidatePortsResource, + AsyncCandidatePortsResource, + CandidatePortsResourceWithRawResponse, + AsyncCandidatePortsResourceWithRawResponse, + CandidatePortsResourceWithStreamingResponse, + AsyncCandidatePortsResourceWithStreamingResponse, +) +from .connected_ports import ( + ConnectedPortsResource, + AsyncConnectedPortsResource, + ConnectedPortsResourceWithRawResponse, + AsyncConnectedPortsResourceWithRawResponse, + ConnectedPortsResourceWithStreamingResponse, + AsyncConnectedPortsResourceWithStreamingResponse, +) +from ....._base_client import make_request_options +from .....types.cloud.reserved_fixed_ip import ReservedFixedIP +from .....types.cloud.reserved_fixed_ips import vip_toggle_params + +__all__ = ["VipResource", "AsyncVipResource"] + + +class VipResource(SyncAPIResource): + @cached_property + def candidate_ports(self) -> CandidatePortsResource: + return CandidatePortsResource(self._client) + + @cached_property + def connected_ports(self) -> ConnectedPortsResource: + return ConnectedPortsResource(self._client) + + @cached_property + def with_raw_response(self) -> VipResourceWithRawResponse: + """ + This property can be used as a prefix for any HTTP method call to return + the raw response object instead of the parsed content. + + For more information, see https://www.github.com/G-Core/gcore-python#accessing-raw-response-data-eg-headers + """ + return VipResourceWithRawResponse(self) + + @cached_property + def with_streaming_response(self) -> VipResourceWithStreamingResponse: + """ + An alternative to `.with_raw_response` that doesn't eagerly read the response body. + + For more information, see https://www.github.com/G-Core/gcore-python#with_streaming_response + """ + return VipResourceWithStreamingResponse(self) + + def toggle( + self, + port_id: str, + *, + project_id: int | None = None, + region_id: int | None = None, + is_vip: bool, + # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. + # The extra values given here take precedence over values defined on the client or passed to this method. + extra_headers: Headers | None = None, + extra_query: Query | None = None, + extra_body: Body | None = None, + timeout: float | httpx.Timeout | None | NotGiven = not_given, + ) -> ReservedFixedIP: + """ + Update the VIP status of a reserved fixed IP. + + Args: + is_vip: If reserved fixed IP should be a VIP + + extra_headers: Send extra headers + + extra_query: Add additional query parameters to the request + + extra_body: Add additional JSON properties to the request + + timeout: Override the client-level default timeout for this request, in seconds + """ + if project_id is None: + project_id = self._client._get_cloud_project_id_path_param() + if region_id is None: + region_id = self._client._get_cloud_region_id_path_param() + if not port_id: + raise ValueError(f"Expected a non-empty value for `port_id` but received {port_id!r}") + return self._patch( + f"/cloud/v1/reserved_fixed_ips/{project_id}/{region_id}/{port_id}", + body=maybe_transform({"is_vip": is_vip}, vip_toggle_params.VipToggleParams), + options=make_request_options( + extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout + ), + cast_to=ReservedFixedIP, + ) + + +class AsyncVipResource(AsyncAPIResource): + @cached_property + def candidate_ports(self) -> AsyncCandidatePortsResource: + return AsyncCandidatePortsResource(self._client) + + @cached_property + def connected_ports(self) -> AsyncConnectedPortsResource: + return AsyncConnectedPortsResource(self._client) + + @cached_property + def with_raw_response(self) -> AsyncVipResourceWithRawResponse: + """ + This property can be used as a prefix for any HTTP method call to return + the raw response object instead of the parsed content. + + For more information, see https://www.github.com/G-Core/gcore-python#accessing-raw-response-data-eg-headers + """ + return AsyncVipResourceWithRawResponse(self) + + @cached_property + def with_streaming_response(self) -> AsyncVipResourceWithStreamingResponse: + """ + An alternative to `.with_raw_response` that doesn't eagerly read the response body. + + For more information, see https://www.github.com/G-Core/gcore-python#with_streaming_response + """ + return AsyncVipResourceWithStreamingResponse(self) + + async def toggle( + self, + port_id: str, + *, + project_id: int | None = None, + region_id: int | None = None, + is_vip: bool, + # Use the following arguments if you need to pass additional parameters to the API that aren't available via kwargs. + # The extra values given here take precedence over values defined on the client or passed to this method. + extra_headers: Headers | None = None, + extra_query: Query | None = None, + extra_body: Body | None = None, + timeout: float | httpx.Timeout | None | NotGiven = not_given, + ) -> ReservedFixedIP: + """ + Update the VIP status of a reserved fixed IP. + + Args: + is_vip: If reserved fixed IP should be a VIP + + extra_headers: Send extra headers + + extra_query: Add additional query parameters to the request + + extra_body: Add additional JSON properties to the request + + timeout: Override the client-level default timeout for this request, in seconds + """ + if project_id is None: + project_id = self._client._get_cloud_project_id_path_param() + if region_id is None: + region_id = self._client._get_cloud_region_id_path_param() + if not port_id: + raise ValueError(f"Expected a non-empty value for `port_id` but received {port_id!r}") + return await self._patch( + f"/cloud/v1/reserved_fixed_ips/{project_id}/{region_id}/{port_id}", + body=await async_maybe_transform({"is_vip": is_vip}, vip_toggle_params.VipToggleParams), + options=make_request_options( + extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout + ), + cast_to=ReservedFixedIP, + ) + + +class VipResourceWithRawResponse: + def __init__(self, vip: VipResource) -> None: + self._vip = vip + + self.toggle = to_raw_response_wrapper( + vip.toggle, + ) + + @cached_property + def candidate_ports(self) -> CandidatePortsResourceWithRawResponse: + return CandidatePortsResourceWithRawResponse(self._vip.candidate_ports) + + @cached_property + def connected_ports(self) -> ConnectedPortsResourceWithRawResponse: + return ConnectedPortsResourceWithRawResponse(self._vip.connected_ports) + + +class AsyncVipResourceWithRawResponse: + def __init__(self, vip: AsyncVipResource) -> None: + self._vip = vip + + self.toggle = async_to_raw_response_wrapper( + vip.toggle, + ) + + @cached_property + def candidate_ports(self) -> AsyncCandidatePortsResourceWithRawResponse: + return AsyncCandidatePortsResourceWithRawResponse(self._vip.candidate_ports) + + @cached_property + def connected_ports(self) -> AsyncConnectedPortsResourceWithRawResponse: + return AsyncConnectedPortsResourceWithRawResponse(self._vip.connected_ports) + + +class VipResourceWithStreamingResponse: + def __init__(self, vip: VipResource) -> None: + self._vip = vip + + self.toggle = to_streamed_response_wrapper( + vip.toggle, + ) + + @cached_property + def candidate_ports(self) -> CandidatePortsResourceWithStreamingResponse: + return CandidatePortsResourceWithStreamingResponse(self._vip.candidate_ports) + + @cached_property + def connected_ports(self) -> ConnectedPortsResourceWithStreamingResponse: + return ConnectedPortsResourceWithStreamingResponse(self._vip.connected_ports) + + +class AsyncVipResourceWithStreamingResponse: + def __init__(self, vip: AsyncVipResource) -> None: + self._vip = vip + + self.toggle = async_to_streamed_response_wrapper( + vip.toggle, + ) + + @cached_property + def candidate_ports(self) -> AsyncCandidatePortsResourceWithStreamingResponse: + return AsyncCandidatePortsResourceWithStreamingResponse(self._vip.candidate_ports) + + @cached_property + def connected_ports(self) -> AsyncConnectedPortsResourceWithStreamingResponse: + return AsyncConnectedPortsResourceWithStreamingResponse(self._vip.connected_ports) diff --git a/src/gcore/types/cloud/reserved_fixed_ips/__init__.py b/src/gcore/types/cloud/reserved_fixed_ips/__init__.py index a437e01d..a796bf7c 100644 --- a/src/gcore/types/cloud/reserved_fixed_ips/__init__.py +++ b/src/gcore/types/cloud/reserved_fixed_ips/__init__.py @@ -2,11 +2,5 @@ from __future__ import annotations -from .candidate_port import CandidatePort as CandidatePort -from .connected_port import ConnectedPort as ConnectedPort from .ip_with_subnet import IPWithSubnet as IPWithSubnet from .vip_toggle_params import VipToggleParams as VipToggleParams -from .candidate_port_list import CandidatePortList as CandidatePortList -from .connected_port_list import ConnectedPortList as ConnectedPortList -from .vip_update_connected_ports_params import VipUpdateConnectedPortsParams as VipUpdateConnectedPortsParams -from .vip_replace_connected_ports_params import VipReplaceConnectedPortsParams as VipReplaceConnectedPortsParams diff --git a/src/gcore/types/cloud/reserved_fixed_ips/vip/__init__.py b/src/gcore/types/cloud/reserved_fixed_ips/vip/__init__.py new file mode 100644 index 00000000..0dd4eb7e --- /dev/null +++ b/src/gcore/types/cloud/reserved_fixed_ips/vip/__init__.py @@ -0,0 +1,10 @@ +# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. + +from __future__ import annotations + +from .candidate_port import CandidatePort as CandidatePort +from .connected_port import ConnectedPort as ConnectedPort +from .candidate_port_list import CandidatePortList as CandidatePortList +from .connected_port_list import ConnectedPortList as ConnectedPortList +from .connected_port_add_params import ConnectedPortAddParams as ConnectedPortAddParams +from .connected_port_replace_params import ConnectedPortReplaceParams as ConnectedPortReplaceParams diff --git a/src/gcore/types/cloud/reserved_fixed_ips/candidate_port.py b/src/gcore/types/cloud/reserved_fixed_ips/vip/candidate_port.py similarity index 82% rename from src/gcore/types/cloud/reserved_fixed_ips/candidate_port.py rename to src/gcore/types/cloud/reserved_fixed_ips/vip/candidate_port.py index 78b5db02..4799c337 100644 --- a/src/gcore/types/cloud/reserved_fixed_ips/candidate_port.py +++ b/src/gcore/types/cloud/reserved_fixed_ips/vip/candidate_port.py @@ -2,9 +2,9 @@ from typing import List -from ..network import Network -from ...._models import BaseModel -from .ip_with_subnet import IPWithSubnet +from ...network import Network +from ....._models import BaseModel +from ..ip_with_subnet import IPWithSubnet __all__ = ["CandidatePort"] diff --git a/src/gcore/types/cloud/reserved_fixed_ips/candidate_port_list.py b/src/gcore/types/cloud/reserved_fixed_ips/vip/candidate_port_list.py similarity index 90% rename from src/gcore/types/cloud/reserved_fixed_ips/candidate_port_list.py rename to src/gcore/types/cloud/reserved_fixed_ips/vip/candidate_port_list.py index d01d7744..3c68db42 100644 --- a/src/gcore/types/cloud/reserved_fixed_ips/candidate_port_list.py +++ b/src/gcore/types/cloud/reserved_fixed_ips/vip/candidate_port_list.py @@ -2,7 +2,7 @@ from typing import List -from ...._models import BaseModel +from ....._models import BaseModel from .candidate_port import CandidatePort __all__ = ["CandidatePortList"] diff --git a/src/gcore/types/cloud/reserved_fixed_ips/connected_port.py b/src/gcore/types/cloud/reserved_fixed_ips/vip/connected_port.py similarity index 82% rename from src/gcore/types/cloud/reserved_fixed_ips/connected_port.py rename to src/gcore/types/cloud/reserved_fixed_ips/vip/connected_port.py index 0920f52c..b368f281 100644 --- a/src/gcore/types/cloud/reserved_fixed_ips/connected_port.py +++ b/src/gcore/types/cloud/reserved_fixed_ips/vip/connected_port.py @@ -2,9 +2,9 @@ from typing import List -from ..network import Network -from ...._models import BaseModel -from .ip_with_subnet import IPWithSubnet +from ...network import Network +from ....._models import BaseModel +from ..ip_with_subnet import IPWithSubnet __all__ = ["ConnectedPort"] diff --git a/src/gcore/types/cloud/reserved_fixed_ips/vip_update_connected_ports_params.py b/src/gcore/types/cloud/reserved_fixed_ips/vip/connected_port_add_params.py similarity index 67% rename from src/gcore/types/cloud/reserved_fixed_ips/vip_update_connected_ports_params.py rename to src/gcore/types/cloud/reserved_fixed_ips/vip/connected_port_add_params.py index d11d10a0..be2667e3 100644 --- a/src/gcore/types/cloud/reserved_fixed_ips/vip_update_connected_ports_params.py +++ b/src/gcore/types/cloud/reserved_fixed_ips/vip/connected_port_add_params.py @@ -4,12 +4,12 @@ from typing_extensions import TypedDict -from ...._types import SequenceNotStr +from ....._types import SequenceNotStr -__all__ = ["VipUpdateConnectedPortsParams"] +__all__ = ["ConnectedPortAddParams"] -class VipUpdateConnectedPortsParams(TypedDict, total=False): +class ConnectedPortAddParams(TypedDict, total=False): project_id: int region_id: int diff --git a/src/gcore/types/cloud/reserved_fixed_ips/connected_port_list.py b/src/gcore/types/cloud/reserved_fixed_ips/vip/connected_port_list.py similarity index 90% rename from src/gcore/types/cloud/reserved_fixed_ips/connected_port_list.py rename to src/gcore/types/cloud/reserved_fixed_ips/vip/connected_port_list.py index cb2079f7..6a0a2251 100644 --- a/src/gcore/types/cloud/reserved_fixed_ips/connected_port_list.py +++ b/src/gcore/types/cloud/reserved_fixed_ips/vip/connected_port_list.py @@ -2,7 +2,7 @@ from typing import List -from ...._models import BaseModel +from ....._models import BaseModel from .connected_port import ConnectedPort __all__ = ["ConnectedPortList"] diff --git a/src/gcore/types/cloud/reserved_fixed_ips/vip_replace_connected_ports_params.py b/src/gcore/types/cloud/reserved_fixed_ips/vip/connected_port_replace_params.py similarity index 66% rename from src/gcore/types/cloud/reserved_fixed_ips/vip_replace_connected_ports_params.py rename to src/gcore/types/cloud/reserved_fixed_ips/vip/connected_port_replace_params.py index cb8aa477..860428e7 100644 --- a/src/gcore/types/cloud/reserved_fixed_ips/vip_replace_connected_ports_params.py +++ b/src/gcore/types/cloud/reserved_fixed_ips/vip/connected_port_replace_params.py @@ -4,12 +4,12 @@ from typing_extensions import TypedDict -from ...._types import SequenceNotStr +from ....._types import SequenceNotStr -__all__ = ["VipReplaceConnectedPortsParams"] +__all__ = ["ConnectedPortReplaceParams"] -class VipReplaceConnectedPortsParams(TypedDict, total=False): +class ConnectedPortReplaceParams(TypedDict, total=False): project_id: int region_id: int diff --git a/tests/api_resources/cloud/reserved_fixed_ips/test_vip.py b/tests/api_resources/cloud/reserved_fixed_ips/test_vip.py index 65011df5..1b905f1f 100644 --- a/tests/api_resources/cloud/reserved_fixed_ips/test_vip.py +++ b/tests/api_resources/cloud/reserved_fixed_ips/test_vip.py @@ -10,10 +10,6 @@ from gcore import Gcore, AsyncGcore from tests.utils import assert_matches_type from gcore.types.cloud import ReservedFixedIP -from gcore.types.cloud.reserved_fixed_ips import ( - CandidatePortList, - ConnectedPortList, -) base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010") @@ -21,154 +17,6 @@ class TestVip: parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"]) - @parametrize - def test_method_list_candidate_ports(self, client: Gcore) -> None: - vip = client.cloud.reserved_fixed_ips.vip.list_candidate_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) - assert_matches_type(CandidatePortList, vip, path=["response"]) - - @parametrize - def test_raw_response_list_candidate_ports(self, client: Gcore) -> None: - response = client.cloud.reserved_fixed_ips.vip.with_raw_response.list_candidate_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) - - assert response.is_closed is True - assert response.http_request.headers.get("X-Stainless-Lang") == "python" - vip = response.parse() - assert_matches_type(CandidatePortList, vip, path=["response"]) - - @parametrize - def test_streaming_response_list_candidate_ports(self, client: Gcore) -> None: - with client.cloud.reserved_fixed_ips.vip.with_streaming_response.list_candidate_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) as response: - assert not response.is_closed - assert response.http_request.headers.get("X-Stainless-Lang") == "python" - - vip = response.parse() - assert_matches_type(CandidatePortList, vip, path=["response"]) - - assert cast(Any, response.is_closed) is True - - @parametrize - def test_path_params_list_candidate_ports(self, client: Gcore) -> None: - with pytest.raises(ValueError, match=r"Expected a non-empty value for `port_id` but received ''"): - client.cloud.reserved_fixed_ips.vip.with_raw_response.list_candidate_ports( - port_id="", - project_id=0, - region_id=0, - ) - - @parametrize - def test_method_list_connected_ports(self, client: Gcore) -> None: - vip = client.cloud.reserved_fixed_ips.vip.list_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - @parametrize - def test_raw_response_list_connected_ports(self, client: Gcore) -> None: - response = client.cloud.reserved_fixed_ips.vip.with_raw_response.list_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) - - assert response.is_closed is True - assert response.http_request.headers.get("X-Stainless-Lang") == "python" - vip = response.parse() - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - @parametrize - def test_streaming_response_list_connected_ports(self, client: Gcore) -> None: - with client.cloud.reserved_fixed_ips.vip.with_streaming_response.list_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) as response: - assert not response.is_closed - assert response.http_request.headers.get("X-Stainless-Lang") == "python" - - vip = response.parse() - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - assert cast(Any, response.is_closed) is True - - @parametrize - def test_path_params_list_connected_ports(self, client: Gcore) -> None: - with pytest.raises(ValueError, match=r"Expected a non-empty value for `port_id` but received ''"): - client.cloud.reserved_fixed_ips.vip.with_raw_response.list_connected_ports( - port_id="", - project_id=0, - region_id=0, - ) - - @parametrize - def test_method_replace_connected_ports(self, client: Gcore) -> None: - vip = client.cloud.reserved_fixed_ips.vip.replace_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - @parametrize - def test_method_replace_connected_ports_with_all_params(self, client: Gcore) -> None: - vip = client.cloud.reserved_fixed_ips.vip.replace_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - port_ids=["351b0dd7-ca09-431c-be53-935db3785067", "bc688791-f1b0-44eb-97d4-07697294b1e1"], - ) - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - @parametrize - def test_raw_response_replace_connected_ports(self, client: Gcore) -> None: - response = client.cloud.reserved_fixed_ips.vip.with_raw_response.replace_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) - - assert response.is_closed is True - assert response.http_request.headers.get("X-Stainless-Lang") == "python" - vip = response.parse() - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - @parametrize - def test_streaming_response_replace_connected_ports(self, client: Gcore) -> None: - with client.cloud.reserved_fixed_ips.vip.with_streaming_response.replace_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) as response: - assert not response.is_closed - assert response.http_request.headers.get("X-Stainless-Lang") == "python" - - vip = response.parse() - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - assert cast(Any, response.is_closed) is True - - @parametrize - def test_path_params_replace_connected_ports(self, client: Gcore) -> None: - with pytest.raises(ValueError, match=r"Expected a non-empty value for `port_id` but received ''"): - client.cloud.reserved_fixed_ips.vip.with_raw_response.replace_connected_ports( - port_id="", - project_id=0, - region_id=0, - ) - @parametrize def test_method_toggle(self, client: Gcore) -> None: vip = client.cloud.reserved_fixed_ips.vip.toggle( @@ -219,216 +67,12 @@ def test_path_params_toggle(self, client: Gcore) -> None: is_vip=True, ) - @parametrize - def test_method_update_connected_ports(self, client: Gcore) -> None: - vip = client.cloud.reserved_fixed_ips.vip.update_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - @parametrize - def test_method_update_connected_ports_with_all_params(self, client: Gcore) -> None: - vip = client.cloud.reserved_fixed_ips.vip.update_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - port_ids=["351b0dd7-ca09-431c-be53-935db3785067", "bc688791-f1b0-44eb-97d4-07697294b1e1"], - ) - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - @parametrize - def test_raw_response_update_connected_ports(self, client: Gcore) -> None: - response = client.cloud.reserved_fixed_ips.vip.with_raw_response.update_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) - - assert response.is_closed is True - assert response.http_request.headers.get("X-Stainless-Lang") == "python" - vip = response.parse() - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - @parametrize - def test_streaming_response_update_connected_ports(self, client: Gcore) -> None: - with client.cloud.reserved_fixed_ips.vip.with_streaming_response.update_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) as response: - assert not response.is_closed - assert response.http_request.headers.get("X-Stainless-Lang") == "python" - - vip = response.parse() - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - assert cast(Any, response.is_closed) is True - - @parametrize - def test_path_params_update_connected_ports(self, client: Gcore) -> None: - with pytest.raises(ValueError, match=r"Expected a non-empty value for `port_id` but received ''"): - client.cloud.reserved_fixed_ips.vip.with_raw_response.update_connected_ports( - port_id="", - project_id=0, - region_id=0, - ) - class TestAsyncVip: parametrize = pytest.mark.parametrize( "async_client", [False, True, {"http_client": "aiohttp"}], indirect=True, ids=["loose", "strict", "aiohttp"] ) - @parametrize - async def test_method_list_candidate_ports(self, async_client: AsyncGcore) -> None: - vip = await async_client.cloud.reserved_fixed_ips.vip.list_candidate_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) - assert_matches_type(CandidatePortList, vip, path=["response"]) - - @parametrize - async def test_raw_response_list_candidate_ports(self, async_client: AsyncGcore) -> None: - response = await async_client.cloud.reserved_fixed_ips.vip.with_raw_response.list_candidate_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) - - assert response.is_closed is True - assert response.http_request.headers.get("X-Stainless-Lang") == "python" - vip = await response.parse() - assert_matches_type(CandidatePortList, vip, path=["response"]) - - @parametrize - async def test_streaming_response_list_candidate_ports(self, async_client: AsyncGcore) -> None: - async with async_client.cloud.reserved_fixed_ips.vip.with_streaming_response.list_candidate_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) as response: - assert not response.is_closed - assert response.http_request.headers.get("X-Stainless-Lang") == "python" - - vip = await response.parse() - assert_matches_type(CandidatePortList, vip, path=["response"]) - - assert cast(Any, response.is_closed) is True - - @parametrize - async def test_path_params_list_candidate_ports(self, async_client: AsyncGcore) -> None: - with pytest.raises(ValueError, match=r"Expected a non-empty value for `port_id` but received ''"): - await async_client.cloud.reserved_fixed_ips.vip.with_raw_response.list_candidate_ports( - port_id="", - project_id=0, - region_id=0, - ) - - @parametrize - async def test_method_list_connected_ports(self, async_client: AsyncGcore) -> None: - vip = await async_client.cloud.reserved_fixed_ips.vip.list_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - @parametrize - async def test_raw_response_list_connected_ports(self, async_client: AsyncGcore) -> None: - response = await async_client.cloud.reserved_fixed_ips.vip.with_raw_response.list_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) - - assert response.is_closed is True - assert response.http_request.headers.get("X-Stainless-Lang") == "python" - vip = await response.parse() - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - @parametrize - async def test_streaming_response_list_connected_ports(self, async_client: AsyncGcore) -> None: - async with async_client.cloud.reserved_fixed_ips.vip.with_streaming_response.list_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) as response: - assert not response.is_closed - assert response.http_request.headers.get("X-Stainless-Lang") == "python" - - vip = await response.parse() - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - assert cast(Any, response.is_closed) is True - - @parametrize - async def test_path_params_list_connected_ports(self, async_client: AsyncGcore) -> None: - with pytest.raises(ValueError, match=r"Expected a non-empty value for `port_id` but received ''"): - await async_client.cloud.reserved_fixed_ips.vip.with_raw_response.list_connected_ports( - port_id="", - project_id=0, - region_id=0, - ) - - @parametrize - async def test_method_replace_connected_ports(self, async_client: AsyncGcore) -> None: - vip = await async_client.cloud.reserved_fixed_ips.vip.replace_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - @parametrize - async def test_method_replace_connected_ports_with_all_params(self, async_client: AsyncGcore) -> None: - vip = await async_client.cloud.reserved_fixed_ips.vip.replace_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - port_ids=["351b0dd7-ca09-431c-be53-935db3785067", "bc688791-f1b0-44eb-97d4-07697294b1e1"], - ) - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - @parametrize - async def test_raw_response_replace_connected_ports(self, async_client: AsyncGcore) -> None: - response = await async_client.cloud.reserved_fixed_ips.vip.with_raw_response.replace_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) - - assert response.is_closed is True - assert response.http_request.headers.get("X-Stainless-Lang") == "python" - vip = await response.parse() - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - @parametrize - async def test_streaming_response_replace_connected_ports(self, async_client: AsyncGcore) -> None: - async with async_client.cloud.reserved_fixed_ips.vip.with_streaming_response.replace_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) as response: - assert not response.is_closed - assert response.http_request.headers.get("X-Stainless-Lang") == "python" - - vip = await response.parse() - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - assert cast(Any, response.is_closed) is True - - @parametrize - async def test_path_params_replace_connected_ports(self, async_client: AsyncGcore) -> None: - with pytest.raises(ValueError, match=r"Expected a non-empty value for `port_id` but received ''"): - await async_client.cloud.reserved_fixed_ips.vip.with_raw_response.replace_connected_ports( - port_id="", - project_id=0, - region_id=0, - ) - @parametrize async def test_method_toggle(self, async_client: AsyncGcore) -> None: vip = await async_client.cloud.reserved_fixed_ips.vip.toggle( @@ -478,59 +122,3 @@ async def test_path_params_toggle(self, async_client: AsyncGcore) -> None: region_id=0, is_vip=True, ) - - @parametrize - async def test_method_update_connected_ports(self, async_client: AsyncGcore) -> None: - vip = await async_client.cloud.reserved_fixed_ips.vip.update_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - @parametrize - async def test_method_update_connected_ports_with_all_params(self, async_client: AsyncGcore) -> None: - vip = await async_client.cloud.reserved_fixed_ips.vip.update_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - port_ids=["351b0dd7-ca09-431c-be53-935db3785067", "bc688791-f1b0-44eb-97d4-07697294b1e1"], - ) - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - @parametrize - async def test_raw_response_update_connected_ports(self, async_client: AsyncGcore) -> None: - response = await async_client.cloud.reserved_fixed_ips.vip.with_raw_response.update_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) - - assert response.is_closed is True - assert response.http_request.headers.get("X-Stainless-Lang") == "python" - vip = await response.parse() - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - @parametrize - async def test_streaming_response_update_connected_ports(self, async_client: AsyncGcore) -> None: - async with async_client.cloud.reserved_fixed_ips.vip.with_streaming_response.update_connected_ports( - port_id="port_id", - project_id=0, - region_id=0, - ) as response: - assert not response.is_closed - assert response.http_request.headers.get("X-Stainless-Lang") == "python" - - vip = await response.parse() - assert_matches_type(ConnectedPortList, vip, path=["response"]) - - assert cast(Any, response.is_closed) is True - - @parametrize - async def test_path_params_update_connected_ports(self, async_client: AsyncGcore) -> None: - with pytest.raises(ValueError, match=r"Expected a non-empty value for `port_id` but received ''"): - await async_client.cloud.reserved_fixed_ips.vip.with_raw_response.update_connected_ports( - port_id="", - project_id=0, - region_id=0, - ) diff --git a/tests/api_resources/cloud/reserved_fixed_ips/vip/__init__.py b/tests/api_resources/cloud/reserved_fixed_ips/vip/__init__.py new file mode 100644 index 00000000..fd8019a9 --- /dev/null +++ b/tests/api_resources/cloud/reserved_fixed_ips/vip/__init__.py @@ -0,0 +1 @@ +# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. diff --git a/tests/api_resources/cloud/reserved_fixed_ips/vip/test_candidate_ports.py b/tests/api_resources/cloud/reserved_fixed_ips/vip/test_candidate_ports.py new file mode 100644 index 00000000..cc679c0e --- /dev/null +++ b/tests/api_resources/cloud/reserved_fixed_ips/vip/test_candidate_ports.py @@ -0,0 +1,116 @@ +# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. + +from __future__ import annotations + +import os +from typing import Any, cast + +import pytest + +from gcore import Gcore, AsyncGcore +from tests.utils import assert_matches_type +from gcore.types.cloud.reserved_fixed_ips.vip import CandidatePortList + +base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010") + + +class TestCandidatePorts: + parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"]) + + @parametrize + def test_method_list(self, client: Gcore) -> None: + candidate_port = client.cloud.reserved_fixed_ips.vip.candidate_ports.list( + port_id="port_id", + project_id=0, + region_id=0, + ) + assert_matches_type(CandidatePortList, candidate_port, path=["response"]) + + @parametrize + def test_raw_response_list(self, client: Gcore) -> None: + response = client.cloud.reserved_fixed_ips.vip.candidate_ports.with_raw_response.list( + port_id="port_id", + project_id=0, + region_id=0, + ) + + assert response.is_closed is True + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + candidate_port = response.parse() + assert_matches_type(CandidatePortList, candidate_port, path=["response"]) + + @parametrize + def test_streaming_response_list(self, client: Gcore) -> None: + with client.cloud.reserved_fixed_ips.vip.candidate_ports.with_streaming_response.list( + port_id="port_id", + project_id=0, + region_id=0, + ) as response: + assert not response.is_closed + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + + candidate_port = response.parse() + assert_matches_type(CandidatePortList, candidate_port, path=["response"]) + + assert cast(Any, response.is_closed) is True + + @parametrize + def test_path_params_list(self, client: Gcore) -> None: + with pytest.raises(ValueError, match=r"Expected a non-empty value for `port_id` but received ''"): + client.cloud.reserved_fixed_ips.vip.candidate_ports.with_raw_response.list( + port_id="", + project_id=0, + region_id=0, + ) + + +class TestAsyncCandidatePorts: + parametrize = pytest.mark.parametrize( + "async_client", [False, True, {"http_client": "aiohttp"}], indirect=True, ids=["loose", "strict", "aiohttp"] + ) + + @parametrize + async def test_method_list(self, async_client: AsyncGcore) -> None: + candidate_port = await async_client.cloud.reserved_fixed_ips.vip.candidate_ports.list( + port_id="port_id", + project_id=0, + region_id=0, + ) + assert_matches_type(CandidatePortList, candidate_port, path=["response"]) + + @parametrize + async def test_raw_response_list(self, async_client: AsyncGcore) -> None: + response = await async_client.cloud.reserved_fixed_ips.vip.candidate_ports.with_raw_response.list( + port_id="port_id", + project_id=0, + region_id=0, + ) + + assert response.is_closed is True + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + candidate_port = await response.parse() + assert_matches_type(CandidatePortList, candidate_port, path=["response"]) + + @parametrize + async def test_streaming_response_list(self, async_client: AsyncGcore) -> None: + async with async_client.cloud.reserved_fixed_ips.vip.candidate_ports.with_streaming_response.list( + port_id="port_id", + project_id=0, + region_id=0, + ) as response: + assert not response.is_closed + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + + candidate_port = await response.parse() + assert_matches_type(CandidatePortList, candidate_port, path=["response"]) + + assert cast(Any, response.is_closed) is True + + @parametrize + async def test_path_params_list(self, async_client: AsyncGcore) -> None: + with pytest.raises(ValueError, match=r"Expected a non-empty value for `port_id` but received ''"): + await async_client.cloud.reserved_fixed_ips.vip.candidate_ports.with_raw_response.list( + port_id="", + project_id=0, + region_id=0, + ) diff --git a/tests/api_resources/cloud/reserved_fixed_ips/vip/test_connected_ports.py b/tests/api_resources/cloud/reserved_fixed_ips/vip/test_connected_ports.py new file mode 100644 index 00000000..7c47c319 --- /dev/null +++ b/tests/api_resources/cloud/reserved_fixed_ips/vip/test_connected_ports.py @@ -0,0 +1,342 @@ +# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details. + +from __future__ import annotations + +import os +from typing import Any, cast + +import pytest + +from gcore import Gcore, AsyncGcore +from tests.utils import assert_matches_type +from gcore.types.cloud.reserved_fixed_ips.vip import ( + ConnectedPortList, +) + +base_url = os.environ.get("TEST_API_BASE_URL", "http://127.0.0.1:4010") + + +class TestConnectedPorts: + parametrize = pytest.mark.parametrize("client", [False, True], indirect=True, ids=["loose", "strict"]) + + @parametrize + def test_method_list(self, client: Gcore) -> None: + connected_port = client.cloud.reserved_fixed_ips.vip.connected_ports.list( + port_id="port_id", + project_id=0, + region_id=0, + ) + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + @parametrize + def test_raw_response_list(self, client: Gcore) -> None: + response = client.cloud.reserved_fixed_ips.vip.connected_ports.with_raw_response.list( + port_id="port_id", + project_id=0, + region_id=0, + ) + + assert response.is_closed is True + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + connected_port = response.parse() + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + @parametrize + def test_streaming_response_list(self, client: Gcore) -> None: + with client.cloud.reserved_fixed_ips.vip.connected_ports.with_streaming_response.list( + port_id="port_id", + project_id=0, + region_id=0, + ) as response: + assert not response.is_closed + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + + connected_port = response.parse() + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + assert cast(Any, response.is_closed) is True + + @parametrize + def test_path_params_list(self, client: Gcore) -> None: + with pytest.raises(ValueError, match=r"Expected a non-empty value for `port_id` but received ''"): + client.cloud.reserved_fixed_ips.vip.connected_ports.with_raw_response.list( + port_id="", + project_id=0, + region_id=0, + ) + + @parametrize + def test_method_add(self, client: Gcore) -> None: + connected_port = client.cloud.reserved_fixed_ips.vip.connected_ports.add( + port_id="port_id", + project_id=0, + region_id=0, + ) + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + @parametrize + def test_method_add_with_all_params(self, client: Gcore) -> None: + connected_port = client.cloud.reserved_fixed_ips.vip.connected_ports.add( + port_id="port_id", + project_id=0, + region_id=0, + port_ids=["351b0dd7-ca09-431c-be53-935db3785067", "bc688791-f1b0-44eb-97d4-07697294b1e1"], + ) + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + @parametrize + def test_raw_response_add(self, client: Gcore) -> None: + response = client.cloud.reserved_fixed_ips.vip.connected_ports.with_raw_response.add( + port_id="port_id", + project_id=0, + region_id=0, + ) + + assert response.is_closed is True + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + connected_port = response.parse() + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + @parametrize + def test_streaming_response_add(self, client: Gcore) -> None: + with client.cloud.reserved_fixed_ips.vip.connected_ports.with_streaming_response.add( + port_id="port_id", + project_id=0, + region_id=0, + ) as response: + assert not response.is_closed + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + + connected_port = response.parse() + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + assert cast(Any, response.is_closed) is True + + @parametrize + def test_path_params_add(self, client: Gcore) -> None: + with pytest.raises(ValueError, match=r"Expected a non-empty value for `port_id` but received ''"): + client.cloud.reserved_fixed_ips.vip.connected_ports.with_raw_response.add( + port_id="", + project_id=0, + region_id=0, + ) + + @parametrize + def test_method_replace(self, client: Gcore) -> None: + connected_port = client.cloud.reserved_fixed_ips.vip.connected_ports.replace( + port_id="port_id", + project_id=0, + region_id=0, + ) + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + @parametrize + def test_method_replace_with_all_params(self, client: Gcore) -> None: + connected_port = client.cloud.reserved_fixed_ips.vip.connected_ports.replace( + port_id="port_id", + project_id=0, + region_id=0, + port_ids=["351b0dd7-ca09-431c-be53-935db3785067", "bc688791-f1b0-44eb-97d4-07697294b1e1"], + ) + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + @parametrize + def test_raw_response_replace(self, client: Gcore) -> None: + response = client.cloud.reserved_fixed_ips.vip.connected_ports.with_raw_response.replace( + port_id="port_id", + project_id=0, + region_id=0, + ) + + assert response.is_closed is True + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + connected_port = response.parse() + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + @parametrize + def test_streaming_response_replace(self, client: Gcore) -> None: + with client.cloud.reserved_fixed_ips.vip.connected_ports.with_streaming_response.replace( + port_id="port_id", + project_id=0, + region_id=0, + ) as response: + assert not response.is_closed + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + + connected_port = response.parse() + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + assert cast(Any, response.is_closed) is True + + @parametrize + def test_path_params_replace(self, client: Gcore) -> None: + with pytest.raises(ValueError, match=r"Expected a non-empty value for `port_id` but received ''"): + client.cloud.reserved_fixed_ips.vip.connected_ports.with_raw_response.replace( + port_id="", + project_id=0, + region_id=0, + ) + + +class TestAsyncConnectedPorts: + parametrize = pytest.mark.parametrize( + "async_client", [False, True, {"http_client": "aiohttp"}], indirect=True, ids=["loose", "strict", "aiohttp"] + ) + + @parametrize + async def test_method_list(self, async_client: AsyncGcore) -> None: + connected_port = await async_client.cloud.reserved_fixed_ips.vip.connected_ports.list( + port_id="port_id", + project_id=0, + region_id=0, + ) + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + @parametrize + async def test_raw_response_list(self, async_client: AsyncGcore) -> None: + response = await async_client.cloud.reserved_fixed_ips.vip.connected_ports.with_raw_response.list( + port_id="port_id", + project_id=0, + region_id=0, + ) + + assert response.is_closed is True + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + connected_port = await response.parse() + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + @parametrize + async def test_streaming_response_list(self, async_client: AsyncGcore) -> None: + async with async_client.cloud.reserved_fixed_ips.vip.connected_ports.with_streaming_response.list( + port_id="port_id", + project_id=0, + region_id=0, + ) as response: + assert not response.is_closed + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + + connected_port = await response.parse() + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + assert cast(Any, response.is_closed) is True + + @parametrize + async def test_path_params_list(self, async_client: AsyncGcore) -> None: + with pytest.raises(ValueError, match=r"Expected a non-empty value for `port_id` but received ''"): + await async_client.cloud.reserved_fixed_ips.vip.connected_ports.with_raw_response.list( + port_id="", + project_id=0, + region_id=0, + ) + + @parametrize + async def test_method_add(self, async_client: AsyncGcore) -> None: + connected_port = await async_client.cloud.reserved_fixed_ips.vip.connected_ports.add( + port_id="port_id", + project_id=0, + region_id=0, + ) + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + @parametrize + async def test_method_add_with_all_params(self, async_client: AsyncGcore) -> None: + connected_port = await async_client.cloud.reserved_fixed_ips.vip.connected_ports.add( + port_id="port_id", + project_id=0, + region_id=0, + port_ids=["351b0dd7-ca09-431c-be53-935db3785067", "bc688791-f1b0-44eb-97d4-07697294b1e1"], + ) + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + @parametrize + async def test_raw_response_add(self, async_client: AsyncGcore) -> None: + response = await async_client.cloud.reserved_fixed_ips.vip.connected_ports.with_raw_response.add( + port_id="port_id", + project_id=0, + region_id=0, + ) + + assert response.is_closed is True + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + connected_port = await response.parse() + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + @parametrize + async def test_streaming_response_add(self, async_client: AsyncGcore) -> None: + async with async_client.cloud.reserved_fixed_ips.vip.connected_ports.with_streaming_response.add( + port_id="port_id", + project_id=0, + region_id=0, + ) as response: + assert not response.is_closed + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + + connected_port = await response.parse() + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + assert cast(Any, response.is_closed) is True + + @parametrize + async def test_path_params_add(self, async_client: AsyncGcore) -> None: + with pytest.raises(ValueError, match=r"Expected a non-empty value for `port_id` but received ''"): + await async_client.cloud.reserved_fixed_ips.vip.connected_ports.with_raw_response.add( + port_id="", + project_id=0, + region_id=0, + ) + + @parametrize + async def test_method_replace(self, async_client: AsyncGcore) -> None: + connected_port = await async_client.cloud.reserved_fixed_ips.vip.connected_ports.replace( + port_id="port_id", + project_id=0, + region_id=0, + ) + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + @parametrize + async def test_method_replace_with_all_params(self, async_client: AsyncGcore) -> None: + connected_port = await async_client.cloud.reserved_fixed_ips.vip.connected_ports.replace( + port_id="port_id", + project_id=0, + region_id=0, + port_ids=["351b0dd7-ca09-431c-be53-935db3785067", "bc688791-f1b0-44eb-97d4-07697294b1e1"], + ) + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + @parametrize + async def test_raw_response_replace(self, async_client: AsyncGcore) -> None: + response = await async_client.cloud.reserved_fixed_ips.vip.connected_ports.with_raw_response.replace( + port_id="port_id", + project_id=0, + region_id=0, + ) + + assert response.is_closed is True + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + connected_port = await response.parse() + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + @parametrize + async def test_streaming_response_replace(self, async_client: AsyncGcore) -> None: + async with async_client.cloud.reserved_fixed_ips.vip.connected_ports.with_streaming_response.replace( + port_id="port_id", + project_id=0, + region_id=0, + ) as response: + assert not response.is_closed + assert response.http_request.headers.get("X-Stainless-Lang") == "python" + + connected_port = await response.parse() + assert_matches_type(ConnectedPortList, connected_port, path=["response"]) + + assert cast(Any, response.is_closed) is True + + @parametrize + async def test_path_params_replace(self, async_client: AsyncGcore) -> None: + with pytest.raises(ValueError, match=r"Expected a non-empty value for `port_id` but received ''"): + await async_client.cloud.reserved_fixed_ips.vip.connected_ports.with_raw_response.replace( + port_id="", + project_id=0, + region_id=0, + )