import datetime

import pickle

import sys

import unittest

from django.conf import settings

from django.core.exceptions import FieldError

from django.db import DatabaseError, connection, connections, DEFAULT_DB_ALIAS

from django.db.models import Count

from django.db.models.query import Q, ITER_CHUNK_SIZE, EmptyQuerySet

from django.test import TestCase

from django.utils.datastructures import SortedDict

from models import (Annotation, Article, Author, Celebrity, Child, Cover, Detail,

    DumbCategory, ExtraInfo, Fan, Item, LeafA, LoopX, LoopZ, ManagedModel,

    Member, NamedCategory, Note, Number, Plaything, PointerA, Ranking, Related,

    Report, ReservedName, Tag, TvChef, Valid, X)

class BaseQuerysetTest(TestCase):

    def assertValueQuerysetEqual(self, qs, values):

        return self.assertQuerysetEqual(qs, values, transform=lambda x: x)

    def assertRaisesMessage(self, exc, msg, func, *args, **kwargs):

        try:

            func(*args, **kwargs)

        except Exception, e:

            self.assertEqual(msg, str(e))

            self.assertTrue(isinstance(e, exc), "Expected %s, got %s" % (exc, type(e)))

        else:

            if hasattr(exc, '__name__'):

                excName = exc.__name__

            else:

                excName = str(exc)

            raise AssertionError, "%s not raised" % excName

