# -*- 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"])