collection.py

SQLAlchemy. 经典的Python ORM框架。学习必看。

            assert_eq()

            try:
                direct |= [e, creator()]
                assert False
            except TypeError:
                assert True

        if hasattr(direct, 'clear'):
            addall(creator(), creator())
            direct.clear()
            control.clear()
            assert_eq()

        if hasattr(direct, 'difference_update'):
            zap()
            e = creator()
            addall(creator(), creator())
            values = set([creator()])

            direct.difference_update(values)
            control.difference_update(values)
            assert_eq()
            values.update(set([e, creator()]))
            direct.difference_update(values)
            control.difference_update(values)
            assert_eq()

        if hasattr(direct, '__isub__'):
            zap()
            e = creator()
            addall(creator(), creator())
            values = set([creator()])

            direct -= values
            control -= values
            assert_eq()
            values.update(set([e, creator()]))
            direct -= values
            control -= values
            assert_eq()

            values = set([creator()])
            obj.attr -= values
            control -= values
            assert_eq()

            try:
                direct -= [e, creator()]
                assert False
            except TypeError:
                assert True

        if hasattr(direct, 'intersection_update'):
            zap()
            e = creator()
            addall(e, creator(), creator())
            values = set(control)

            direct.intersection_update(values)
            control.intersection_update(values)
            assert_eq()

            values.update(set([e, creator()]))
            direct.intersection_update(values)
            control.intersection_update(values)
            assert_eq()

        if hasattr(direct, '__iand__'):
            zap()
            e = creator()
            addall(e, creator(), creator())
            values = set(control)

            direct &= values
            control &= values
            assert_eq()

            values.update(set([e, creator()]))
            direct &= values
            control &= values
            assert_eq()

            values.update(set([creator()]))
            obj.attr &= values
            control &= values
            assert_eq()

            try:
                direct &= [e, creator()]
                assert False
            except TypeError:
                assert True

        if hasattr(direct, 'symmetric_difference_update'):
            zap()
            e = creator()
            addall(e, creator(), creator())

            values = set([e, creator()])
            direct.symmetric_difference_update(values)
            control.symmetric_difference_update(values)
            assert_eq()

            e = creator()
            addall(e)
            values = set([e])
            direct.symmetric_difference_update(values)
            control.symmetric_difference_update(values)
            assert_eq()

            values = set()
            direct.symmetric_difference_update(values)
            control.symmetric_difference_update(values)
            assert_eq()

        if hasattr(direct, '__ixor__'):
            zap()
            e = creator()
            addall(e, creator(), creator())

            values = set([e, creator()])
            direct ^= values
            control ^= values
            assert_eq()

            e = creator()
            addall(e)
            values = set([e])
            direct ^= values
            control ^= values
            assert_eq()

            values = set()
            direct ^= values
            control ^= values
            assert_eq()

            values = set([creator()])
            obj.attr ^= values
            control ^= values
            assert_eq()

            try:
                direct ^= [e, creator()]
                assert False
            except TypeError:
                assert True

    def _test_set_bulk(self, typecallable, creator=entity_maker):
        class Foo(object):
            pass

        canary = Canary()
        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'attr', True, extension=canary,
                                   typecallable=typecallable, useobject=True)

        obj = Foo()
        direct = obj.attr

        e1 = creator()
        obj.attr.add(e1)

        like_me = typecallable()
        e2 = creator()
        like_me.add(e2)

        self.assert_(obj.attr is direct)
        obj.attr = like_me
        self.assert_(obj.attr is not direct)
        self.assert_(obj.attr is not like_me)
        self.assert_(obj.attr == set([e2]))
        self.assert_(e1 in canary.removed)
        self.assert_(e2 in canary.added)

        e3 = creator()
        real_set = set([e3])
        obj.attr = real_set
        self.assert_(obj.attr is not real_set)
        self.assert_(obj.attr == set([e3]))
        self.assert_(e2 in canary.removed)
        self.assert_(e3 in canary.added)

        e4 = creator()
        try:
            obj.attr = [e4]
            self.assert_(False)
        except TypeError:
            self.assert_(e4 not in canary.data)
            self.assert_(e3 in canary.data)

    def test_set(self):
        self._test_adapter(set)
        self._test_set(set)
        self._test_set_bulk(set)

    def test_set_subclass(self):
        class MySet(set):
            pass
        self._test_adapter(MySet)
        self._test_set(MySet)
        self._test_set_bulk(MySet)
        self.assert_(getattr(MySet, '_sa_instrumented') == id(MySet))

    def test_set_duck(self):
        class SetLike(object):
            def __init__(self):
                self.data = set()
            def add(self, item):
                self.data.add(item)
            def remove(self, item):
                self.data.remove(item)
            def discard(self, item):
                self.data.discard(item)
            def pop(self):
                return self.data.pop()
            def update(self, other):
                self.data.update(other)
            def __iter__(self):
                return iter(self.data)
            def __eq__(self, other):
                return self.data == other

        self._test_adapter(SetLike)
        self._test_set(SetLike)
        self._test_set_bulk(SetLike)
        self.assert_(getattr(SetLike, '_sa_instrumented') == id(SetLike))

    def test_set_emulates(self):
        class SetIsh(object):
            __emulates__ = py_set
            def __init__(self):
                self.data = set()
            def add(self, item):
                self.data.add(item)
            def remove(self, item):
                self.data.remove(item)
            def discard(self, item):
                self.data.discard(item)
            def pop(self):
                return self.data.pop()
            def update(self, other):
                self.data.update(other)
            def __iter__(self):
                return iter(self.data)
            def __eq__(self, other):
                return self.data == other

        self._test_adapter(SetIsh)
        self._test_set(SetIsh)
        self._test_set_bulk(SetIsh)
        self.assert_(getattr(SetIsh, '_sa_instrumented') == id(SetIsh))

    def _test_dict(self, typecallable, creator=dictable_entity):
        class Foo(object):
            pass

        canary = Canary()
        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'attr', True, extension=canary,
                                   typecallable=typecallable, useobject=True)

        obj = Foo()
        adapter = collections.collection_adapter(obj.attr)
        direct = obj.attr
        control = dict()

        def assert_eq():
            self.assert_(set(direct.values()) == canary.data)
            self.assert_(set(adapter) == canary.data)
            self.assert_(direct == control)

        def addall(*values):
            for item in values:
                direct.set(item)
                control[item.a] = item
            assert_eq()
        def zap():
            for item in list(adapter):
                direct.remove(item)
            control.clear()

        # assume an 'set' method is available for tests
        addall(creator())

        if hasattr(direct, '__setitem__'):
            e = creator()
            direct[e.a] = e
            control[e.a] = e
            assert_eq()

            e = creator(e.a, e.b)
            direct[e.a] = e
            control[e.a] = e
            assert_eq()

        if hasattr(direct, '__delitem__'):
            e = creator()
            addall(e)

            del direct[e.a]
            del control[e.a]
            assert_eq()

            e = creator()
            try:
                del direct[e.a]
            except KeyError:
                self.assert_(e not in canary.removed)

        if hasattr(direct, 'clear'):
            addall(creator(), creator(), creator())

            direct.clear()
            control.clear()
            assert_eq()

            direct.clear()
            control.clear()
            assert_eq()

        if hasattr(direct, 'pop'):
            e = creator()
            addall(e)

            direct.pop(e.a)
            control.pop(e.a)
            assert_eq()

            e = creator()
            try:
                direct.pop(e.a)
            except KeyError:
                self.assert_(e not in canary.removed)

        if hasattr(direct, 'popitem'):
            zap()
            e = creator()
            addall(e)

            direct.popitem()
            control.popitem()
            assert_eq()

        if hasattr(direct, 'setdefault'):
            e = creator()

            val_a = direct.setdefault(e.a, e)
            val_b = control.setdefault(e.a, e)
            assert_eq()
            self.assert_(val_a is val_b)

            val_a = direct.setdefault(e.a, e)
            val_b = control.setdefault(e.a, e)
            assert_eq()
            self.assert_(val_a is val_b)

        if hasattr(direct, 'update'):
            e = creator()
            d = dict([(ee.a, ee) for ee in [e, creator(), creator()]])
            addall(e, creator())

            direct.update(d)
            control.update(d)
            assert_eq()

            if sys.version_info >= (2, 4):
                kw = dict([(ee.a, ee) for ee in [e, creator()]])
                direct.update(**kw)
                control.update(**kw)
                assert_eq()

    def _test_dict_bulk(self, typecallable, creator=dictable_entity):
        class Foo(object):
            pass

        canary = Canary()
        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'attr', True, extension=canary,
                                   typecallable=typecallable, useobject=True)

        obj = Foo()
        direct = obj.attr

        e1 = creator()
        collections.collection_adapter(direct).append_with_event(e1)

        like_me = typecallable()
        e2 = creator()
        like_me.set(e2)

        self.assert_(obj.attr is direct)
        obj.attr = like_me
        self.assert_(obj.attr is not direct)
        self.assert_(obj.attr is not like_me)
        self.assert_(set(collections.collection_adapter(obj.attr)) == set([e2]))
        self.assert_(e1 in canary.removed)
        self.assert_(e2 in canary.added)


        # key validity on bulk assignment is a basic feature of MappedCollection
        # but is not present in basic, @converter-less dict collections.
        e3 = creator()
        if isinstance(obj.attr, collections.MappedCollection):
            real_dict = dict(badkey=e3)
            try:
                obj.attr = real_dict
                self.assert_(False)
            except TypeError:
                pass
            self.assert_(obj.attr is not real_dict)
            self.assert_('badkey' not in obj.attr)
            self.assertEquals(set(collections.collection_adapter(obj.attr)),
                              set([e2]))
            self.assert_(e3 not in canary.added)
        else:
            real_dict = dict(keyignored1=e3)
            obj.attr = real_dict
            self.assert_(obj.attr is not real_dict)
            self.assert_('keyignored1' not in obj.attr)
            self.assertEquals(set(collections.collection_adapter(obj.attr)),
                              set([e3]))
            self.assert_(e2 in canary.removed)
            self.assert_(e3 in canary.added)

        obj.attr = typecallable()
        self.assertEquals(list(collections.collection_adapter(obj.attr)), [])

        e4 = creator()
        try:
            obj.attr = [e4]
            self.assert_(False)
        except TypeError:
            self.assert_(e4 not in canary.data)

    def test_dict(self):
        try:
            self._test_adapter(dict, dictable_entity,
                               to_set=lambda c: set(c.values()))
            self.assert_(False)
        except exceptions.ArgumentError, e:
            self.assert_(e.args[0] == 'Type InstrumentedDict must elect an appender method to be a collection class')

        try:
            self._test_dict(dict)
            self.assert_(False)
        except exceptions.ArgumentError, e:
            self.assert_(e.args[0] == 'Type InstrumentedDict must elect an appender method to be a collection class')

    def test_dict_subclass(self):
        class MyDict(dict):
            @collection.appender
            @collection.internally_instrumented
            def set(self, item, _sa_initiator=None):
                self.__setitem__(item.a, item, _sa_initiator=_sa_initiator)
            @collection.remover
            @collection.internally_instrumented
            def _remove(self, item, _sa_initiator=None):
                self.__delitem__(item.a, _sa_initiator=_sa_initiator)

        self._test_adapter(MyDict, dictable_entity,
                           to_set=lambda c: set(c.values()))
        self._test_dict(MyDict)
        self._test_dict_bulk(MyDict)
        self.assert_(getattr(MyDict, '_sa_instrumented') == id(MyDict))

    def test_dict_subclass2(self):
        class MyEasyDict(collections.MappedCollection):
            def __init__(self):
                super(MyEasyDict, self).__init__(lambda e: e.a)

        self._test_adapter(MyEasyDict, dictable_entity,
                           to_set=lambda c: set(c.values()))
        self._test_dict(MyEasyDict)
        self._test_dict_bulk(MyEasyDict)
        self.assert_(getattr(MyEasyDict, '_sa_instrumented') == id(MyEasyDict))

    def test_dict_subclass3(self):
        class MyOrdered(util.OrderedDict, collections.MappedCollection):
            def __init__(self):