Resolve "Implement client for SSHKey endpoint"

Author: LKaemmerling

hcloud/hcloud.py

@@ -7,6 +7,7 @@
 from hcloud.isos.client import IsosClient
 from hcloud.servers.client import ServersClient
 from hcloud.server_types.client import ServerTypesClient
+from hcloud.ssh_keys.client import SSHKeysClient
 from hcloud.volumes.client import VolumesClient
 from hcloud.images.client import ImagesClient
 from hcloud.locations.client import LocationsClient
@@ -37,6 +38,7 @@ def __init__(self, token):
         self.actions = ActionsClient(self)
         self.images = ImagesClient(self)
         self.isos = IsosClient(self)
+        self.ssh_keys = SSHKeysClient(self)
 
     def _get_user_agent(self):
         return "hcloud-python/" + self.version

hcloud/servers/client.py

@@ -146,7 +146,7 @@ def create(self,
                name,                      # type: str
                server_type,               # type: str
                image,                     # type: Image
-               ssh_keys=None,             # type: Optional[List[str]]
+               ssh_keys=None,             # type: Optional[List[SSHKey]]
                volumes=None,              # type: Optional[List[Volume]]
                user_data=None,            # type: Optional[str]
                labels=None,               # type: Optional[Dict[str, str]]
@@ -184,7 +184,7 @@ def create(self,
             data['datacenter'] = datacenter.id_or_name
 
         if ssh_keys is not None:
-            data['ssh_keys'] = ssh_keys
+            data['ssh_keys'] = [str(ssh_key.id_or_name) for ssh_key in ssh_keys]
         if volumes is not None:
             data['volumes'] = [str(volume.id) for volume in volumes]
         if user_data is not None:

hcloud/ssh_keys/__init__.py

@@ -0,0 +1 @@
+# -*- coding: utf-8 -*-

hcloud/ssh_keys/client.py

@@ -0,0 +1,64 @@
+# -*- coding: utf-8 -*-
+from hcloud.core.client import ClientEntityBase, BoundModelBase
+
+from hcloud.ssh_keys.domain import SSHKey
+
+
+class BoundSSHKey(BoundModelBase):
+    model = SSHKey
+
+    def update(self, name=None, labels=None):
+        # type: (Optional[str], Optional[Dict[str, str]]) -> BoundSSHKey
+        return self._client.update(self, name, labels)
+
+    def delete(self):
+        # type: () -> bool
+        return self._client.delete(self)
+
+
+class SSHKeysClient(ClientEntityBase):
+
+    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]
+        params = {}
+        if name is not None:
+            params['name'] = name
+        if fingerprint is not None:
+            params['fingerprint'] = fingerprint
+        if label_selector is not None:
+            params['label_selector'] = label_selector
+
+        response = self._client.request(url="/ssh_keys", method="GET", params=params)
+        return [BoundSSHKey(self, location_data) for location_data in response['ssh_keys']]
+
+    def create(self, name, public_key, labels=None):
+        # type: (str, str, Optional[Dict[str, str]]) -> BoundSSHKey
+        data = {
+            'name': name,
+            'public_key': public_key
+        }
+        if labels is not None:
+            data['labels'] = labels
+        response = self._client.request(url="/ssh_keys", method="POST", json=data)
+        return BoundSSHKey(self, response['ssh_key'])
+
+    def update(self, ssh_key, name=None, labels=None):
+        # type: (SSHKey,  Optional[str],  Optional[Dict[str, str]]) -> BoundSSHKey
+        data = {}
+        if name is not None:
+            data['name'] = name
+        if labels is not None:
+            data['labels'] = labels
+        response = self._client.request(url="/ssh_keys/{ssh_key_id}".format(ssh_key_id=ssh_key.id), method="PUT", json=data)
+        return BoundSSHKey(self, response['ssh_key'])
+
+    def delete(self, ssh_key):
+        # type: (SSHKey) -> bool
+        self._client.request(url="/ssh_keys/{ssh_key_id}".format(ssh_key_id=ssh_key.id), method="DELETE")
+        # Return allays true, because the API does not return an action for it. When an error occurs a HcloudAPIException will be raised
+        return True

hcloud/ssh_keys/domain.py

@@ -0,0 +1,27 @@
+# -*- coding: utf-8 -*-
+from hcloud.core.domain import BaseDomain
+
+
+class SSHKey(BaseDomain):
+
+    __slots__ = (
+        "id",
+        "name",
+        "fingerprint",
+        "public_key",
+        "labels"
+    )
+
+    def __init__(
+        self,
+        id=None,
+        name=None,
+        fingerprint=None,
+        public_key=None,
+        labels=None
+    ):
+        self.id = id
+        self.name = name
+        self.fingerprint = fingerprint
+        self.public_key = public_key
+        self.labels = labels

hcloud/volumes/client.py

@@ -15,14 +15,34 @@ def __init__(self, client, data, complete=True):
             data['location'] = BoundLocation(client._client.locations, location)
         super(BoundVolume, self).__init__(client, data, complete)
 
-    def attach(self, server):
-        # type: (Union[Server, BoundServer]) -> Action
-        return self._client.attach(server, self)
+    def get_actions(self, sort=None):
+        # type: (Optional[List[str]]) -> List[BoundAction]
+        return self._client.get_actions(self, sort)
+
+    def update(self, name=None, labels=None):
+        # type: (Optional[str], Optional[Dict[str, str]]) -> BoundAction
+        return self._client.update(self, name, labels)
+
+    def delete(self):
+        # type: () -> BoundAction
+        return self._client.delete(self)
+
+    def attach(self, server, automount=None):
+        # type: (Union[Server, BoundServer]) -> BoundAction
+        return self._client.attach(self, server, automount)
 
     def detach(self):
         # type: () -> BoundAction
         return self._client.detach(self)
 
+    def resize(self, size):
+        # type: (int) -> BoundAction
+        return self._client.resize(self, size)
+
+    def change_protection(self, delete=None):
+        # type: (Optional[bool]) -> BoundAction
+        return self._client.change_protection(self, delete)
+
 
 class VolumesClient(ClientEntityBase):
 
@@ -82,12 +102,55 @@ def create(self,
         )
         return result
 
-    def attach(self, server, volume):
-        # type: (Union[Server, BoundServer], Union[Volume, BoundVolume]) -> Action
-        data = self._client.request(url="/volumes/{volume_id}/actions/attach".format(volume_id=volume.id), json={'server': server.id}, method="POST")
+    def get_actions(self, volume, sort=None):
+        # type: (Union[Volume, BoundVolume], Optional[List[str]]) -> List[BoundAction]
+        params = {}
+
+        if sort is not None:
+            params.update({"sort": sort})
+        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']]
+
+    def update(self, volume, name=None, labels=None):
+        # type:(Union[Volume, BoundVolume],  Optional[str],  Optional[Dict[str, str]]) -> BoundVolume
+        data = {}
+        if name is not None:
+            data.update({"name": name})
+        if labels is not None:
+            data.update({"labels": labels})
+        response = self._client.request(url="/volumes/{volume_id}".format(volume_id=volume.id), method="PUT", json=data)
+        return BoundVolume(self, response['volume'])
+
+    def delete(self, volume):
+        # type: (Union[Volume, BoundVolume]) -> BoundAction
+        self._client.request(url="/volumes/{volume_id}".format(volume_id=volume.id), method="DELETE")
+        return True
+
+    def resize(self, volume, size):
+        # type: (Union[Volume, BoundVolume], int) -> BoundAction
+        data = self._client.request(url="/volumes/{volume_id}/actions/resize".format(volume_id=volume.id), json={'size': size}, method="POST")
+        return BoundAction(self._client.actions, data['action'])
+
+    def attach(self, volume, server, automount=None):
+        # type: (Union[Volume, BoundVolume], Union[Server, BoundServer], Optional[bool]) -> BoundAction
+        data = {'server': server.id}
+        if automount is not None:
+            data["automount"] = automount
+
+        data = self._client.request(url="/volumes/{volume_id}/actions/attach".format(volume_id=volume.id), json=data, method="POST")
         return BoundAction(self._client.actions, data['action'])
 
     def detach(self, volume):
-        # type: (Union[Volume, BoundVolume]) -> Action
+        # type: (Union[Volume, BoundVolume]) -> BoundAction
         data = self._client.request(url="/volumes/{volume_id}/actions/detach".format(volume_id=volume.id), method="POST")
         return BoundAction(self._client.actions, data['action'])
+
+    def change_protection(self, volume, delete=None):
+        # type: (Union[Volume, BoundVolume], Optional[bool], Optional[bool]) -> BoundAction
+        data = {}
+        if delete is not None:
+            data.update({"delete": delete})
+
+        response = self._client.request(url="/volumes/{volume_id}/actions/change_protection".format(volume_id=volume.id),
+                                        method="POST", json=data)
+        return BoundAction(self._client.actions, response['action'])

tests/integration/servers/test_servers.py

@@ -3,6 +3,7 @@
 
 from hcloud.servers.client import BoundServer
 from hcloud.servers.domain import Server
+from hcloud.ssh_keys.domain import SSHKey
 from hcloud.volumes.domain import Volume
 from hcloud.images.domain import Image
 from hcloud.isos.domain import Iso
@@ -163,7 +164,7 @@ def test_create(self, hetzner_client):
             "my-server",
             "cx11",
             image=Image(name="ubuntu-16.04"),
-            ssh_keys=["my-ssh-key"],
+            ssh_keys=[SSHKey(name="my-ssh-key")],
             volumes=[Volume(id=1)],
             user_data="#cloud-config\\nruncmd:\\n- [touch, /root/cloud-init-worked]\\n",
             location=Location(name="nbg1"),

tests/integration/ssh_keys/__init__.py

@@ -0,0 +1,57 @@
+import mock
+import pytest
+
+from hcloud.ssh_keys.client import BoundSSHKey
+from hcloud.ssh_keys.domain import SSHKey
+
+
+class TestBoundSSHKey(object):
+    @pytest.fixture()
+    def bound_ssh_key(self, hetzner_client):
+        return BoundSSHKey(client=hetzner_client.ssh_keys, data=dict(id=42))
+
+    def test_update(self, bound_ssh_key):
+        ssh_key = bound_ssh_key.update(name="New name")
+
+        assert ssh_key.id == 2323
+        assert ssh_key.name == "New name"
+
+    def test_delete(self, bound_ssh_key):
+        delete_success = bound_ssh_key.delete()
+
+        assert delete_success is True
+
+
+class TestSSHKeysClient(object):
+    def test_get_by_id(self, hetzner_client):
+        ssh_key = hetzner_client.ssh_keys.get_by_id(1)
+        assert ssh_key.id == 2323
+        assert ssh_key.name == "My ssh key"
+        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):
+        ssh_keys = hetzner_client.ssh_keys.get_all()
+        assert ssh_keys[0].id == 2323
+        assert ssh_keys[0].name == "My ssh key"
+        assert ssh_keys[0].fingerprint == "b7:2f:30:a0:2f:6c:58:6c:21:04:58:61:ba:06:3b:2f"
+        assert ssh_keys[0].public_key == "ssh-rsa AAAjjk76kgf...Xt"
+
+    def test_create(self, hetzner_client):
+        ssh_key = hetzner_client.ssh_keys.create(name="My ssh key", public_key="ssh-rsa AAAjjk76kgf...Xt")
+
+        assert ssh_key.id == 2323
+        assert ssh_key.name == "My ssh key"
+
+    @pytest.mark.parametrize("ssh_key", [SSHKey(id=1), BoundSSHKey(mock.MagicMock(), dict(id=1))])
+    def test_update(self, hetzner_client, ssh_key):
+        ssh_key = hetzner_client.ssh_keys.update(ssh_key, name="New name")
+
+        assert ssh_key.id == 2323
+        assert ssh_key.name == "New name"
+
+    @pytest.mark.parametrize("ssh_key", [SSHKey(id=1), BoundSSHKey(mock.MagicMock(), dict(id=1))])
+    def test_delete(self, hetzner_client, ssh_key):
+        delete_success = hetzner_client.ssh_keys.delete(ssh_key)
+
+        assert delete_success is True

