# 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())