diff --git a/hcloud/storage_boxes/client.py b/hcloud/storage_boxes/client.py index c76f7468..498963ff 100644 --- a/hcloud/storage_boxes/client.py +++ b/hcloud/storage_boxes/client.py @@ -2,7 +2,7 @@ from typing import TYPE_CHECKING, Any, NamedTuple -from ..actions import BoundAction +from ..actions import ActionsPageResult, BoundAction, ResourceActionsClient from ..core import BoundModelBase, Meta, ResourceClientBase from ..locations import BoundLocation, Location from ..storage_box_types import BoundStorageBoxType, StorageBoxType @@ -12,6 +12,7 @@ StorageBox, StorageBoxAccessSettings, StorageBoxFoldersResponse, + StorageBoxSnapshot, StorageBoxSnapshotPlan, StorageBoxStats, ) @@ -55,6 +56,52 @@ def __init__( super().__init__(client, data, complete) + def get_actions_list( + self, + *, + status: list[str] | None = None, + sort: list[str] | None = None, + page: int | None = None, + per_page: int | None = None, + ) -> ActionsPageResult: + """ + Returns all Actions for the Storage Box for a specific page. + + See https://docs.hetzner.cloud/reference/hetzner#storage-box-actions-list-actions + + :param status: Filter the actions by status. The response will only contain actions matching the specified statuses. + :param sort: Sort resources by field and direction. + :param page: Page number to return. + :param per_page: Maximum number of entries returned per page. + """ + return self._client.get_actions_list( + self, + status=status, + sort=sort, + page=page, + per_page=per_page, + ) + + def get_actions( + self, + *, + status: list[str] | None = None, + sort: list[str] | None = None, + ) -> list[BoundAction]: + """ + Returns all Actions for the Storage Box. + + See https://docs.hetzner.cloud/reference/hetzner#storage-box-actions-list-actions + + :param status: Filter the actions by status. The response will only contain actions matching the specified statuses. + :param sort: Sort resources by field and direction. + """ + return self._client.get_actions( + self, + status=status, + sort=sort, + ) + # TODO: implement bound methods @@ -72,9 +119,16 @@ class StorageBoxesClient(ResourceClientBase): _base_url = "/storage_boxes" + actions: ResourceActionsClient + """Storage Boxes scoped actions client + + :type: :class:`ResourceActionsClient ` + """ + def __init__(self, client: Client): super().__init__(client) self._client = client._client_hetzner + self.actions = ResourceActionsClient(self, self._base_url) def get_by_id(self, id: int) -> BoundStorageBox: """ @@ -241,7 +295,7 @@ def delete( """ Deletes a Storage Box. - See https://docs.hetzner.cloud/reference/hetzner#storage-boxes-delete-storage-box + See https://docs.hetzner.cloud/reference/hetzner#storage-boxes-delete-a-storage-box :param storage_box: Storage Box to delete. """ @@ -264,7 +318,7 @@ def get_folders( Files are not part of the response. - See https://docs.hetzner.cloud/reference/hetzner#storage-boxes-list-content-of-storage-box + See https://docs.hetzner.cloud/reference/hetzner#storage-boxes-list-folders-of-a-storage-box :param storage_box: Storage Box to list the folders from. :param path: Relative path to list the folders from. @@ -280,3 +334,225 @@ def get_folders( ) return StorageBoxFoldersResponse(folders=response["folders"]) + + def get_actions_list( + self, + storage_box: StorageBox | BoundStorageBox, + *, + status: list[str] | None = None, + sort: list[str] | None = None, + page: int | None = None, + per_page: int | None = None, + ) -> ActionsPageResult: + """ + Returns all Actions for a Storage Box for a specific page. + + See https://docs.hetzner.cloud/reference/hetzner#storage-box-actions-list-actions-for-a-storage-box + + :param storage_box: Storage Box to fetch the Actions from. + :param status: Filter the actions by status. The response will only contain actions matching the specified statuses. + :param sort: Sort resources by field and direction. + :param page: Page number to return. + :param per_page: Maximum number of entries returned per page. + """ + params: dict[str, Any] = {} + if status is not None: + params["status"] = status + if sort is not None: + params["sort"] = sort + if page is not None: + params["page"] = page + if per_page is not None: + params["per_page"] = per_page + + response = self._client.request( + method="GET", + url=f"/storage_boxes/{storage_box.id}/actions", + params=params, + ) + return ActionsPageResult( + actions=[BoundAction(self._parent.actions, o) for o in response["actions"]], + meta=Meta.parse_meta(response), + ) + + def get_actions( + self, + storage_box: StorageBox | BoundStorageBox, + *, + status: list[str] | None = None, + sort: list[str] | None = None, + ) -> list[BoundAction]: + """ + Returns all Actions for a Storage Box. + + See https://docs.hetzner.cloud/reference/hetzner#storage-box-actions-list-actions-for-a-storage-box + + :param storage_box: Storage Box to fetch the Actions from. + :param status: Filter the actions by status. The response will only contain actions matching the specified statuses. + :param sort: Sort resources by field and direction. + """ + return self._iter_pages( + self.get_actions_list, + storage_box, + status=status, + sort=sort, + ) + + def change_protection( + self, + storage_box: StorageBox | BoundStorageBox, + *, + delete: bool | None = None, + ) -> BoundAction: + """ + Changes the protection of a Storage Box. + + See https://docs.hetzner.cloud/reference/hetzner#storage-box-actions-change-protection + + :param storage_box: Storage Box to update. + :param delete: Prevents the Storage Box from being deleted. + """ + data: dict[str, Any] = {} + if delete is not None: + data["delete"] = delete + + response = self._client.request( + method="POST", + url=f"{self._base_url}/{storage_box.id}/actions/change_protection", + json=data, + ) + return BoundAction(self._parent.actions, response["action"]) + + def change_type( + self, + storage_box: StorageBox | BoundStorageBox, + storage_box_type: StorageBoxType | BoundStorageBoxType, + ) -> BoundAction: + """ + Changes the type of a Storage Box. + + See https://docs.hetzner.cloud/reference/hetzner#storage-box-actions-change-type + + :param storage_box: Storage Box to update. + :param storage_box_type: Storage Box Type to change to. + """ + data: dict[str, Any] = { + "storage_box_type": storage_box_type.id_or_name, + } + + response = self._client.request( + method="POST", + url=f"{self._base_url}/{storage_box.id}/actions/change_type", + json=data, + ) + return BoundAction(self._parent.actions, response["action"]) + + def reset_password( + self, + storage_box: StorageBox | BoundStorageBox, + *, + password: str, + ) -> BoundAction: + """ + Reset the password of a Storage Box. + + See https://docs.hetzner.cloud/reference/hetzner#storage-box-actions-reset-password + + :param storage_box: Storage Box to update. + :param password: New password. + """ + data: dict[str, Any] = { + "password": password, + } + + response = self._client.request( + method="POST", + url=f"{self._base_url}/{storage_box.id}/actions/reset_password", + json=data, + ) + return BoundAction(self._parent.actions, response["action"]) + + def update_access_settings( + self, + storage_box: StorageBox | BoundStorageBox, + access_settings: StorageBoxAccessSettings, + ) -> BoundAction: + """ + Reset the password of a Storage Box. + + See https://docs.hetzner.cloud/reference/hetzner#storage-box-actions-update-access-settings + + :param storage_box: Storage Box to update. + :param access_settings: New access settings for the Storage Box. + """ + data: dict[str, Any] = access_settings.to_payload() + + response = self._client.request( + method="POST", + url=f"{self._base_url}/{storage_box.id}/actions/update_access_settings", + json=data, + ) + return BoundAction(self._parent.actions, response["action"]) + + def rollback_snapshot( + self, + storage_box: StorageBox | BoundStorageBox, + snapshot: StorageBoxSnapshot, # TODO: Add BoundStorageBoxSnapshot + ) -> BoundAction: + """ + Rollback the Storage Box to the given snapshot. + + See https://docs.hetzner.cloud/reference/hetzner#storage-box-actions-rollback-snapshot + + :param storage_box: Storage Box to update. + :param snapshot: Snapshot to rollback to. + """ + data: dict[str, Any] = { + "snapshot": snapshot.id_or_name, + } + + response = self._client.request( + method="POST", + url=f"{self._base_url}/{storage_box.id}/actions/rollback_snapshot", + json=data, + ) + return BoundAction(self._parent.actions, response["action"]) + + def disable_snapshot_plan( + self, + storage_box: StorageBox | BoundStorageBox, + ) -> BoundAction: + """ + Disable the snapshot plan a Storage Box. + + See https://docs.hetzner.cloud/reference/hetzner#storage-box-actions-disable-snapshot-plan + + :param storage_box: Storage Box to update. + """ + response = self._client.request( + method="POST", + url=f"{self._base_url}/{storage_box.id}/actions/disable_snapshot_plan", + ) + return BoundAction(self._parent.actions, response["action"]) + + def enable_snapshot_plan( + self, + storage_box: StorageBox | BoundStorageBox, + snapshot_plan: StorageBoxSnapshotPlan, + ) -> BoundAction: + """ + Enable the snapshot plan a Storage Box. + + See https://docs.hetzner.cloud/reference/hetzner#storage-box-actions-enable-snapshot-plan + + :param storage_box: Storage Box to update. + :param snapshot_plan: Snapshot Plan to enable. + """ + data: dict[str, Any] = snapshot_plan.to_payload() + + response = self._client.request( + method="POST", + url=f"{self._base_url}/{storage_box.id}/actions/enable_snapshot_plan", + json=data, + ) + return BoundAction(self._parent.actions, response["action"]) diff --git a/hcloud/storage_boxes/domain.py b/hcloud/storage_boxes/domain.py index d324b6f4..6f01f406 100644 --- a/hcloud/storage_boxes/domain.py +++ b/hcloud/storage_boxes/domain.py @@ -157,8 +157,8 @@ class StorageBoxSnapshotPlan(BaseDomain): __api_properties__ = ( "max_snapshots", - "minute", "hour", + "minute", "day_of_week", "day_of_month", ) @@ -166,18 +166,32 @@ class StorageBoxSnapshotPlan(BaseDomain): def __init__( self, - max_snapshots: int | None = None, - minute: int | None = None, - hour: int | None = None, + max_snapshots: int, + hour: int, + minute: int, day_of_week: int | None = None, day_of_month: int | None = None, ): self.max_snapshots = max_snapshots - self.minute = minute self.hour = hour + self.minute = minute self.day_of_week = day_of_week self.day_of_month = day_of_month + def to_payload(self) -> dict[str, Any]: + """ + Generates the request payload from this domain object. + """ + payload: dict[str, Any] = { + "max_snapshots": self.max_snapshots, + "hour": self.hour, + "minute": self.minute, + "day_of_week": self.day_of_week, # API default is null + "day_of_month": self.day_of_month, # API default is null + } + + return payload + class CreateStorageBoxResponse(BaseDomain): """ @@ -227,3 +241,28 @@ def __init__( folders: list[str], ): self.folders = folders + + +# Snapshots +############################################################################### + + +class StorageBoxSnapshot(BaseDomain, DomainIdentityMixin): + """ + Storage Box Snapshot Domain. + """ + + # TODO: full domain + __api_properties__ = ( + "id", + "name", + ) + __slots__ = __api_properties__ + + def __init__( + self, + id: int | None = None, + name: str | None = None, + ): + self.id = id + self.name = name diff --git a/tests/unit/actions/test_client.py b/tests/unit/actions/test_client.py index 6c09f7dd..0b708cfb 100644 --- a/tests/unit/actions/test_client.py +++ b/tests/unit/actions/test_client.py @@ -22,6 +22,7 @@ from hcloud.networks import BoundNetwork, NetworksClient from hcloud.primary_ips import BoundPrimaryIP, PrimaryIPsClient from hcloud.servers import BoundServer, ServersClient +from hcloud.storage_boxes import BoundStorageBox, StorageBoxesClient from hcloud.volumes import BoundVolume, VolumesClient from hcloud.zones import BoundZone, ZonesClient @@ -38,6 +39,7 @@ "servers": (ServersClient, BoundServer), "volumes": (VolumesClient, BoundVolume), "zones": (ZonesClient, BoundZone), + "storage_boxes": (StorageBoxesClient, BoundStorageBox), } diff --git a/tests/unit/storage_boxes/test_client.py b/tests/unit/storage_boxes/test_client.py index 77ff925f..b1eab37c 100644 --- a/tests/unit/storage_boxes/test_client.py +++ b/tests/unit/storage_boxes/test_client.py @@ -14,8 +14,9 @@ BoundStorageBox, StorageBox, StorageBoxesClient, + StorageBoxSnapshotPlan, ) -from hcloud.storage_boxes.domain import StorageBoxAccessSettings +from hcloud.storage_boxes.domain import StorageBoxAccessSettings, StorageBoxSnapshot from ..conftest import BoundModelTestCase, assert_bound_action1 @@ -39,7 +40,9 @@ def resource_client(self, client: Client) -> StorageBoxesClient: @pytest.fixture() def bound_model( - self, resource_client: StorageBoxesClient, storage_box1 + self, + resource_client: StorageBoxesClient, + storage_box1, ) -> BoundStorageBox: return BoundStorageBox(resource_client, data=storage_box1) @@ -309,3 +312,164 @@ def test_get_folders( ) assert result.folders == ["dir1", "dir2"] + + def test_change_protection( + self, + request_mock: mock.MagicMock, + resource_client: StorageBoxesClient, + action_response, + ): + request_mock.return_value = action_response + + action = resource_client.change_protection(StorageBox(id=42), delete=True) + + request_mock.assert_called_with( + method="POST", + url="/storage_boxes/42/actions/change_protection", + json={"delete": True}, + ) + + assert_bound_action1(action, resource_client._parent.actions) + + def test_change_type( + self, + request_mock: mock.MagicMock, + resource_client: StorageBoxesClient, + action_response, + ): + request_mock.return_value = action_response + + action = resource_client.change_type( + StorageBox(id=42), + StorageBoxType(name="bx21"), + ) + + request_mock.assert_called_with( + method="POST", + url="/storage_boxes/42/actions/change_type", + json={"storage_box_type": "bx21"}, + ) + + assert_bound_action1(action, resource_client._parent.actions) + + def test_reset_password( + self, + request_mock: mock.MagicMock, + resource_client: StorageBoxesClient, + action_response, + ): + request_mock.return_value = action_response + + action = resource_client.reset_password( + StorageBox(id=42), + password="password", + ) + + request_mock.assert_called_with( + method="POST", + url="/storage_boxes/42/actions/reset_password", + json={"password": "password"}, + ) + + assert_bound_action1(action, resource_client._parent.actions) + + def test_update_access_settings( + self, + request_mock: mock.MagicMock, + resource_client: StorageBoxesClient, + action_response, + ): + request_mock.return_value = action_response + + action = resource_client.update_access_settings( + StorageBox(id=42), + StorageBoxAccessSettings( + reachable_externally=True, + ssh_enabled=True, + webdav_enabled=False, + ), + ) + + request_mock.assert_called_with( + method="POST", + url="/storage_boxes/42/actions/update_access_settings", + json={ + "reachable_externally": True, + "ssh_enabled": True, + "webdav_enabled": False, + }, + ) + + assert_bound_action1(action, resource_client._parent.actions) + + def test_rollback_snapshot( + self, + request_mock: mock.MagicMock, + resource_client: StorageBoxesClient, + action_response, + ): + request_mock.return_value = action_response + + action = resource_client.rollback_snapshot( + StorageBox(id=42), + StorageBoxSnapshot(id=32), + ) + + request_mock.assert_called_with( + method="POST", + url="/storage_boxes/42/actions/rollback_snapshot", + json={"snapshot": 32}, + ) + + assert_bound_action1(action, resource_client._parent.actions) + + def test_disable_snapshot_plan( + self, + request_mock: mock.MagicMock, + resource_client: StorageBoxesClient, + action_response, + ): + request_mock.return_value = action_response + + action = resource_client.disable_snapshot_plan( + StorageBox(id=42), + ) + + request_mock.assert_called_with( + method="POST", + url="/storage_boxes/42/actions/disable_snapshot_plan", + ) + + assert_bound_action1(action, resource_client._parent.actions) + + def test_enable_snapshot_plan( + self, + request_mock: mock.MagicMock, + resource_client: StorageBoxesClient, + action_response, + ): + request_mock.return_value = action_response + + action = resource_client.enable_snapshot_plan( + StorageBox(id=42), + StorageBoxSnapshotPlan( + max_snapshots=10, + hour=3, + minute=30, + day_of_week=None, + ), + ) + + request_mock.assert_called_with( + method="POST", + url="/storage_boxes/42/actions/enable_snapshot_plan", + json={ + "max_snapshots": 10, + "hour": 3, + "minute": 30, + "day_of_week": None, + "day_of_month": None, + }, + ) + + assert_bound_action1(action, resource_client._parent.actions)