<commit_before># Copyright 2012-2015 TxMongo Developers
#
# 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.

"""Test the collection module.
Based on pymongo driver's test_collection.py
"""

from bson.son import SON
from pymongo import errors

from twisted.internet import defer
from twisted.trial import unittest

import txmongo

from txmongo import filter
from txmongo.collection import Collection

mongo_host = "localhost"
mongo_port = 27017


class TestIndexInfo(unittest.TestCase):
    
    timeout = 5

    @defer.inlineCallbacks
    def setUp(self):
        self.conn = yield txmongo.MongoConnection(mongo_host, mongo_port)
        self.db = self.conn.mydb
        self.coll = self.db.mycol

    @defer.inlineCallbacks
    def tearDown(self):
        yield self.coll.drop()
        yield self.conn.disconnect()

    @defer.inlineCallbacks
    def test_collection(self):
        self.assertRaises(TypeError, Collection, self.db, 5)

        def make_col(base, name):
            return base[name]

        self.assertRaises(errors.InvalidName, make_col, self.db, "")
        self.assertRaises(errors.InvalidName, make_col, self.db, "te$t")
        self.assertRaises(errors.InvalidName, make_col, self.db, ".test")
        self.assertRaises(errors.InvalidName, make_col, self.db, "test.")
        self.assertRaises(errors.InvalidName, make_col, self.db, "tes..t")
        self.assertRaises(errors.InvalidName, make_col, self.db.test, "")
        self.assertRaises(errors.InvalidName, make_col, self.db.test, "te$t")
        self.assertRaises(errors.InvalidName, make_col, self.db.test, ".test")
        self.assertRaises(errors.InvalidName, make_col, self.db.test, "test.")
        self.assertRaises(errors.InvalidName, make_col, self.db.test, "tes..t")
        self.assertRaises(errors.InvalidName, make_col, self.db.test, "tes\x00t")
        self.assertRaises(TypeError, self.coll.save, "test")
        self.assertFailure(self.coll.filemd5("test"), ValueError)
        self.assertFailure(self.db.test.find(spec="test"), TypeError)
        self.assertFailure(self.db.test.find(fields="test"), TypeError)
        self.assertFailure(self.db.test.find(skip="test"), TypeError)
        self.assertFailure(self.db.test.find(limit="test"), TypeError)
        self.assertFailure(self.db.test.insert([1]), TypeError)
        self.assertFailure(self.db.test.insert(1), TypeError)
        self.assertFailure(self.db.test.update(1, 1), TypeError)
        self.assertFailure(self.db.test.update({}, 1), TypeError)
        self.assertFailure(self.db.test.update({}, {}, 'a'), TypeError)

        self.assert_(isinstance(self.db.test, Collection))
        self.assertEqual(NotImplemented, self.db.test.__cmp__(7))
        self.assertNotEqual(cmp(self.db.test, 7), 0)
        self.assertEqual(self.db.test, Collection(self.db, "test"))
        self.assertEqual(self.db.test.mike, self.db["test.mike"])
        self.assertEqual(self.db.test["mike"], self.db["test.mike"])
        self.assertEqual(repr(self.db.test), "Collection(mydb, test)")
        self.assertEqual(self.db.test.test, self.db.test("test"))

        options = yield self.db.test.options()
        self.assertTrue(isinstance(options, dict))

        yield self.db.drop_collection("test")
        collection_names = yield self.db.collection_names()
        self.assertFalse("test" in collection_names)

    @defer.inlineCallbacks
    def test_create_index(self):
        db = self.db
        coll = self.coll

        self.assertFailure(coll.create_index(5), TypeError)
        self.assertFailure(coll.create_index({"hello": 1}), TypeError)

        yield coll.insert({'c': 1})  # make sure collection exists.

        yield coll.drop_indexes()
        count = yield db.system.indexes.count({"ns": u"mydb.mycol"})
        self.assertEqual(count, 1)

        yield coll.create_index(filter.sort(filter.ASCENDING("hello")))
        yield coll.create_index(filter.sort(filter.ASCENDING("hello") +
                                filter.DESCENDING("world")))

        count = yield db.system.indexes.count({"ns": u"mydb.mycol"}) 
        self.assertEqual(count, 3)

        yield coll.drop_indexes()
        ix = yield coll.create_index(filter.sort(filter.ASCENDING("hello") +
                                     filter.DESCENDING("world")), name="hello_world")
        self.assertEquals(ix, "hello_world")

        yield coll.drop_indexes()
        count = yield db.system.indexes.count({"ns": u"mydb.mycol"}) 
        self.assertEqual(count, 1)
        
        yield coll.create_index(filter.sort(filter.ASCENDING("hello")))
        indices = yield db.system.indexes.find({"ns": u"mydb.mycol"}) 
        self.assert_(u"hello_1" in [a["name"] for a in indices])

        yield coll.drop_indexes()
        count = yield db.system.indexes.count({"ns": u"mydb.mycol"}) 
        self.assertEqual(count, 1)

        ix = yield coll.create_index(filter.sort(filter.ASCENDING("hello") +
                                     filter.DESCENDING("world")))
        self.assertEquals(ix, "hello_1_world_-1")
    
    @defer.inlineCallbacks
    def test_create_index_nodup(self):
        coll = self.coll

        yield coll.drop()
        yield coll.insert({'b': 1})
        yield coll.insert({'b': 1})

        ix = coll.create_index(filter.sort(filter.ASCENDING("b")), unique=True)
        yield self.assertFailure(ix, errors.DuplicateKeyError)

    @defer.inlineCallbacks
    def test_create_index_dropdups(self):
        # dropDups was removed from MongoDB v3.0
        ismaster = yield self.db.command("ismaster")
        if ismaster["maxWireVersion"] >= 3:
            raise unittest.SkipTest()

        yield self.coll.drop()
        yield self.coll.insert([{'b': 1}, {'b': 1}])

        ix = yield self.coll.create_index(filter.sort(filter.ASCENDING('b')),<commit_msg>Added description for SkipTest to calm DeprecationWarning<commit_after># Copyright 2012-2015 TxMongo Developers
#
# 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.

"""Test the collection module.
Based on pymongo driver's test_collection.py
"""

from bson.son import SON
from pymongo import errors

from twisted.internet import defer
from twisted.trial import unittest

import txmongo

from txmongo import filter
from txmongo.collection import Collection

mongo_host = "localhost"
mongo_port = 27017


class TestIndexInfo(unittest.TestCase):
    
    timeout = 5

    @defer.inlineCallbacks
    def setUp(self):
        self.conn = yield txmongo.MongoConnection(mongo_host, mongo_port)
        self.db = self.conn.mydb
        self.coll = self.db.mycol

    @defer.inlineCallbacks
    def tearDown(self):
        yield self.coll.drop()
        yield self.conn.disconnect()

    @defer.inlineCallbacks
    def test_collection(self):
        self.assertRaises(TypeError, Collection, self.db, 5)

        def make_col(base, name):
            return base[name]

        self.assertRaises(errors.InvalidName, make_col, self.db, "")
        self.assertRaises(errors.InvalidName, make_col, self.db, "te$t")
        self.assertRaises(errors.InvalidName, make_col, self.db, ".test")
        self.assertRaises(errors.InvalidName, make_col, self.db, "test.")
        self.assertRaises(errors.InvalidName, make_col, self.db, "tes..t")
        self.assertRaises(errors.InvalidName, make_col, self.db.test, "")
        self.assertRaises(errors.InvalidName, make_col, self.db.test, "te$t")
        self.assertRaises(errors.InvalidName, make_col, self.db.test, ".test")
        self.assertRaises(errors.InvalidName, make_col, self.db.test, "test.")
        self.assertRaises(errors.InvalidName, make_col, self.db.test, "tes..t")
        self.assertRaises(errors.InvalidName, make_col, self.db.test, "tes\x00t")
        self.assertRaises(TypeError, self.coll.save, "test")
        self.assertFailure(self.coll.filemd5("test"), ValueError)
        self.assertFailure(self.db.test.find(spec="test"), TypeError)
        self.assertFailure(self.db.test.find(fields="test"), TypeError)
        self.assertFailure(self.db.test.find(skip="test"), TypeError)
        self.assertFailure(self.db.test.find(limit="test"), TypeError)
        self.assertFailure(self.db.test.insert([1]), TypeError)
        self.assertFailure(self.db.test.insert(1), TypeError)
        self.assertFailure(self.db.test.update(1, 1), TypeError)
        self.assertFailure(self.db.test.update({}, 1), TypeError)
        self.assertFailure(self.db.test.update({}, {}, 'a'), TypeError)

        self.assert_(isinstance(self.db.test, Collection))
        self.assertEqual(NotImplemented, self.db.test.__cmp__(7))
        self.assertNotEqual(cmp(self.db.test, 7), 0)
        self.assertEqual(self.db.test, Collection(self.db, "test"))
        self.assertEqual(self.db.test.mike, self.db["test.mike"])
        self.assertEqual(self.db.test["mike"], self.db["test.mike"])
        self.assertEqual(repr(self.db.test), "Collection(mydb, test)")
        self.assertEqual(self.db.test.test, self.db.test("test"))

        options = yield self.db.test.options()
        self.assertTrue(isinstance(options, dict))

        yield self.db.drop_collection("test")
        collection_names = yield self.db.collection_names()
        self.assertFalse("test" in collection_names)

    @defer.inlineCallbacks
    def test_create_index(self):
        db = self.db
        coll = self.coll

        self.assertFailure(coll.create_index(5), TypeError)
        self.assertFailure(coll.create_index({"hello": 1}), TypeError)

        yield coll.insert({'c': 1})  # make sure collection exists.

        yield coll.drop_indexes()
        count = yield db.system.indexes.count({"ns": u"mydb.mycol"})
        self.assertEqual(count, 1)

        yield coll.create_index(filter.sort(filter.ASCENDING("hello")))
        yield coll.create_index(filter.sort(filter.ASCENDING("hello") +
                                filter.DESCENDING("world")))

        count = yield db.system.indexes.count({"ns": u"mydb.mycol"}) 
        self.assertEqual(count, 3)

        yield coll.drop_indexes()
        ix = yield coll.create_index(filter.sort(filter.ASCENDING("hello") +
                                     filter.DESCENDING("world")), name="hello_world")
        self.assertEquals(ix, "hello_world")

        yield coll.drop_indexes()
        count = yield db.system.indexes.count({"ns": u"mydb.mycol"}) 
        self.assertEqual(count, 1)
        
        yield coll.create_index(filter.sort(filter.ASCENDING("hello")))
        indices = yield db.system.indexes.find({"ns": u"mydb.mycol"}) 
        self.assert_(u"hello_1" in [a["name"] for a in indices])

        yield coll.drop_indexes()
        count = yield db.system.indexes.count({"ns": u"mydb.mycol"}) 
        self.assertEqual(count, 1)

        ix = yield coll.create_index(filter.sort(filter.ASCENDING("hello") +
                                     filter.DESCENDING("world")))
        self.assertEquals(ix, "hello_1_world_-1")
    
    @defer.inlineCallbacks
    def test_create_index_nodup(self):
        coll = self.coll

        yield coll.drop()
        yield coll.insert({'b': 1})
        yield coll.insert({'b': 1})

        ix = coll.create_index(filter.sort(filter.ASCENDING("b")), unique=True)
        yield self.assertFailure(ix, errors.DuplicateKeyError)

    @defer.inlineCallbacks
    def test_create_index_dropdups(self):
        # dropDups was removed from MongoDB v3.0
        ismaster = yield self.db.command("ismaster")
        if ismaster["maxWireVersion"] >= 3:
            raise unittest.SkipTest("dropDups was removed from MongoDB 3")

        yield self.coll.drop()
        yield self.coll.insert([{'b': 1}, {'b': 1}])

        ix = yield self.coll.create_index(filter.sort(filter.ASCENDING('b')),
{'fraction_non_alphanumeric': 0.13073954477435615, 'fraction_numerical': 0.00748208238166496, 'mean_word_length': 3.0672645739910314, 'pattern_counts': {'":': 12, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7626310', 'n_tokens_mistral': 4013, 'n_tokens_neox': 3864, 'n_words': 778}