# Licensed 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 random
from functools import partial

import mango
import copy
import unittest

DOCS = [
    {"_id": "1", "name": "Jimi", "age": 10, "cars": 1},
    {"_id": "2", "name": "kate", "age": 8, "cars": 0},
]


class IndexCrudTests(mango.DbPerClass):
    def setUp(self):
        self.db.recreate()

    def test_bad_fields(self):
        bad_fields = [
            None,
            True,
            False,
            "bing",
            2.0,
            {"foo": "bar"},
            [{"foo": 2}],
            [{"foo": "asc", "bar": "desc"}],
            [{"foo": "asc"}, {"bar": "desc"}],
            [""],
        ]
        for fields in bad_fields:
            try:
                self.db.create_index(fields)
            except Exception as e:
                self.assertEqual(e.response.status_code, 400)
            else:
                raise AssertionError("bad create index")

    def test_bad_types(self):
        bad_types = [
            None,
            True,
            False,
            1.5,
            "foo",  # Future support
            "geo",  # Future support
            {"foo": "bar"},
            ["baz", 3.0],
        ]
        for bt in bad_types:
            try:
                self.db.create_index(["foo"], idx_type=bt)
            except Exception as e:
                self.assertEqual(
                    e.response.status_code, 400, (bt, e.response.status_code)
                )
            else:
                raise AssertionError("bad create index")

    def test_bad_names(self):
        bad_names = ["", True, False, 1.5, {"foo": "bar"}, [None, False]]
        for bn in bad_names:
            try:
                self.db.create_index(["foo"], name=bn)
            except Exception as e:
                self.assertEqual(e.response.status_code, 400)
            else:
                raise AssertionError("bad create index")

    def test_bad_ddocs(self):
        bad_ddocs = ["", True, False, 1.5, {"foo": "bar"}, [None, False]]
        for bd in bad_ddocs:
            try:
                self.db.create_index(["foo"], ddoc=bd)
            except Exception as e:
                self.assertEqual(e.response.status_code, 400)
            else:
                raise AssertionError("bad create index")

    def test_bad_urls(self):
        # These are only the negative test cases because ideally the
        # positive ones are implicitly tested by other ones.

        all_methods = [
            ("PUT", self.db.sess.put),
            ("GET", self.db.sess.get),
            ("POST", self.db.sess.post),
            ("PATCH", self.db.sess.get),
            ("DELETE", self.db.sess.delete),
            ("HEAD", self.db.sess.head),
            ("COPY", partial(self.db.sess.request, "COPY")),
            ("OPTIONS", partial(self.db.sess.request, "OPTIONS")),
            ("TRACE", partial(self.db.sess.request, "TRACE")),
            ("CONNECT", partial(self.db.sess.request, "CONNECT")),
        ]

        def all_but(method):
            return list(filter(lambda x: x[0] != method, all_methods))

        # Three-element subpaths are used as a shorthand to delete
        # indexes via design documents, see below.
        for subpath in ["a", "a/b", "a/b/c/d", "a/b/c/d/e", "a/b/c/d/e/f"]:
            path = self.db.path("_index/{}".format(subpath))
            for method_name, method in all_methods:
                with self.subTest(path=path, method=method_name):
                    r = method(path)
                    self.assertEqual(r.status_code, 404)

        for method_name, method in all_but("POST"):
            path = self.db.path("_index/_bulk_delete")
            with self.subTest(path=path, method=method_name):
                r = method(path)
                self.assertEqual(r.status_code, 405)

        fields = ["foo", "bar"]
        ddoc = "dd"
        idx = "idx_01"
        ret = self.db.create_index(fields, name=idx, ddoc=ddoc)
        assert ret is True
        for subpath in [
            "{}/json/{}".format(ddoc, idx),
            "_design/{}/json/{}".format(ddoc, idx),
        ]:
            path = self.db.path("_index/{}".format(subpath))
            for method_name, method in all_but("DELETE"):
                r = method(path)
                with self.subTest(path=path, method=method_name):
                    self.assertEqual(r.status_code, 405)

    def test_create_idx_01(self):
        fields = ["foo", "bar"]
        ret = self.db.create_index(fields, name="idx_01")
        assert ret is True
        for idx in self.db.list_indexes():
            if idx["name"] != "idx_01":
                continue
            self.assertEqual(idx["def"]["fields"], [{"foo": "asc"}, {"bar": "asc"}])
            return
        raise AssertionError("index not created")

    def test_create_idx_01_exists(self):
        fields = ["foo", "bar"]
        ret = self.db.create_index(fields, name="idx_01")
        assert ret is True
        ret = self.db.create_index(fields, name="idx_01")
        assert ret is False

    def test_create_idx_02(self):
        fields = ["baz", "foo"]
        ret = self.db.create_index(fields, name="idx_02")
        assert ret is True
        for idx in self.db.list_indexes():
            if idx["name"] != "idx_02":
                continue
            self.assertEqual(idx["def"]["fields"], [{"baz": "asc"}, {"foo": "asc"}])
            return
        raise AssertionError("index not created")

    def test_read_idx_doc(self):
        self.db.create_index(["foo", "bar"], name="idx_01")
        self.db.create_index(["hello", "bar"])
        for idx in self.db.list_indexes():
            if idx["type"] == "special":
                continue
            ddocid = idx["ddoc"]
            doc = self.db.open_doc(ddocid)
            self.assertEqual(doc["_id"], ddocid)
            info = self.db.ddoc_info(ddocid)
            self.assertEqual(info["name"], ddocid.split("_design/")[-1])

    def test_delete_idx_escaped(self):
        self.db.create_index(["foo", "bar"], name="idx_01")
        pre_indexes = self.db.list_indexes()
        ret = self.db.create_index(["bing"], name="idx_del_1")
        assert ret is True
        for idx in self.db.list_indexes():
            if idx["name"] != "idx_del_1":
                continue
            self.assertEqual(idx["def"]["fields"], [{"bing": "asc"}])
            self.db.delete_index(idx["ddoc"].replace("/", "%2F"), idx["name"])
        post_indexes = self.db.list_indexes()
        self.assertEqual(pre_indexes, post_indexes)

    def test_delete_idx_unescaped(self):
        pre_indexes = self.db.list_indexes()
        ret = self.db.create_index(["bing"], name="idx_del_2")
        assert ret is True
        for idx in self.db.list_indexes():
            if idx["name"] != "idx_del_2":
                continue
            self.assertEqual(idx["def"]["fields"], [{"bing": "asc"}])
            self.db.delete_index(idx["ddoc"], idx["name"])
        post_indexes = self.db.list_indexes()
        self.assertEqual(pre_indexes, post_indexes)

    def test_delete_idx_no_design(self):
        pre_indexes = self.db.list_indexes()
        ret = self.db.create_index(["bing"], name="idx_del_3")
        assert ret is True
        for idx in self.db.list_indexes():
            if idx["name"] != "idx_del_3":
                continue
            self.assertEqual(idx["def"]["fields"], [{"bing": "asc"}])
            self.db.delete_index(idx["ddoc"].split("/")[-1], idx["name"])
        post_indexes = self.db.list_indexes()
        self.assertEqual(pre_indexes, post_indexes)

    def test_bulk_delete(self):
        fields = ["field1"]
        ret = self.db.create_index(fields, name="idx_01")
        assert ret is True

        fields = ["field2"]
        ret = self.db.create_index(fields, name="idx_02")
        assert ret is True

        fields = ["field3"]
        ret = self.db.create_index(fields, name="idx_03")
        assert ret is True

        docids = []

        for idx in self.db.list_indexes():
            if idx["ddoc"] is not None:
                docids.append(idx["ddoc"])

        docids.append("_design/this_is_not_an_index_name")

        ret = self.db.bulk_delete(docids)

        self.assertEqual(ret["fail"][0]["id"], "_design/this_is_not_an_index_name")
        self.assertEqual(len(ret["success"]), 3)

        for idx in self.db.list_indexes():
            assert idx["type"] != "json"
            assert idx["type"] != "text"

    def test_recreate_index(self):
        pre_indexes = self.db.list_indexes()
        for i in range(5):
            ret = self.db.create_index(["bing"], name="idx_recreate")
            assert ret is True
            for idx in self.db.list_indexes():
                if idx["name"] != "idx_recreate":
                    continue
                self.assertEqual(idx["def"]["fields"], [{"bing": "asc"}])
                self.db.delete_index(idx["ddoc"], idx["name"])
                break
            post_indexes = self.db.list_indexes()
            self.assertEqual(pre_indexes, post_indexes)

    def test_delete_missing(self):
        # Missing design doc
        try:
            self.db.delete_index("this_is_not_a_design_doc_id", "foo")
        except Exception as e:
            self.assertEqual(e.response.status_code, 404)
        else:
            raise AssertionError("bad index delete")

        # Missing view name
        ret = self.db.create_index(["fields"], name="idx_01")
        indexes = self.db.list_indexes()
        not_special = [idx for idx in indexes if idx["type"] != "special"]
        idx = random.choice(not_special)
        ddocid = idx["ddoc"].split("/")[-1]
        try:
            self.db.delete_index(ddocid, "this_is_not_an_index_name")
        except Exception as e:
            self.assertEqual(e.response.status_code, 404)
        else:
            raise AssertionError("bad index delete")

        # Bad view type
        try:
            self.db.delete_index(ddocid, idx["name"], idx_type="not_a_real_type")
        except Exception as e:
            self.assertEqual(e.response.status_code, 404)
        else:
            raise AssertionError("bad index delete")

    def test_limit_skip_index(self):
        fields = ["field1"]
        ret = self.db.create_index(fields, name="idx_01")
        assert ret is True

        fields = ["field2"]
        ret = self.db.create_index(fields, name="idx_02")
        assert ret is True

        fields = ["field3"]
        ret = self.db.create_index(fields, name="idx_03")
        assert ret is True

        fields = ["field4"]
        ret = self.db.create_index(fields, name="idx_04")
        assert ret is True

        fields = ["field5"]
        ret = self.db.create_index(fields, name="idx_05")
        assert ret is True

        self.assertEqual(len(self.db.list_indexes(limit=2)), 2)
        self.assertEqual(len(self.db.list_indexes(limit=5, skip=4)), 2)
        self.assertEqual(len(self.db.list_indexes(skip=5)), 1)
        self.assertEqual(len(self.db.list_indexes(skip=6)), 0)
        self.assertEqual(len(self.db.list_indexes(skip=100)), 0)
        self.assertEqual(len(self.db.list_indexes(limit=10000000)), 6)

        try:
            self.db.list_indexes(skip=-1)
        except Exception as e:
            self.assertEqual(e.response.status_code, 500)

        try:
            self.db.list_indexes(limit=0)
        except Exception as e:
            self.assertEqual(e.response.status_code, 500)

    def test_out_of_sync(self):
        self.db.save_docs(copy.deepcopy(DOCS))
        self.db.create_index(["age"], name="age")

        selector = {"age": {"$gt": 0}}
        docs = self.db.find(
            selector, use_index="_design/a017b603a47036005de93034ff689bbbb6a873c4"
        )
        self.assertEqual(len(docs), 2)

        self.db.delete_doc("1")

        docs1 = self.db.find(
            selector,
            update="False",
            use_index="_design/a017b603a47036005de93034ff689bbbb6a873c4",
        )
        self.assertEqual(len(docs1), 1)


