0
0
mirror of https://github.com/thegeeklab/corenetworks.git synced 2024-11-21 21:20:38 +00:00

add commit method and more testing

This commit is contained in:
Robert Kaussow 2020-04-11 12:05:42 +02:00
parent 6d9fa820fb
commit ae40c99fcf
6 changed files with 101 additions and 19 deletions

View File

@ -10,7 +10,6 @@ from requests import Session
from requests.auth import AuthBase
from .exceptions import AuthError
from .exceptions import CorenetworksError
class CoreNetworksBasicAuth(AuthBase):
@ -50,10 +49,6 @@ class CoreNetworksBasicAuth(AuthBase):
session = Session()
handle = session.send(prepared_request)
handle.raise_for_status()
except ConnectionError as e:
raise CorenetworksError(
"Server unreachable: {reason}".format(reason=e.message.reason), payload=e
)
except HTTPError as e:
raise AuthError(
"Login failed: {code} {reason}".format(
@ -61,6 +56,8 @@ class CoreNetworksBasicAuth(AuthBase):
),
payload=e
)
except ConnectionError as e:
raise
response = handle.json()

View File

@ -166,7 +166,7 @@ class CoreNetworks():
See https://beta.api.core-networks.de/doc/#functon_dnszones_records_add
Returns:
list: List of removed records.
list: Empty list.
"""
schema = copy.deepcopy(self._schema)
@ -191,6 +191,23 @@ class CoreNetworks():
return self.__normalize(result)
def commit(self, zone):
"""
Commit changed records to the given DNS zone.
Args:
zone (str): Name of the target DNS zone.
Returns:
list: Empty list.
"""
result = self.__rest_helper(
"/dnszones/{zone}/records/commit".format(zone=zone), method="POST"
)
return self.__normalize(result)
def __rest_helper(self, url, data=None, params=None, method="GET"):
"""Handle requests to the Core Networks API."""
url = self.__endpoint + url
@ -228,10 +245,6 @@ class CoreNetworks():
handle = session.send(request)
handle.raise_for_status()
except ConnectionError as e:
raise CorenetworksError(
"Server unreachable: {reason}".format(reason=e.message.reason), payload=e
)
except HTTPError as e:
raise CorenetworksError(
"Invalid response: {code} {reason}".format(
@ -239,6 +252,8 @@ class CoreNetworks():
),
payload=e
)
except ConnectionError as e:
raise
if handle.status_code == 200:
response = handle.json()

View File

@ -5,13 +5,10 @@
class CoreNetworksException(Exception):
"""The main exception class."""
def __init__(self, message, payload=None):
self.message = message
def __init__(self, msg, payload=None):
super(CoreNetworksException, self).__init__(msg)
self.payload = payload
def __str__(self): # noqa
return str(self.message)
class CorenetworksError(CoreNetworksException):
"""Authentication errors exception class."""

View File

@ -2,6 +2,7 @@
import json
from requests import ConnectionError
from six.moves.urllib.parse import parse_qs # noqa
from six.moves.urllib.parse import unquote # noqa
@ -19,10 +20,13 @@ def records_get_callback(request, context):
"data": "::1"
}]
if "missing.com" in request.path:
if "missing" in request.path:
context.status_code = 404
return "[]"
if "dict" in request.path:
return "{}"
query_raw = parse_qs(request.query)
if query_raw:
query = dict((k.replace("[]", ""), v) for k, v in query_raw.items())
@ -35,3 +39,7 @@ def records_get_callback(request, context):
def records_post_callback(request, context):
return "[]"
def records_error_callback(request, context):
raise ConnectionError

View File

@ -0,0 +1,23 @@
"""Test authenticator classes."""
from corenetworks.authenticators import CoreNetworksBasicAuth
from corenetworks.authenticators import CoreNetworksTokenAuth
def test_basic_auth(requests_mock):
requests_mock.post(
"https://beta.api.core-networks.de/auth/token",
json={"token": "mytoken"},
)
auth = CoreNetworksBasicAuth(
user="test", password="test", endpoint="https://beta.api.core-networks.de"
)
assert auth.token == "mytoken"
def test_token_auth(requests_mock):
auth = CoreNetworksTokenAuth(token="mytoken")
assert auth.token == "mytoken"

View File

@ -1,23 +1,44 @@
"""Test client class."""
import pytest
import requests
from six.moves.urllib.parse import parse_qs # noqa
from six.moves.urllib.parse import unquote # noqa
from corenetworks import CoreNetworks
from corenetworks.authenticators import CoreNetworksBasicAuth
from corenetworks.exceptions import AuthError
from corenetworks.exceptions import CorenetworksError
from corenetworks.exceptions import ValidationError
from corenetworks.test.fixtures.callback import records_error_callback
from corenetworks.test.fixtures.callback import records_get_callback
from corenetworks.test.fixtures.callback import records_post_callback
@pytest.fixture
def client(mocker):
client = CoreNetworks(api_token="secure")
mocker.patch.object(CoreNetworksBasicAuth, "_login", return_value="testtoken")
client = CoreNetworks(user="testuser", password="testpass")
return client
def test_auth_error():
with pytest.raises(AuthError) as e:
assert CoreNetworks(user="test")
assert str(e.value) == "Insufficient authentication details provided"
def test_request_error(requests_mock, client):
requests_mock.post(
"https://beta.api.core-networks.de/dnszones/example.com/records/commit",
text=records_error_callback,
)
with pytest.raises(requests.ConnectionError):
assert client.commit(zone="example.com")
def test_records(requests_mock, client):
requests_mock.get(
"https://beta.api.core-networks.de/dnszones/example.com/records/",
@ -44,15 +65,26 @@ def test_records(requests_mock, client):
def test_no_records(requests_mock, client):
requests_mock.get(
"https://beta.api.core-networks.de/dnszones/missing.com/records/",
"https://beta.api.core-networks.de/dnszones/missing/records/",
text=records_get_callback,
)
with pytest.raises(CorenetworksError) as e:
assert client.records(zone="missing.com")
assert client.records(zone="missing")
assert str(e.value) == "Invalid response: 404 None"
def test_type_records(requests_mock, client):
requests_mock.get(
"https://beta.api.core-networks.de/dnszones/dict/records/",
text=records_get_callback,
)
dictresp = client.records(zone="dict")
assert dictresp == {}
def test_filter_records(requests_mock, client):
requests_mock.get(
"https://beta.api.core-networks.de/dnszones/example.com/records/",
@ -118,3 +150,13 @@ def test_delete_record_invalid(requests_mock, client):
with pytest.raises(ValidationError) as empty:
assert client.delete_record(zone="example.com", params={})
assert str(empty.value).startswith("Dataset invalid:")
def test_commit(requests_mock, client):
requests_mock.post(
"https://beta.api.core-networks.de/dnszones/example.com/records/commit",
text=records_post_callback,
)
resp = client.commit(zone="example.com")
assert resp == []