| # 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 |
| # limitations under the License. |
| |
| import json |
| import sys |
| import unittest |
| |
| from libcloud.utils.py3 import httplib |
| from libcloud.common.gandi_live import ( |
| GandiLiveBaseError, |
| JsonParseError, |
| InvalidRequestError, |
| ) |
| from libcloud.dns.types import ( |
| RecordType, |
| RecordError, |
| ZoneDoesNotExistError, |
| RecordDoesNotExistError, |
| ZoneAlreadyExistsError, |
| RecordAlreadyExistsError, |
| ) |
| from libcloud.dns.drivers.gandi_live import GandiLiveDNSDriver |
| from libcloud.dns.base import Zone, Record |
| from libcloud.test.file_fixtures import DNSFileFixtures |
| from libcloud.test.secrets import DNS_GANDI_LIVE |
| from libcloud.test.common.test_gandi_live import BaseGandiLiveMockHttp |
| |
| |
| class GandiLiveTests(unittest.TestCase): |
| def setUp(self): |
| GandiLiveDNSDriver.connectionCls.conn_class = GandiLiveMockHttp |
| GandiLiveMockHttp.type = None |
| self.driver = GandiLiveDNSDriver(*DNS_GANDI_LIVE) |
| self.test_zone = Zone( |
| id="example.com", |
| type="master", |
| ttl=None, |
| domain="example.com", |
| extra={"zone_uuid": "a53re"}, |
| driver=self, |
| ) |
| self.test_bad_zone = Zone( |
| id="badexample.com", |
| type="master", |
| ttl=None, |
| domain="badexample.com", |
| extra={"zone_uuid": "a53rf"}, |
| driver=self, |
| ) |
| self.test_record = Record( |
| id="A:bob", |
| type=RecordType.A, |
| name="bob", |
| zone=self.test_zone, |
| data="127.0.0.1", |
| driver=self, |
| extra={}, |
| ) |
| self.test_bad_record = Record( |
| id="A:jane", |
| type=RecordType.A, |
| name="jane", |
| zone=self.test_bad_zone, |
| data="127.0.0.1", |
| driver=self, |
| extra={}, |
| ) |
| |
| def test_list_zones(self): |
| zones = self.driver.list_zones() |
| self.assertEqual(len(zones), 2) |
| zone = zones[0] |
| self.assertEqual(zone.id, "example.com") |
| self.assertEqual(zone.type, "master") |
| self.assertEqual(zone.domain, "example.com") |
| self.assertIsNone(zone.ttl) |
| zone = zones[1] |
| self.assertEqual(zone.id, "example.net") |
| self.assertEqual(zone.type, "master") |
| self.assertEqual(zone.domain, "example.net") |
| self.assertIsNone(zone.ttl) |
| |
| def test_create_zone(self): |
| zone = self.driver.create_zone("example.org", extra={"name": "Example"}) |
| self.assertEqual(zone.id, "example.org") |
| self.assertEqual(zone.domain, "example.org") |
| self.assertEqual(zone.extra["zone_uuid"], "54321") |
| |
| def test_create_zone_without_name(self): |
| zone = self.driver.create_zone("example.org") |
| self.assertEqual(zone.id, "example.org") |
| self.assertEqual(zone.domain, "example.org") |
| self.assertEqual(zone.extra["zone_uuid"], "54321") |
| |
| def test_get_zone(self): |
| zone = self.driver.get_zone("example.com") |
| self.assertEqual(zone.id, "example.com") |
| self.assertEqual(zone.type, "master") |
| self.assertEqual(zone.domain, "example.com") |
| self.assertIsNone(zone.ttl) |
| |
| def test_list_records(self): |
| records = self.driver.list_records(self.test_zone) |
| self.assertEqual(len(records), 3) |
| record = records[0] |
| self.assertEqual(record.id, "A:@") |
| self.assertEqual(record.name, "@") |
| self.assertEqual(record.type, RecordType.A) |
| self.assertEqual(record.data, "127.0.0.1") |
| record = records[1] |
| self.assertEqual(record.id, "CNAME:www") |
| self.assertEqual(record.name, "www") |
| self.assertEqual(record.type, RecordType.CNAME) |
| self.assertEqual(record.data, "bob.example.com.") |
| record = records[2] |
| self.assertEqual(record.id, "A:bob") |
| self.assertEqual(record.name, "bob") |
| self.assertEqual(record.type, RecordType.A) |
| self.assertEqual(record.data, "127.0.1.1") |
| |
| def test_get_record(self): |
| record = self.driver.get_record(self.test_zone.id, "A:bob") |
| self.assertEqual(record.id, "A:bob") |
| self.assertEqual(record.name, "bob") |
| self.assertEqual(record.type, RecordType.A) |
| self.assertEqual(record.data, "127.0.1.1") |
| |
| def test_create_record(self): |
| record = self.driver.create_record( |
| "alice", self.test_zone, "AAAA", "::1", extra={"ttl": 400} |
| ) |
| self.assertEqual(record.id, "AAAA:alice") |
| self.assertEqual(record.name, "alice") |
| self.assertEqual(record.type, RecordType.AAAA) |
| self.assertEqual(record.data, "::1") |
| |
| def test_create_record_doesnt_throw_if_ttl_is_not_provided(self): |
| record = self.driver.create_record("alice", self.test_zone, "AAAA", "::1") |
| self.assertEqual(record.id, "AAAA:alice") |
| self.assertEqual(record.name, "alice") |
| self.assertEqual(record.type, RecordType.AAAA) |
| self.assertEqual(record.data, "::1") |
| |
| def test_bad_record_validation(self): |
| with self.assertRaises(RecordError) as ctx: |
| self.driver.create_record( |
| "alice", self.test_zone, "AAAA", "1" * 1025, extra={"ttl": 400} |
| ) |
| self.assertTrue("Record data must be" in str(ctx.exception)) |
| with self.assertRaises(RecordError) as ctx: |
| self.driver.create_record( |
| "alice", self.test_zone, "AAAA", "::1", extra={"ttl": 10} |
| ) |
| self.assertTrue("TTL must be at least" in str(ctx.exception)) |
| with self.assertRaises(RecordError) as ctx: |
| self.driver.create_record( |
| "alice", self.test_zone, "AAAA", "::1", extra={"ttl": 31 * 24 * 60 * 60} |
| ) |
| self.assertTrue("TTL must not exceed" in str(ctx.exception)) |
| |
| def test_update_record(self): |
| record = self.driver.update_record( |
| self.test_record, "bob", RecordType.A, "192.168.0.2", {"ttl": 500} |
| ) |
| self.assertEqual(record.id, "A:bob") |
| self.assertEqual(record.name, "bob") |
| self.assertEqual(record.type, RecordType.A) |
| self.assertEqual(record.data, "192.168.0.2") |
| |
| def test_delete_record(self): |
| success = self.driver.delete_record(self.test_record) |
| self.assertTrue(success) |
| |
| def test_export_bind(self): |
| bind_export = self.driver.export_zone_to_bind_format(self.test_zone) |
| bind_lines = bind_export.decode("utf8").split("\n") |
| self.assertEqual(bind_lines[0], "@ 10800 IN A 127.0.0.1") |
| |
| def test_bad_json_response(self): |
| with self.assertRaises(JsonParseError): |
| self.driver.get_zone("badexample.com") |
| |
| def test_no_record_found(self): |
| with self.assertRaises(RecordDoesNotExistError): |
| self.driver.get_record(self.test_zone.id, "A:none") |
| |
| def test_record_already_exists(self): |
| with self.assertRaises(RecordAlreadyExistsError): |
| self.driver.create_record( |
| "bob", self.test_bad_zone, "A", "127.0.0.1", extra={"ttl": 400} |
| ) |
| |
| def test_no_zone_found(self): |
| with self.assertRaises(ZoneDoesNotExistError): |
| self.driver.get_zone("nosuchzone.com") |
| |
| def test_zone_already_exists(self): |
| with self.assertRaises(ZoneAlreadyExistsError): |
| self.driver.create_zone("badexample.com") |
| |
| # This is somewhat spurious - the driver should be preventing any invalid |
| # request error from arising. |
| def test_suberrors(self): |
| with self.assertRaises(InvalidRequestError) as ctx: |
| self.driver.update_record( |
| self.test_bad_record, "jane", RecordType.A, "192.168.0.2", {"ttl": 500} |
| ) |
| self.assertTrue("is not a foo" in str(ctx.exception)) |
| |
| def test_other_error(self): |
| with self.assertRaises(GandiLiveBaseError): |
| self.driver.list_records(self.test_bad_zone) |
| |
| def test_mx_record(self): |
| record = self.driver.get_record(self.test_zone.id, "MX:lists") |
| self.assertEqual(record.extra["priority"], "10") |
| self.assertTrue("_other_records" in record.extra) |
| other_record = record.extra["_other_records"][0] |
| self.assertEqual(other_record["extra"]["priority"], "20") |
| |
| def test_ex_create_multivalue_record(self): |
| records = self.driver.ex_create_multi_value_record( |
| "alice", self.test_zone, "AAAA", ["::1", "::2"], extra={"ttl": 400} |
| ) |
| self.assertEqual(records[0].id, "AAAA:alice") |
| self.assertEqual(records[0].name, "alice") |
| self.assertEqual(records[0].type, RecordType.AAAA) |
| self.assertEqual(records[0].data, "::1") |
| self.assertEqual(records[1].id, "AAAA:alice") |
| self.assertEqual(records[1].name, "alice") |
| self.assertEqual(records[1].type, RecordType.AAAA) |
| self.assertEqual(records[1].data, "::2") |
| |
| def test_update_multivalue_record(self): |
| record = self.driver.get_record(self.test_zone.id, "MX:lists") |
| updated = self.driver.update_record( |
| record, None, None, "mail1", {"ttl": 400, "priority": 10} |
| ) |
| self.assertEqual(updated.extra["priority"], "10") |
| self.assertEqual(updated.data, "mail1") |
| self.assertTrue("_other_records" in record.extra) |
| other_record = record.extra["_other_records"][0] |
| self.assertEqual(other_record["extra"]["priority"], "20") |
| |
| def test_ex_update_gandi_zone_name(self): |
| updated = self.driver.ex_update_gandi_zone_name("111111", "Foo") |
| self.assertTrue(updated) |
| |
| def test_ex_delete_gandi_zone(self): |
| deleted = self.driver.ex_delete_gandi_zone("111111") |
| self.assertTrue(deleted) |
| |
| |
| class GandiLiveMockHttp(BaseGandiLiveMockHttp): |
| fixtures = DNSFileFixtures("gandi_live") |
| |
| def _json_api_v5_domains_get(self, method, url, body, headers): |
| body = self.fixtures.load("list_zones.json") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _json_api_v5_domains_example_com_get(self, method, url, body, headers): |
| body = self.fixtures.load("get_zone.json") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _json_api_v5_domains_badexample_com_get(self, method, url, body, headers): |
| body = self.fixtures.load("get_bad_zone.json") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _json_api_v5_domains_nosuchzone_com_get(self, method, url, body, headers): |
| body = self.fixtures.load("get_nonexistent_zone.json") |
| return (httplib.NOT_FOUND, body, {}, httplib.responses[httplib.NOT_FOUND]) |
| |
| def _json_api_v5_zones_post(self, method, url, body, headers): |
| input = json.loads(body) |
| if "badexample" in input["name"]: |
| body = self.fixtures.load("create_bad_zone.json") |
| return (httplib.CONFLICT, body, {}, httplib.responses[httplib.CONFLICT]) |
| else: |
| body = self.fixtures.load("create_zone.json") |
| return ( |
| httplib.OK, |
| body, |
| {"Location": "/zones/54321"}, |
| httplib.responses[httplib.OK], |
| ) |
| |
| def _json_api_v5_zones_111111_patch(self, method, url, body, headers): |
| body = self.fixtures.load("update_gandi_zone.json") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _json_api_v5_zones_111111_delete(self, method, url, body, headers): |
| return (httplib.NO_CONTENT, "", {}, httplib.responses[httplib.OK]) |
| |
| def _json_api_v5_domains_example_org_patch(self, method, url, body, headers): |
| body = self.fixtures.load("create_domain.json") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _json_api_v5_domains_example_com_records_get(self, method, url, body, headers): |
| body = self.fixtures.load("list_records.json") |
| resp_headers = {} |
| if ( |
| headers is not None |
| and "Accept" in headers |
| and headers["Accept"] == "text/plain" |
| ): |
| body = self.fixtures.load("list_records_bind.txt") |
| resp_headers["Content-Type"] = "text/plain" |
| return (httplib.OK, body, resp_headers, httplib.responses[httplib.OK]) |
| |
| def _json_api_v5_domains_example_com_records_bob_A_get( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("get_record.json") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _json_api_v5_domains_example_com_records_none_A_get( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("get_nonexistent_record.json") |
| return (httplib.NOT_FOUND, body, {}, httplib.responses[httplib.NOT_FOUND]) |
| |
| def _json_api_v5_domains_example_com_records_lists_MX_get( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("get_mx_record.json") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _json_api_v5_domains_example_com_records_lists_MX_put( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("update_mx_record.json") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _json_api_v5_domains_example_com_records_post(self, method, url, body, headers): |
| body = self.fixtures.load("create_record.json") |
| return ( |
| httplib.OK, |
| body, |
| {"Location": "/zones/12345/records/alice/AAAA"}, |
| httplib.responses[httplib.OK], |
| ) |
| |
| def _json_api_v5_domains_badexample_com_records_post( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("create_existing_record.json") |
| return (httplib.CONFLICT, body, {}, httplib.responses[httplib.CONFLICT]) |
| |
| def _json_api_v5_domains_badexample_com_records_get( |
| self, method, url, body, headers |
| ): |
| return ( |
| httplib.INTERNAL_SERVER_ERROR, |
| body, |
| {}, |
| httplib.responses[httplib.INTERNAL_SERVER_ERROR], |
| ) |
| |
| def _json_api_v5_domains_badexample_com_records_jane_A_put( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("update_bad_record.json") |
| return (httplib.BAD_REQUEST, body, {}, httplib.responses[httplib.BAD_REQUEST]) |
| |
| def _json_api_v5_domains_example_com_records_bob_A_put( |
| self, method, url, body, headers |
| ): |
| body = self.fixtures.load("update_record.json") |
| return (httplib.OK, body, {}, httplib.responses[httplib.OK]) |
| |
| def _json_api_v5_domains_example_com_records_bob_A_delete( |
| self, method, url, body, headers |
| ): |
| return (httplib.NO_CONTENT, "", {}, httplib.responses[httplib.OK]) |
| |
| |
| if __name__ == "__main__": |
| sys.exit(unittest.main()) |
