diff --git a/hcloud/storage_boxes/__init__.py b/hcloud/storage_boxes/__init__.py index 5f654466..50276417 100644 --- a/hcloud/storage_boxes/__init__.py +++ b/hcloud/storage_boxes/__init__.py @@ -2,8 +2,10 @@ from .client import ( BoundStorageBox, + BoundStorageBoxSnapshot, StorageBoxesClient, StorageBoxesPageResult, + StorageBoxSnapshotsPageResult, ) from .domain import ( CreateStorageBoxResponse, @@ -11,12 +13,14 @@ StorageBox, StorageBoxAccessSettings, StorageBoxFoldersResponse, + StorageBoxSnapshot, StorageBoxSnapshotPlan, StorageBoxStats, ) __all__ = [ "BoundStorageBox", + "BoundStorageBoxSnapshot", "CreateStorageBoxResponse", "DeleteStorageBoxResponse", "StorageBox", @@ -24,6 +28,8 @@ "StorageBoxesClient", "StorageBoxesPageResult", "StorageBoxFoldersResponse", + "StorageBoxSnapshot", "StorageBoxSnapshotPlan", + "StorageBoxSnapshotsPageResult", "StorageBoxStats", ] diff --git a/hcloud/storage_boxes/client.py b/hcloud/storage_boxes/client.py index f9390ac9..c1ec9c26 100644 --- a/hcloud/storage_boxes/client.py +++ b/hcloud/storage_boxes/client.py @@ -8,12 +8,15 @@ from ..storage_box_types import BoundStorageBoxType, StorageBoxType from .domain import ( CreateStorageBoxResponse, + CreateStorageBoxSnapshotResponse, DeleteStorageBoxResponse, + DeleteStorageBoxSnapshotResponse, StorageBox, StorageBoxAccessSettings, StorageBoxFoldersResponse, StorageBoxSnapshot, StorageBoxSnapshotPlan, + StorageBoxSnapshotStats, StorageBoxStats, ) @@ -105,11 +108,42 @@ def get_actions( # TODO: implement bound methods +class BoundStorageBoxSnapshot(BoundModelBase, StorageBoxSnapshot): + _client: StorageBoxesClient + + model = StorageBoxSnapshot + + def __init__( + self, + client: StorageBoxesClient, + data: dict[str, Any], + complete: bool = True, + ): + raw = data.get("storage_box") + if raw is not None: + data["storage_box"] = BoundStorageBox( + client, data={"id": raw}, complete=False + ) + + raw = data.get("stats") + if raw is not None: + data["stats"] = StorageBoxSnapshotStats.from_dict(raw) + + super().__init__(client, data, complete) + + # TODO: implement bound methods + + class StorageBoxesPageResult(NamedTuple): storage_boxes: list[BoundStorageBox] meta: Meta +class StorageBoxSnapshotsPageResult(NamedTuple): + snapshots: list[BoundStorageBoxSnapshot] + meta: Meta + + class StorageBoxesClient(ResourceClientBase): """ A client for the Storage Boxes API. @@ -556,3 +590,198 @@ def enable_snapshot_plan( json=data, ) return BoundAction(self._parent.actions, response["action"]) + + # Snapshots + ########################################################################### + + def get_snapshot_by_id( + self, + storage_box: StorageBox | BoundStorageBox, + id: int, + ) -> BoundStorageBoxSnapshot: + """ + Returns a single Snapshot from a Storage Box. + + See https://docs.hetzner.cloud/reference/hetzner#storage-box-snapshots-get-a-snapshot + + :param storage_box: Storage Box to get the Snapshot from. + :param id: ID of the Snapshot. + """ + response = self._client.request( + method="GET", + url=f"{self._base_url}/{storage_box.id}/snapshots/{id}", + ) + return BoundStorageBoxSnapshot(self, response["snapshot"]) + + def get_snapshot_by_name( + self, + storage_box: StorageBox | BoundStorageBox, + name: str, + ) -> BoundStorageBoxSnapshot: + """ + Returns a single Snapshot from a Storage Box. + + See https://docs.hetzner.cloud/reference/hetzner#storage-box-snapshots-list-snapshots + + :param storage_box: Storage Box to get the Snapshot from. + :param name: Name of the Snapshot. + """ + return self._get_first_by(self.get_snapshot_list, storage_box, name=name) + + def get_snapshot_list( + self, + storage_box: StorageBox | BoundStorageBox, + *, + name: str | None = None, + is_automatic: bool | None = None, + label_selector: str | None = None, + sort: list[str] | None = None, + ) -> StorageBoxSnapshotsPageResult: + """ + Returns all Snapshots for a Storage Box. + + See https://docs.hetzner.cloud/reference/hetzner#storage-box-snapshots-list-snapshots + + :param storage_box: Storage Box to get the Snapshots from. + :param name: Filter resources by their name. The response will only contain the resources matching exactly the specified name. + :param is_automatic: Filter wether the snapshot was made by a Snapshot Plan. + :param label_selector: Filter resources by labels. The response will only contain resources matching the label selector. + :param sort: Sort resources by field and direction. + """ + params: dict[str, Any] = {} + if name is not None: + params["name"] = name + if is_automatic is not None: + params["is_automatic"] = is_automatic + if label_selector is not None: + params["label_selector"] = label_selector + if sort is not None: + params["sort"] = sort + + response = self._client.request( + method="GET", + url=f"{self._base_url}/{storage_box.id}/snapshots", + params=params, + ) + return StorageBoxSnapshotsPageResult( + snapshots=[ + BoundStorageBoxSnapshot(self, item) for item in response["snapshots"] + ], + meta=Meta.parse_meta(response), + ) + + def get_snapshot_all( + self, + storage_box: StorageBox | BoundStorageBox, + *, + name: str | None = None, + is_automatic: bool | None = None, + label_selector: str | None = None, + sort: list[str] | None = None, + ) -> list[BoundStorageBoxSnapshot]: + """ + Returns all Snapshots for a Storage Box. + + See https://docs.hetzner.cloud/reference/hetzner#storage-box-snapshots-list-snapshots + + :param storage_box: Storage Box to get the Snapshots from. + :param name: Filter resources by their name. The response will only contain the resources matching exactly the specified name. + :param is_automatic: Filter wether the snapshot was made by a Snapshot Plan. + :param label_selector: Filter resources by labels. The response will only contain resources matching the label selector. + :param sort: Sort resources by field and direction. + """ + # The endpoint does not have pagination, forward to the list method. + result, _ = self.get_snapshot_list( + storage_box, + name=name, + is_automatic=is_automatic, + label_selector=label_selector, + sort=sort, + ) + return result + + def create_snapshot( + self, + storage_box: StorageBox | BoundStorageBox, + *, + description: str | None = None, + labels: dict[str, str] | None = None, + ) -> CreateStorageBoxSnapshotResponse: + """ + Creates a Snapshot of the Storage Box. + + See https://docs.hetzner.cloud/reference/hetzner#storage-box-snapshots-create-a-snapshot + + :param storage_box: Storage Box to create a Snapshot from. + :param description: Description of the Snapshot. + :param labels: User-defined labels (key/value pairs) for the Resource. + """ + data: dict[str, Any] = {} + if description is not None: + data["description"] = description + if labels is not None: + data["labels"] = labels + + response = self._client.request( + method="POST", + url=f"{self._base_url}/{storage_box.id}/snapshots", + json=data, + ) + return CreateStorageBoxSnapshotResponse( + snapshot=BoundStorageBoxSnapshot(self, response["snapshot"]), + action=BoundAction(self._parent.actions, response["action"]), + ) + + def update_snapshot( + self, + snapshot: StorageBoxSnapshot | BoundStorageBoxSnapshot, + *, + description: str | None = None, + labels: dict[str, str] | None = None, + ) -> BoundStorageBoxSnapshot: + """ + Updates a Storage Box Snapshot. + + See https://docs.hetzner.cloud/reference/hetzner#storage-box-snapshots-update-a-snapshot + + :param snapshot: Storage Box Snapshot to update. + :param description: Description of the Snapshot. + :param labels: User-defined labels (key/value pairs) for the Resource. + """ + if snapshot.storage_box is None: + raise ValueError("snapshot storage_box property is none") + + data: dict[str, Any] = {} + if description is not None: + data["description"] = description + if labels is not None: + data["labels"] = labels + + response = self._client.request( + method="PUT", + url=f"{self._base_url}/{snapshot.storage_box.id}/snapshots/{snapshot.id}", + json=data, + ) + return BoundStorageBoxSnapshot(self, response["snapshot"]) + + def delete_snapshot( + self, + snapshot: StorageBoxSnapshot | BoundStorageBoxSnapshot, + ) -> DeleteStorageBoxSnapshotResponse: + """ + Deletes a Storage Box Snapshot. + + See https://docs.hetzner.cloud/reference/hetzner#storage-box-snapshots-delete-a-snapshot + + :param snapshot: Storage Box Snapshot to delete. + """ + if snapshot.storage_box is None: + raise ValueError("snapshot storage_box property is none") + + response = self._client.request( + method="DELETE", + url=f"{self._base_url}/{snapshot.storage_box.id}/snapshots/{snapshot.id}", + ) + return DeleteStorageBoxSnapshotResponse( + action=BoundAction(self._parent.actions, response["action"]), + ) diff --git a/hcloud/storage_boxes/domain.py b/hcloud/storage_boxes/domain.py index 6f01f406..5414cbf1 100644 --- a/hcloud/storage_boxes/domain.py +++ b/hcloud/storage_boxes/domain.py @@ -10,7 +10,7 @@ from ..storage_box_types import BoundStorageBoxType, StorageBoxType if TYPE_CHECKING: - from .client import BoundStorageBox + from .client import BoundStorageBox, BoundStorageBoxSnapshot StorageBoxStatus = Literal[ "active", @@ -252,10 +252,15 @@ class StorageBoxSnapshot(BaseDomain, DomainIdentityMixin): Storage Box Snapshot Domain. """ - # TODO: full domain __api_properties__ = ( "id", "name", + "description", + "is_automatic", + "labels", + "storage_box", + "created", + "stats", ) __slots__ = __api_properties__ @@ -263,6 +268,73 @@ def __init__( self, id: int | None = None, name: str | None = None, + description: str | None = None, + is_automatic: bool | None = None, + labels: dict[str, str] | None = None, + storage_box: BoundStorageBox | StorageBox | None = None, + created: str | None = None, + stats: StorageBoxSnapshotStats | None = None, ): self.id = id self.name = name + self.description = description + self.is_automatic = is_automatic + self.labels = labels + self.storage_box = storage_box + self.created = isoparse(created) if created else None + self.stats = stats + + +class StorageBoxSnapshotStats(BaseDomain): + """ + Storage Box Snapshot Stats Domain. + """ + + __api_properties__ = ( + "size", + "size_filesystem", + ) + __slots__ = __api_properties__ + + def __init__( + self, + size: int, + size_filesystem: int, + ): + self.size = size + self.size_filesystem = size_filesystem + + +class CreateStorageBoxSnapshotResponse(BaseDomain): + """ + Create Storage Box Snapshot Response Domain. + """ + + __api_properties__ = ( + "snapshot", + "action", + ) + __slots__ = __api_properties__ + + def __init__( + self, + snapshot: BoundStorageBoxSnapshot, + action: BoundAction, + ): + self.snapshot = snapshot + self.action = action + + +class DeleteStorageBoxSnapshotResponse(BaseDomain): + """ + Delete Storage Box Snapshot Response Domain. + """ + + __api_properties__ = ("action",) + __slots__ = __api_properties__ + + def __init__( + self, + action: BoundAction, + ): + self.action = action diff --git a/tests/unit/storage_boxes/conftest.py b/tests/unit/storage_boxes/conftest.py index 6b3a67fb..561123dc 100644 --- a/tests/unit/storage_boxes/conftest.py +++ b/tests/unit/storage_boxes/conftest.py @@ -87,3 +87,45 @@ def storage_box2(): "labels": {}, "protection": {"delete": False}, } + + +@pytest.fixture() +def storage_box_snapshot1(): + return { + "id": 34, + "name": "storage-box-snapshot1", + "description": "", + "is_automatic": False, + "stats": { + "size": 394957594, + "size_filesystem": 3949572745, + }, + "labels": { + "key": "value", + }, + "protection": { + "delete": False, + }, + "created": "2025-11-10T19:16:57Z", + "storage_box": 42, + } + + +@pytest.fixture() +def storage_box_snapshot2(): + return { + "id": 35, + "name": "storage-box-snapshot2", + "description": "", + "is_automatic": True, + "stats": { + "size": 0, + "size_filesystem": 0, + }, + "labels": {}, + "protection": { + "delete": False, + }, + "created": "2025-11-10T19:18:57Z", + "storage_box": 42, + } diff --git a/tests/unit/storage_boxes/test_client.py b/tests/unit/storage_boxes/test_client.py index b1eab37c..9a477258 100644 --- a/tests/unit/storage_boxes/test_client.py +++ b/tests/unit/storage_boxes/test_client.py @@ -12,16 +12,18 @@ from hcloud.storage_box_types import StorageBoxType from hcloud.storage_boxes import ( BoundStorageBox, + BoundStorageBoxSnapshot, StorageBox, + StorageBoxAccessSettings, StorageBoxesClient, + StorageBoxSnapshot, StorageBoxSnapshotPlan, ) -from hcloud.storage_boxes.domain import StorageBoxAccessSettings, StorageBoxSnapshot from ..conftest import BoundModelTestCase, assert_bound_action1 -def assert_bound_model( +def assert_bound_storage_box( o: BoundStorageBox, resource_client: StorageBoxesClient, ): @@ -31,6 +33,16 @@ def assert_bound_model( assert o.name == "storage-box1" +def assert_bound_storage_box_snapshot( + o: BoundStorageBox, + resource_client: StorageBoxesClient, +): + assert isinstance(o, BoundStorageBoxSnapshot) + assert o._client is resource_client + assert o.id == 34 + assert o.name == "storage-box-snapshot1" + + class TestBoundStorageBox(BoundModelTestCase): methods = [] @@ -46,11 +58,66 @@ def bound_model( ) -> BoundStorageBox: return BoundStorageBox(resource_client, data=storage_box1) - def test_init(self, bound_model, resource_client): + def test_init(self, bound_model: BoundStorageBox, resource_client): + o = bound_model + + assert_bound_storage_box(o, resource_client) + + assert o.storage_box_type.id == 42 + assert o.storage_box_type.name == "bx11" + assert o.location.id == 1 + assert o.location.name == "fsn1" + assert o.system == "FSN1-BX355" + assert o.server == "u1337.your-storagebox.de" + assert o.username == "u12345" + assert o.labels == {"key": "value"} + assert o.protection == {"delete": False} + assert o.snapshot_plan.max_snapshots == 20 + assert o.snapshot_plan.minute == 0 + assert o.snapshot_plan.hour == 7 + assert o.snapshot_plan.day_of_week == 7 + assert o.snapshot_plan.day_of_month is None + assert o.access_settings.reachable_externally is False + assert o.access_settings.samba_enabled is False + assert o.access_settings.ssh_enabled is False + assert o.access_settings.webdav_enabled is False + assert o.access_settings.zfs_enabled is False + assert o.stats.size == 2342236717056 + assert o.stats.size_data == 2102612983808 + assert o.stats.size_snapshots == 239623733248 + assert o.status == "active" + assert o.created == isoparse("2025-01-30T23:55:00Z") + + +class TestBoundStorageBoxSnapshot(BoundModelTestCase): + methods = [] + + @pytest.fixture() + def resource_client(self, client: Client) -> StorageBoxesClient: + return client.storage_boxes + + @pytest.fixture() + def bound_model( + self, + resource_client: StorageBoxesClient, + storage_box_snapshot1, + ) -> BoundStorageBoxSnapshot: + return BoundStorageBoxSnapshot(resource_client, data=storage_box_snapshot1) + + def test_init(self, bound_model: BoundStorageBoxSnapshot, resource_client): o = bound_model - assert_bound_model(o, resource_client) - # TODO: test all properties + assert_bound_storage_box_snapshot(o, resource_client) + + assert isinstance(o.storage_box, BoundStorageBox) + assert o.storage_box.id == 42 + + assert o.description == "" + assert o.is_automatic is False + assert o.labels == {"key": "value"} + assert o.stats.size == 394957594 + assert o.stats.size_filesystem == 3949572745 + assert o.created == isoparse("2025-11-10T19:16:57Z") class TestStorageBoxClient: @@ -73,31 +140,7 @@ def test_get_by_id( url="/storage_boxes/42", ) - assert_bound_model(result, resource_client) - assert result.storage_box_type.id == 42 - assert result.storage_box_type.name == "bx11" - assert result.location.id == 1 - assert result.location.name == "fsn1" - assert result.system == "FSN1-BX355" - assert result.server == "u1337.your-storagebox.de" - assert result.username == "u12345" - assert result.labels == {"key": "value"} - assert result.protection == {"delete": False} - assert result.snapshot_plan.max_snapshots == 20 - assert result.snapshot_plan.minute == 0 - assert result.snapshot_plan.hour == 7 - assert result.snapshot_plan.day_of_week == 7 - assert result.snapshot_plan.day_of_month is None - assert result.access_settings.reachable_externally is False - assert result.access_settings.samba_enabled is False - assert result.access_settings.ssh_enabled is False - assert result.access_settings.webdav_enabled is False - assert result.access_settings.zfs_enabled is False - assert result.stats.size == 2342236717056 - assert result.stats.size_data == 2102612983808 - assert result.stats.size_snapshots == 239623733248 - assert result.status == "active" - assert result.created == isoparse("2025-01-30T23:55:00Z") + assert_bound_storage_box(result, resource_client) @pytest.mark.parametrize( "params", @@ -194,7 +237,7 @@ def test_get_by_name( params=params, ) - assert_bound_model(result, resource_client) + assert_bound_storage_box(result, resource_client) def test_create( self, @@ -240,7 +283,7 @@ def test_create( }, ) - assert_bound_model(result.storage_box, resource_client) + assert_bound_storage_box(result.storage_box, resource_client) def test_update( self, @@ -267,7 +310,7 @@ def test_update( }, ) - assert_bound_model(result, resource_client) + assert_bound_storage_box(result, resource_client) def test_delete( self, @@ -473,3 +516,217 @@ def test_enable_snapshot_plan( ) assert_bound_action1(action, resource_client._parent.actions) + + # Snapshots + ########################################################################### + + def test_get_snapshot_by_id( + self, + request_mock: mock.MagicMock, + resource_client: StorageBoxesClient, + storage_box_snapshot1, + ): + request_mock.return_value = {"snapshot": storage_box_snapshot1} + + result = resource_client.get_snapshot_by_id(StorageBox(42), 34) + + request_mock.assert_called_with( + method="GET", + url="/storage_boxes/42/snapshots/34", + ) + + assert_bound_storage_box_snapshot(result, resource_client) + + @pytest.mark.parametrize( + "params", + [ + {"name": "storage-box-snapshot1"}, + {}, + ], + ) + def test_get_snapshot_list( + self, + request_mock: mock.MagicMock, + resource_client: StorageBoxesClient, + storage_box_snapshot1, + storage_box_snapshot2, + params, + ): + request_mock.return_value = { + "snapshots": [storage_box_snapshot1, storage_box_snapshot2] + } + + result = resource_client.get_snapshot_list(StorageBox(42), **params) + + request_mock.assert_called_with( + url="/storage_boxes/42/snapshots", + method="GET", + params=params, + ) + + assert result.meta is not None + assert len(result.snapshots) == 2 + + result1 = result.snapshots[0] + result2 = result.snapshots[1] + + assert result1._client is resource_client + assert result1.id == 34 + assert result1.name == "storage-box-snapshot1" + assert isinstance(result1.storage_box, BoundStorageBox) + assert result1.storage_box.id == 42 + + assert result2._client is resource_client + assert result2.id == 35 + assert result2.name == "storage-box-snapshot2" + assert isinstance(result2.storage_box, BoundStorageBox) + assert result2.storage_box.id == 42 + + @pytest.mark.parametrize( + "params", + [ + {"name": "storage-box-snapshot1"}, + {}, + ], + ) + def test_get_snapshot_all( + self, + request_mock: mock.MagicMock, + resource_client: StorageBoxesClient, + storage_box_snapshot1, + storage_box_snapshot2, + params, + ): + request_mock.return_value = { + "snapshots": [storage_box_snapshot1, storage_box_snapshot2] + } + + result = resource_client.get_snapshot_all(StorageBox(42), **params) + + request_mock.assert_called_with( + url="/storage_boxes/42/snapshots", + method="GET", + params=params, + ) + + assert len(result) == 2 + + result1 = result[0] + result2 = result[1] + + assert result1._client is resource_client + assert result1.id == 34 + assert result1.name == "storage-box-snapshot1" + assert isinstance(result1.storage_box, BoundStorageBox) + assert result1.storage_box.id == 42 + + assert result2._client is resource_client + assert result2.id == 35 + assert result2.name == "storage-box-snapshot2" + assert isinstance(result2.storage_box, BoundStorageBox) + assert result2.storage_box.id == 42 + + def test_get_snapshot_by_name( + self, + request_mock: mock.MagicMock, + resource_client: StorageBoxesClient, + storage_box_snapshot1, + ): + request_mock.return_value = {"snapshots": [storage_box_snapshot1]} + + result = resource_client.get_snapshot_by_name( + StorageBox(42), "storage-box-snapshot1" + ) + + request_mock.assert_called_with( + method="GET", + url="/storage_boxes/42/snapshots", + params={"name": "storage-box-snapshot1"}, + ) + + assert_bound_storage_box_snapshot(result, resource_client) + + def test_create_snapshot( + self, + request_mock: mock.MagicMock, + resource_client: StorageBoxesClient, + storage_box_snapshot1: dict, + action1_running, + ): + request_mock.return_value = { + "snapshot": { + # Only a partial snapshot is returned + key: storage_box_snapshot1[key] + for key in ["id", "storage_box"] + }, + "action": action1_running, + } + + result = resource_client.create_snapshot( + StorageBox(42), + description="something", + labels={"key": "value"}, + ) + + request_mock.assert_called_with( + method="POST", + url="/storage_boxes/42/snapshots", + json={ + "description": "something", + "labels": {"key": "value"}, + }, + ) + + assert isinstance(result.snapshot, BoundStorageBoxSnapshot) + assert result.snapshot._client is resource_client + assert result.snapshot.id == 34 + + assert_bound_action1(result.action, resource_client._parent.actions) + + def test_update_snapshot( + self, + request_mock: mock.MagicMock, + resource_client: StorageBoxesClient, + storage_box_snapshot1, + ): + request_mock.return_value = { + "snapshot": storage_box_snapshot1, + } + + result = resource_client.update_snapshot( + StorageBoxSnapshot(id=34, storage_box=StorageBox(42)), + description="something", + labels={"key": "value"}, + ) + + request_mock.assert_called_with( + method="PUT", + url="/storage_boxes/42/snapshots/34", + json={ + "description": "something", + "labels": {"key": "value"}, + }, + ) + + assert_bound_storage_box_snapshot(result, resource_client) + + def test_delete_snapshot( + self, + request_mock: mock.MagicMock, + resource_client: StorageBoxesClient, + action1_running, + ): + request_mock.return_value = { + "action": action1_running, + } + + result = resource_client.delete_snapshot( + StorageBoxSnapshot(id=34, storage_box=StorageBox(42)) + ) + + request_mock.assert_called_with( + method="DELETE", + url="/storage_boxes/42/snapshots/34", + ) + + assert_bound_action1(result.action, resource_client._parent.actions)