tests/integration/ssh_keys/test_ssh_keys.py

@@ -14,6 +14,22 @@ class TestBoundVolume(object):
     def bound_volume(self, hetzner_client):
         return BoundVolume(client=hetzner_client.volumes, data=dict(id=4711))
 
+    def test_get_actions(self, bound_volume):
+        actions = bound_volume.get_actions()
+
+        assert len(actions) == 1
+        assert actions[0].id == 13
+        assert actions[0].command == "attach_volume"
+
+    def test_update(self, bound_volume):
+        volume = bound_volume.update(name="new-name", labels={})
+        assert volume.id == 4711
+        assert volume.name == "new-name"
+
+    def test_delete(self, bound_volume):
+        delete_success = bound_volume.delete()
+        assert delete_success is True
+
     @pytest.mark.parametrize("server",
                              (Server(id=1), BoundServer(mock.MagicMock(), dict(id=1))))
     def test_attach(self, hetzner_client, bound_volume, server):
@@ -28,6 +44,12 @@ def test_detach(self, hetzner_client, bound_volume):
         assert action.progress == 0
         assert action.command == "detach_volume"
 
+    def test_resize(self, hetzner_client, bound_volume):
+        action = bound_volume.resize(50)
+        assert action.id == 13
+        assert action.progress == 0
+        assert action.command == "resize_volume"
+
 
 class TestVolumesClient(object):
 
