<s> import mock <EOL> import pickle <EOL> import re <EOL> from unittest import skipUnless <EOL> import pytest <EOL> from django . contrib . auth . models import User <EOL> from django . db . models . query import QuerySet <EOL> from django . template import Context , Template <EOL> from django . test import TestCase <EOL> from django . test . utils import override_settings <EOL> from django_mysql . models import ( <EOL> ApproximateInt , SmartIterator , add_QuerySetMixin <EOL> ) <EOL> from django_mysql . utils import have_program , index_name <EOL> from testapp . models import ( <EOL> Author , AuthorExtra , AuthorMultiIndex , Book , NameAuthor , NameAuthorExtra , <EOL> VanillaAuthor <EOL> ) <EOL> from testapp . utils import CaptureLastQuery , captured_stdout , used_indexes <EOL> class MixinQuerysetTests ( TestCase ) : <EOL> def test_User ( self ) : <EOL> qs = User . objects . all ( ) <EOL> assert not hasattr ( qs , '<STR_LIT>' ) <EOL> qs2 = add_QuerySetMixin ( qs ) <EOL> assert hasattr ( qs2 , '<STR_LIT>' ) <EOL> assert qs . __class__ != qs2 . __class__ <EOL> assert qs . __class__ in qs2 . __class__ . __mro__ <EOL> class ApproximateCountTests ( TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( ApproximateCountTests , self ) . setUp ( ) <EOL> Author . objects . bulk_create ( [ Author ( ) for i in range ( <NUM_LIT:10> ) ] ) <EOL> def test_activation_deactivation ( self ) : <EOL> qs = Author . objects . all ( ) <EOL> assert not qs . _count_tries_approx <EOL> qs2 = qs . count_tries_approx ( min_size = <NUM_LIT:2> ) <EOL> assert qs != qs2 <EOL> assert qs2 . _count_tries_approx <EOL> count = qs2 . count ( ) <EOL> assert isinstance ( count , ApproximateInt ) <EOL> qs3 = qs2 . count_tries_approx ( False ) <EOL> assert qs2 != qs3 <EOL> assert not qs3 . _count_tries_approx <EOL> def test_activation_but_fallback ( self ) : <EOL> qs = Author . objects . exclude ( name = '<STR_LIT>' ) . count_tries_approx ( ) <EOL> count = qs . count ( ) <EOL> assert count == <NUM_LIT:10> <EOL> assert not isinstance ( count , ApproximateInt ) <EOL> def test_activation_but_fallback_due_to_min_size ( self ) : <EOL> qs = Author . objects . count_tries_approx ( ) <EOL> count = qs . count ( ) <EOL> assert count == <NUM_LIT:10> <EOL> assert not isinstance ( count , ApproximateInt ) <EOL> def test_output_in_templates ( self ) : <EOL> approx_count = Author . objects . approx_count ( min_size = <NUM_LIT:1> ) <EOL> text = Template ( '<STR_LIT>' ) . render ( Context ( { '<STR_LIT>' : approx_count } ) ) <EOL> assert text . startswith ( '<STR_LIT>' ) <EOL> approx_count2 = Author . objects . approx_count ( <EOL> min_size = <NUM_LIT:1> , <EOL> return_approx_int = False <EOL> ) <EOL> text = Template ( '<STR_LIT>' ) . render ( Context ( { '<STR_LIT>' : approx_count2 } ) ) <EOL> assert not text . startswith ( '<STR_LIT>' ) <EOL> def test_fallback_with_filters ( self ) : <EOL> filtered = Author . objects . filter ( name = '<STR_LIT>' ) <EOL> assert filtered . approx_count ( fall_back = True ) == <NUM_LIT:10> <EOL> with pytest . raises ( ValueError ) : <EOL> filtered . approx_count ( fall_back = False ) <EOL> def test_fallback_with_slice ( self ) : <EOL> assert Author . objects . all ( ) [ : <NUM_LIT:100> ] . approx_count ( ) == <NUM_LIT:10> <EOL> with pytest . raises ( ValueError ) : <EOL> Author . objects . all ( ) [ : <NUM_LIT:100> ] . approx_count ( fall_back = False ) <EOL> def test_fallback_with_distinct ( self ) : <EOL> assert Author . objects . distinct ( ) . approx_count ( ) == <NUM_LIT:10> <EOL> with pytest . raises ( ValueError ) : <EOL> Author . objects . distinct ( ) . approx_count ( fall_back = False ) <EOL> def test_fallback_with_arbitrary_extra ( self ) : <EOL> assert Author . objects . extra ( where = [ '<STR_LIT>' ] ) . approx_count ( ) == <NUM_LIT:10> <EOL> with pytest . raises ( ValueError ) : <EOL> Author . objects . extra ( where = [ '<STR_LIT>' ] ) . approx_count ( fall_back = False ) <EOL> def test_approx_count_with_label ( self ) : <EOL> assert Author . objects . label ( '<STR_LIT>' ) . approx_count ( fall_back = False ) == <NUM_LIT:10> <EOL> def test_approx_count_with_straight_join ( self ) : <EOL> assert ( <EOL> Author . objects . straight_join ( ) . approx_count ( fall_back = False ) == <NUM_LIT:10> <EOL> ) <EOL> class QueryHintTests ( TestCase ) : <EOL> def test_label ( self ) : <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . label ( "<STR_LIT>" ) . all ( ) ) <EOL> assert cap . query . startswith ( "<STR_LIT>" ) <EOL> def test_label_twice ( self ) : <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . label ( "<STR_LIT>" ) <EOL> . label ( "<STR_LIT>" ) <EOL> . all ( ) ) <EOL> assert cap . query . startswith ( <EOL> "<STR_LIT>" <EOL> ) <EOL> def test_label_star ( self ) : <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . label ( "<STR_LIT>" ) . label ( "<STR_LIT:*>" ) . all ( ) ) <EOL> assert cap . query . startswith ( "<STR_LIT>" ) <EOL> def test_label_update ( self ) : <EOL> Author . objects . create ( name = '<STR_LIT>' ) <EOL> with CaptureLastQuery ( ) as cap : <EOL> Author . objects . label ( "<STR_LIT>" ) . update ( name = '<STR_LIT>' ) <EOL> assert cap . query . startswith ( "<STR_LIT>" ) <EOL> def test_label_bad ( self ) : <EOL> with self . assertRaises ( ValueError ) : <EOL> Author . objects . label ( "<STR_LIT>" ) <EOL> def test_label_and_straight_join ( self ) : <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . label ( "<STR_LIT>" ) <EOL> . straight_join ( ) <EOL> . all ( ) ) <EOL> assert cap . query . startswith ( <EOL> "<STR_LIT>" <EOL> ) <EOL> def test_straight_join ( self ) : <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . filter ( books__title__startswith = '<STR_LIT:A>' ) <EOL> . straight_join ( ) ) <EOL> assert cap . query . startswith ( "<STR_LIT>" ) <EOL> def test_straight_join_with_distinct ( self ) : <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . filter ( tutor = None ) <EOL> . distinct ( ) <EOL> . values ( '<STR_LIT>' ) <EOL> . straight_join ( ) ) <EOL> assert cap . query . startswith ( "<STR_LIT>" ) <EOL> @ override_settings ( DJANGO_MYSQL_REWRITE_QUERIES = False ) <EOL> def test_can_disable_setting ( self ) : <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . all ( ) . straight_join ( ) ) <EOL> assert not cap . query . startswith ( "<STR_LIT>" ) <EOL> def test_sql_cache ( self ) : <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . sql_cache ( ) . all ( ) ) <EOL> assert cap . query . startswith ( "<STR_LIT>" ) <EOL> def test_sql_no_cache ( self ) : <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . sql_no_cache ( ) . all ( ) ) <EOL> assert cap . query . startswith ( "<STR_LIT>" ) <EOL> def test_sql_small_result ( self ) : <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . sql_small_result ( ) . all ( ) ) <EOL> assert cap . query . startswith ( "<STR_LIT>" ) <EOL> def test_sql_big_result ( self ) : <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . sql_big_result ( ) . all ( ) ) <EOL> assert cap . query . startswith ( "<STR_LIT>" ) <EOL> def test_sql_buffer_result ( self ) : <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . sql_buffer_result ( ) . all ( ) ) <EOL> assert cap . query . startswith ( "<STR_LIT>" ) <EOL> def test_adding_many ( self ) : <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . straight_join ( ) <EOL> . sql_cache ( ) <EOL> . sql_big_result ( ) <EOL> . sql_buffer_result ( ) ) <EOL> assert cap . query . startswith ( <EOL> "<STR_LIT>" <EOL> ) <EOL> def test_complex_query_1 ( self ) : <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . distinct ( ) <EOL> . straight_join ( ) <EOL> . filter ( books__title__startswith = "<STR_LIT:A>" ) <EOL> . exclude ( books__id__lte = <NUM_LIT:1> ) <EOL> . prefetch_related ( '<STR_LIT>' ) <EOL> . filter ( bio__gt = '<STR_LIT>' ) <EOL> . exclude ( bio__startswith = '<STR_LIT>' ) ) <EOL> assert cap . query . startswith ( "<STR_LIT>" ) <EOL> def test_complex_query_2 ( self ) : <EOL> subq = Book . objects . straight_join ( ) . filter ( title__startswith = "<STR_LIT:A>" ) <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . straight_join ( ) <EOL> . filter ( books__in = subq ) ) <EOL> assert cap . query . startswith ( "<STR_LIT>" ) <EOL> def test_use_index ( self ) : <EOL> name_idx = index_name ( Author , '<STR_LIT:name>' ) <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . filter ( name__gt = '<STR_LIT>' ) <EOL> . use_index ( name_idx ) ) <EOL> assert ( '<STR_LIT>' + name_idx + '<STR_LIT>' ) in cap . query <EOL> used = used_indexes ( cap . query ) <EOL> assert len ( used ) == <NUM_LIT:0> or name_idx in used <EOL> def test_use_index_primary ( self ) : <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . use_index ( '<STR_LIT>' ) ) <EOL> assert ( '<STR_LIT>' ) in cap . query <EOL> used = used_indexes ( cap . query ) <EOL> assert len ( used ) == <NUM_LIT:0> or '<STR_LIT>' in used <EOL> def test_force_index ( self ) : <EOL> name_idx = index_name ( Author , '<STR_LIT:name>' ) <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . filter ( name__gt = '<STR_LIT>' ) <EOL> . force_index ( name_idx ) ) <EOL> assert ( '<STR_LIT>' + name_idx + '<STR_LIT>' ) in cap . query <EOL> assert name_idx in used_indexes ( cap . query ) <EOL> def test_force_index_primary ( self ) : <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . force_index ( '<STR_LIT>' ) ) <EOL> assert ( '<STR_LIT>' ) in cap . query <EOL> used = used_indexes ( cap . query ) <EOL> assert len ( used ) == <NUM_LIT:0> or '<STR_LIT>' in used <EOL> def test_ignore_index ( self ) : <EOL> name_idx = index_name ( Author , '<STR_LIT:name>' ) <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . filter ( name__gt = '<STR_LIT>' ) . ignore_index ( name_idx ) ) <EOL> assert ( '<STR_LIT>' + name_idx + '<STR_LIT>' ) in cap . query <EOL> assert name_idx not in used_indexes ( cap . query ) <EOL> def test_ignore_index_multiple ( self ) : <EOL> name_idx = index_name ( AuthorMultiIndex , '<STR_LIT:name>' ) <EOL> name_country_idx = index_name ( AuthorMultiIndex , '<STR_LIT:name>' , '<STR_LIT>' ) <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( AuthorMultiIndex . objects <EOL> . filter ( name__gt = '<STR_LIT>' ) <EOL> . ignore_index ( name_idx , name_country_idx ) ) <EOL> assert ( <EOL> '<STR_LIT>' + name_idx + '<STR_LIT>' + name_country_idx + '<STR_LIT>' <EOL> in cap . query <EOL> ) <EOL> used = used_indexes ( cap . query ) <EOL> assert name_idx not in used <EOL> assert name_country_idx not in used <EOL> def test_ignore_index_primary ( self ) : <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . filter ( name__gt = '<STR_LIT>' ) . ignore_index ( '<STR_LIT>' ) ) <EOL> assert ( '<STR_LIT>' ) in cap . query <EOL> assert '<STR_LIT>' not in used_indexes ( cap . query ) <EOL> def test_force_index_at_least_one ( self ) : <EOL> with pytest . raises ( ValueError ) as excinfo : <EOL> Author . objects . force_index ( ) <EOL> assert ( <EOL> str ( excinfo . value ) == <EOL> "<STR_LIT>" <EOL> ) <EOL> def test_force_index_invalid_for ( self ) : <EOL> with pytest . raises ( ValueError ) as excinfo : <EOL> Author . objects . force_index ( '<STR_LIT:a>' , for_ = '<STR_LIT>' ) <EOL> assert "<STR_LIT>" in str ( excinfo . value ) <EOL> def test_force_index_invalid_kwarg ( self ) : <EOL> with pytest . raises ( ValueError ) as excinfo : <EOL> Author . objects . force_index ( '<STR_LIT:a>' , nonexistent = True ) <EOL> assert ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" in <EOL> str ( excinfo . value ) <EOL> ) <EOL> def test_index_hint_force_order_by ( self ) : <EOL> name_idx = index_name ( Author , '<STR_LIT:name>' ) <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . force_index ( name_idx , for_ = '<STR_LIT>' ) <EOL> . order_by ( '<STR_LIT:name>' ) ) <EOL> assert ( '<STR_LIT>' + name_idx + "<STR_LIT>" ) in cap . query <EOL> assert name_idx in used_indexes ( cap . query ) <EOL> def test_use_index_none ( self ) : <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects . values_list ( '<STR_LIT:name>' ) . distinct ( ) . use_index ( ) ) <EOL> assert '<STR_LIT>' in cap . query <EOL> assert used_indexes ( cap . query ) == set ( ) <EOL> def test_use_index_table_name ( self ) : <EOL> extra_table = '<STR_LIT>' <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects <EOL> . select_related ( '<STR_LIT>' ) <EOL> . use_index ( '<STR_LIT>' , table_name = extra_table ) ) <EOL> assert '<STR_LIT>' + extra_table + '<STR_LIT>' in cap . query <EOL> def test_force_index_table_name_doesnt_exist_ignored ( self ) : <EOL> with CaptureLastQuery ( ) as cap : <EOL> list ( Author . objects <EOL> . select_related ( '<STR_LIT>' ) <EOL> . force_index ( '<STR_LIT>' , table_name = '<STR_LIT>' ) ) <EOL> assert '<STR_LIT>' not in cap . query <EOL> class FoundRowsTests ( TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( FoundRowsTests , self ) . setUp ( ) <EOL> Author . objects . bulk_create ( [ Author ( ) for i in range ( <NUM_LIT:10> ) ] ) <EOL> def test_found_rows_requires_sql_calc_found_rows ( self ) : <EOL> authors = Author . objects . all ( ) <EOL> with pytest . raises ( ValueError ) as excinfo : <EOL> authors . found_rows <EOL> assert "<STR_LIT>" in str ( excinfo . value ) <EOL> def test_found_rows_requires_iteration ( self ) : <EOL> authors = Author . objects . sql_calc_found_rows ( ) [ : <NUM_LIT:5> ] <EOL> with self . assertRaises ( RuntimeError ) : <EOL> authors . found_rows <EOL> def test_it_working ( self ) : <EOL> with self . assertNumQueries ( <NUM_LIT:2> ) : <EOL> authors = Author . objects . sql_calc_found_rows ( ) [ : <NUM_LIT:5> ] <EOL> list ( authors ) <EOL> assert authors . found_rows == <NUM_LIT:10> <EOL> def test_no_repeats ( self ) : <EOL> with self . assertNumQueries ( <NUM_LIT:2> ) : <EOL> authors = ( <EOL> Author . objects . sql_calc_found_rows ( ) <EOL> . sql_calc_found_rows ( ) [ : <NUM_LIT:5> ] <EOL> ) <EOL> list ( authors ) <EOL> assert authors . found_rows == <NUM_LIT:10> <EOL> def test_it_working_prefetching ( self ) : <EOL> with self . assertNumQueries ( <NUM_LIT:3> ) : <EOL> authors = ( <EOL> Author . objects . sql_calc_found_rows ( ) <EOL> . prefetch_related ( '<STR_LIT>' ) [ : <NUM_LIT:5> ] <EOL> ) <EOL> list ( authors ) <EOL> assert authors . found_rows == <NUM_LIT:10> <EOL> def test_pickleability ( self ) : <EOL> authors = Author . objects . sql_calc_found_rows ( ) <EOL> pickle . dumps ( authors ) <EOL> def test_cloning_doesnt_copy_found_rows_number ( self ) : <EOL> authors = Author . objects . sql_calc_found_rows ( ) <EOL> list ( authors ) <EOL> assert authors . found_rows == <NUM_LIT:10> <EOL> authors2 = authors . filter ( id__gte = <NUM_LIT:1> ) <EOL> with pytest . raises ( RuntimeError ) : <EOL> authors2 . found_rows <EOL> with self . assertNumQueries ( <NUM_LIT:2> ) : <EOL> list ( authors2 ) <EOL> assert authors2 . found_rows == <NUM_LIT:10> <EOL> class SmartIteratorTests ( TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( SmartIteratorTests , self ) . setUp ( ) <EOL> Author . objects . bulk_create ( [ Author ( id = i + <NUM_LIT:1> ) for i in range ( <NUM_LIT:10> ) ] ) <EOL> def test_bad_querysets ( self ) : <EOL> with pytest . raises ( ValueError ) as excinfo : <EOL> Author . objects . all ( ) . order_by ( '<STR_LIT:name>' ) . iter_smart_chunks ( ) <EOL> assert "<STR_LIT>" in str ( excinfo . value ) <EOL> with pytest . raises ( ValueError ) as excinfo : <EOL> Author . objects . all ( ) [ : <NUM_LIT:5> ] . iter_smart_chunks ( ) <EOL> assert "<STR_LIT>" in str ( excinfo . value ) <EOL> with pytest . raises ( ValueError ) as excinfo : <EOL> NameAuthor . objects . all ( ) . iter_smart_chunks ( ) <EOL> assert "<STR_LIT>" in str ( excinfo . value ) <EOL> def test_chunks ( self ) : <EOL> seen = [ ] <EOL> for authors in Author . objects . iter_smart_chunks ( ) : <EOL> seen . extend ( author . id for author in authors ) <EOL> all_ids = list ( Author . objects . order_by ( '<STR_LIT:id>' ) <EOL> . values_list ( '<STR_LIT:id>' , flat = True ) ) <EOL> assert seen == all_ids <EOL> def test_objects ( self ) : <EOL> seen = [ author . id for author in Author . objects . iter_smart ( ) ] <EOL> all_ids = list ( Author . objects . order_by ( '<STR_LIT:id>' ) <EOL> . values_list ( '<STR_LIT:id>' , flat = True ) ) <EOL> assert seen == all_ids <EOL> def test_objects_non_atomic ( self ) : <EOL> seen = [ author . id for author in <EOL> Author . objects . iter_smart ( atomically = False ) ] <EOL> all_ids = list ( Author . objects . order_by ( '<STR_LIT:id>' ) <EOL> . values_list ( '<STR_LIT:id>' , flat = True ) ) <EOL> assert seen == all_ids <EOL> def test_objects_reverse ( self ) : <EOL> seen = [ author . id for author in Author . objects . reverse ( ) . iter_smart ( ) ] <EOL> all_ids = list ( Author . objects . order_by ( '<STR_LIT>' ) <EOL> . values_list ( '<STR_LIT:id>' , flat = True ) ) <EOL> assert seen == all_ids <EOL> def test_objects_pk_range_all ( self ) : <EOL> seen = [ author . id for author in <EOL> Author . objects . iter_smart ( pk_range = '<STR_LIT:all>' ) ] <EOL> all_ids = list ( Author . objects . order_by ( '<STR_LIT:id>' ) <EOL> . values_list ( '<STR_LIT:id>' , flat = True ) ) <EOL> assert seen == all_ids <EOL> def test_objects_pk_range_tuple ( self ) : <EOL> min_id = Author . objects . earliest ( '<STR_LIT:id>' ) . id <EOL> max_id = Author . objects . order_by ( '<STR_LIT:id>' ) [ <NUM_LIT:5> ] . id <EOL> seen = [ author . id for author in <EOL> Author . objects . iter_smart ( pk_range = ( min_id , max_id ) ) ] <EOL> cut_ids = list ( Author . objects . order_by ( '<STR_LIT:id>' ) <EOL> . filter ( id__gte = min_id , id__lte = max_id ) <EOL> . values_list ( '<STR_LIT:id>' , flat = True ) ) <EOL> assert seen == cut_ids <EOL> def test_objects_pk_range_tuple_0_0 ( self ) : <EOL> seen = [ author . id for author in <EOL> Author . objects . iter_smart ( pk_range = ( <NUM_LIT:0> , <NUM_LIT:0> ) ) ] <EOL> assert seen == [ ] <EOL> def test_objects_pk_range_reversed ( self ) : <EOL> min_id = Author . objects . earliest ( '<STR_LIT:id>' ) . id <EOL> max_id = Author . objects . latest ( '<STR_LIT:id>' ) . id <EOL> seen = [ author . id for author in <EOL> Author . objects . iter_smart ( pk_range = ( max_id , min_id ) ) ] <EOL> all_ids = list ( Author . objects . order_by ( '<STR_LIT>' ) <EOL> . values_list ( '<STR_LIT:id>' , flat = True ) ) <EOL> assert seen == all_ids <EOL> def test_objects_pk_range_bad ( self ) : <EOL> with pytest . raises ( ValueError ) as excinfo : <EOL> list ( Author . objects . iter_smart ( pk_range = "<STR_LIT>" ) ) <EOL> assert "<STR_LIT>" in str ( excinfo . value ) <EOL> def test_pk_range_race_condition ( self ) : <EOL> getitem = QuerySet . __getitem__ <EOL> def fail_second_slice ( * args , ** kwargs ) : <EOL> fail_second_slice . calls += <NUM_LIT:1> <EOL> if fail_second_slice . calls == <NUM_LIT:2> : <EOL> Author . objects . all ( ) . delete ( ) <EOL> return getitem ( * args , ** kwargs ) <EOL> fail_second_slice . calls = <NUM_LIT:0> <EOL> path = '<STR_LIT>' <EOL> with mock . patch ( path , fail_second_slice ) : <EOL> seen = [ author . id for author in Author . objects . iter_smart ( ) ] <EOL> assert seen == [ ] <EOL> def test_objects_chunk_size ( self ) : <EOL> smart = iter ( Author . objects . iter_smart_chunks ( chunk_size = <NUM_LIT:3> ) ) <EOL> chunk = next ( smart ) <EOL> assert len ( list ( chunk ) ) == <NUM_LIT:3> <EOL> def test_objects_chunk_size_1 ( self ) : <EOL> smart = iter ( Author . objects . iter_smart_chunks ( chunk_size = <NUM_LIT:1> ) ) <EOL> chunk = next ( smart ) <EOL> assert len ( list ( chunk ) ) == <NUM_LIT:1> <EOL> def test_objects_max_size ( self ) : <EOL> seen = [ ] <EOL> for chunk in Author . objects . iter_smart_chunks ( chunk_max = <NUM_LIT:3> ) : <EOL> ids = [ author . id for author in chunk ] <EOL> assert len ( ids ) <= <NUM_LIT:3> <EOL> seen . extend ( ids ) <EOL> all_ids = list ( Author . objects . order_by ( '<STR_LIT:id>' ) <EOL> . values_list ( '<STR_LIT:id>' , flat = True ) ) <EOL> assert seen == all_ids <EOL> def test_objects_max_size_1 ( self ) : <EOL> seen = [ ] <EOL> for chunk in Author . objects . iter_smart_chunks ( chunk_max = <NUM_LIT:1> ) : <EOL> ids = [ author . id for author in chunk ] <EOL> assert len ( ids ) == <NUM_LIT:1> <EOL> seen . extend ( ids ) <EOL> all_ids = list ( Author . objects . order_by ( '<STR_LIT:id>' ) <EOL> . values_list ( '<STR_LIT:id>' , flat = True ) ) <EOL> assert seen == all_ids <EOL> def test_objects_max_size_bounds_chunk_size ( self ) : <EOL> smart = iter ( Author . objects . iter_smart_chunks ( chunk_max = <NUM_LIT:5> , <EOL> chunk_size = <NUM_LIT:1000> ) ) <EOL> seen = [ ] <EOL> for chunk in smart : <EOL> ids = [ author . id for author in chunk ] <EOL> assert len ( ids ) <= <NUM_LIT:5> <EOL> seen . extend ( ids ) <EOL> all_ids = list ( Author . objects . order_by ( '<STR_LIT:id>' ) <EOL> . values_list ( '<STR_LIT:id>' , flat = True ) ) <EOL> assert seen == all_ids <EOL> def test_objects_min_size_bounds_chunk_size ( self ) : <EOL> smart = iter ( Author . objects . iter_smart_chunks ( chunk_min = <NUM_LIT:5> , <EOL> chunk_size = <NUM_LIT:1> ) ) <EOL> seen = [ ] <EOL> for chunk in smart : <EOL> ids = [ author . id for author in chunk ] <EOL> assert len ( ids ) >= <NUM_LIT:5> <EOL> seen . extend ( ids ) <EOL> all_ids = list ( Author . objects . order_by ( '<STR_LIT:id>' ) <EOL> . values_list ( '<STR_LIT:id>' , flat = True ) ) <EOL> assert seen == all_ids <EOL> def test_min_size_greater_than_max ( self ) : <EOL> with pytest . raises ( AssertionError ) : <EOL> Author . objects . iter_smart_chunks ( chunk_min = <NUM_LIT:2> , chunk_max = <NUM_LIT:1> ) <EOL> def test_no_matching_objects ( self ) : <EOL> seen = [ author . id for author in <EOL> Author . objects . filter ( name = "<STR_LIT>" ) . iter_smart ( ) ] <EOL> assert seen == [ ] <EOL> def test_no_objects ( self ) : <EOL> Author . objects . all ( ) . delete ( ) <EOL> seen = [ author . id for author in Author . objects . iter_smart ( ) ] <EOL> assert seen == [ ] <EOL> def test_pk_hole ( self ) : <EOL> first = Author . objects . earliest ( '<STR_LIT:id>' ) <EOL> last = Author . objects . latest ( '<STR_LIT:id>' ) <EOL> Author . objects . filter ( id__gt = first . id , id__lt = last . id ) . delete ( ) <EOL> seen = [ author . id for author in Author . objects . iter_smart ( ) ] <EOL> assert seen == [ first . id , last . id ] <EOL> def test_iter_smart_pk_range ( self ) : <EOL> seen = [ ] <EOL> for start_pk , end_pk in Author . objects . iter_smart_pk_ranges ( ) : <EOL> seen . extend ( <EOL> Author . objects . filter ( id__gte = start_pk , id__lt = end_pk ) <EOL> . values_list ( '<STR_LIT:id>' , flat = True ) <EOL> ) <EOL> all_ids = list ( Author . objects . order_by ( '<STR_LIT:id>' ) <EOL> . values_list ( '<STR_LIT:id>' , flat = True ) ) <EOL> assert seen == all_ids <EOL> def test_iter_smart_pk_range_with_raw ( self ) : <EOL> seen = [ ] <EOL> for start_pk , end_pk in Author . objects . iter_smart_pk_ranges ( ) : <EOL> authors = Author . objects . raw ( """<STR_LIT>""" . format ( Author . _meta . db_table ) , ( start_pk , end_pk ) ) <EOL> seen . extend ( author . id for author in authors ) <EOL> all_ids = list ( Author . objects . order_by ( '<STR_LIT:id>' ) <EOL> . values_list ( '<STR_LIT:id>' , flat = True ) ) <EOL> assert seen == all_ids <EOL> def test_iter_smart_fk_primary_key ( self ) : <EOL> author , author2 = Author . objects . all ( ) [ : <NUM_LIT:2> ] <EOL> AuthorExtra . objects . create ( author = author , legs = <NUM_LIT:2> ) <EOL> AuthorExtra . objects . create ( author = author2 , legs = <NUM_LIT:1> ) <EOL> seen_author_ids = [ ] <EOL> for extra in AuthorExtra . objects . iter_smart ( ) : <EOL> seen_author_ids . append ( extra . author_id ) <EOL> assert seen_author_ids == [ author . id , author2 . id ] <EOL> def test_iter_smart_fk_string_primary_key_fails ( self ) : <EOL> with pytest . raises ( ValueError ) as excinfo : <EOL> NameAuthorExtra . objects . iter_smart ( ) <EOL> assert "<STR_LIT>" in str ( excinfo . value ) <EOL> def test_reporting ( self ) : <EOL> with captured_stdout ( ) as output : <EOL> qs = Author . objects . all ( ) <EOL> for authors in qs . iter_smart_chunks ( report_progress = True ) : <EOL> list ( authors ) <EOL> lines = output . getvalue ( ) . split ( '<STR_LIT:\n>' ) <EOL> reports = lines [ <NUM_LIT:0> ] . split ( '<STR_LIT:\r>' ) <EOL> for report in reports : <EOL> assert re . match ( <EOL> r"<STR_LIT>" <EOL> r"<STR_LIT>" <EOL> r"<STR_LIT>" , <EOL> report <EOL> ) <EOL> assert re . match ( <EOL> r'<STR_LIT>' , <EOL> lines [ <NUM_LIT:1> ] , <EOL> ) <EOL> def test_reporting_reverse ( self ) : <EOL> Author . objects . create ( id = <NUM_LIT> ) <EOL> with captured_stdout ( ) as output : <EOL> qs = Author . objects . reverse ( ) <EOL> for authors in qs . iter_smart_chunks ( report_progress = True , <EOL> chunk_min = <NUM_LIT:100> ) : <EOL> list ( authors ) <EOL> lines = output . getvalue ( ) . split ( '<STR_LIT:\n>' ) <EOL> reports = lines [ <NUM_LIT:0> ] . split ( '<STR_LIT:\r>' ) <EOL> assert len ( reports ) > <NUM_LIT:0> <EOL> for report in reports : <EOL> assert re . match ( <EOL> r"<STR_LIT>" <EOL> r"<STR_LIT>" <EOL> r"<STR_LIT>" , <EOL> report <EOL> ) <EOL> assert re . match ( <EOL> r'<STR_LIT>' , <EOL> lines [ <NUM_LIT:1> ] , <EOL> ) <EOL> def test_reporting_with_total ( self ) : <EOL> with captured_stdout ( ) as output : <EOL> qs = Author . objects . all ( ) <EOL> for authors in qs . iter_smart_chunks ( report_progress = True , total = <NUM_LIT:4> ) : <EOL> list ( authors ) <EOL> lines = output . getvalue ( ) . split ( '<STR_LIT:\n>' ) <EOL> reports = lines [ <NUM_LIT:0> ] . split ( '<STR_LIT:\r>' ) <EOL> for report in reports : <EOL> assert re . match ( <EOL> r"<STR_LIT>" <EOL> r"<STR_LIT>" <EOL> r"<STR_LIT>" , <EOL> report <EOL> ) <EOL> assert re . match ( <EOL> r'<STR_LIT>' , <EOL> lines [ <NUM_LIT:1> ] , <EOL> ) <EOL> def test_reporting_on_uncounted_qs ( self ) : <EOL> Author . objects . create ( name = "<STR_LIT>" ) <EOL> with captured_stdout ( ) as output : <EOL> qs = Author . objects . filter ( name = "<STR_LIT>" ) <EOL> for authors in qs . iter_smart_chunks ( report_progress = True ) : <EOL> authors . delete ( ) <EOL> lines = output . getvalue ( ) . split ( '<STR_LIT:\n>' ) <EOL> reports = lines [ <NUM_LIT:0> ] . split ( '<STR_LIT:\r>' ) <EOL> for report in reports : <EOL> assert re . match ( <EOL> r"<STR_LIT>" <EOL> r"<STR_LIT>" , <EOL> report <EOL> ) <EOL> assert re . match ( <EOL> r'<STR_LIT>' , <EOL> lines [ <NUM_LIT:1> ] , <EOL> ) <EOL> def test_filter_and_delete ( self ) : <EOL> VanillaAuthor . objects . create ( name = "<STR_LIT>" ) <EOL> VanillaAuthor . objects . create ( name = "<STR_LIT>" ) <EOL> VanillaAuthor . objects . create ( name = "<STR_LIT>" ) <EOL> VanillaAuthor . objects . create ( name = "<STR_LIT>" ) <EOL> bad_authors = VanillaAuthor . objects . filter ( name = "<STR_LIT>" ) <EOL> assert bad_authors . count ( ) == <NUM_LIT:2> <EOL> with captured_stdout ( ) : <EOL> for author in SmartIterator ( bad_authors , report_progress = True ) : <EOL> author . delete ( ) <EOL> assert bad_authors . count ( ) == <NUM_LIT:0> <EOL> @ skipUnless ( have_program ( '<STR_LIT>' ) , <EOL> "<STR_LIT>" ) <EOL> class VisualExplainTests ( TestCase ) : <EOL> def test_basic ( self ) : <EOL> with captured_stdout ( ) as capture : <EOL> Author . objects . all ( ) . pt_visual_explain ( ) <EOL> output = capture . getvalue ( ) <EOL> assert "<STR_LIT>" in output <EOL> assert "<STR_LIT>" in output <EOL> assert "<STR_LIT>" in output <EOL> def test_basic_no_display ( self ) : <EOL> output = Author . objects . all ( ) . pt_visual_explain ( display = False ) <EOL> assert "<STR_LIT>" in output <EOL> assert "<STR_LIT>" in output <EOL> assert "<STR_LIT>" in output <EOL> def test_subquery ( self ) : <EOL> subq = Author . objects . all ( ) . values_list ( '<STR_LIT:id>' , flat = True ) <EOL> output = Author . objects . filter ( id__in = subq ) . pt_visual_explain ( display = False ) <EOL> assert "<STR_LIT>" in output <EOL> assert "<STR_LIT>" in output <EOL> assert "<STR_LIT>" in output <EOL> assert "<STR_LIT>" in output </s>
