src/mango/test/04-key-tests.py

# -*- coding: latin-1 -*-
# 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 mango
import unittest

TEST_DOCS = [
    {"type": "complex_key", "title": "normal key"},
    {
        "type": "complex_key",
        "title": "key with dot",
        "dot.key": "dot's value",
        "none": {"dot": "none dot's value"},
        "name.first": "Kvothe",
    },
    {
        "type": "complex_key",
        "title": "key with peso",
        "$key": "peso",
        "deep": {"$key": "deep peso"},
        "name": {"first": "Master Elodin"},
    },
    {"type": "complex_key", "title": "unicode key", "": "apple"},
    {
        "title": "internal_fields_format",
        "utf8-1[]:string": "string",
        "utf8-2[]:boolean[]": True,
        "utf8-3[]:number": 9,
        "utf8-3[]:null": None,
    },
]


@unittest.skipUnless(mango.has_text_service(), "requires text service")
class KeyTests(mango.DbPerClass):
    @classmethod
    def setUpClass(klass):
        super(KeyTests, klass).setUpClass()
        klass.db.save_docs(TEST_DOCS, w=3)
        klass.db.create_index(["type"], ddoc="view")
        if mango.has_text_service():
            klass.db.create_text_index(ddoc="text")

    def run_check(self, query, check, fields=None, indexes=None):
        if indexes is None:
            indexes = ["view", "text"]
        for idx in indexes:
            docs = self.db.find(query, fields=fields, use_index=idx)
            check(docs)

    def test_dot_key(self):
        query = {"type": "complex_key"}
        fields = ["title", "dot\\.key", "none.dot"]

        def check(docs):
            assert len(docs) == 4
            assert "dot.key" in docs[1]
            assert docs[1]["dot.key"] == "dot's value"
            assert "none" in docs[1]
            assert docs[1]["none"]["dot"] == "none dot's value"

        self.run_check(query, check, fields=fields)

    def test_peso_key(self):
        query = {"type": "complex_key"}
        fields = ["title", "$key", "deep.$key"]

        def check(docs):
            assert len(docs) == 4
            assert "$key" in docs[2]
            assert docs[2]["$key"] == "peso"
            assert "deep" in docs[2]
            assert docs[2]["deep"]["$key"] == "deep peso"

        self.run_check(query, check, fields=fields)

    def test_unicode_in_fieldname(self):
        query = {"type": "complex_key"}
        fields = ["title", ""]

        def check(docs):
            assert len(docs) == 4
            # note:  == \uf8ff
            assert "\uf8ff" in docs[3]
            assert docs[3]["\uf8ff"] == "apple"

        self.run_check(query, check, fields=fields)

    # The rest of these tests are only run against the text
    # indexes because view indexes don't have to worry about
    # field *name* escaping in the index.

    def test_unicode_in_selector_field(self):
        query = {"": "apple"}

        def check(docs):
            assert len(docs) == 1
            assert docs[0]["\uf8ff"] == "apple"

        self.run_check(query, check, indexes=["text"])

    def test_internal_field_tests(self):
        queries = [
            {"utf8-1[]:string": "string"},
            {"utf8-2[]:boolean[]": True},
            {"utf8-3[]:number": 9},
            {"utf8-3[]:null": None},
        ]

        def check(docs):
            assert len(docs) == 1
            assert docs[0]["title"] == "internal_fields_format"

        for query in queries:
            self.run_check(query, check, indexes=["text"])

    def test_escape_period(self):
        query = {"name\\.first": "Kvothe"}

        def check(docs):
            assert len(docs) == 1
            assert docs[0]["name.first"] == "Kvothe"

        self.run_check(query, check, indexes=["text"])

        query = {"name.first": "Kvothe"}

        def check_empty(docs):
            assert len(docs) == 0

        self.run_check(query, check_empty, indexes=["text"])

    def test_object_period(self):
        query = {"name.first": "Master Elodin"}

        def check(docs):
            assert len(docs) == 1
            assert docs[0]["title"] == "key with peso"

        self.run_check(query, check, indexes=["text"])

        query = {"name\\.first": "Master Elodin"}

        def check_empty(docs):
            assert len(docs) == 0

        self.run_check(query, check_empty, indexes=["text"])