execute the passed query against the passed model and check the output compare the results of a raw query against expected results all values on the model are equal this includes checking that they are the same type the results of a raw query contain no annotations the passed raw query results contain the expected annotations basic test of raw query with a simple database query raw queries are lazy they aren t actually executed until they re iterated over test of a simple raw query against a model containing a foreign key test of a simple raw query against a model containing a field with dbcolumn defined a raw query with a model that has a pk dbcolumn with mixed case test of raw raw query s tolerance for columns being returned in any order test of raw query s optional ability to translate unexpected result column names to specific model fields test passing optional query parameters test passing optional query parameters test representation of raw query with parameters test of a simple raw query against a model containing a m2m field extra translations are ignored query select from rawquery translations something else s objects all self assertsuccessfulrawquery query s translationstranslations def testmissingfieldsself query select id firstname dob from rawquery for in objects rawquery self assertisnotnone firstname lastname isn t given but it will be retrieved on demand self assertisnotnone lastname def testmissingfieldswithoutpkself query select firstname dob from rawquery msg raw query must include the primary key with self assertraisesmessagefielddoesnotexist msg list objects rawquery def testannotationsself query select a countb id as bookcount from rawquery a left join rawquerybook b on a id b id group by a id a firstname a lastname a dob order by a id expectedannotations bookcount 3 bookcount 0 bookcount 1 bookcount 0 s objects orderbypk self assertsuccessfulrawquery query s expectedannotations def testwhitespacequeryself query select from rawquery s objects all self assertsuccessfulrawquery query s def testmultipleiterationsself query select from rawquery normals objects all raws objects rawquery first iteration firstiterations 0 for index raw in enumerateraws self assertequalnormalsindex raw firstiterations 1 second iteration seconditerations 0 for index raw in enumerateraws self assertequalnormalsindex raw seconditerations 1 self assertequalfirstiterations seconditerations def testgetitemself indexing on rawquerysets query select from rawquery order by id asc third objects rawquery2 self assertequalthird firstname bob firsttwo objects rawquery0 2 self assertequallenfirsttwo 2 with self assertraisestypeerror objects rawquerytest def testinheritanceself f friendly objects create firstnamewesley lastnamechun dobdate1962 10 28 query select from rawqueryfriendly self assertequalo pk for o in friendly objects rawquery f pk def testquerycountself self assertnumqueries 1 list objects rawselect from rawquery def testsubqueryinrawsqlself list book objects raw select id from select from rawquerybook where paperback is not null sq def testdbcolumnnameisusedinrawqueryself b bookfkaspk objects createbookself b1 self assertequal list bookfkaspk objects raw select notthedefault from rawquerybookfkaspk b def testdecimalparameterself c coffee objects createbrandstarbucks price20 5 qs coffee objects raw select from rawquerycoffee where price s paramsdecimal20 self assertequallistqs c def testresultcachingself with self assertnumqueries1 books book objects rawselect from rawquerybook listbooks listbooks def testiteratorself with self assertnumqueries2 books book objects rawselect from rawquerybook listbooks iterator listbooks iterator def testboolself self assertisboolbook objects rawselect from rawquerybook true self assertis boolbook objects rawselect from rawquerybook where id 0 false def testlenself self assertequallenbook objects rawselect from rawquerybook 4 self assertequal lenbook objects rawselect from rawquerybook where id 0 0 execute the passed query against the passed model and check the output compare the results of a raw query against expected results all values on the model are equal this includes checking that they are the same type the results of a raw query contain no annotations the passed raw query results contain the expected annotations basic test of raw query with a simple database query raw queries are lazy they aren t actually executed until they re iterated over test of a simple raw query against a model containing a foreign key test of a simple raw query against a model containing a field with db_column defined a raw query with a model that has a pk db_column with mixed case test of raw raw query s tolerance for columns being returned in any order test of raw query s optional ability to translate unexpected result column names to specific model fields test passing optional query parameters test passing optional query parameters test representation of raw query with parameters test of a simple raw query against a model containing a m2m field extra translations are ignored last_name isn t given but it will be retrieved on demand first iteration second iteration indexing on rawquerysets regression test that ensures the column attribute on the field is used to generate the list of fields included in the query as opposed to the attname this is important when the primary key is a foreignkey field because attname and column are not necessarily the same
from datetime import date
from decimal import Decimal

