Add Pagination to volumes and ssh keys

Author: LKaemmerling

hcloud/ssh_keys/client.py

@@ -17,24 +17,41 @@ def delete(self):
 
 
 class SSHKeysClient(ClientEntityBase):
+    results_list_attribute_name = 'ssh_keys'
 
     def get_by_id(self, id):
         # type: (int) -> BoundSSHKey
         response = self._client.request(url="/ssh_keys/{ssh_key_id}".format(ssh_key_id=id), method="GET")
         return BoundSSHKey(self, response['ssh_key'])
 
-    def get_all(self, name=None, fingerprint=None, label_selector=None):
-        # type: (Optional[str], Optional[str], Optional[str]) -> List[BoundSSHKey]
+    def get_list(self,
+                 name=None,  # type: Optional[str]
+                 fingerprint=None,  # type: Optional[str]
+                 label_selector=None,  # type: Optional[str]
+                 page=None,  # type: Optional[int]
+                 per_page=None  # type: Optional[int]
+                 ):
+        # type: (...) -> PageResults[List[BoundSSHKey], Meta]
         params = {}
-        if name is not None:
+        if name:
             params['name'] = name
-        if fingerprint is not None:
+        if fingerprint:
             params['fingerprint'] = fingerprint
-        if label_selector is not None:
+        if label_selector:
             params['label_selector'] = label_selector
+        if page:
+            params['page'] = page
+        if per_page:
+            params['per_page'] = per_page
 
         response = self._client.request(url="/ssh_keys", method="GET", params=params)
-        return [BoundSSHKey(self, location_data) for location_data in response['ssh_keys']]
+
+        ass_ssh_keys = [BoundSSHKey(self, server_data) for server_data in response['ssh_keys']]
+        return self.add_meta_to_result(ass_ssh_keys, response)
+
+    def get_all(self, name=None, fingerprint=None, label_selector=None):
+        # type: (Optional[str], Optional[str], Optional[str]) -> List[BoundSSHKey]
+        return super(SSHKeysClient, self).get_all(name=name, fingerprint=fingerprint, label_selector=label_selector)
 
     def create(self, name, public_key, labels=None):
         # type: (str, str, Optional[Dict[str, str]]) -> BoundSSHKey

hcloud/volumes/client.py

@@ -2,6 +2,7 @@
 from hcloud.core.client import ClientEntityBase, BoundModelBase
 
 from hcloud.actions.client import BoundAction
+from hcloud.core.domain import add_meta_to_result
 from hcloud.volumes.domain import Volume, CreateVolumeResponse
 from hcloud.locations.client import BoundLocation
 
@@ -15,6 +16,10 @@ def __init__(self, client, data, complete=True):
             data['location'] = BoundLocation(client._client.locations, location)
         super(BoundVolume, self).__init__(client, data, complete)
 
+    def get_actions_list(self, sort=None, page=None, per_page=None):
+        # type: (Optional[List[str]], Optional[int], Optional[int]) -> PageResults[List[BoundAction, Meta]]
+        return self._client.get_actions_list(self, sort, page, per_page)
+
     def get_actions(self, sort=None):
         # type: (Optional[List[str]]) -> List[BoundAction]
         return self._client.get_actions(self, sort)
@@ -112,14 +117,23 @@ def create(self,
         )
         return result
 
-    def get_actions(self, volume, sort=None):
-        # type: (Union[Volume, BoundVolume], Optional[List[str]]) -> List[BoundAction]
+    def get_actions_list(self, volume, sort=None, page=None, per_page=None):
+        # type: (Volume, Optional[List[str]], Optional[int], Optional[int]) -> PageResults[List[BoundAction], Meta]
         params = {}
-
         if sort is not None:
-            params.update({"sort": sort})
+            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(url="/volumes/{volume_id}/actions".format(volume_id=volume.id), method="GET", params=params)
-        return [BoundAction(self._client.actions, action_data) for action_data in response['actions']]
+        actions = [BoundAction(self._client.actions, action_data) for action_data in response['actions']]
+        return add_meta_to_result(actions, response, 'actions')
+
+    def get_actions(self, volume, sort=None):
+        # type: (Union[Volume, BoundVolume], Optional[List[str]]) -> List[BoundAction]
+        return super(VolumesClient, self).get_actions(volume, sort=sort)
 
     def update(self, volume, name=None, labels=None):
         # type:(Union[Volume, BoundVolume],  Optional[str],  Optional[Dict[str, str]]) -> BoundVolume

tests/integration/ssh_keys/test_ssh_keys.py

@@ -30,7 +30,7 @@ def test_get_by_id(self, hetzner_client):
         assert ssh_key.fingerprint == "b7:2f:30:a0:2f:6c:58:6c:21:04:58:61:ba:06:3b:2f"
         assert ssh_key.public_key == "ssh-rsa AAAjjk76kgf...Xt"
 
-    def test_get_all(self, hetzner_client):
+    def test_get_list(self, hetzner_client):
         ssh_keys = hetzner_client.ssh_keys.get_all()
         assert ssh_keys[0].id == 2323
         assert ssh_keys[0].name == "My ssh key"

tests/unit/ssh_keys/test_client.py

