diff --git a/PKG-INFO b/PKG-INFO
index 54c033c0908982655b29123ccf983f210d3b27c3..3926662beb4e6a7785c952432edeab38bf2a935f 100644
--- a/PKG-INFO
+++ b/PKG-INFO
@@ -1,6 +1,6 @@
Metadata-Version: 2.1
Name: hcloud
-Version: 1.24.0
+Version: 1.25.0
Summary: Official Hetzner Cloud python library
Home-page: https://github.com/hetznercloud/hcloud-python
Author: Hetzner Cloud GmbH
diff --git a/hcloud.egg-info/PKG-INFO b/hcloud.egg-info/PKG-INFO
index 54c033c0908982655b29123ccf983f210d3b27c3..3926662beb4e6a7785c952432edeab38bf2a935f 100644
--- a/hcloud.egg-info/PKG-INFO
+++ b/hcloud.egg-info/PKG-INFO
@@ -1,6 +1,6 @@
Metadata-Version: 2.1
Name: hcloud
-Version: 1.24.0
+Version: 1.25.0
Summary: Official Hetzner Cloud python library
Home-page: https://github.com/hetznercloud/hcloud-python
Author: Hetzner Cloud GmbH
diff --git a/hcloud.egg-info/SOURCES.txt b/hcloud.egg-info/SOURCES.txt
index e0597a65102ca376c24824797f03f7a094ad56b7..23e2990ff9a117992babd40181e344cb5ad7b5e6 100644
--- a/hcloud.egg-info/SOURCES.txt
+++ b/hcloud.egg-info/SOURCES.txt
@@ -36,6 +36,7 @@ docs/_static/logo-hetzner-online.svg
docs/_static/js/open_links_in_new_tab.js
hcloud/__init__.py
hcloud/__version__.py
+hcloud/_client.py
hcloud/_exceptions.py
hcloud/hcloud.py
hcloud.egg-info/PKG-INFO
@@ -105,6 +106,7 @@ hcloud/volumes/domain.py
tests/__init__.py
tests/unit/__init__.py
tests/unit/conftest.py
+tests/unit/test_client.py
tests/unit/test_hcloud.py
tests/unit/actions/__init__.py
tests/unit/actions/conftest.py
diff --git a/hcloud/__init__.py b/hcloud/__init__.py
index 592ff64d6d27c1b3f4b4f69492025a6a726fa1f2..5beda91d29f79eb24886eec78c3c0a6afbeadf56 100644
--- a/hcloud/__init__.py
+++ b/hcloud/__init__.py
@@ -1,2 +1,2 @@
+from ._client import Client # noqa
from ._exceptions import APIException, HCloudException # noqa
-from .hcloud import Client # noqa
diff --git a/hcloud/__version__.py b/hcloud/__version__.py
index 0c992210039dd305c397baa838d333543a0475a8..61abd10ca66c40a1c7c24bcbb38717a881d209ee 100644
--- a/hcloud/__version__.py
+++ b/hcloud/__version__.py
@@ -1 +1 @@
-VERSION = "1.24.0" # x-release-please-version
+VERSION = "1.25.0" # x-release-please-version
diff --git a/hcloud/_client.py b/hcloud/_client.py
new file mode 100644
index 0000000000000000000000000000000000000000..e62624f1610bad9a3fd0d015d8403ac7c165e9ce
--- /dev/null
+++ b/hcloud/_client.py
@@ -0,0 +1,222 @@
+import time
+from typing import Optional, Union
+
+import requests
+
+from .__version__ import VERSION
+from ._exceptions import APIException
+from .actions.client import ActionsClient
+from .certificates.client import CertificatesClient
+from .datacenters.client import DatacentersClient
+from .firewalls.client import FirewallsClient
+from .floating_ips.client import FloatingIPsClient
+from .images.client import ImagesClient
+from .isos.client import IsosClient
+from .load_balancer_types.client import LoadBalancerTypesClient
+from .load_balancers.client import LoadBalancersClient
+from .locations.client import LocationsClient
+from .networks.client import NetworksClient
+from .placement_groups.client import PlacementGroupsClient
+from .primary_ips.client import PrimaryIPsClient
+from .server_types.client import ServerTypesClient
+from .servers.client import ServersClient
+from .ssh_keys.client import SSHKeysClient
+from .volumes.client import VolumesClient
+
+
+class Client:
+ """Base Client for accessing the Hetzner Cloud API"""
+
+ _version = VERSION
+ _retry_wait_time = 0.5
+ __user_agent_prefix = "hcloud-python"
+
+ def __init__(
+ self,
+ token: str,
+ api_endpoint: str = "https://api.hetzner.cloud/v1",
+ application_name: Optional[str] = None,
+ application_version: Optional[str] = None,
+ poll_interval: int = 1,
+ ):
+ """Create an new Client instance
+
+ :param token: Hetzner Cloud API token
+ :param api_endpoint: Hetzner Cloud API endpoint
+ :param application_name: Your application name
+ :param application_version: Your application _version
+ :param poll_interval: Interval for polling information from Hetzner Cloud API in seconds
+ """
+ self.token = token
+ self._api_endpoint = api_endpoint
+ self._application_name = application_name
+ self._application_version = application_version
+ self._requests_session = requests.Session()
+ self.poll_interval = poll_interval
+
+ self.datacenters = DatacentersClient(self)
+ """DatacentersClient Instance
+
+ :type: :class:`DatacentersClient <hcloud.datacenters.client.DatacentersClient>`
+ """
+ self.locations = LocationsClient(self)
+ """LocationsClient Instance
+
+ :type: :class:`LocationsClient <hcloud.locations.client.LocationsClient>`
+ """
+ self.servers = ServersClient(self)
+ """ServersClient Instance
+
+ :type: :class:`ServersClient <hcloud.servers.client.ServersClient>`
+ """
+ self.server_types = ServerTypesClient(self)
+ """ServerTypesClient Instance
+
+ :type: :class:`ServerTypesClient <hcloud.server_types.client.ServerTypesClient>`
+ """
+ self.volumes = VolumesClient(self)
+ """VolumesClient Instance
+
+ :type: :class:`VolumesClient <hcloud.volumes.client.VolumesClient>`
+ """
+ self.actions = ActionsClient(self)
+ """ActionsClient Instance
+
+ :type: :class:`ActionsClient <hcloud.actions.client.ActionsClient>`
+ """
+ self.images = ImagesClient(self)
+ """ImagesClient Instance
+
+ :type: :class:`ImagesClient <hcloud.images.client.ImagesClient>`
+ """
+ self.isos = IsosClient(self)
+ """ImagesClient Instance
+
+ :type: :class:`IsosClient <hcloud.isos.client.IsosClient>`
+ """
+ self.ssh_keys = SSHKeysClient(self)
+ """SSHKeysClient Instance
+
+ :type: :class:`SSHKeysClient <hcloud.ssh_keys.client.SSHKeysClient>`
+ """
+ self.floating_ips = FloatingIPsClient(self)
+ """FloatingIPsClient Instance
+
+ :type: :class:`FloatingIPsClient <hcloud.floating_ips.client.FloatingIPsClient>`
+ """
+ self.primary_ips = PrimaryIPsClient(self)
+ """PrimaryIPsClient Instance
+
+ :type: :class:`PrimaryIPsClient <hcloud.primary_ips.client.PrimaryIPsClient>`
+ """
+ self.networks = NetworksClient(self)
+ """NetworksClient Instance
+
+ :type: :class:`NetworksClient <hcloud.networks.client.NetworksClient>`
+ """
+ self.certificates = CertificatesClient(self)
+ """CertificatesClient Instance
+
+ :type: :class:`CertificatesClient <hcloud.certificates.client.CertificatesClient>`
+ """
+
+ self.load_balancers = LoadBalancersClient(self)
+ """LoadBalancersClient Instance
+
+ :type: :class:`LoadBalancersClient <hcloud.load_balancers.client.LoadBalancersClient>`
+ """
+
+ self.load_balancer_types = LoadBalancerTypesClient(self)
+ """LoadBalancerTypesClient Instance
+
+ :type: :class:`LoadBalancerTypesClient <hcloud.load_balancer_types.client.LoadBalancerTypesClient>`
+ """
+
+ self.firewalls = FirewallsClient(self)
+ """FirewallsClient Instance
+
+ :type: :class:`FirewallsClient <hcloud.firewalls.client.FirewallsClient>`
+ """
+
+ self.placement_groups = PlacementGroupsClient(self)
+ """PlacementGroupsClient Instance
+
+ :type: :class:`PlacementGroupsClient <hcloud.placement_groups.client.PlacementGroupsClient>`
+ """
+
+ def _get_user_agent(self) -> str:
+ """Get the user agent of the hcloud-python instance with the user application name (if specified)
+
+ :return: The user agent of this hcloud-python instance
+ """
+ user_agents = []
+ for name, version in [
+ (self._application_name, self._application_version),
+ (self.__user_agent_prefix, self._version),
+ ]:
+ if name is not None:
+ user_agents.append(name if version is None else f"{name}/{version}")
+
+ return " ".join(user_agents)
+
+ def _get_headers(self) -> dict:
+ headers = {
+ "User-Agent": self._get_user_agent(),
+ "Authorization": f"Bearer {self.token}",
+ }
+ return headers
+
+ def _raise_exception_from_response(self, response: requests.Response):
+ raise APIException(
+ code=response.status_code,
+ message=response.reason,
+ details={"content": response.content},
+ )
+
+ def _raise_exception_from_content(self, content: dict):
+ raise APIException(
+ code=content["error"]["code"],
+ message=content["error"]["message"],
+ details=content["error"]["details"],
+ )
+
+ def request(
+ self,
+ method: str,
+ url: str,
+ tries: int = 1,
+ **kwargs,
+ ) -> Union[bytes, dict]:
+ """Perform a request to the Hetzner Cloud API, wrapper around requests.request
+
+ :param method: HTTP Method to perform the Request
+ :param url: URL of the Endpoint
+ :param tries: Tries of the request (used internally, should not be set by the user)
+ :return: Response
+ """
+ response = self._requests_session.request(
+ method=method,
+ url=self._api_endpoint + url,
+ headers=self._get_headers(),
+ **kwargs,
+ )
+
+ content = response.content
+ try:
+ if len(content) > 0:
+ content = response.json()
+ except (TypeError, ValueError):
+ self._raise_exception_from_response(response)
+
+ if not response.ok:
+ if content:
+ if content["error"]["code"] == "rate_limit_exceeded" and tries < 5:
+ time.sleep(tries * self._retry_wait_time)
+ tries = tries + 1
+ return self.request(method, url, tries, **kwargs)
+ else:
+ self._raise_exception_from_content(content)
+ else:
+ self._raise_exception_from_response(response)
+
+ return content
diff --git a/hcloud/_exceptions.py b/hcloud/_exceptions.py
index 36fee5deeda740b1df1dd706e5d87a25731fd253..d0801e951b0fa470e1edd54a9c2082f323862e27 100644
--- a/hcloud/_exceptions.py
+++ b/hcloud/_exceptions.py
@@ -6,9 +6,7 @@ class APIException(HCloudException):
"""There was an error while performing an API Request"""
def __init__(self, code, message, details):
+ super().__init__(message)
self.code = code
self.message = message
self.details = details
-
- def __str__(self):
- return self.message
diff --git a/hcloud/actions/domain.py b/hcloud/actions/domain.py
index 0d91eaca1c1ee1441e75d9bac6c4c8153d2b653d..1cc25d475e4087770b828c3e4af909f53771ecf7 100644
--- a/hcloud/actions/domain.py
+++ b/hcloud/actions/domain.py
@@ -61,12 +61,18 @@ class ActionException(HCloudException):
"""A generic action exception"""
def __init__(self, action):
+ message = self.__doc__
+ if action.error is not None and "message" in action.error:
+ message += f": {action.error['message']}"
+
+ super().__init__(message)
+ self.message = message
self.action = action
class ActionFailedException(ActionException):
- """The Action you were waiting for failed"""
+ """The pending action failed"""
class ActionTimeoutException(ActionException):
- """The Action you were waiting for timed out"""
+ """The pending action timed out"""
diff --git a/hcloud/hcloud.py b/hcloud/hcloud.py
index e62624f1610bad9a3fd0d015d8403ac7c165e9ce..af4e5c2a51004e51fd139a9b0dde764db2004780 100644
--- a/hcloud/hcloud.py
+++ b/hcloud/hcloud.py
@@ -1,222 +1,9 @@
-import time
-from typing import Optional, Union
+import warnings
-import requests
+warnings.warn(
+ "The 'hcloud.hcloud' module is deprecated, please import from the 'hcloud' module instead (e.g. 'from hcloud import Client').",
+ DeprecationWarning,
+ stacklevel=2,
+)
-from .__version__ import VERSION
-from ._exceptions import APIException
-from .actions.client import ActionsClient
-from .certificates.client import CertificatesClient
-from .datacenters.client import DatacentersClient
-from .firewalls.client import FirewallsClient
-from .floating_ips.client import FloatingIPsClient
-from .images.client import ImagesClient
-from .isos.client import IsosClient
-from .load_balancer_types.client import LoadBalancerTypesClient
-from .load_balancers.client import LoadBalancersClient
-from .locations.client import LocationsClient
-from .networks.client import NetworksClient
-from .placement_groups.client import PlacementGroupsClient
-from .primary_ips.client import PrimaryIPsClient
-from .server_types.client import ServerTypesClient
-from .servers.client import ServersClient
-from .ssh_keys.client import SSHKeysClient
-from .volumes.client import VolumesClient
-
-
-class Client:
- """Base Client for accessing the Hetzner Cloud API"""
-
- _version = VERSION
- _retry_wait_time = 0.5
- __user_agent_prefix = "hcloud-python"
-
- def __init__(
- self,
- token: str,
- api_endpoint: str = "https://api.hetzner.cloud/v1",
- application_name: Optional[str] = None,
- application_version: Optional[str] = None,
- poll_interval: int = 1,
- ):
- """Create an new Client instance
-
- :param token: Hetzner Cloud API token
- :param api_endpoint: Hetzner Cloud API endpoint
- :param application_name: Your application name
- :param application_version: Your application _version
- :param poll_interval: Interval for polling information from Hetzner Cloud API in seconds
- """
- self.token = token
- self._api_endpoint = api_endpoint
- self._application_name = application_name
- self._application_version = application_version
- self._requests_session = requests.Session()
- self.poll_interval = poll_interval
-
- self.datacenters = DatacentersClient(self)
- """DatacentersClient Instance
-
- :type: :class:`DatacentersClient <hcloud.datacenters.client.DatacentersClient>`
- """
- self.locations = LocationsClient(self)
- """LocationsClient Instance
-
- :type: :class:`LocationsClient <hcloud.locations.client.LocationsClient>`
- """
- self.servers = ServersClient(self)
- """ServersClient Instance
-
- :type: :class:`ServersClient <hcloud.servers.client.ServersClient>`
- """
- self.server_types = ServerTypesClient(self)
- """ServerTypesClient Instance
-
- :type: :class:`ServerTypesClient <hcloud.server_types.client.ServerTypesClient>`
- """
- self.volumes = VolumesClient(self)
- """VolumesClient Instance
-
- :type: :class:`VolumesClient <hcloud.volumes.client.VolumesClient>`
- """
- self.actions = ActionsClient(self)
- """ActionsClient Instance
-
- :type: :class:`ActionsClient <hcloud.actions.client.ActionsClient>`
- """
- self.images = ImagesClient(self)
- """ImagesClient Instance
-
- :type: :class:`ImagesClient <hcloud.images.client.ImagesClient>`
- """
- self.isos = IsosClient(self)
- """ImagesClient Instance
-
- :type: :class:`IsosClient <hcloud.isos.client.IsosClient>`
- """
- self.ssh_keys = SSHKeysClient(self)
- """SSHKeysClient Instance
-
- :type: :class:`SSHKeysClient <hcloud.ssh_keys.client.SSHKeysClient>`
- """
- self.floating_ips = FloatingIPsClient(self)
- """FloatingIPsClient Instance
-
- :type: :class:`FloatingIPsClient <hcloud.floating_ips.client.FloatingIPsClient>`
- """
- self.primary_ips = PrimaryIPsClient(self)
- """PrimaryIPsClient Instance
-
- :type: :class:`PrimaryIPsClient <hcloud.primary_ips.client.PrimaryIPsClient>`
- """
- self.networks = NetworksClient(self)
- """NetworksClient Instance
-
- :type: :class:`NetworksClient <hcloud.networks.client.NetworksClient>`
- """
- self.certificates = CertificatesClient(self)
- """CertificatesClient Instance
-
- :type: :class:`CertificatesClient <hcloud.certificates.client.CertificatesClient>`
- """
-
- self.load_balancers = LoadBalancersClient(self)
- """LoadBalancersClient Instance
-
- :type: :class:`LoadBalancersClient <hcloud.load_balancers.client.LoadBalancersClient>`
- """
-
- self.load_balancer_types = LoadBalancerTypesClient(self)
- """LoadBalancerTypesClient Instance
-
- :type: :class:`LoadBalancerTypesClient <hcloud.load_balancer_types.client.LoadBalancerTypesClient>`
- """
-
- self.firewalls = FirewallsClient(self)
- """FirewallsClient Instance
-
- :type: :class:`FirewallsClient <hcloud.firewalls.client.FirewallsClient>`
- """
-
- self.placement_groups = PlacementGroupsClient(self)
- """PlacementGroupsClient Instance
-
- :type: :class:`PlacementGroupsClient <hcloud.placement_groups.client.PlacementGroupsClient>`
- """
-
- def _get_user_agent(self) -> str:
- """Get the user agent of the hcloud-python instance with the user application name (if specified)
-
- :return: The user agent of this hcloud-python instance
- """
- user_agents = []
- for name, version in [
- (self._application_name, self._application_version),
- (self.__user_agent_prefix, self._version),
- ]:
- if name is not None:
- user_agents.append(name if version is None else f"{name}/{version}")
-
- return " ".join(user_agents)
-
- def _get_headers(self) -> dict:
- headers = {
- "User-Agent": self._get_user_agent(),
- "Authorization": f"Bearer {self.token}",
- }
- return headers
-
- def _raise_exception_from_response(self, response: requests.Response):
- raise APIException(
- code=response.status_code,
- message=response.reason,
- details={"content": response.content},
- )
-
- def _raise_exception_from_content(self, content: dict):
- raise APIException(
- code=content["error"]["code"],
- message=content["error"]["message"],
- details=content["error"]["details"],
- )
-
- def request(
- self,
- method: str,
- url: str,
- tries: int = 1,
- **kwargs,
- ) -> Union[bytes, dict]:
- """Perform a request to the Hetzner Cloud API, wrapper around requests.request
-
- :param method: HTTP Method to perform the Request
- :param url: URL of the Endpoint
- :param tries: Tries of the request (used internally, should not be set by the user)
- :return: Response
- """
- response = self._requests_session.request(
- method=method,
- url=self._api_endpoint + url,
- headers=self._get_headers(),
- **kwargs,
- )
-
- content = response.content
- try:
- if len(content) > 0:
- content = response.json()
- except (TypeError, ValueError):
- self._raise_exception_from_response(response)
-
- if not response.ok:
- if content:
- if content["error"]["code"] == "rate_limit_exceeded" and tries < 5:
- time.sleep(tries * self._retry_wait_time)
- tries = tries + 1
- return self.request(method, url, tries, **kwargs)
- else:
- self._raise_exception_from_content(content)
- else:
- self._raise_exception_from_response(response)
-
- return content
+from ._client import * # noqa
diff --git a/tests/unit/actions/test_domain.py b/tests/unit/actions/test_domain.py
index 02959da7eacaf80fdd6f2bed773f67a16f700008..54b730c79800caaa66963eb1484d209beb0229f3 100644
--- a/tests/unit/actions/test_domain.py
+++ b/tests/unit/actions/test_domain.py
@@ -1,7 +1,14 @@
import datetime
from datetime import timezone
-from hcloud.actions.domain import Action
+import pytest
+
+from hcloud.actions.domain import (
+ Action,
+ ActionException,
+ ActionFailedException,
+ ActionTimeoutException,
+)
class TestAction:
@@ -15,3 +22,43 @@ class TestAction:
assert action.finished == datetime.datetime(
2016, 3, 30, 23, 50, tzinfo=timezone.utc
)
+
+
+def test_action_exceptions():
+ with pytest.raises(
+ ActionException,
+ match=r"The pending action failed: Server does not exist anymore",
+ ):
+ raise ActionFailedException(
+ action=Action(
+ **{
+ "id": 1084730887,
+ "command": "change_server_type",
+ "status": "error",
+ "progress": 100,
+ "resources": [{"id": 34574042, "type": "server"}],
+ "error": {
+ "code": "server_does_not_exist_anymore",
+ "message": "Server does not exist anymore",
+ },
+ "started": "2023-07-06T14:52:42+00:00",
+ "finished": "2023-07-06T14:53:08+00:00",
+ }
+ )
+ )
+
+ with pytest.raises(ActionException, match=r"The pending action timed out"):
+ raise ActionTimeoutException(
+ action=Action(
+ **{
+ "id": 1084659545,
+ "command": "create_server",
+ "status": "running",
+ "progress": 50,
+ "started": "2023-07-06T13:58:38+00:00",
+ "finished": None,
+ "resources": [{"id": 34572291, "type": "server"}],
+ "error": None,
+ }
+ )
+ )
diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py
index d15d36b5223e49d1b4bbf59038988f365f954a9a..0ab7490608c026420fe58ac904914db2fbeeab2c 100644
--- a/tests/unit/conftest.py
+++ b/tests/unit/conftest.py
@@ -7,7 +7,7 @@ from hcloud import Client
@pytest.fixture(autouse=True, scope="function")
def mocked_requests():
- patcher = mock.patch("hcloud.hcloud.requests")
+ patcher = mock.patch("hcloud._client.requests")
mocked_requests = patcher.start()
yield mocked_requests
patcher.stop()
diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py
new file mode 100644
index 0000000000000000000000000000000000000000..5f5bf8114b57be52a2bb33f142a0a7bc40adbd68
--- /dev/null
+++ b/tests/unit/test_client.py
@@ -0,0 +1,182 @@
+import json
+from unittest.mock import MagicMock
+
+import pytest
+import requests
+
+from hcloud import APIException, Client
+
+
+class TestHetznerClient:
+ @pytest.fixture()
+ def client(self):
+ Client._version = "0.0.0"
+ client = Client(token="project_token")
+
+ client._requests_session = MagicMock()
+ return client
+
+ @pytest.fixture()
+ def response(self):
+ response = requests.Response()
+ response.status_code = 200
+ response._content = json.dumps({"result": "data"}).encode("utf-8")
+ return response
+
+ @pytest.fixture()
+ def fail_response(self, response):
+ response.status_code = 422
+ error = {
+ "code": "invalid_input",
+ "message": "invalid input in field 'broken_field': is too long",
+ "details": {
+ "fields": [{"name": "broken_field", "messages": ["is too long"]}]
+ },
+ }
+ response._content = json.dumps({"error": error}).encode("utf-8")
+ return response
+
+ @pytest.fixture()
+ def rate_limit_response(self, response):
+ response.status_code = 422
+ error = {
+ "code": "rate_limit_exceeded",
+ "message": "limit of 10 requests per hour reached",
+ "details": {},
+ }
+ response._content = json.dumps({"error": error}).encode("utf-8")
+ return response
+
+ def test__get_user_agent(self, client):
+ user_agent = client._get_user_agent()
+ assert user_agent == "hcloud-python/0.0.0"
+
+ def test__get_user_agent_with_application_name(self, client):
+ client = Client(token="project_token", application_name="my-app")
+ user_agent = client._get_user_agent()
+ assert user_agent == "my-app hcloud-python/0.0.0"
+
+ def test__get_user_agent_with_application_name_and_version(self, client):
+ client = Client(
+ token="project_token",
+ application_name="my-app",
+ application_version="1.0.0",
+ )
+ user_agent = client._get_user_agent()
+ assert user_agent == "my-app/1.0.0 hcloud-python/0.0.0"
+
+ def test__get_headers(self, client):
+ headers = client._get_headers()
+ assert headers == {
+ "User-Agent": "hcloud-python/0.0.0",
+ "Authorization": "Bearer project_token",
+ }
+
+ def test_request_library_mocked(self, client):
+ response = client.request("POST", "url", params={"1": 2})
+ assert response.__class__.__name__ == "MagicMock"
+
+ def test_request_ok(self, client, response):
+ client._requests_session.request.return_value = response
+ response = client.request(
+ "POST", "/servers", params={"argument": "value"}, timeout=2
+ )
+ client._requests_session.request.assert_called_once_with(
+ method="POST",
+ url="https://api.hetzner.cloud/v1/servers",
+ headers={
+ "User-Agent": "hcloud-python/0.0.0",
+ "Authorization": "Bearer project_token",
+ },
+ params={"argument": "value"},
+ timeout=2,
+ )
+ assert response == {"result": "data"}
+
+ def test_request_fails(self, client, fail_response):
+ client._requests_session.request.return_value = fail_response
+ with pytest.raises(APIException) as exception_info:
+ client.request(
+ "POST", "http://url.com", params={"argument": "value"}, timeout=2
+ )
+ error = exception_info.value
+ assert error.code == "invalid_input"
+ assert error.message == "invalid input in field 'broken_field': is too long"
+ assert error.details["fields"][0]["name"] == "broken_field"
+
+ def test_request_500(self, client, fail_response):
+ fail_response.status_code = 500
+ fail_response.reason = "Internal Server Error"
+ fail_response._content = "Internal Server Error"
+ client._requests_session.request.return_value = fail_response
+ with pytest.raises(APIException) as exception_info:
+ client.request(
+ "POST", "http://url.com", params={"argument": "value"}, timeout=2
+ )
+ error = exception_info.value
+ assert error.code == 500
+ assert error.message == "Internal Server Error"
+ assert error.details["content"] == "Internal Server Error"
+
+ def test_request_broken_json_200(self, client, response):
+ content = b"{'key': 'value'"
+ response.reason = "OK"
+ response._content = content
+ client._requests_session.request.return_value = response
+ with pytest.raises(APIException) as exception_info:
+ client.request(
+ "POST", "http://url.com", params={"argument": "value"}, timeout=2
+ )
+ error = exception_info.value
+ assert error.code == 200
+ assert error.message == "OK"
+ assert error.details["content"] == content
+
+ def test_request_empty_content_200(self, client, response):
+ content = ""
+ response.reason = "OK"
+ response._content = content
+ client._requests_session.request.return_value = response
+ response = client.request(
+ "POST", "http://url.com", params={"argument": "value"}, timeout=2
+ )
+ assert response == ""
+
+ def test_request_500_empty_content(self, client, fail_response):
+ fail_response.status_code = 500
+ fail_response.reason = "Internal Server Error"
+ fail_response._content = ""
+ client._requests_session.request.return_value = fail_response
+ with pytest.raises(APIException) as exception_info:
+ client.request(
+ "POST", "http://url.com", params={"argument": "value"}, timeout=2
+ )
+ error = exception_info.value
+ assert error.code == 500
+ assert error.message == "Internal Server Error"
+ assert error.details["content"] == ""
+ assert str(error) == "Internal Server Error"
+
+ def test_request_limit(self, client, rate_limit_response):
+ client._retry_wait_time = 0
+ client._requests_session.request.return_value = rate_limit_response
+ with pytest.raises(APIException) as exception_info:
+ client.request(
+ "POST", "http://url.com", params={"argument": "value"}, timeout=2
+ )
+ error = exception_info.value
+ assert client._requests_session.request.call_count == 5
+ assert error.code == "rate_limit_exceeded"
+ assert error.message == "limit of 10 requests per hour reached"
+
+ def test_request_limit_then_success(self, client, rate_limit_response):
+ client._retry_wait_time = 0
+ response = requests.Response()
+ response.status_code = 200
+ response._content = json.dumps({"result": "data"}).encode("utf-8")
+ client._requests_session.request.side_effect = [rate_limit_response, response]
+
+ client.request(
+ "POST", "http://url.com", params={"argument": "value"}, timeout=2
+ )
+ assert client._requests_session.request.call_count == 2
diff --git a/tests/unit/test_hcloud.py b/tests/unit/test_hcloud.py
index 5f5bf8114b57be52a2bb33f142a0a7bc40adbd68..87ab50aa9d8e905d37f3c183ee3e099610dcef51 100644
--- a/tests/unit/test_hcloud.py
+++ b/tests/unit/test_hcloud.py
@@ -1,182 +1,6 @@
-import json
-from unittest.mock import MagicMock
-
import pytest
-import requests
-
-from hcloud import APIException, Client
-
-
-class TestHetznerClient:
- @pytest.fixture()
- def client(self):
- Client._version = "0.0.0"
- client = Client(token="project_token")
-
- client._requests_session = MagicMock()
- return client
-
- @pytest.fixture()
- def response(self):
- response = requests.Response()
- response.status_code = 200
- response._content = json.dumps({"result": "data"}).encode("utf-8")
- return response
-
- @pytest.fixture()
- def fail_response(self, response):
- response.status_code = 422
- error = {
- "code": "invalid_input",
- "message": "invalid input in field 'broken_field': is too long",
- "details": {
- "fields": [{"name": "broken_field", "messages": ["is too long"]}]
- },
- }
- response._content = json.dumps({"error": error}).encode("utf-8")
- return response
-
- @pytest.fixture()
- def rate_limit_response(self, response):
- response.status_code = 422
- error = {
- "code": "rate_limit_exceeded",
- "message": "limit of 10 requests per hour reached",
- "details": {},
- }
- response._content = json.dumps({"error": error}).encode("utf-8")
- return response
-
- def test__get_user_agent(self, client):
- user_agent = client._get_user_agent()
- assert user_agent == "hcloud-python/0.0.0"
-
- def test__get_user_agent_with_application_name(self, client):
- client = Client(token="project_token", application_name="my-app")
- user_agent = client._get_user_agent()
- assert user_agent == "my-app hcloud-python/0.0.0"
-
- def test__get_user_agent_with_application_name_and_version(self, client):
- client = Client(
- token="project_token",
- application_name="my-app",
- application_version="1.0.0",
- )
- user_agent = client._get_user_agent()
- assert user_agent == "my-app/1.0.0 hcloud-python/0.0.0"
-
- def test__get_headers(self, client):
- headers = client._get_headers()
- assert headers == {
- "User-Agent": "hcloud-python/0.0.0",
- "Authorization": "Bearer project_token",
- }
-
- def test_request_library_mocked(self, client):
- response = client.request("POST", "url", params={"1": 2})
- assert response.__class__.__name__ == "MagicMock"
-
- def test_request_ok(self, client, response):
- client._requests_session.request.return_value = response
- response = client.request(
- "POST", "/servers", params={"argument": "value"}, timeout=2
- )
- client._requests_session.request.assert_called_once_with(
- method="POST",
- url="https://api.hetzner.cloud/v1/servers",
- headers={
- "User-Agent": "hcloud-python/0.0.0",
- "Authorization": "Bearer project_token",
- },
- params={"argument": "value"},
- timeout=2,
- )
- assert response == {"result": "data"}
-
- def test_request_fails(self, client, fail_response):
- client._requests_session.request.return_value = fail_response
- with pytest.raises(APIException) as exception_info:
- client.request(
- "POST", "http://url.com", params={"argument": "value"}, timeout=2
- )
- error = exception_info.value
- assert error.code == "invalid_input"
- assert error.message == "invalid input in field 'broken_field': is too long"
- assert error.details["fields"][0]["name"] == "broken_field"
-
- def test_request_500(self, client, fail_response):
- fail_response.status_code = 500
- fail_response.reason = "Internal Server Error"
- fail_response._content = "Internal Server Error"
- client._requests_session.request.return_value = fail_response
- with pytest.raises(APIException) as exception_info:
- client.request(
- "POST", "http://url.com", params={"argument": "value"}, timeout=2
- )
- error = exception_info.value
- assert error.code == 500
- assert error.message == "Internal Server Error"
- assert error.details["content"] == "Internal Server Error"
-
- def test_request_broken_json_200(self, client, response):
- content = b"{'key': 'value'"
- response.reason = "OK"
- response._content = content
- client._requests_session.request.return_value = response
- with pytest.raises(APIException) as exception_info:
- client.request(
- "POST", "http://url.com", params={"argument": "value"}, timeout=2
- )
- error = exception_info.value
- assert error.code == 200
- assert error.message == "OK"
- assert error.details["content"] == content
-
- def test_request_empty_content_200(self, client, response):
- content = ""
- response.reason = "OK"
- response._content = content
- client._requests_session.request.return_value = response
- response = client.request(
- "POST", "http://url.com", params={"argument": "value"}, timeout=2
- )
- assert response == ""
-
- def test_request_500_empty_content(self, client, fail_response):
- fail_response.status_code = 500
- fail_response.reason = "Internal Server Error"
- fail_response._content = ""
- client._requests_session.request.return_value = fail_response
- with pytest.raises(APIException) as exception_info:
- client.request(
- "POST", "http://url.com", params={"argument": "value"}, timeout=2
- )
- error = exception_info.value
- assert error.code == 500
- assert error.message == "Internal Server Error"
- assert error.details["content"] == ""
- assert str(error) == "Internal Server Error"
-
- def test_request_limit(self, client, rate_limit_response):
- client._retry_wait_time = 0
- client._requests_session.request.return_value = rate_limit_response
- with pytest.raises(APIException) as exception_info:
- client.request(
- "POST", "http://url.com", params={"argument": "value"}, timeout=2
- )
- error = exception_info.value
- assert client._requests_session.request.call_count == 5
- assert error.code == "rate_limit_exceeded"
- assert error.message == "limit of 10 requests per hour reached"
- def test_request_limit_then_success(self, client, rate_limit_response):
- client._retry_wait_time = 0
- response = requests.Response()
- response.status_code = 200
- response._content = json.dumps({"result": "data"}).encode("utf-8")
- client._requests_session.request.side_effect = [rate_limit_response, response]
- client.request(
- "POST", "http://url.com", params={"argument": "value"}, timeout=2
- )
- assert client._requests_session.request.call_count == 2
+def test_deprecated_hcloud_hcloud_module():
+ with pytest.deprecated_call():
+ from hcloud.hcloud import Client # noqa