@unittest.skipUnless(mango.has_text_service(), "requires text service")
class IndexCrudTextTests(mango.DbPerClass):
    def setUp(self):
        self.db.recreate()

    def test_create_text_idx(self):
        fields = [
            {"name": "stringidx", "type": "string"},
            {"name": "booleanidx", "type": "boolean"},
        ]
        ret = self.db.create_text_index(fields=fields, name="text_idx_01")
        assert ret is True
        for idx in self.db.list_indexes():
            if idx["name"] != "text_idx_01":
                continue
            self.assertEqual(
                idx["def"]["fields"],
                [{"stringidx": "string"}, {"booleanidx": "boolean"}],
            )
            return
        raise AssertionError("index not created")

    def test_create_bad_text_idx(self):
        bad_fields = [
            True,
            False,
            "bing",
            2.0,
            ["foo", "bar"],
            [{"name": "foo2"}],
            [{"name": "foo3", "type": "garbage"}],
            [{"type": "number"}],
            [{"name": "age", "type": "number"}, {"name": "bad"}],
            [{"name": "age", "type": "number"}, "bla"],
            [{"name": "", "type": "number"}, "bla"],
        ]
        for fields in bad_fields:
            try:
                self.db.create_text_index(fields=fields)
            except Exception as e:
                self.assertEqual(e.response.status_code, 400)
            else:
                raise AssertionError("bad create text index")

    def test_limit_skip_index(self):
        fields = ["field1"]
        ret = self.db.create_index(fields, name="idx_01")
        assert ret is True

        fields = ["field2"]
        ret = self.db.create_index(fields, name="idx_02")
        assert ret is True

        fields = ["field3"]
        ret = self.db.create_index(fields, name="idx_03")
        assert ret is True

        fields = ["field4"]
        ret = self.db.create_index(fields, name="idx_04")
        assert ret is True

        fields = [
            {"name": "stringidx", "type": "string"},
            {"name": "booleanidx", "type": "boolean"},
        ]
        ret = self.db.create_text_index(fields=fields, name="idx_05")
        assert ret is True

        self.assertEqual(len(self.db.list_indexes(limit=2)), 2)
        self.assertEqual(len(self.db.list_indexes(limit=5, skip=4)), 2)
        self.assertEqual(len(self.db.list_indexes(skip=5)), 1)
        self.assertEqual(len(self.db.list_indexes(skip=6)), 0)
        self.assertEqual(len(self.db.list_indexes(skip=100)), 0)
        self.assertEqual(len(self.db.list_indexes(limit=10000000)), 6)

        try:
            self.db.list_indexes(skip=-1)
        except Exception as e:
            self.assertEqual(e.response.status_code, 500)

        try:
            self.db.list_indexes(limit=0)
        except Exception as e:
            self.assertEqual(e.response.status_code, 500)