Resolve "Implement client for FloatingIP endpoint"

Author: LKaemmerling

hcloud/floating_ips/__init__.py

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

hcloud/floating_ips/client.py

@@ -0,0 +1,183 @@
+# -*- coding: utf-8 -*-
+from hcloud.actions.client import BoundAction
+from hcloud.core.client import BoundModelBase, ClientEntityBase
+from hcloud.core.domain import add_meta_to_result
+
+from hcloud.floating_ips.domain import FloatingIP, CreateFloatingIPResponse
+from hcloud.locations.client import BoundLocation
+
+
+class BoundFloatingIP(BoundModelBase):
+    model = FloatingIP
+
+    def __init__(self, client, data):
+        from hcloud.servers.client import BoundServer
+        server = data.get("server")
+        if server is not None:
+            data['server'] = BoundServer(client._client.servers, {"id": server}, complete=False)
+
+        home_location = data.get("home_location")
+        if home_location is not None:
+            data['home_location'] = BoundLocation(client._client.locations, home_location)
+
+        super(BoundFloatingIP, self).__init__(client, data)
+
+    def get_actions_list(self, sort=None, page=None, per_page=None):
+        # type: (Optional[List[str]], Optional[int], Optional[int]) -> PageResult[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)
+
+    def update(self, description=None, labels=None):
+        # type: (Optional[str], Optional[Dict[str, str]]) -> BoundFloatingIP
+        return self._client.update(self, description, labels)
+
+    def delete(self):
+        # type: () -> bool
+        return self._client.delete(self)
+
+    def change_protection(self, delete=None):
+        # type: (Optional[bool]) -> BoundAction
+        return self._client.change_protection(self, delete)
+
+    def assign(self, server):
+        # type: (Server) -> BoundAction
+        return self._client.assign(self, server)
+
+    def unassign(self):
+        # type: () -> BoundAction
+        return self._client.unassign(self)
+
+    def change_dns_ptr(self, ip, dns_ptr):
+        # type: (str, str) -> BoundAction
+        return self._client.change_dns_ptr(self, ip, dns_ptr)
+
+
+class FloatingIPsClient(ClientEntityBase):
+    results_list_attribute_name = 'floating_ips'
+
+    def get_actions_list(self,
+                         floating_ip,         # type: FloatingIP
+                         sort=None,     # type: Optional[List[str]]
+                         page=None,     # type: Optional[int]
+                         per_page=None  # type: Optional[int]
+                         ):
+        # type: (...) -> PageResults[List[BoundAction], Meta]
+        params = {}
+        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(url="/floating_ips/{floating_ip_id}/actions".format(floating_ip_id=floating_ip.id), method="GET", params=params)
+        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,
+                    floating_ip,   # type: FloatingIP
+                    sort=None,     # type: Optional[List[str]]
+                    ):
+        # type: (...) -> List[BoundAction]
+        return super(FloatingIPsClient, self).get_actions(floating_ip, sort=sort)
+
+    def get_by_id(self, id):
+        # type: (int) -> BoundFloatingIP
+        response = self._client.request(url="/floating_ips/{floating_ip_id}".format(floating_ip_id=id), method="GET")
+        return BoundFloatingIP(self, response['floating_ip'])
+
+    def get_list(self,
+                 label_selector=None,  # type: Optional[str]
+                 page=None,            # type: Optional[int]
+                 per_page=None         # type: Optional[int]
+                 ):
+        # type: (...) -> PageResults[List[BoundFloatingIP]]
+        params = {}
+
+        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="/floating_ips", method="GET", params=params)
+        floating_ips = [BoundFloatingIP(self, floating_ip_data) for floating_ip_data in response['floating_ips']]
+
+        return self.add_meta_to_result(floating_ips, response)
+
+    def get_all(self, label_selector=None):
+        # type: (Optional[str]) -> List[BoundFloatingIP]
+        return super(FloatingIPsClient, self).get_all(label_selector=label_selector)
+
+    def create(self,
+               type,                # type: str
+               description=None,    # type: Optional[str]
+               labels=None,         # type: Optional[str]
+               home_location=None,  # type: Optional[Location]
+               server=None,         # type: Optional[Server]
+               ):
+        # type: (...) -> CreateFloatingIPResponse
+
+        data = {
+            'type': type
+        }
+        if description is not None:
+            data['description'] = description
+        if labels is not None:
+            data['labels'] = labels
+        if home_location is not None:
+            data['home_location'] = home_location.id_or_name
+        if server is not None:
+            data['server'] = server.id
+
+        response = self._client.request(url="/floating_ips", json=data, method="POST")
+
+        result = CreateFloatingIPResponse(
+            floating_ip=BoundFloatingIP(self, response['floating_ip']),
+            action=BoundAction(self._client.actions, response['action'])
+        )
+        return result
+
+    def update(self, floating_ip, description=None, labels=None):
+        # type: (FloatingIP,  Optional[str], Optional[Dict[str, str]]) -> BoundFloatingIP
+        data = {}
+        if description is not None:
+            data['description'] = description
+        if labels is not None:
+            data['labels'] = labels
+
+        response = self._client.request(url="/floating_ips/{floating_ip_id}".format(floating_ip_id=floating_ip.id), method="PUT", json=data)
+        return BoundFloatingIP(self, response['floating_ip'])
+
+    def delete(self, floating_ip):
+        # type: (FloatingIP) -> bool
+        self._client.request(url="/floating_ips/{floating_ip_id}".format(floating_ip_id=floating_ip.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
+
+    def change_protection(self, floating_ip, delete=None):
+        # type: (FloatingIP, Optional[bool], Optional[bool]) -> BoundAction
+        data = {}
+        if delete is not None:
+            data.update({"delete": delete})
+
+        response = self._client.request(url="/floating_ips/{floating_ip_id}/actions/change_protection".format(floating_ip_id=floating_ip.id), method="POST", json=data)
+        return BoundAction(self._client.actions, response['action'])
+
+    def assign(self, floating_ip, server):
+        # type: (FloatingIP, Server) -> BoundAction
+        response = self._client.request(url="/floating_ips/{floating_ip_id}/actions/assign".format(floating_ip_id=floating_ip.id), method="POST", json={"server": server.id})
+        return BoundAction(self._client.actions, response['action'])
+
+    def unassign(self, floating_ip):
+        # type: (FloatingIP) -> BoundAction
+        response = self._client.request(url="/floating_ips/{floating_ip_id}/actions/unassign".format(floating_ip_id=floating_ip.id), method="POST")
+        return BoundAction(self._client.actions, response['action'])
+
+    def change_dns_ptr(self, floating_ip, ip, dns_ptr):
+        # type: (FloatingIP, str, str) -> BoundAction
+        response = self._client.request(url="/floating_ips/{floating_ip_id}/actions/change_dns_ptr".format(floating_ip_id=floating_ip.id), method="POST", json={"ip": ip, "dns_ptr": dns_ptr})
+        return BoundAction(self._client.actions, response['action'])

hcloud/floating_ips/domain.py

@@ -0,0 +1,58 @@
+# -*- coding: utf-8 -*-
+from hcloud.core.domain import BaseDomain
+
+
+class FloatingIP(BaseDomain):
+
+    __slots__ = (
+        "id",
+        "type",
+        "description",
+        "ip",
+        "server",
+        "dns_ptr",
+        "home_location",
+        "blocked",
+        "protection",
+        "labels"
+    )
+
+    def __init__(
+        self,
+        id=None,
+        type=None,
+        description=None,
+        ip=None,
+        server=None,
+        dns_ptr=None,
+        home_location=None,
+        blocked=None,
+        protection=None,
+        labels=None
+
+    ):
+        self.id = id
+        self.type = type
+        self.description = description
+        self.ip = ip
+        self.server = server
+        self.dns_ptr = dns_ptr
+        self.home_location = home_location
+        self.blocked = blocked
+        self.protection = protection
+        self.labels = labels
+
+
+class CreateFloatingIPResponse(BaseDomain):
+    __slots__ = (
+        "floating_ip",
+        "action"
+    )
+
+    def __init__(
+            self,
+            floating_ip,     # type: BoundFloatingIP
+            action,          # type: BoundAction
+    ):
+        self.floating_ip = floating_ip
+        self.action = action

hcloud/hcloud.py

@@ -4,6 +4,7 @@
 import requests
 
 from hcloud.actions.client import ActionsClient
+from hcloud.floating_ips.client import FloatingIPsClient
 from hcloud.isos.client import IsosClient
 from hcloud.servers.client import ServersClient
 from hcloud.server_types.client import ServerTypesClient
@@ -39,6 +40,7 @@ def __init__(self, token):
         self.images = ImagesClient(self)
         self.isos = IsosClient(self)
         self.ssh_keys = SSHKeysClient(self)
+        self.floating_ips = FloatingIPsClient(self)
 
     def _get_user_agent(self):
         return "hcloud-python/" + self.version

tests/integration/floating_ips/__init__.py

@@ -0,0 +1,129 @@
+import pytest
+import mock
+
+from hcloud.servers.client import BoundServer
+from hcloud.servers.domain import Server
+from hcloud.floating_ips.client import BoundFloatingIP
+from hcloud.floating_ips.domain import FloatingIP
+
+
+class TestBoundFloatingIPs(object):
+
+    @pytest.fixture()
+    def bound_floating_ip(self, hetzner_client):
+        return BoundFloatingIP(client=hetzner_client.floating_ips, data=dict(id=4711))
+
+    def test_get_actions(self, bound_floating_ip):
+        actions = bound_floating_ip.get_actions()
+
+        assert len(actions) == 1
+        assert actions[0].id == 13
+        assert actions[0].command == "assign_floating_ip"
+
+    def test_update(self, bound_floating_ip):
+        floating_ip = bound_floating_ip.update(description="New description", labels={})
+        assert floating_ip.id == 4711
+        assert floating_ip.description == "New description"
+
+    def test_delete(self, bound_floating_ip):
+        delete_success = bound_floating_ip.delete()
+        assert delete_success is True
+
+    @pytest.mark.parametrize("server",
+                             (Server(id=1), BoundServer(mock.MagicMock(), dict(id=1))))
+    def test_assign(self, hetzner_client, bound_floating_ip, server):
+        action = bound_floating_ip.assign(server)
+        assert action.id == 13
+        assert action.progress == 0
+        assert action.command == "assign_floating_ip"
+
+    def test_unassign(self, hetzner_client, bound_floating_ip):
+        action = bound_floating_ip.unassign()
+        assert action.id == 13
+        assert action.progress == 0
+        assert action.command == "unassign_floating_ip"
+
+    def test_change_dns_ptr(self, hetzner_client, bound_floating_ip):
+        action = bound_floating_ip.change_dns_ptr("1.2.3.4", "server02.example.com")
+        assert action.id == 13
+        assert action.progress == 0
+        assert action.command == "change_dns_ptr"
+
+
+class TestFloatingIPsClient(object):
+
+    def test_get_by_id(self, hetzner_client):
+        bound_floating_ip = hetzner_client.floating_ips.get_by_id(4711)
+        assert bound_floating_ip.id == 4711
+        assert bound_floating_ip.description == "Web Frontend"
+        assert bound_floating_ip.type == "ipv4"
+
+    def test_get_list(self, hetzner_client):
+        result = hetzner_client.floating_ips.get_list()
+        bound_floating_ips = result.floating_ips
+        assert bound_floating_ips[0].id == 4711
+        assert bound_floating_ips[0].description == "Web Frontend"
+        assert bound_floating_ips[0].type == "ipv4"
+
+    @pytest.mark.parametrize("server", [Server(id=1), BoundServer(mock.MagicMock(), dict(id=1))])
+    def test_create(self, hetzner_client, server):
+        response = hetzner_client.floating_ips.create(
+            type="ipv4",
+            description="Web Frontend",
+            # home_location=Location(description="fsn1"),
+            server=server
+        )
+
+        floating_ip = response.floating_ip
+        action = response.action
+
+        assert floating_ip.id == 4711
+        assert floating_ip.description == "Web Frontend"
+        assert floating_ip.type == "ipv4"
+
+        assert action.id == 13
+        assert action.command == "assign_floating_ip"
+
+    @pytest.mark.parametrize("floating_ip", [FloatingIP(id=1), BoundFloatingIP(mock.MagicMock(), dict(id=1))])
+    def test_get_actions(self, hetzner_client, floating_ip):
+        actions = hetzner_client.floating_ips.get_actions(floating_ip)
+
+        assert len(actions) == 1
+        assert actions[0].id == 13
+        assert actions[0].command == "assign_floating_ip"
+
+    @pytest.mark.parametrize("floating_ip", [FloatingIP(id=1), BoundFloatingIP(mock.MagicMock(), dict(id=1))])
+    def test_update(self, hetzner_client, floating_ip):
+        floating_ip = hetzner_client.floating_ips.update(floating_ip, description="New description", labels={})
+
+        assert floating_ip.id == 4711
+        assert floating_ip.description == "New description"
+
+    @pytest.mark.parametrize("floating_ip", [FloatingIP(id=1), BoundFloatingIP(mock.MagicMock(), dict(id=1))])
+    def test_delete(self, hetzner_client, floating_ip):
+        delete_success = hetzner_client.floating_ips.delete(floating_ip)
+
+        assert delete_success is True
+
+    @pytest.mark.parametrize("server,floating_ip",
+                             [(Server(id=43), FloatingIP(id=4711)),
+                              (BoundServer(mock.MagicMock(), dict(id=43)), BoundFloatingIP(mock.MagicMock(), dict(id=4711)))])
+    def test_assign(self, hetzner_client, server, floating_ip):
+        action = hetzner_client.floating_ips.assign(floating_ip, server)
+        assert action.id == 13
+        assert action.progress == 0
+        assert action.command == "assign_floating_ip"
+
+    @pytest.mark.parametrize("floating_ip", [FloatingIP(id=4711), BoundFloatingIP(mock.MagicMock(), dict(id=4711))])
+    def test_unassign(self, hetzner_client, floating_ip):
+        action = hetzner_client.floating_ips.unassign(floating_ip)
+        assert action.id == 13
+        assert action.progress == 0
+        assert action.command == "unassign_floating_ip"
+
+    @pytest.mark.parametrize("floating_ip", [FloatingIP(id=4711), BoundFloatingIP(mock.MagicMock(), dict(id=4711))])
+    def test_change_dns_ptr(self, hetzner_client, floating_ip):
+        action = hetzner_client.floating_ips.change_dns_ptr(floating_ip, "1.2.3.4", "server02.example.com")
+        assert action.id == 13
+        assert action.progress == 0
+        assert action.command == "change_dns_ptr"