Fabio/certbot-dns-ionos
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Unit tests

Browse source
This commit is contained in:
Helge 2020-12-30 00:29:20 +01:00
parent 66f87007c0
commit 7928413673
3 changed files with 85 additions and 77 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.vscode/launch.json vendored
View file

@ -10,7 +10,7 @@
"request": "launch",
"program": "/home/pi/dev/certbot/venv3/bin/certbot",
"console": "integratedTerminal",
"args": ["certonly", "-a", "dns-ionos", "-d", "*.examaple.de", "--dns-ionos-credentials", "/home/pi/dev/certbot-dns-ionos/my_debug/secrets/credentials.ini", "--config-dir", "my_debug/config", "--work-dir", "my_debug/work", "--logs-dir", "my_debug/logs"]
"args": ["certonly", "-a", "dns-ionos", "-d", "*.erbehome.de", "--dns-ionos-credentials", "/home/pi/dev/certbot-dns-ionos/my_debug/secrets/credentials.ini", "--config-dir", "my_debug/config", "--work-dir", "my_debug/work", "--logs-dir", "my_debug/logs"]
}
]
}

3
certbot_dns_ionos/dns_ionos.py
View file

@ -120,7 +120,8 @@ class _ionosClient(object):
)
logger.debug("API REquest to URL: %s", url)
if resp.status_code != 200:
error_msg = resp.reason + " " + resp.text['message']
content = json.loads(resp.content)
error_msg = resp.reason + " " + content['message']
raise errors.PluginError(
"HTTP Error during request {0}:{1}".format(resp.status_code, error_msg)
)

129
certbot_dns_ionos/dns_ionos_test.py
View file