from django.core.exceptions import FieldDoesNotExist
from django.db.models.query import RawQuerySet
from django.test import TestCase, skipUnlessDBFeature

from .models import (
    Author,
    Book,
    BookFkAsPk,
    Coffee,
    FriendlyAuthor,
    MixedCaseIDColumn,
    Reviewer,
)


class RawQueryTests(TestCase):
    @classmethod
    def setUpTestData(cls):
        cls.a1 = Author.objects.create(
            first_name="Joe", last_name="Smith", dob=date(1950, 9, 20)
        )
        cls.a2 = Author.objects.create(
            first_name="Jill", last_name="Doe", dob=date(1920, 4, 2)
        )
        cls.a3 = Author.objects.create(
            first_name="Bob", last_name="Smith", dob=date(1986, 1, 25)
        )
        cls.a4 = Author.objects.create(
            first_name="Bill", last_name="Jones", dob=date(1932, 5, 10)
        )
        cls.b1 = Book.objects.create(
            title="The awesome book",
            author=cls.a1,
            paperback=False,
            opening_line=(
                "It was a bright cold day in April and the clocks were striking "
                "thirteen."
            ),
        )
        cls.b2 = Book.objects.create(
            title="The horrible book",
            author=cls.a1,
            paperback=True,
            opening_line=(
                "On an evening in the latter part of May a middle-aged man "
                "was walking homeward from Shaston to the village of Marlott, "
                "in the adjoining Vale of Blakemore, or Blackmoor."
            ),
        )
        cls.b3 = Book.objects.create(
            title="Another awesome book",
            author=cls.a1,
            paperback=False,
            opening_line="A squat gray building of only thirty-four stories.",
        )
        cls.b4 = Book.objects.create(
            title="Some other book",
            author=cls.a3,
            paperback=True,
            opening_line="It was the day my grandmother exploded.",
        )
        cls.c1 = Coffee.objects.create(brand="dunkin doughnuts")
        cls.c2 = Coffee.objects.create(brand="starbucks")
        cls.r1 = Reviewer.objects.create()
        cls.r2 = Reviewer.objects.create()
        cls.r1.reviewed.add(cls.b2, cls.b3, cls.b4)

    def assertSuccessfulRawQuery(
        self,
        model,
        query,
        expected_results,
        expected_annotations=(),
        params=[],
        translations=None,
    ):
        
        results = list(
            model.objects.raw(query, params=params, translations=translations)
        )
        self.assertProcessed(model, results, expected_results, expected_annotations)
        self.assertAnnotations(results, expected_annotations)

    def assertProcessed(self, model, results, orig, expected_annotations=()):
        
        self.assertEqual(len(results), len(orig))
        for index, item in enumerate(results):
            orig_item = orig[index]
            for annotation in expected_annotations:
                setattr(orig_item, *annotation)

            for field in model._meta.fields:
                
                self.assertEqual(
                    getattr(item, field.attname), getattr(orig_item, field.attname)
                )
                
                self.assertEqual(
                    type(getattr(item, field.attname)),
                    type(getattr(orig_item, field.attname)),
                )

    def assertNoAnnotations(self, results):
        
        self.assertAnnotations(results, ())

    def assertAnnotations(self, results, expected_annotations):
        
        if expected_annotations:
            for index, result in enumerate(results):
                annotation, value = expected_annotations[index]
                self.assertTrue(hasattr(result, annotation))
                self.assertEqual(getattr(result, annotation), value)

    def test_rawqueryset_repr(self):
        queryset = RawQuerySet(raw_query="SELECT * FROM raw_query_author")
        self.assertEqual(
            repr(queryset), "<RawQuerySet: SELECT * FROM raw_query_author>"
        )
        self.assertEqual(
            repr(queryset.query), "<RawQuery: SELECT * FROM raw_query_author>"
        )

    def test_simple_raw_query(self):
        
        query = "SELECT * FROM raw_query_author"
        authors = Author.objects.all()
        self.assertSuccessfulRawQuery(Author, query, authors)

    def test_raw_query_lazy(self):
        
        q = Author.objects.raw("SELECT * FROM raw_query_author")
        self.assertIsNone(q.query.cursor)
        list(q)
        self.assertIsNotNone(q.query.cursor)

    def test_FK_raw_query(self):
        
        query = "SELECT * FROM raw_query_book"
        books = Book.objects.all()
        self.assertSuccessfulRawQuery(Book, query, books)

    def test_db_column_handler(self):
        
        query = "SELECT * FROM raw_query_coffee"
        coffees = Coffee.objects.all()
        self.assertSuccessfulRawQuery(Coffee, query, coffees)

    def test_pk_with_mixed_case_db_column(self):
        
        query = "SELECT * FROM raw_query_mixedcaseidcolumn"
        queryset = MixedCaseIDColumn.objects.all()
        self.assertSuccessfulRawQuery(MixedCaseIDColumn, query, queryset)

    def test_order_handler(self):
        
        selects = (
            ("dob, last_name, first_name, id"),
            ("last_name, dob, first_name, id"),
            ("first_name, last_name, dob, id"),
        )

        for select in selects:
            query = "SELECT %s FROM raw_query_author" % select
            authors = Author.objects.all()
            self.assertSuccessfulRawQuery(Author, query, authors)

    def test_translations(self):
        
        query = (
            "SELECT first_name AS first, last_name AS last, dob, id "
            "FROM raw_query_author"
        )
        translations = {"first": "first_name", "last": "last_name"}
        authors = Author.objects.all()
        self.assertSuccessfulRawQuery(Author, query, authors, translations=translations)

    def test_params(self):
        
        query = "SELECT * FROM raw_query_author WHERE first_name = %s"
        author = Author.objects.all()[2]
        params = [author.first_name]
        qset = Author.objects.raw(query, params=params)
        results = list(qset)
        self.assertProcessed(Author, results, [author])
        self.assertNoAnnotations(results)
        self.assertEqual(len(results), 1)
        self.assertIsInstance(repr(qset), str)

    def test_params_none(self):
        query = "SELECT * FROM raw_query_author WHERE first_name like 'J%'"
        qset = Author.objects.raw(query, params=None)
        self.assertEqual(len(qset), 2)

    def test_escaped_percent(self):
        query = "SELECT * FROM raw_query_author WHERE first_name like 'J%%'"
        qset = Author.objects.raw(query)
        self.assertEqual(len(qset), 2)

    @skipUnlessDBFeature("supports_paramstyle_pyformat")
    def test_pyformat_params(self):
        
        query = "SELECT * FROM raw_query_author WHERE first_name = %(first)s"
        author = Author.objects.all()[2]
        params = {"first": author.first_name}
        qset = Author.objects.raw(query, params=params)
        results = list(qset)
        self.assertProcessed(Author, results, [author])
        self.assertNoAnnotations(results)
        self.assertEqual(len(results), 1)
        self.assertIsInstance(repr(qset), str)

    def test_query_representation(self):
        
        query = "SELECT * FROM raw_query_author WHERE last_name = %(last)s"
        qset = Author.objects.raw(query, {"last": "foo"})
        self.assertEqual(
            repr(qset),
            "<RawQuerySet: SELECT * FROM raw_query_author WHERE last_name = foo>",
        )
        self.assertEqual(
            repr(qset.query),
            "<RawQuery: SELECT * FROM raw_query_author WHERE last_name = foo>",
        )

        query = "SELECT * FROM raw_query_author WHERE last_name = %s"
        qset = Author.objects.raw(query, {"foo"})
        self.assertEqual(
            repr(qset),
            "<RawQuerySet: SELECT * FROM raw_query_author WHERE last_name = foo>",
        )
        self.assertEqual(
            repr(qset.query),
            "<RawQuery: SELECT * FROM raw_query_author WHERE last_name = foo>",
        )

    def test_many_to_many(self):
        
        query = "SELECT * FROM raw_query_reviewer"
        reviewers = Reviewer.objects.all()
        self.assertSuccessfulRawQuery(Reviewer, query, reviewers)

    def test_extra_conversions(self):
        
        query = "SELECT * FROM raw_query_author"
        translations = {"something": "else"}
        authors = Author.objects.all()
        self.assertSuccessfulRawQuery(Author, query, authors, translations=translations)

    def test_missing_fields(self):
        query = "SELECT id, first_name, dob FROM raw_query_author"
        for author in Author.objects.raw(query):
            self.assertIsNotNone(author.first_name)
            
            self.assertIsNotNone(author.last_name)

    def test_missing_fields_without_PK(self):
        query = "SELECT first_name, dob FROM raw_query_author"
        msg = "Raw query must include the primary key"
        with self.assertRaisesMessage(FieldDoesNotExist, msg):
            list(Author.objects.raw(query))

    def test_annotations(self):
        query = (
            "SELECT a.*, count(b.id) as book_count "
            "FROM raw_query_author a "
            "LEFT JOIN raw_query_book b ON a.id = b.author_id "
            "GROUP BY a.id, a.first_name, a.last_name, a.dob ORDER BY a.id"
        )
        expected_annotations = (
            ("book_count", 3),
            ("book_count", 0),
            ("book_count", 1),
            ("book_count", 0),
        )
        authors = Author.objects.order_by("pk")
        self.assertSuccessfulRawQuery(Author, query, authors, expected_annotations)

    def test_white_space_query(self):
        query = "    SELECT * FROM raw_query_author"
        authors = Author.objects.all()
        self.assertSuccessfulRawQuery(Author, query, authors)

    def test_multiple_iterations(self):
        query = "SELECT * FROM raw_query_author"
        normal_authors = Author.objects.all()
        raw_authors = Author.objects.raw(query)

        
        first_iterations = 0
        for index, raw_author in enumerate(raw_authors):
            self.assertEqual(normal_authors[index], raw_author)
            first_iterations += 1

        
        second_iterations = 0
        for index, raw_author in enumerate(raw_authors):
            self.assertEqual(normal_authors[index], raw_author)
            second_iterations += 1

        self.assertEqual(first_iterations, second_iterations)

    def test_get_item(self):
        
        query = "SELECT * FROM raw_query_author ORDER BY id ASC"
        third_author = Author.objects.raw(query)[2]
        self.assertEqual(third_author.first_name, "Bob")

        first_two = Author.objects.raw(query)[0:2]
        self.assertEqual(len(first_two), 2)

        with self.assertRaises(TypeError):
            Author.objects.raw(query)["test"]

    def test_inheritance(self):
        f = FriendlyAuthor.objects.create(
            first_name="Wesley", last_name="Chun", dob=date(1962, 10, 28)
        )
        query = "SELECT * FROM raw_query_friendlyauthor"
        self.assertEqual([o.pk for o in FriendlyAuthor.objects.raw(query)], [f.pk])

    def test_query_count(self):
        self.assertNumQueries(
            1, list, Author.objects.raw("SELECT * FROM raw_query_author")
        )

    def test_subquery_in_raw_sql(self):
        list(
            Book.objects.raw(
                "SELECT id FROM "
                "(SELECT * FROM raw_query_book WHERE paperback IS NOT NULL) sq"
            )
        )

    def test_db_column_name_is_used_in_raw_query(self):
        
        b = BookFkAsPk.objects.create(book=self.b1)
        self.assertEqual(
            list(
                BookFkAsPk.objects.raw(
                    "SELECT not_the_default FROM raw_query_bookfkaspk"
                )
            ),
            [b],
        )

    def test_decimal_parameter(self):
        c = Coffee.objects.create(brand="starbucks", price=20.5)
        qs = Coffee.objects.raw(
            "SELECT * FROM raw_query_coffee WHERE price >= %s", params=[Decimal(20)]
        )
        self.assertEqual(list(qs), [c])

    def test_result_caching(self):
        with self.assertNumQueries(1):
            books = Book.objects.raw("SELECT * FROM raw_query_book")
            list(books)
            list(books)

    def test_iterator(self):
        with self.assertNumQueries(2):
            books = Book.objects.raw("SELECT * FROM raw_query_book")
            list(books.iterator())
            list(books.iterator())

    def test_bool(self):
        self.assertIs(bool(Book.objects.raw("SELECT * FROM raw_query_book")), True)
        self.assertIs(
            bool(Book.objects.raw("SELECT * FROM raw_query_book WHERE id = 0")), False
        )

    def test_len(self):
        self.assertEqual(len(Book.objects.raw("SELECT * FROM raw_query_book")), 4)
        self.assertEqual(
            len(Book.objects.raw("SELECT * FROM raw_query_book WHERE id = 0")), 0
        )