@@ -0,0 +1,2075 @@
+import sqlalchemy as sa%0Afrom sqlalchemy_utils.aggregates import aggregated%0Afrom tests import TestCase%0A%0A%0Aclass TestAggregatesWithManyToManyRelationships(TestCase):%0A    dns = 'postgres://postgres@localhost/sqlalchemy_utils_test'%0A%0A    def create_models(self):%0A        user_group = sa.Table('user_group', self.Base.metadata,%0A            sa.Column('user_id', sa.Integer, sa.ForeignKey('user.id')),%0A            sa.Column('group_id', sa.Integer, sa.ForeignKey('group.id'))%0A        )%0A%0A        class User(self.Base):%0A            __tablename__ = 'user'%0A            id = sa.Column(sa.Integer, primary_key=True)%0A            name = sa.Column(sa.Unicode(255))%0A%0A            @aggregated('groups', sa.Column(sa.Integer, default=0))%0A            def group_count(self):%0A                return sa.func.count('1')%0A%0A            groups = sa.orm.relationship(%0A                'Group',%0A                backref='users',%0A                secondary=user_group%0A            )%0A%0A        class Group(self.Base):%0A            __tablename__ = 'group'%0A            id = sa.Column(sa.Integer, primary_key=True)%0A            name = sa.Column(sa.Unicode(255))%0A%0A        self.User = User%0A        self.Group = Group%0A%0A    def test_assigns_aggregates_on_insert(self):%0A        user = self.User(%0A            name=u'John Matrix'%0A        )%0A        self.session.add(user)%0A        self.session.commit()%0A        group = self.Group(%0A            name=u'Some group',%0A            users=%5Buser%5D%0A        )%0A        self.session.add(group)%0A        self.session.commit()%0A        self.session.refresh(user)%0A        assert user.group_count == 1%0A%0A    def test_updates_aggregates_on_delete(self):%0A        user = self.User(%0A            name=u'John Matrix'%0A        )%0A        self.session.add(user)%0A        self.session.commit()%0A        group = self.Group(%0A            name=u'Some group',%0A            users=%5Buser%5D%0A        )%0A        self.session.add(group)%0A        self.session.commit()%0A        self.session.refresh(user)%0A        user.groups = %5B%5D%0A        self.session.commit()%0A        self.session.refresh(user)%0A        assert user.group_count == 0%0A

Add tests for many to many aggregates