@ -77,91 +77,98 @@ class ionosClientTest(unittest.TestCase):
def setUp(self):
from certbot_dns_ionos.dns_ionos import _ionosClient
self.adapter = requests_mock.Adapter()
self.client = _ionosClient(FAKE_ENDPOINT, FAKE_PREFIX, FAKE_SECRET)
self.client.session.mount("mock", self.adapter)
def _register_response(
self, ep_id, response=None, message=None, additional_matcher=None, **kwargs
):
resp = {"code": "ok", "message": message, "response": response}
if message is not None:
resp["code"] = "remote_failure"
def add_matcher(request):
data = json.loads(request.text)
add_result = True
if additional_matcher is not None:
add_result = additionsal_matcher(request)
return (
(
("prefix" in data and data["prefix"] == FAKE_PREFIX)
and ("prefix" in data and data["secret"] == FAKE_SECRET)
)
or data["session_id"] == "FAKE_SESSION"
) and add_result
self.adapter.register_uri(
requests_mock.ANY,
"{0}?{1}".format(FAKE_ENDPOINT, ep_id),
text=json.dumps(resp),
additional_matcher=add_matcher,
**kwargs
)
def test_add_txt_record(self):
self._register_response("login", response="FAKE_SESSION")
self._register_response("dns_zone_get_id", response=23)
self._register_response("dns_txt_add", response=99)
self._register_response(
"dns_zone_get", response={"zone_id": 102, "server_id": 1}
)
self._register_response("dns_rr_get_all_by_zone", response=[])
with requests_mock.Mocker() as m:
mock_response = [{
"id": "11af3414-ebba-11e9-8df5-66fbe8a334b4",
"name": "example.com",
"type": "NATIVE"}]
m.register_uri('GET', 'mock://endpoint/dns/v1/zones', status_code=200, reason="OK", json=mock_response)
mock_response = {
"id": "11af3414-ebba-11e9-8df5-66fbe8a334b4",
"name": "example.com",
"type": "NATIVE",
"records": [
{
"id": "22af3414-abbe-9e11-5df5-66fbe8e334b4",
"name": "foo",
"rootName": "string",
"type": "TXT",
"content": "string",
"changeDate": "string",
"ttl": 0,
"prio": 0,
"disabled": False
}
]
}
m.register_uri('GET', 'mock://endpoint/dns/v1/zones/11af3414-ebba-11e9-8df5-66fbe8a334b4', status_code=200, reason="OK", json=mock_response)
m.register_uri('PUT', 'mock://endpoint/dns/v1/zones/11af3414-ebba-11e9-8df5-66fbe8a334b4/records/22af3414-abbe-9e11-5df5-66fbe8e334b4', status_code=200, reason="OK")
try:
self.client.add_txt_record(
DOMAIN, self.record_name, self.record_content, self.record_ttl
)
except:
self.fail("No exeption expected")
def test_add_txt_record_fail_to_find_domain(self):
self._register_response("login", response="FAKE_SESSION")
self._register_response("dns_zone_get_id", message="Not Found")
with requests_mock.Mocker() as m:
mock_response = [{
"id": "11af3414-ebba-11e9-8df5-66fbe8a334b4",
"name": "test.com",
"type": "NATIVE"}]
m.register_uri('GET', 'mock://endpoint/dns/v1/zones', status_code=200, reason="OK", json=mock_response)
with self.assertRaises(errors.PluginError) as context:
self.client.add_txt_record(
DOMAIN, self.record_name, self.record_content, self.record_ttl
)
def test_add_txt_record_fail_to_authenticate(self):
self._register_response("login", message="FAILED")
with requests_mock.Mocker() as m:
mock_response = {'message': 'Missing or invalid API key.'}
m.register_uri('GET', 'mock://endpoint/dns/v1/zones', status_code=401, reason="Unauthorized", json=mock_response)
with self.assertRaises(errors.PluginError) as context:
self.client.add_txt_record(
DOMAIN, self.record_name, self.record_content, self.record_ttl
)
def test_del_txt_record(self):
self._register_response("login", response="FAKE_SESSION")
self._register_response("dns_zone_get_id", response=23)
self._register_response("dns_rr_get_all_by_zone", response=[])
self._register_response("dns_txt_delete", response="")
with requests_mock.Mocker() as m:
mock_response = [{
"id": "11af3414-ebba-11e9-8df5-66fbe8a334b4",
"name": "example.com",
"type": "NATIVE"}]
m.register_uri('GET', 'mock://endpoint/dns/v1/zones', status_code=200, reason="OK", json=mock_response)
mock_response = {
"id": "11af3414-ebba-11e9-8df5-66fbe8a334b4",
"name": "example.com",
"type": "NATIVE",
"records": [
{
"id": "22af3414-abbe-9e11-5df5-66fbe8e334b4",
"name": "foo",
"rootName": "string",
"type": "TXT",
"content": "string",
"changeDate": "string",
"ttl": 0,
"prio": 0,
"disabled": False
}
]
}
m.register_uri('GET', 'mock://endpoint/dns/v1/zones/11af3414-ebba-11e9-8df5-66fbe8a334b4', status_code=200, reason="OK", json=mock_response)
m.register_uri('DELETE', 'mock://endpoint/dns/v1/zones/11af3414-ebba-11e9-8df5-66fbe8a334b4/records/22af3414-abbe-9e11-5df5-66fbe8e334b4', status_code=200, reason="OK")
try:
self.client.del_txt_record(
DOMAIN, self.record_name, self.record_content, self.record_ttl
)
except:
self.fail("No exeption expected")
def test_del_txt_record_fail_to_find_domain(self):
self._register_response("login", response="FAKE_SESSION")
self._register_response("dns_zone_get_id", message="Not Found")
with self.assertRaises(errors.PluginError) as context:
self.client.del_txt_record(
DOMAIN, self.record_name, self.record_content, self.record_ttl
)
def test_del_txt_record_fail_to_authenticate(self):
self._register_response("login", message="FAILED")
with self.assertRaises(errors.PluginError) as context:
self.client.del_txt_record(
DOMAIN, self.record_name, self.record_content, self.record_ttl
)
if __name__ == "__main__":