@@ -52,11 +52,19 @@ def test_get_by_id(self, ssh_keys_client, ssh_key_response):
         assert ssh_key.id == 2323
         assert ssh_key.name == "My ssh key"
 
-    def test_get_all_no_params(self, ssh_keys_client, two_ssh_keys_response):
+    @pytest.mark.parametrize(
+        "params",
+        [
+            {'name': "My ssh key", "fingerprint": "b7:2f:30:a0:2f:6c:58:6c:21:04:58:61:ba:06:3b:2f", "label_selector": "k==v", 'page': 1, 'per_page': 10},
+            {}
+        ]
+    )
+    def test_get_list(self, ssh_keys_client, two_ssh_keys_response, params):
         ssh_keys_client._client.request.return_value = two_ssh_keys_response
-        ssh_keys = ssh_keys_client.get_all()
-        ssh_keys_client._client.request.assert_called_with(url="/ssh_keys", method="GET", params={})
+        result = ssh_keys_client.get_list(**params)
+        ssh_keys_client._client.request.assert_called_with(url="/ssh_keys", method="GET", params=params)
 
+        ssh_keys = result.ssh_keys
         assert len(ssh_keys) == 2
 
         ssh_keys1 = ssh_keys[0]
@@ -70,11 +78,33 @@ def test_get_all_no_params(self, ssh_keys_client, two_ssh_keys_response):
         assert ssh_keys2.id == 2324
         assert ssh_keys2.name == "SSH-Key"
 
-    def test_get_all_with_params(self, ssh_keys_client):
-        params = {'name': "My ssh key", "fingerprint": "b7:2f:30:a0:2f:6c:58:6c:21:04:58:61:ba:06:3b:2f", "label_selector": "k==v"}
-        ssh_keys_client.get_all(**params)
+    @pytest.mark.parametrize(
+        "params",
+        [
+            {'name': "My ssh key", 'label_selector': "label1"},
+            {}
+        ]
+    )
+    def test_get_all(self, ssh_keys_client, two_ssh_keys_response, params):
+        ssh_keys_client._client.request.return_value = two_ssh_keys_response
+        ssh_keys = ssh_keys_client.get_all(**params)
+
+        params.update({'page': 1, 'per_page': 50})
         ssh_keys_client._client.request.assert_called_with(url="/ssh_keys", method="GET", params=params)
 
+        assert len(ssh_keys) == 2
+
+        ssh_keys1 = ssh_keys[0]
+        ssh_keys2 = ssh_keys[1]
+
+        assert ssh_keys1._client is ssh_keys_client
+        assert ssh_keys1.id == 2323
+        assert ssh_keys1.name == "SSH-Key"
+
+        assert ssh_keys2._client is ssh_keys_client
+        assert ssh_keys2.id == 2324
+        assert ssh_keys2.name == "SSH-Key"
+
     def test_create(self, ssh_keys_client, ssh_key_response):
         ssh_keys_client._client.request.return_value = ssh_key_response
         ssh_key = ssh_keys_client.create(name="My ssh key", public_key="ssh-rsa AAAjjk76kgf...Xt")

tests/unit/volumes/test_client.py

@@ -45,7 +45,7 @@ def test_bound_volume_init(self, volume_response):
     def test_get_actions(self, hetzner_client, bound_volume, response_get_actions):
         hetzner_client.request.return_value = response_get_actions
         actions = bound_volume.get_actions(sort="id")
-        hetzner_client.request.assert_called_with(url="/volumes/14/actions", method="GET", params={"sort": "id"})
+        hetzner_client.request.assert_called_with(url="/volumes/14/actions", method="GET", params={"page": 1, "per_page": 50, "sort": "id"})
 
         assert len(actions) == 1
         assert isinstance(actions[0], BoundAction)
@@ -137,7 +137,13 @@ def test_get_by_id(self, volumes_client, volume_response):
         assert bound_volume.id == 1
         assert bound_volume.name == "database-storage"
 
-    @pytest.mark.parametrize("params", [{'label_selector': "label1", 'page': 1, 'per_page': 10}])
+    @pytest.mark.parametrize(
+        "params",
+        [
+            {'label_selector': "label1", 'page': 1, 'per_page': 10},
+            {}
+        ]
+    )
     def test_get_list(self, volumes_client, two_volumes_response, params):
         volumes_client._client.request.return_value = two_volumes_response
         result = volumes_client.get_list(**params)
@@ -258,11 +264,12 @@ def test_create_wrong_location_server_combination(self, volumes_client, location
         volumes_client._client.request.assert_not_called()
 
     @pytest.mark.parametrize("volume", [Volume(id=1), BoundVolume(mock.MagicMock(), dict(id=1))])
-    def test_get_actions(self, volumes_client, volume, response_get_actions):
+    def test_get_actions_list(self, volumes_client, volume, response_get_actions):
         volumes_client._client.request.return_value = response_get_actions
-        actions = volumes_client.get_actions(volume)
-        volumes_client._client.request.assert_called_with(url="/volumes/1/actions", method="GET", params={})
+        result = volumes_client.get_actions_list(volume, sort="id")
+        volumes_client._client.request.assert_called_with(url="/volumes/1/actions", method="GET", params={"sort": "id"})
 
+        actions = result.actions
         assert len(actions) == 1
         assert isinstance(actions[0], BoundAction)