Source

djsqlalchemy / djsqlalchemy / tests / __init__.py

Full commit
from django import test
from django.db import transaction

from djsqlalchemy.models import Category
from djsqlalchemy.models import Item

from sqlalchemy.sql import select
from djsqlalchemy.api import get_engine
from djsqlalchemy.api import get_tables
from djsqlalchemy.api import get_one


class MapperTest(test.TestCase):
    def test_simple_queries(self):
        engine = get_engine()

        tables = get_tables()
        item_table = tables['djsqlalchemy_item']
        category_table = tables['djsqlalchemy_category']

        category = Category.objects.create(name='category')
        item = Item.objects.create(name='item1', category1=category,
            category2=category, category3=category)
        transaction.commit()

        s = item_table.select()
        self.assertTrue(bool(engine.execute(s).fetchall()))

        s = category_table.select()
        self.assertTrue(bool(engine.execute(s).fetchall()))

    def test_mapper(self):
        tables = get_tables()
        item_table = tables['djsqlalchemy_item']
        category_table = tables['djsqlalchemy_category']

        category1 = Category.objects.create(name='category1')
        category2 = Category.objects.create(name='category2')
        category3 = Category.objects.create(name='category3')

        item = Item.objects.create(name='item1', category1=category1,
            category2=category2, category3=category3)

        transaction.commit()

        category1_table = category_table.alias('category1')
        category2_table = category_table.alias('category2')
        category3_table = category_table.alias('category3')
        s = select([item_table, category1_table, category2_table, category3_table],
            from_obj=[
                item_table.join(category1_table,
                                category1_table.c.id == item_table.c.category1_id)
                          .join(category2_table,
                                category2_table.c.id == item_table.c.category2_id)
                          .join(category3_table,
                                category3_table.c.id == item_table.c.category3_id)
            ], use_labels=True)

        item_obj = get_one(s, Item)
        self.assertEqual(item_obj, item)
        self.assertEqual(item_obj._category1_cache, category1)
        self.assertEqual(item_obj._category2_cache, category2)
        self.assertEqual(item_obj._category3_cache, category3)

        s = select([category_table], use_labels=True)
        self.assertRaises(Category.MultipleObjectsReturned, get_one,
                          s, Category)

        s = select([category_table], category_table.c.id == 999,
                   use_labels=True)
        self.assertRaises(Category.DoesNotExist, get_one,
                          s, Category)

    def test_manager(self):
        sample_category = Category.objects.create(name='sample category')
        sample_item = Item.objects.create(name='sample item',
                                          category2=sample_category)

        item = Item.complex_query.get()
        self.assertEqual(item, sample_item)
        self.assertEqual(item._category2_cache, sample_category)