github/corenetworks
0
0
Fork 0
mirror of https://github.com/thegeeklab/corenetworks.git synced 2024-06-02 16:59:41 +02:00
Code Issues Releases Activity

add some more tests

Browse Source
This commit is contained in:
Robert Kaussow 2020-04-04 23:21:32 +02:00
parent e987750c4a
commit 43ed62b7ff
4 changed files with 156 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

86
corenetworks/client.py
View File

@ -47,6 +47,52 @@ class CoreNetworks():
},
}
self._filter_schema = {
"type": "object",
"properties": {
"name": {
"anyOf": [{
"type": "string",
}, {
"type": "array",
"items": {
"type": "string"
},
}],
},
"ttl": {
"anyOf": [{
"type": "number",
}, {
"type": "array",
"items": {
"type": "number"
},
}],
},
"type": {
"anyOf": [{
"type": "string",
}, {
"type": "array",
"items": {
"type": "string"
},
}],
},
"data": {
"anyOf": [{
"type": "string",
}, {
"type": "array",
"items": {
"type": "string"
},
}],
},
},
}
if api_token:
self._auth = CoreNetworksTokenAuth(api_token)
else:
@ -57,7 +103,7 @@ class CoreNetworks():
# RECORDS
def records(self, zone, data={}):
def records(self, zone, params={}):
"""
Get the list of records for the specific domain.
@ -74,10 +120,10 @@ class CoreNetworks():
list: List of entry dicts.
"""
schema = copy.deepcopy(self._schema)
self.__validate(data, schema)
schema = copy.deepcopy(self._filter_schema)
self.__validate(params, schema)
filter_string = self.__json_to_filter(data)
filter_string = self.__json_to_filter(params)
result = self.__rest_helper(
"/dnszones/{zone}/records/{filter}".format(zone=zone, filter=filter_string),
method="GET"
@ -85,7 +131,7 @@ class CoreNetworks():
return self.__normalize(result)
def add_record(self, zone, data):
def add_record(self, zone, params):
"""
Create a record for the given domain.
@ -99,15 +145,17 @@ class CoreNetworks():
"""
schema = copy.deepcopy(self._schema)
schema["required"] = ["name", "type", "data"]
self.__validate(data, schema)
self.__validate(params, schema)
result = self.__rest_helper(
"/dnszones/{zone}/records/".format(zone=zone), data=data, method="POST"
self.__rest_helper(
"/dnszones/{zone}/records/".format(zone=zone), data=params, method="POST"
)
result = self.records(zone=zone, params=params)
return self.__normalize(result)
def delete_record(self, zone, data):
def delete_record(self, zone, params):
"""
Delete all DNS records of a zone that match the data.
@ -116,7 +164,7 @@ class CoreNetworks():
"""
schema = copy.deepcopy(self._schema)
schema["properties"]["force_all"] = {"type": "boolean"}
schema["oneOf"] = [{
schema["anyOf"] = [{
"required": ["name"]
}, {
"required": ["type"]
@ -125,18 +173,18 @@ class CoreNetworks():
}, {
"required": ["force_all"]
}]
self.__validate(data, schema)
self.__validate(params, schema)
if data.get("force_all"):
data = {}
if params.get("force_all"):
params = {}
print(data)
print(params)
# result = self.__rest_helper(
# "/dnszones/{zone}/records/delete/".format(zone=zone), data=data, method="POST"
# )
result = self.__rest_helper(
"/dnszones/{zone}/records/delete/".format(zone=zone), data=params, method="POST"
)
# return self.__normalize(result)
return self.__normalize(result)
def __rest_helper(self, url, data=None, params=None, method="GET"):
"""Handle requests to the Core Networks API."""
@ -211,7 +259,7 @@ class CoreNetworks():
if isinstance(value, list):
for item in value:
filter_list.append("{key}[]={value}".format(key=key, value=item))
elif isinstance(value, str):
elif isinstance(value, str) or isinstance(value, int):
filter_list.append("{key}={value}".format(key=key, value=value))
else:
raise CorenetworksError("Unknown type: {}".format(type(value)))

1
corenetworks/tests/fixtures/__init__.py vendored Normal file
View File

@ -0,0 +1 @@
"""Misc pytest fixtures."""

37
corenetworks/tests/fixtures/callback.py vendored Normal file
View File

@ -0,0 +1,37 @@
"""Requests mock callback functions."""
import json
from six.moves.urllib.parse import parse_qs # noqa
from six.moves.urllib.parse import unquote # noqa
def records_get_callback(request, context):
records = [{
"type": "A",
"ttl": 1800,
"name": "test",
"data": "127.0.0.1"
}, {
"type": "AAAA",
"ttl": 1800,
"name": "test",
"data": "::1"
}]
if "missing.com" in request.path:
context.status_code = 404
return "[]"
query_raw = parse_qs(request.query)
if query_raw:
query = dict((k.replace("[]", ""), v) for k, v in query_raw.items())
res = [d for d in records if d["type"].lower() in query["type"]]
return "{}".format(json.dumps(res))
else:
return "{}".format(json.dumps(records))
def records_post_callback(request, context):
return "[]"

58
corenetworks/tests/unit/test_client.py
View File

@ -1,9 +1,13 @@
"""Test client class."""
import pytest
from six.moves.urllib.parse import parse_qs # noqa
from six.moves.urllib.parse import unquote # noqa
from corenetworks import CoreNetworks
from corenetworks.exceptions import CorenetworksError
from corenetworks.tests.fixtures.callback import records_get_callback
from corenetworks.tests.fixtures.callback import records_post_callback
@pytest.fixture
@ -16,24 +20,64 @@ def client(mocker):
def test_records(requests_mock, client):
requests_mock.get(
"https://beta.api.core-networks.de/dnszones/example.com/records/",
text='[{"test": "test"}]'
text=records_get_callback
)
exp = [{
"type": "A",
"ttl": 1800,
"name": "test",
"data": "127.0.0.1"
}, {
"type": "AAAA",
"ttl": 1800,
"name": "test",
"data": "::1"
}]
resp = client.records(zone="example.com")
assert resp == [{"test": "test"}]
assert resp == exp
def test_no_records(requests_mock, client):
def failure_callback(request, context):
context.status_code = 404
return "[]"
requests_mock.get(
"https://beta.api.core-networks.de/dnszones/missing.com/records/",
text=failure_callback,
text=records_get_callback,
)
with pytest.raises(CorenetworksError) as e:
assert client.records(zone="missing.com")
assert str(e.value) == "Invalid response: 404 None"
def test_filter_records(requests_mock, client):
requests_mock.get(
"https://beta.api.core-networks.de/dnszones/example.com/records/",
text=records_get_callback,
)
resp = client.records(zone="example.com", params={"type": ["A"]})
assert resp == [{"type": "A", "ttl": 1800, "name": "test", "data": "127.0.0.1"}]
def test_add_record(requests_mock, client):
requests_mock.post(
"https://beta.api.core-networks.de/dnszones/example.com/records/",
text=records_post_callback,
)
requests_mock.get(
"https://beta.api.core-networks.de/dnszones/example.com/records/",
text=records_get_callback,
)
resp = client.add_record(
zone="example.com", params={
"type": "A",
"ttl": 1800,
"name": "test",
"data": "127.0.0.1"
}
)
assert resp == [{"type": "A", "ttl": 1800, "name": "test", "data": "127.0.0.1"}]