Why Gemfury? Push, build, and install  RubyGems npm packages Python packages Maven artifacts PHP packages Go Modules Debian packages RPM packages NuGet packages

Repository URL to install this package:

Details    
mongokit / tests / test_descriptors.py
Size: Mime:
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (c) 2009-2011, Nicolas Clairon
# All rights reserved.
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
#     * Redistributions of source code must retain the above copyright
#       notice, this list of conditions and the following disclaimer.
#     * Redistributions in binary form must reproduce the above copyright
#       notice, this list of conditions and the following disclaimer in the
#       documentation and/or other materials provided with the distribution.
#     * Neither the name of the University of California, Berkeley nor the
#       names of its contributors may be used to endorse or promote products
#       derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND ANY
# EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
# DISCLAIMED. IN NO EVENT SHALL THE REGENTS AND CONTRIBUTORS BE LIABLE FOR ANY
# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

import unittest

from mongokit import *

class DescriptorsTestCase(unittest.TestCase):
    def setUp(self):
        self.connection = Connection()
        self.col = self.connection['test']['mongokit']

    def tearDown(self):
        self.connection.drop_database('test')

    def test_duplicate_required(self):
        failed = False
        try:
            class MyDoc(Document):
                structure = {"foo":unicode}
                required_fields = ["foo", "foo"]
        except DuplicateRequiredError, e:
            self.assertEqual(str(e), "duplicate required_fields : ['foo', 'foo']")
            failed = True
        self.assertEqual(failed, True)

    def test_flat_required(self):
        class MyDoc(Document):
            structure = {
                "foo":unicode,
            }
            required_fields = ["foo"]
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        self.assertRaises(RequireFieldError, mydoc.validate )
        mydoc['foo'] = u'bla'
        mydoc.validate()

    def test_nested_required(self):
        class MyDoc(Document):
            structure = {
                "bla":{
                    "foo":unicode,
                },
            }
            required_fields = ["bla.foo"]
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        self.assertRaises(RequireFieldError, mydoc.validate )
        mydoc['bla']['foo'] = u'bla'
        mydoc.validate()

    def test_list_required(self):
        class MyDoc(Document):
            structure = {
                "foo":[]
            }
            required_fields = ["foo"]
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        self.assertRaises(RequireFieldError, mydoc.validate )
        mydoc['foo'] = [1,2,3]
        mydoc.validate()

    def test_dict_required2(self):
        class MyDoc(Document):
            structure = {
                "foo":dict
            }
            required_fields = ["foo"]
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        self.assertRaises(RequireFieldError, mydoc.validate )
        mydoc['foo'] = {u"3":[u'bla']}
        mydoc.validate()

    def test_dict_required(self):
        class MyDoc(Document):
            structure = {
                "foo":{}
            }
            required_fields = ["foo"]
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        self.assertRaises(RequireFieldError, mydoc.validate )
        mydoc['foo'] = {u'bar':u'bla'}
        self.assertRaises(StructureError, mydoc.validate )

    def test_dict_nested_required(self):
        class MyDoc(Document):
            structure = {
                "foo":{unicode:{"bar":int}}
            }
            required_fields = ["foo.$unicode.bar"]
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        self.assertRaises(RequireFieldError, mydoc.validate )

    def test_default_values(self):
        class MyDoc(Document):
            structure = {
                "foo":int,
                "bla":unicode,
            }
            default_values = {"foo":42}
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        assert mydoc["foo"] == 42
        assert mydoc == {'foo':42, 'bla':None}, mydoc

    def test_default_values_nested(self):
        class MyDoc(Document):
            structure = {
                "bar":{
                    "foo":int,
                    "bla":unicode,
                }
            }
            default_values = {"bar.foo":42}
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        assert mydoc['bar']["foo"] == 42
        assert mydoc == {'bar':{'foo':42, 'bla':None}}, mydoc

    def test_default_values_nested_inheritance(self):
        import datetime
        class Core(Document):
            structure = {
                "core":{
                    "creation_date":datetime.datetime,
                }
            }
            default_values = {
                "core.creation_date": datetime.datetime(2010, 1, 1),
            }

        class MyDoc(Core):
            structure = {
                "bar":{
                    "foo":int,
                    "bla":unicode,
                }
            }
            default_values = {"bar.foo":42}

        class MyDoc2(MyDoc):
            structure = {
                "mydoc2":{
                    "toto":int
                }
            }
        self.connection.register([MyDoc2])
        mydoc = self.col.MyDoc2()
        assert mydoc['bar']["foo"] == 42
        assert mydoc == {'mydoc2': {'toto': None}, 'core': {'creation_date': datetime.datetime(2010, 1, 1, 0, 0)}, 'bar': {'foo': 42, 'bla': None}}

    def test_default_values_from_function(self):
        import time
        class MyDoc(Document):
            structure = {
                "foo":float
            }
            default_values = {"foo":time.time}
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        mydoc.validate()

    def test_default_values_from_function2(self):
        import time
        class Doc( Document ):
            structure = {
                "doc":{
                    "creation_date":float,
                    "updated_date": float,
                }
            }
            default_values = {
                "doc.creation_date": time.time,
                "doc.updated_date": time.time
            }
        doc = Doc()
        assert isinstance(doc['doc']['creation_date'], float), doc['doc']['creation_date']
        assert isinstance(doc['doc']['updated_date'], float)

    def test_default_values_from_function_nested(self):
        import time
        class MyDoc(Document):
            structure = {
                "foo":{"bar":float}
            }
            default_values = {"foo.bar":time.time}
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        mydoc.validate()
        assert mydoc['foo']['bar'] > 0

    def _test_default_values_from_function_througt_types(self):
        # XXX TODO
