| # Licensed to the Apache Software Foundation (ASF) under one or more |
| # contributor license agreements. See the NOTICE file distributed with |
| # this work for additional information regarding copyright ownership. |
| # The ASF licenses this file to You under the Apache License, Version 2.0 |
| # (the "License"); you may not use this file except in compliance with |
| # the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| |
| import sys |
| import unittest |
| |
| from libcloud.dns.drivers.digitalocean import DigitalOceanDNSDriver |
| from libcloud.dns.types import RecordType |
| from libcloud.test import LibcloudTestCase, MockHttp |
| from libcloud.test.file_fixtures import DNSFileFixtures |
| from libcloud.test.secrets import DIGITALOCEAN_v2_PARAMS |
| from libcloud.utils.py3 import httplib |
| |
| |
| class DigitalOceanDNSTests(LibcloudTestCase): |
| def setUp(self): |
| DigitalOceanDNSDriver.connectionCls.conn_class = DigitalOceanDNSMockHttp |
| DigitalOceanDNSMockHttp.type = None |
| self.driver = DigitalOceanDNSDriver(*DIGITALOCEAN_v2_PARAMS) |
| |
| def test_list_zones(self): |
| zones = self.driver.list_zones() |
| self.assertTrue(len(zones) >= 1) |
| |
| def test_get_zone(self): |
| zone = self.driver.get_zone("testdomain") |
| self.assertEqual(zone.id, "testdomain") |
| |
| def test_get_zone_not_found(self): |
| DigitalOceanDNSMockHttp.type = "NOT_FOUND" |
| self.assertRaises(Exception, self.driver.get_zone, "testdomain") |
| |
| def test_list_records(self): |
| zone = self.driver.get_zone("testdomain") |
| records = self.driver.list_records(zone) |
| self.assertTrue(len(records) >= 1) |
| self.assertEqual(records[1].ttl, 1800) |
| self.assertEqual(records[4].ttl, None) |
| |
| def test_get_record(self): |
| record = self.driver.get_record("testdomain", "1234564") |
| self.assertEqual(record.id, "1234564") |
| self.assertEqual(record.type, RecordType.A) |
| self.assertEqual(record.data, "123.45.67.89") |
| self.assertEqual(record.ttl, 1800) |
| |
| def test_get_record_not_found(self): |
| DigitalOceanDNSMockHttp.type = "NOT_FOUND" |
| self.assertRaises(Exception, self.driver.get_zone, "testdomain") |
| |
| def test_create_zone(self): |
| DigitalOceanDNSMockHttp.type = "CREATE" |
| zone = self.driver.create_zone("testdomain") |
| self.assertEqual(zone.id, "testdomain") |
| |
| def test_create_record(self): |
| zone = self.driver.get_zone("testdomain") |
| |
| DigitalOceanDNSMockHttp.type = "CREATE" |
| record = self.driver.create_record( |
| "sub", zone, RecordType.A, "234.56.78.90", extra={"ttl": 60} |
| ) |
| self.assertEqual(record.id, "1234565") |
| self.assertEqual(record.type, RecordType.A) |
| self.assertEqual(record.data, "234.56.78.90") |
| self.assertEqual(record.ttl, 60) |
| |
| def test_update_record(self): |
| record = self.driver.get_record("testdomain", "1234564") |
| |
| DigitalOceanDNSMockHttp.type = "UPDATE" |
| record = self.driver.update_record( |
| record, data="234.56.78.90", extra={"ttl": 60} |
| ) |
| self.assertEqual(record.id, "1234564") |
| self.assertEqual(record.data, "234.56.78.90") |
| self.assertEqual(record.ttl, 60) |
| |
| def test_delete_zone(self): |
| zone = self.driver.get_zone("testdomain") |
| |
| DigitalOceanDNSMockHttp.type = "DELETE" |
| self.assertTrue(self.driver.delete_zone(zone)) |
| |
| def test_delete_record(self): |
| record = self.driver.get_record("testdomain", "1234564") |
| |
| DigitalOceanDNSMockHttp.type = "DELETE" |
| self.assertTrue(self.driver.delete_record(record)) |
| |
| |
| class DigitalOceanDNSMockHttp(MockHttp): |
| fixtures = DNSFileFixtures("digitalocean") |
| |
| response_map = { |
| None: httplib.OK, |
| "CREATE": httplib.CREATED, |
| "DELETE": httplib.NO_CONTENT, |
| "EMPTY": httplib.OK, |
| "NOT_FOUND": httplib.NOT_FOUND, |
| "UNAUTHORIZED": httplib.UNAUTHORIZED, |
| "UPDATE": httplib.OK, |
| "UNPROCESSABLE": httplib.UNPROCESSABLE_ENTITY, |
| } |
| |
| def _v2_domains(self, method, url, body, headers): |
| body = self.fixtures.load("_v2_domains.json") |
| return ( |
| self.response_map[self.type], |
| body, |
| {}, |
| httplib.responses[self.response_map[self.type]], |
| ) |
| |
| def _v2_domains_CREATE(self, method, url, body, headers): |
| if body is None: |
| body = self.fixtures.load("_v2_domains_UNPROCESSABLE_ENTITY.json") |
| return ( |
| self.response_map["UNPROCESSABLE"], |
| body, |
| {}, |
| httplib.responses[self.response_map["UNPROCESSABLE"]], |
| ) |
| body = self.fixtures.load("_v2_domains_CREATE.json") |
| return ( |
| self.response_map[self.type], |
| body, |
| {}, |
| httplib.responses[self.response_map[self.type]], |
| ) |
| |
| def _v2_domains_testdomain(self, method, url, body, headers): |
| body = self.fixtures.load("_v2_domains_testdomain.json") |
| return ( |
| self.response_map[self.type], |
| body, |
| {}, |
| httplib.responses[self.response_map[self.type]], |
| ) |
| |
| def _v2_domains_testdomain_DELETE(self, method, url, body, headers): |
| return ( |
| self.response_map[self.type], |
| body, |
| {}, |
| httplib.responses[self.response_map[self.type]], |
| ) |
| |
| def _v2_domains_testdomain_NOT_FOUND(self, method, url, body, headers): |
| body = self.fixtures.load("_v2_domains_testdomain_NOT_FOUND.json") |
| return ( |
| self.response_map[self.type], |
| body, |
| {}, |
| httplib.responses[self.response_map[self.type]], |
| ) |
| |
| def _v2_domains_testdomain_records(self, method, url, body, headers): |
| body = self.fixtures.load("_v2_domains_testdomain_records.json") |
| return ( |
| self.response_map[self.type], |
| body, |
| {}, |
| httplib.responses[self.response_map[self.type]], |
| ) |
| |
| def _v2_domains_testdomain_records_CREATE(self, method, url, body, headers): |
| if body is None: |
| body = self.fixtures.load("_v2_domains_UNPROCESSABLE_ENTITY.json") |
| return ( |
| self.response_map["UNPROCESSABLE"], |
| body, |
| {}, |
| httplib.responses[self.response_map["UNPROCESSABLE"]], |
| ) |
| body = self.fixtures.load("_v2_domains_testdomain_records_CREATE.json") |
| return ( |
| self.response_map[self.type], |
| body, |
| {}, |
| httplib.responses[self.response_map[self.type]], |
| ) |
| |
| def _v2_domains_testdomain_records_1234564(self, method, url, body, headers): |
| body = self.fixtures.load("_v2_domains_testdomain_records_1234564.json") |
| return ( |
| self.response_map[self.type], |
| body, |
| {}, |
| httplib.responses[self.response_map[self.type]], |
| ) |
| |
| def _v2_domains_testdomain_records_1234564_DELETE(self, method, url, body, headers): |
| self.type = "DELETE" |
| return ( |
| self.response_map[self.type], |
| body, |
| {}, |
| httplib.responses[self.response_map[self.type]], |
| ) |
| |
| def _v2_domains_testdomain_records_1234564_UPDATE(self, method, url, body, headers): |
| if body is None: |
| body = self.fixtures.load("_v2_domains_UNPROCESSABLE_ENTITY.json") |
| return ( |
| self.response_map["UNPROCESSABLE"], |
| body, |
| {}, |
| httplib.responses[self.response_map["UNPROCESSABLE"]], |
| ) |
| body = self.fixtures.load("_v2_domains_testdomain_records_1234564_UPDATE.json") |
| return ( |
| self.response_map[self.type], |
| body, |
| {}, |
| httplib.responses[self.response_map[self.type]], |
| ) |
| |
| |
| if __name__ == "__main__": |
| sys.exit(unittest.main()) |