class Queries1Tests(BaseQuerysetTest):

    def setUp(self):

        generic = NamedCategory.objects.create(name="Generic")

        self.t1 = Tag.objects.create(name='t1', category=generic)

        self.t2 = Tag.objects.create(name='t2', parent=self.t1, category=generic)

        self.t3 = Tag.objects.create(name='t3', parent=self.t1)

        t4 = Tag.objects.create(name='t4', parent=self.t3)

        self.t5 = Tag.objects.create(name='t5', parent=self.t3)

        self.n1 = Note.objects.create(note='n1', misc='foo', id=1)

        n2 = Note.objects.create(note='n2', misc='bar', id=2)

        self.n3 = Note.objects.create(note='n3', misc='foo', id=3)

        ann1 = Annotation.objects.create(name='a1', tag=self.t1)

        ann1.notes.add(self.n1)

        ann2 = Annotation.objects.create(name='a2', tag=t4)

        ann2.notes.add(n2, self.n3)

        # Create these out of order so that sorting by 'id' will be different to sorting

        # by 'info'. Helps detect some problems later.

        self.e2 = ExtraInfo.objects.create(info='e2', note=n2)

        e1 = ExtraInfo.objects.create(info='e1', note=self.n1)

        self.a1 = Author.objects.create(name='a1', num=1001, extra=e1)

        self.a2 = Author.objects.create(name='a2', num=2002, extra=e1)

        a3 = Author.objects.create(name='a3', num=3003, extra=self.e2)

        self.a4 = Author.objects.create(name='a4', num=4004, extra=self.e2)

        self.time1 = datetime.datetime(2007, 12, 19, 22, 25, 0)

        self.time2 = datetime.datetime(2007, 12, 19, 21, 0, 0)

        time3 = datetime.datetime(2007, 12, 20, 22, 25, 0)

        time4 = datetime.datetime(2007, 12, 20, 21, 0, 0)

        self.i1 = Item.objects.create(name='one', created=self.time1, modified=self.time1, creator=self.a1, note=self.n3)

        self.i1.tags = [self.t1, self.t2]

        self.i2 = Item.objects.create(name='two', created=self.time2, creator=self.a2, note=n2)

        self.i2.tags = [self.t1, self.t3]

        self.i3 = Item.objects.create(name='three', created=time3, creator=self.a2, note=self.n3)

        i4 = Item.objects.create(name='four', created=time4, creator=self.a4, note=self.n3)

        i4.tags = [t4]

        self.r1 = Report.objects.create(name='r1', creator=self.a1)

        Report.objects.create(name='r2', creator=a3)

        Report.objects.create(name='r3')

        # Ordering by 'rank' gives us rank2, rank1, rank3. Ordering by the Meta.ordering

        # will be rank3, rank2, rank1.

        self.rank1 = Ranking.objects.create(rank=2, author=self.a2)

        Cover.objects.create(title="first", item=i4)

        Cover.objects.create(title="second", item=self.i2)

    def test_ticket1050(self):

        self.assertQuerysetEqual(

            Item.objects.filter(tags__isnull=True),

            ['<Item: three>']

        )

        self.assertQuerysetEqual(

            Item.objects.filter(tags__id__isnull=True),

            ['<Item: three>']

        )

    def test_ticket1801(self):

        self.assertQuerysetEqual(

            Author.objects.filter(item=self.i2),

            ['<Author: a2>']

        )

        self.assertQuerysetEqual(

            Author.objects.filter(item=self.i3),

            ['<Author: a2>']

        )

        self.assertQuerysetEqual(

            Author.objects.filter(item=self.i2) & Author.objects.filter(item=self.i3),

            ['<Author: a2>']

        )

    def test_ticket2306(self):

        # Checking that no join types are "left outer" joins.

        query = Item.objects.filter(tags=self.t2).query

        self.assertTrue(query.LOUTER not in [x[2] for x in query.alias_map.values()])

        self.assertQuerysetEqual(

            Item.objects.filter(Q(tags=self.t1)).order_by('name'),

            ['<Item: one>', '<Item: two>']

        )

        self.assertQuerysetEqual(

            Item.objects.filter(Q(tags=self.t1)).filter(Q(tags=self.t2)),

            ['<Item: one>']

        )

        self.assertQuerysetEqual(

            Item.objects.filter(Q(tags=self.t1)).filter(Q(creator__name='fred')|Q(tags=self.t2)),

            ['<Item: one>']

        )

        # Each filter call is processed "at once" against a single table, so this is

        # different from the previous example as it tries to find tags that are two

        # things at once (rather than two tags).

        self.assertQuerysetEqual(

            Item.objects.filter(Q(tags=self.t1) & Q(tags=self.t2)),

            []

        )

        self.assertQuerysetEqual(

            Item.objects.filter(Q(tags=self.t1), Q(creator__name='fred')|Q(tags=self.t2)),

            []

        )

        qs = Author.objects.filter(ranking__rank=2, ranking__id=self.rank1.id)

        self.assertQuerysetEqual(list(qs), ['<Author: a2>'])

        self.assertEqual(2, qs.query.count_active_tables(), 2)

        qs = Author.objects.filter(ranking__rank=2).filter(ranking__id=self.rank1.id)

        self.assertEqual(qs.query.count_active_tables(), 3)

    def test_ticket4464(self):

        self.assertQuerysetEqual(

            Item.objects.filter(tags=self.t1).filter(tags=self.t2),

            ['<Item: one>']

        )

        self.assertQuerysetEqual(

            Item.objects.filter(tags__in=[self.t1, self.t2]).distinct().order_by('name'),

            ['<Item: one>', '<Item: two>']

        )

        self.assertQuerysetEqual(

            Item.objects.filter(tags__in=[self.t1, self.t2]).filter(tags=self.t3),

            ['<Item: two>']

        )

        # Make sure .distinct() works with slicing (this was broken in Oracle).

        self.assertQuerysetEqual(

            Item.objects.filter(tags__in=[self.t1, self.t2]).order_by('name')[:3],

            ['<Item: one>', '<Item: one>', '<Item: two>']

        )

        self.assertQuerysetEqual(

            Item.objects.filter(tags__in=[self.t1, self.t2]).distinct().order_by('name')[:3],

            ['<Item: one>', '<Item: two>']

        )

    def test_tickets_2080_3592(self):

        self.assertQuerysetEqual(

            Author.objects.filter(item__name='one') | Author.objects.filter(name='a3'),

            ['<Author: a1>', '<Author: a3>']

        )

        self.assertQuerysetEqual(

            Author.objects.filter(Q(item__name='one') | Q(name='a3')),

            ['<Author: a1>', '<Author: a3>']

        )

        self.assertQuerysetEqual(

            Author.objects.filter(Q(name='a3') | Q(item__name='one')),

            ['<Author: a1>', '<Author: a3>']

        )

        self.assertQuerysetEqual(

            Author.objects.filter(Q(item__name='three') | Q(report__name='r3')),

            ['<Author: a2>']

        )

    def test_ticket6074(self):

        # Merging two empty result sets shouldn't leave a queryset with no constraints

        # (which would match everything).

        self.assertQuerysetEqual(Author.objects.filter(Q(id__in=[])), [])

        self.assertQuerysetEqual(

            Author.objects.filter(Q(id__in=[])|Q(id__in=[])),

            []

        )

    def test_tickets_1878_2939(self):

        self.assertEqual(Item.objects.values('creator').distinct().count(), 3)

        # Create something with a duplicate 'name' so that we can test multi-column

        # cases (which require some tricky SQL transformations under the covers).

        xx = Item(name='four', created=self.time1, creator=self.a2, note=self.n1)

        xx.save()

        self.assertEqual(

            Item.objects.exclude(name='two').values('creator', 'name').distinct().count(),

            4

        )

        self.assertEqual(

            Item.objects.exclude(name='two').extra(select={'foo': '%s'}, select_params=(1,)).values('creator', 'name', 'foo').distinct().count(),

            4

        )

        self.assertEqual(

            Item.objects.exclude(name='two').extra(select={'foo': '%s'}, select_params=(1,)).values('creator', 'name').distinct().count(),

            4

        )

        xx.delete()

    def test_ticket7323(self):

        self.assertEqual(Item.objects.values('creator', 'name').count(), 4)

    def test_ticket2253(self):

        q1 = Item.objects.order_by('name')

        q2 = Item.objects.filter(id=self.i1.id)

        self.assertQuerysetEqual(

            q1,

            ['<Item: four>', '<Item: one>', '<Item: three>', '<Item: two>']

        )

        self.assertQuerysetEqual(q2, ['<Item: one>'])

        self.assertQuerysetEqual(

            (q1 | q2).order_by('name'),

            ['<Item: four>', '<Item: one>', '<Item: three>', '<Item: two>']

        )

        self.assertQuerysetEqual((q1 & q2).order_by('name'), ['<Item: one>'])

        # FIXME: This is difficult to fix and very much an edge case, so punt for now.

        # This is related to the order_by() tests, below, but the old bug exhibited

        # itself here (q2 was pulling too many tables into the combined query with the

        # new ordering, but only because we have evaluated q2 already).

        #

        #self.assertEqual(len((q1 & q2).order_by('name').query.tables), 1)

        q1 = Item.objects.filter(tags=self.t1)

        q2 = Item.objects.filter(note=self.n3, tags=self.t2)

        q3 = Item.objects.filter(creator=self.a4)

        self.assertQuerysetEqual(

            ((q1 & q2) | q3).order_by('name'),

            ['<Item: four>', '<Item: one>']

        )

    def test_tickets_4088_4306(self):

        self.assertQuerysetEqual(

            Report.objects.filter(creator=1001),

            ['<Report: r1>']

        )

        self.assertQuerysetEqual(

            Report.objects.filter(creator__num=1001),

            ['<Report: r1>']

        )

        self.assertQuerysetEqual(Report.objects.filter(creator__id=1001), [])

        self.assertQuerysetEqual(

            Report.objects.filter(creator__id=self.a1.id),

            ['<Report: r1>']

        )

        self.assertQuerysetEqual(

            Report.objects.filter(creator__name='a1'),

            ['<Report: r1>']

        )

    def test_ticket4510(self):

        self.assertQuerysetEqual(

            Author.objects.filter(report__name='r1'),

            ['<Author: a1>']

        )

    def test_ticket7378(self):

        self.assertQuerysetEqual(self.a1.report_set.all(), ['<Report: r1>'])

    def test_tickets_5324_6704(self):

        self.assertQuerysetEqual(

            Item.objects.filter(tags__name='t4'),

            ['<Item: four>']

        )

        self.assertQuerysetEqual(

            Item.objects.exclude(tags__name='t4').order_by('name').distinct(),

            ['<Item: one>', '<Item: three>', '<Item: two>']

        )

        self.assertQuerysetEqual(

            Item.objects.exclude(tags__name='t4').order_by('name').distinct().reverse(),

            ['<Item: two>', '<Item: three>', '<Item: one>']

        )

        self.assertQuerysetEqual(

            Author.objects.exclude(item__name='one').distinct().order_by('name'),

            ['<Author: a2>', '<Author: a3>', '<Author: a4>']

        )

        # Excluding across a m2m relation when there is more than one related

        # object associated was problematic.

        self.assertQuerysetEqual(

            Item.objects.exclude(tags__name='t1').order_by('name'),

            ['<Item: four>', '<Item: three>']

        )

        self.assertQuerysetEqual(

            Item.objects.exclude(tags__name='t1').exclude(tags__name='t4'),

            ['<Item: three>']

        )

        # Excluding from a relation that cannot be NULL should not use outer joins.

        query = Item.objects.exclude(creator__in=[self.a1, self.a2]).query

        self.assertTrue(query.LOUTER not in [x[2] for x in query.alias_map.values()])

        # Similarly, when one of the joins cannot possibly, ever, involve NULL

        # values (Author -> ExtraInfo, in the following), it should never be

        # promoted to a left outer join. So the following query should only

        # involve one "left outer" join (Author -> Item is 0-to-many).

        qs = Author.objects.filter(id=self.a1.id).filter(Q(extra__note=self.n1)|Q(item__note=self.n3))

        self.assertEqual(

            len([x[2] for x in qs.query.alias_map.values() if x[2] == query.LOUTER and qs.query.alias_refcount[x[1]]]),

            1

        )

        # The previous changes shouldn't affect nullable foreign key joins.

        self.assertQuerysetEqual(

            Tag.objects.filter(parent__isnull=True).order_by('name'),

            ['<Tag: t1>']

        )

        self.assertQuerysetEqual(

            Tag.objects.exclude(parent__isnull=True).order_by('name'),

            ['<Tag: t2>', '<Tag: t3>', '<Tag: t4>', '<Tag: t5>']

        )

        self.assertQuerysetEqual(

            Tag.objects.exclude(Q(parent__name='t1') | Q(parent__isnull=True)).order_by('name'),

            ['<Tag: t4>', '<Tag: t5>']

        )

        self.assertQuerysetEqual(

            Tag.objects.exclude(Q(parent__isnull=True) | Q(parent__name='t1')).order_by('name'),

            ['<Tag: t4>', '<Tag: t5>']

        )

        self.assertQuerysetEqual(

            Tag.objects.exclude(Q(parent__parent__isnull=True)).order_by('name'),

            ['<Tag: t4>', '<Tag: t5>']

        )

        self.assertQuerysetEqual(

            Tag.objects.filter(~Q(parent__parent__isnull=True)).order_by('name'),

            ['<Tag: t4>', '<Tag: t5>']

        )

    def test_ticket2091(self):

        t = Tag.objects.get(name='t4')

        self.assertQuerysetEqual(

            Item.objects.filter(tags__in=[t]),

            ['<Item: four>']

        )

    def test_heterogeneous_qs_combination(self):

        # Combining querysets built on different models should behave in a well-defined

        # fashion. We raise an error.

        self.assertRaisesMessage(

            AssertionError,

            'Cannot combine queries on two different base models.',

            lambda: Author.objects.all() & Tag.objects.all()

        )

        self.assertRaisesMessage(

            AssertionError,

            'Cannot combine queries on two different base models.',

            lambda: Author.objects.all() | Tag.objects.all()

        )

    def test_ticket3141(self):

        self.assertEqual(Author.objects.extra(select={'foo': '1'}).count(), 4)

        self.assertEqual(

            Author.objects.extra(select={'foo': '%s'}, select_params=(1,)).count(),

            4

        )

    def test_ticket2400(self):

        self.assertQuerysetEqual(

            Author.objects.filter(item__isnull=True),

            ['<Author: a3>']

        )

        self.assertQuerysetEqual(

            Tag.objects.filter(item__isnull=True),

            ['<Tag: t5>']

        )

    def test_ticket2496(self):

        self.assertQuerysetEqual(

            Item.objects.extra(tables=['queries_author']).select_related().order_by('name')[:1],

            ['<Item: four>']

        )

    def test_tickets_2076_7256(self):

        # Ordering on related tables should be possible, even if the table is

        # not otherwise involved.

        self.assertQuerysetEqual(

            Item.objects.order_by('note__note', 'name'),

            ['<Item: two>', '<Item: four>', '<Item: one>', '<Item: three>']

        )

        # Ordering on a related field should use the remote model's default

        # ordering as a final step.

        self.assertQuerysetEqual(

            Author.objects.order_by('extra', '-name'),

            ['<Author: a2>', '<Author: a1>', '<Author: a4>', '<Author: a3>']

        )

        # Using remote model default ordering can span multiple models (in this

        # case, Cover is ordered by Item's default, which uses Note's default).

        self.assertQuerysetEqual(

            Cover.objects.all(),

            ['<Cover: first>', '<Cover: second>']

        )

        # If the remote model does not have a default ordering, we order by its 'id'

        # field.

        self.assertQuerysetEqual(

            Item.objects.order_by('creator', 'name'),

            ['<Item: one>', '<Item: three>', '<Item: two>', '<Item: four>']

        )

        # Ordering by a many-valued attribute (e.g. a many-to-many or reverse

        # ForeignKey) is legal, but the results might not make sense. That

        # isn't Django's problem. Garbage in, garbage out.

        self.assertQuerysetEqual(

            Item.objects.filter(tags__isnull=False).order_by('tags', 'id'),

            ['<Item: one>', '<Item: two>', '<Item: one>', '<Item: two>', '<Item: four>']

        )

        # If we replace the default ordering, Django adjusts the required

        # tables automatically. Item normally requires a join with Note to do

        # the default ordering, but that isn't needed here.

        qs = Item.objects.order_by('name')

        self.assertQuerysetEqual(

            qs,

            ['<Item: four>', '<Item: one>', '<Item: three>', '<Item: two>']

        )

        self.assertEqual(len(qs.query.tables), 1)

    def test_tickets_2874_3002(self):

        qs = Item.objects.select_related().order_by('note__note', 'name')

        self.assertQuerysetEqual(

            qs,

            ['<Item: two>', '<Item: four>', '<Item: one>', '<Item: three>']

        )

        # This is also a good select_related() test because there are multiple

        # Note entries in the SQL. The two Note items should be different.

        self.assertTrue(repr(qs[0].note), '<Note: n2>')

        self.assertEqual(repr(qs[0].creator.extra.note), '<Note: n1>')

    def test_ticket3037(self):

        self.assertQuerysetEqual(

            Item.objects.filter(Q(creator__name='a3', name='two')|Q(creator__name='a4', name='four')),

            ['<Item: four>']

        )

    def test_tickets_5321_7070(self):

        # Ordering columns must be included in the output columns. Note that

        # this means results that might otherwise be distinct are not (if there

        # are multiple values in the ordering cols), as in this example. This

        # isn't a bug; it's a warning to be careful with the selection of

        # ordering columns.

        self.assertValueQuerysetEqual(

            Note.objects.values('misc').distinct().order_by('note', '-misc'),

            [{'misc': u'foo'}, {'misc': u'bar'}, {'misc': u'foo'}]

        )

    def test_ticket4358(self):

        # If you don't pass any fields to values(), relation fields are

        # returned as "foo_id" keys, not "foo". For consistency, you should be

        # able to pass "foo_id" in the fields list and have it work, too. We

        # actually allow both "foo" and "foo_id".

        # The *_id version is returned by default.

        self.assertTrue('note_id' in ExtraInfo.objects.values()[0])

        # You can also pass it in explicitly.

        self.assertValueQuerysetEqual(

            ExtraInfo.objects.values('note_id'),

            [{'note_id': 1}, {'note_id': 2}]

        )

        # ...or use the field name.

        self.assertValueQuerysetEqual(

            ExtraInfo.objects.values('note'),

            [{'note': 1}, {'note': 2}]

        )

    def test_ticket2902(self):

        # Parameters can be given to extra_select, *if* you use a SortedDict.

        # (First we need to know which order the keys fall in "naturally" on

        # your system, so we can put things in the wrong way around from

        # normal. A normal dict would thus fail.)

        s = [('a', '%s'), ('b', '%s')]

        params = ['one', 'two']

        if {'a': 1, 'b': 2}.keys() == ['a', 'b']:

            s.reverse()

            params.reverse()

        # This slightly odd comparison works around the fact that PostgreSQL will

        # return 'one' and 'two' as strings, not Unicode objects. It's a side-effect of

        # using constants here and not a real concern.

        d = Item.objects.extra(select=SortedDict(s), select_params=params).values('a', 'b')[0]

        self.assertEqual(d, {'a': u'one', 'b': u'two'})

        # Order by the number of tags attached to an item.

        l = Item.objects.extra(select={'count': 'select count(*) from queries_item_tags where queries_item_tags.item_id = queries_item.id'}).order_by('-count')

        self.assertEqual([o.count for o in l], [2, 2, 1, 0])

    def test_ticket6154(self):

        # Multiple filter statements are joined using "AND" all the time.

        self.assertQuerysetEqual(

            Author.objects.filter(id=self.a1.id).filter(Q(extra__note=self.n1)|Q(item__note=self.n3)),

            ['<Author: a1>']

        )

        self.assertQuerysetEqual(

                Author.objects.filter(Q(extra__note=self.n1)|Q(item__note=self.n3)).filter(id=self.a1.id),

            ['<Author: a1>']

        )

    def test_ticket6981(self):

        self.assertQuerysetEqual(

            Tag.objects.select_related('parent').order_by('name'),

            ['<Tag: t1>', '<Tag: t2>', '<Tag: t3>', '<Tag: t4>', '<Tag: t5>']

        )