#        class MyDoc(Document):
#            structure = {
#                "foo":{int:float}
#            }
#            default_values = {"foo.$int":time.time}
#        mydoc = MyDoc()
#        mydoc.validate()
#        # can't go through types, because no values
#        assert mydoc['foo'] == {}

        # but
        import time
        class MyDoc(Document):
            structure = {
                "foo":{int:float}
            }
            default_values = {"foo":{3:time.time}}
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        mydoc.validate()
        assert mydoc['foo'][3] > 0

    def test_default_list_values(self):
        class MyDoc(Document):
            structure = {
                "foo":[int]
            }
            default_values = {"foo":[42,3]}
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        assert mydoc["foo"] == [42,3]
        mydoc['foo'] = [1,2,3]
        mydoc.save()
        mydoc = self.col.MyDoc()
        assert mydoc['foo'] == [42,3]

    def test_default_list_values_empty(self):
        class MyDoc(Document):
            structure = {
                "foo":list
            }
            default_values = {"foo":[3]}
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        assert mydoc["foo"] == [3]
        mydoc['foo'].append(2)
        mydoc.save()
        mydoc = self.col.MyDoc()
        assert mydoc['foo'] == [3], mydoc


    def test_default_list_values_with_callable(self):
        def get_truth():
            return 42
        class MyDoc(Document):
            structure = {
                "foo":[int]
            }
            default_values = {"foo":[get_truth,3]}
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        assert mydoc["foo"] == [42,3]
        mydoc.validate()


    def test_default_list_nested_values(self):
        class MyDoc(Document):
            structure = {
                "foo":{
                    "bar":[int]
                }
            }
            default_values = {"foo.bar":[42,3]}
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        assert mydoc["foo"]["bar"] == [42,3]

    def test_default_dict_values(self):
        class MyDoc(Document):
            structure = {
                "foo":dict
            }
            default_values = {"foo":{"bar":42}}
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        assert mydoc["foo"] == {"bar":42}, mydoc
        mydoc['foo'] = {'bar':1}
        mydoc.save()
        mydoc = self.col.MyDoc()
        assert mydoc["foo"] == {"bar":42}, mydoc

    def test_default_dict_values_empty(self):
        class MyDoc(Document):
            structure = {
                "foo":dict
            }
            default_values = {"foo":{}}
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        print id(mydoc.structure['foo']), id(mydoc['foo']), id(mydoc.default_values['foo'])
        assert mydoc["foo"] == {}, mydoc
        mydoc['foo'][u'bar'] = 1
        mydoc.save()
        mydoc2 = self.col.MyDoc()
        print id(mydoc2.structure['foo']), id(mydoc2['foo']), id(mydoc2.default_values['foo'])
        assert mydoc2["foo"] == {}, mydoc

        class MyDoc(Document):
            structure = {
                "foo":{unicode:int}
            }
            default_values = {"foo":{}}
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        assert mydoc["foo"] == {}, mydoc
        mydoc['foo'][u'bar'] = 1
        mydoc.save()
        mydoc2 = self.col.MyDoc()
        assert mydoc2["foo"] == {}, mydoc


    def test_default_dict_values_with_callable(self):
        def get_truth():
            return {'bar':42}
        class MyDoc(Document):
            structure = {
                "foo":{}
            }
            default_values = {"foo":get_truth}
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        assert mydoc["foo"] == {"bar":42}, mydoc

    def test_default_dict_checked_values(self):
        class MyDoc(Document):
            structure = {
                "foo":{unicode:int}
            }
            default_values = {"foo":{u"bar":42}}
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        assert mydoc["foo"] == {"bar":42}, mydoc

    def test_default_dict_nested_checked_values(self):
        class MyDoc(Document):
            structure = {
                "foo":{unicode:{"bla":int, "ble":unicode}}
            }
            default_values = {"foo":{u"bar":{"bla":42, "ble":u"arf"}}}
        mydoc = MyDoc()
        assert mydoc["foo"] == {u"bar":{"bla":42, "ble":u"arf"}}, mydoc

    def test_default_values_with_dict_in_list(self):
        @self.connection.register
        class MyDoc(Document):
            structure = {
                'bar': [{'foo':unicode}]
            }
            default_values = {
                'bar': [{'foo': u'bla'}]
            }
        doc = self.col.MyDoc()
        assert doc['bar'] == [{'foo': u'bla'}]

    def test_validators(self):
        class MyDoc(Document):
            structure = {
                "foo":unicode,
                "bar":{
                    "bla":int
                }
            }
            validators = {
                "foo":lambda x: x.startswith("http://"),
                "bar.bla": lambda x: x > 5
            }
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        mydoc["foo"] = u"google.com"
        self.assertRaises(ValidationError, mydoc.validate)
        mydoc["foo"] = u"http://google.com"
        mydoc.validate()
        mydoc['bar']['bla'] = 2
        self.assertRaises(ValidationError, mydoc.validate)
        mydoc['bar']['bla'] = 42
        mydoc.validate()

    def test_validators_througt_types(self):
        class MyDoc(Document):
            structure = {
                "bar":{
                    int:{"bla":int}
                }
            }
            validators = {
                "bar.$int.bla": lambda x: x > 5
            }
        mydoc = MyDoc()
        mydoc['bar'].update({3:{'bla': 15}})
        self.assertRaises(InvalidDocument, mydoc.validate)

    def test_multiple_validators(self):
        class MyDoc(Document):
            structure = {
                "foo":unicode,
            }
            validators = {
                "foo":[lambda x: x.startswith("http://"),lambda x: x.endswith(".com")],
            }
        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        mydoc["foo"] = u"google.com"
        self.assertRaises(ValidationError, mydoc.validate)
        mydoc["foo"] = u"http://google.fr"
        self.assertRaises(ValidationError, mydoc.validate)
        mydoc["foo"] = u"http://google.com"
        mydoc.validate()

    def test_validators_with_custom_validation_message(self):
        class MinLengthValidator(object):
            def __init__(self, min_length):
                self.min_length = min_length

            def __call__(self, value):
                if len(value) >= self.min_length:
                    return True
                else:
                    raise Exception('%s must be atleast ' + str(self.min_length) + ' characters long.')

        class Client(Document):
            structure = {
              'first_name': unicode
            }
            validators = {
              'first_name': MinLengthValidator(2)
            }
        self.connection.register([Client])
        client = self.col.Client()
        client['first_name'] = u'Georges'
        client.validate()
        client['first_name'] = u'J'
        self.assertRaises(Exception, client.validate)
        message = ""
        try:
            client.validate()
        except Exception, e:
            message = unicode(e)
        assert message == "first_name must be atleast 2 characters long.", message

    def test_complexe_validation(self):
        class MyDoc(Document):
            structure = {
                "foo":unicode,
                "bar":{
                    "bla":int
                }
            }
            def validate(self):
                if self['bar']['bla']:
                    self['foo'] = unicode(self['bar']['bla'])
                else:
                    self['foo'] = None
                super(MyDoc, self).validate()

        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        mydoc['bar']['bla'] = 4
        assert mydoc['foo'] is None
        mydoc.validate()
        assert mydoc['foo'] == "4", mydoc['foo']
        mydoc['bar']['bla'] = None
        mydoc.validate()
        assert mydoc['foo'] is None

    def test_complexe_validation2(self):

        class MyDoc(Document):
            structure = {
                "foo":unicode,
                "bar":{"bla":unicode}
            }
            default_values = {"bar.bla":3}
            def validate(self):
                self["bar"]["bla"] = unicode(self["bar"]["bla"])
                self["foo"] = unicode(self["foo"])
                super(MyDoc, self).validate()

        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        mydoc['foo'] = 4
        mydoc.validate()
        assert mydoc['foo'] == "4", mydoc['foo']
        assert mydoc["bar"]["bla"] == "3", mydoc

    def test_complexe_validation3(self):
        class MyDoc(Document):
            structure = {
                "foo":unicode,
                "bar":{
                    "bla":int
                },
                "ble":unicode,
            }
            def validate(self):
                if self['bar']['bla'] is not None:
                    self['foo'] = unicode(self['bar']['bla'])
                else:
                    self['foo'] = None
                self["ble"] = self["foo"]
                super(MyDoc, self).validate()

        self.connection.register([MyDoc])
        mydoc = self.col.MyDoc()
        mydoc['bar']['bla'] = 4
        assert mydoc['foo'] is None
        mydoc.validate()
        assert mydoc['foo'] == "4"
        assert mydoc["ble"] == "4"
        mydoc['bar']['bla'] = None
        mydoc.validate()
        assert mydoc['foo'] is None
        assert mydoc['ble'] is None

    def test_bad_default_values(self):
        failed = False
        try:
            class MyDoc(Document):
                structure = {
                    "foo":{"bar":int},
                }
                default_values = {"foo.bla":2}
        except ValueError, e:
            failed = True
            self.assertEqual(str(e), "Error in default_values: can't find foo.bla in structure")
        self.assertEqual(failed, True)

    def test_bad_validators(self):
        failed = False
        try:
            class MyDoc(Document):
                structure = {
                    "foo":{"bar":int},
                }
                validators = {"foo.bla":lambda x:x}
        except ValueError, e:
            failed = True
            self.assertEqual(str(e), "Error in validators: can't find foo.bla in structure")
        self.assertEqual(failed, True)

    def test_bad_required(self):
        failed = False
        try:
            class MyDoc(Document):
                db_name = "test"
                collection_name = "mongokit"
                structure = {
                    "profil":{
                        "screen_name":unicode,
                        "age":int
                    }
                }
                required_fields = ['profil.screen_nam']
        except ValueError, e:
            failed = True
            self.assertEqual(str(e), "Error in required_fields: can't find profil.screen_nam in structure")
        self.assertEqual(failed, True)

    def test_nested_structure2(self):
        class MyDoc(Document):
            db_name = "test"
            collection_name = "mongokit"
            structure = {
                unicode:{int:int}
            }

        mydoc = MyDoc()
        assert mydoc._namespaces == ['$unicode', '$unicode.$int']