@@ -68,11 +90,32 @@ def test_create(self, hetzner_client, server):
         assert next_actions[0].id == 13
         assert next_actions[0].command == "start_server"
 
+    @pytest.mark.parametrize("volume", [Volume(id=1), BoundVolume(mock.MagicMock(), dict(id=1))])
+    def test_get_actions(self, hetzner_client, volume):
+        actions = hetzner_client.volumes.get_actions(volume)
+
+        assert len(actions) == 1
+        assert actions[0].id == 13
+        assert actions[0].command == "attach_volume"
+
+    @pytest.mark.parametrize("volume", [Volume(id=1), BoundVolume(mock.MagicMock(), dict(id=1))])
+    def test_update(self, hetzner_client, volume):
+        volume = hetzner_client.volumes.update(volume, name="new-name", labels={})
+
+        assert volume.id == 4711
+        assert volume.name == "new-name"
+
+    @pytest.mark.parametrize("volume", [Volume(id=1), BoundVolume(mock.MagicMock(), dict(id=1))])
+    def test_delete(self, hetzner_client, volume):
+        delete_success = hetzner_client.volumes.delete(volume)
+
+        assert delete_success is True
+
     @pytest.mark.parametrize("server,volume",
                              [(Server(id=43), Volume(id=4711)),
                               (BoundServer(mock.MagicMock(), dict(id=43)), BoundVolume(mock.MagicMock(), dict(id=4711)))])
     def test_attach(self, hetzner_client, server, volume):
-        action = hetzner_client.volumes.attach(server, volume)
+        action = hetzner_client.volumes.attach(volume, server)
         assert action.id == 13
         assert action.progress == 0
         assert action.command == "attach_volume"
@@ -83,3 +126,10 @@ def test_detach(self, hetzner_client, volume):
         assert action.id == 13
         assert action.progress == 0
         assert action.command == "detach_volume"
+
+    @pytest.mark.parametrize("volume", [Volume(id=4711), BoundVolume(mock.MagicMock(), dict(id=4711))])
+    def test_resize(self, hetzner_client, volume):
+        action = hetzner_client.volumes.resize(volume, 50)
+        assert action.id == 13
+        assert action.progress == 0
+        assert action.command == "resize_volume"