Skip to content
Snippets Groups Projects
Commit f4ab504b authored by Leo Antunes's avatar Leo Antunes
Browse files

New upstream version 1.25.0

parent 03fab50f
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
Showing
with 478 additions and 410 deletions
PKG-INFO
+ 1
− 1
View file @ f4ab504b
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
......
hcloud.egg-info/PKG-INFO
+ 1
− 1
View file @ f4ab504b
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
......
hcloud.egg-info/SOURCES.txt
+ 2
− 0
View file @ f4ab504b
......@@ -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
......
hcloud/__init__.py
+ 1
− 1
View file @ f4ab504b
from ._client import Client # noqa
from ._exceptions import APIException, HCloudException # noqa
from .hcloud import Client # noqa
hcloud/__version__.py
+ 1
− 1
View file @ f4ab504b
VERSION = "1.24.0" # x-release-please-version
VERSION = "1.25.0" # x-release-please-version
hcloud/_client.py 0 → 100644
+ 222
− 0
View file @ f4ab504b
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
hcloud/_exceptions.py
+ 1
− 3
View file @ f4ab504b
......@@ -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
hcloud/actions/domain.py
+ 8
− 2
View file @ f4ab504b
......@@ -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"""
hcloud/hcloud.py
+ 7
− 220
View file @ f4ab504b
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
tests/unit/actions/test_domain.py
+ 48
− 1
View file @ f4ab504b
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,
}
)
)
tests/unit/conftest.py
+ 1
− 1
View file @ f4ab504b
......@@ -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()
......
tests/unit/test_client.py 0 → 100644
+ 182
− 0
View file @ f4ab504b
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
tests/unit/test_hcloud.py
+ 3
− 179
View file @ f4ab504b
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
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment