<s> from tests . integration import use_singledc , PROTOCOL_VERSION <EOL> import logging <EOL> log = logging . getLogger ( __name__ ) <EOL> try : <EOL> import unittest2 as unittest <EOL> except ImportError : <EOL> import unittest <EOL> from itertools import cycle , count <EOL> from six . moves import range <EOL> from threading import Event <EOL> from cassandra . cluster import Cluster <EOL> from cassandra . concurrent import execute_concurrent , execute_concurrent_with_args <EOL> from cassandra . policies import HostDistance <EOL> from cassandra . query import SimpleStatement <EOL> def setup_module ( ) : <EOL> use_singledc ( ) <EOL> class QueryPagingTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> if PROTOCOL_VERSION < <NUM_LIT:2> : <EOL> raise unittest . SkipTest ( <EOL> "<STR_LIT>" <EOL> % ( PROTOCOL_VERSION , ) ) <EOL> self . cluster = Cluster ( protocol_version = PROTOCOL_VERSION ) <EOL> if PROTOCOL_VERSION < <NUM_LIT:3> : <EOL> self . cluster . set_core_connections_per_host ( HostDistance . LOCAL , <NUM_LIT:1> ) <EOL> self . session = self . cluster . connect ( ) <EOL> self . session . execute ( "<STR_LIT>" ) <EOL> def tearDown ( self ) : <EOL> self . cluster . shutdown ( ) <EOL> def test_paging ( self ) : <EOL> statements_and_params = zip ( cycle ( [ "<STR_LIT>" ] ) , <EOL> [ ( i , ) for i in range ( <NUM_LIT:100> ) ] ) <EOL> execute_concurrent ( self . session , list ( statements_and_params ) ) <EOL> prepared = self . session . prepare ( "<STR_LIT>" ) <EOL> for fetch_size in ( <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:7> , <NUM_LIT:10> , <NUM_LIT> , <NUM_LIT:100> , <NUM_LIT> , <NUM_LIT> ) : <EOL> self . session . default_fetch_size = fetch_size <EOL> self . assertEqual ( <NUM_LIT:100> , len ( list ( self . session . execute ( "<STR_LIT>" ) ) ) ) <EOL> statement = SimpleStatement ( "<STR_LIT>" ) <EOL> self . assertEqual ( <NUM_LIT:100> , len ( list ( self . session . execute ( statement ) ) ) ) <EOL> self . assertEqual ( <NUM_LIT:100> , len ( list ( self . session . execute ( prepared ) ) ) ) <EOL> def test_paging_verify_writes ( self ) : <EOL> statements_and_params = zip ( cycle ( [ "<STR_LIT>" ] ) , <EOL> [ ( i , ) for i in range ( <NUM_LIT:100> ) ] ) <EOL> execute_concurrent ( self . session , statements_and_params ) <EOL> prepared = self . session . prepare ( "<STR_LIT>" ) <EOL> for fetch_size in ( <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:7> , <NUM_LIT:10> , <NUM_LIT> , <NUM_LIT:100> , <NUM_LIT> , <NUM_LIT> ) : <EOL> self . session . default_fetch_size = fetch_size <EOL> results = self . session . execute ( "<STR_LIT>" ) <EOL> result_array = set ( ) <EOL> result_set = set ( ) <EOL> for result in results : <EOL> result_array . add ( result . k ) <EOL> result_set . add ( result . v ) <EOL> self . assertEqual ( set ( range ( <NUM_LIT:100> ) ) , result_array ) <EOL> self . assertEqual ( set ( [ <NUM_LIT:0> ] ) , result_set ) <EOL> statement = SimpleStatement ( "<STR_LIT>" ) <EOL> results = self . session . execute ( statement ) <EOL> result_array = set ( ) <EOL> result_set = set ( ) <EOL> for result in results : <EOL> result_array . add ( result . k ) <EOL> result_set . add ( result . v ) <EOL> self . assertEqual ( set ( range ( <NUM_LIT:100> ) ) , result_array ) <EOL> self . assertEqual ( set ( [ <NUM_LIT:0> ] ) , result_set ) <EOL> results = self . session . execute ( prepared ) <EOL> result_array = set ( ) <EOL> result_set = set ( ) <EOL> for result in results : <EOL> result_array . add ( result . k ) <EOL> result_set . add ( result . v ) <EOL> self . assertEqual ( set ( range ( <NUM_LIT:100> ) ) , result_array ) <EOL> self . assertEqual ( set ( [ <NUM_LIT:0> ] ) , result_set ) <EOL> def test_paging_verify_with_composite_keys ( self ) : <EOL> ddl = '''<STR_LIT>''' <EOL> self . session . execute ( ddl ) <EOL> statements_and_params = zip ( cycle ( [ "<STR_LIT>" <EOL> "<STR_LIT>" ] ) , <EOL> [ ( i , i + <NUM_LIT:1> ) for i in range ( <NUM_LIT:100> ) ] ) <EOL> execute_concurrent ( self . session , statements_and_params ) <EOL> prepared = self . session . prepare ( "<STR_LIT>" ) <EOL> for fetch_size in ( <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:7> , <NUM_LIT:10> , <NUM_LIT> , <NUM_LIT:100> , <NUM_LIT> , <NUM_LIT> ) : <EOL> self . session . default_fetch_size = fetch_size <EOL> results = self . session . execute ( "<STR_LIT>" ) <EOL> result_array = [ ] <EOL> value_array = [ ] <EOL> for result in results : <EOL> result_array . append ( result . k2 ) <EOL> value_array . append ( result . v ) <EOL> self . assertSequenceEqual ( range ( <NUM_LIT:100> ) , result_array ) <EOL> self . assertSequenceEqual ( range ( <NUM_LIT:1> , <NUM_LIT> ) , value_array ) <EOL> statement = SimpleStatement ( "<STR_LIT>" ) <EOL> results = self . session . execute ( statement ) <EOL> result_array = [ ] <EOL> value_array = [ ] <EOL> for result in results : <EOL> result_array . append ( result . k2 ) <EOL> value_array . append ( result . v ) <EOL> self . assertSequenceEqual ( range ( <NUM_LIT:100> ) , result_array ) <EOL> self . assertSequenceEqual ( range ( <NUM_LIT:1> , <NUM_LIT> ) , value_array ) <EOL> results = self . session . execute ( prepared ) <EOL> result_array = [ ] <EOL> value_array = [ ] <EOL> for result in results : <EOL> result_array . append ( result . k2 ) <EOL> value_array . append ( result . v ) <EOL> self . assertSequenceEqual ( range ( <NUM_LIT:100> ) , result_array ) <EOL> self . assertSequenceEqual ( range ( <NUM_LIT:1> , <NUM_LIT> ) , value_array ) <EOL> def test_async_paging ( self ) : <EOL> statements_and_params = zip ( cycle ( [ "<STR_LIT>" ] ) , <EOL> [ ( i , ) for i in range ( <NUM_LIT:100> ) ] ) <EOL> execute_concurrent ( self . session , list ( statements_and_params ) ) <EOL> prepared = self . session . prepare ( "<STR_LIT>" ) <EOL> for fetch_size in ( <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:7> , <NUM_LIT:10> , <NUM_LIT> , <NUM_LIT:100> , <NUM_LIT> , <NUM_LIT> ) : <EOL> self . session . default_fetch_size = fetch_size <EOL> self . assertEqual ( <NUM_LIT:100> , len ( list ( self . session . execute_async ( "<STR_LIT>" ) . result ( ) ) ) ) <EOL> statement = SimpleStatement ( "<STR_LIT>" ) <EOL> self . assertEqual ( <NUM_LIT:100> , len ( list ( self . session . execute_async ( statement ) . result ( ) ) ) ) <EOL> self . assertEqual ( <NUM_LIT:100> , len ( list ( self . session . execute_async ( prepared ) . result ( ) ) ) ) <EOL> def test_async_paging_verify_writes ( self ) : <EOL> ddl = '''<STR_LIT>''' <EOL> self . session . execute ( ddl ) <EOL> statements_and_params = zip ( cycle ( [ "<STR_LIT>" <EOL> "<STR_LIT>" ] ) , <EOL> [ ( i , i + <NUM_LIT:1> ) for i in range ( <NUM_LIT:100> ) ] ) <EOL> execute_concurrent ( self . session , statements_and_params ) <EOL> prepared = self . session . prepare ( "<STR_LIT>" ) <EOL> for fetch_size in ( <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:7> , <NUM_LIT:10> , <NUM_LIT> , <NUM_LIT:100> , <NUM_LIT> , <NUM_LIT> ) : <EOL> self . session . default_fetch_size = fetch_size <EOL> results = self . session . execute_async ( "<STR_LIT>" ) . result ( ) <EOL> result_array = [ ] <EOL> value_array = [ ] <EOL> for result in results : <EOL> result_array . append ( result . k2 ) <EOL> value_array . append ( result . v ) <EOL> self . assertSequenceEqual ( range ( <NUM_LIT:100> ) , result_array ) <EOL> self . assertSequenceEqual ( range ( <NUM_LIT:1> , <NUM_LIT> ) , value_array ) <EOL> statement = SimpleStatement ( "<STR_LIT>" ) <EOL> results = self . session . execute_async ( statement ) . result ( ) <EOL> result_array = [ ] <EOL> value_array = [ ] <EOL> for result in results : <EOL> result_array . append ( result . k2 ) <EOL> value_array . append ( result . v ) <EOL> self . assertSequenceEqual ( range ( <NUM_LIT:100> ) , result_array ) <EOL> self . assertSequenceEqual ( range ( <NUM_LIT:1> , <NUM_LIT> ) , value_array ) <EOL> results = self . session . execute_async ( prepared ) . result ( ) <EOL> result_array = [ ] <EOL> value_array = [ ] <EOL> for result in results : <EOL> result_array . append ( result . k2 ) <EOL> value_array . append ( result . v ) <EOL> self . assertSequenceEqual ( range ( <NUM_LIT:100> ) , result_array ) <EOL> self . assertSequenceEqual ( range ( <NUM_LIT:1> , <NUM_LIT> ) , value_array ) <EOL> def test_paging_callbacks ( self ) : <EOL> statements_and_params = zip ( cycle ( [ "<STR_LIT>" ] ) , <EOL> [ ( i , ) for i in range ( <NUM_LIT:100> ) ] ) <EOL> execute_concurrent ( self . session , list ( statements_and_params ) ) <EOL> prepared = self . session . prepare ( "<STR_LIT>" ) <EOL> for fetch_size in ( <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:7> , <NUM_LIT:10> , <NUM_LIT> , <NUM_LIT:100> , <NUM_LIT> , <NUM_LIT> ) : <EOL> self . session . default_fetch_size = fetch_size <EOL> future = self . session . execute_async ( "<STR_LIT>" ) <EOL> event = Event ( ) <EOL> counter = count ( ) <EOL> def handle_page ( rows , future , counter ) : <EOL> for row in rows : <EOL> next ( counter ) <EOL> if future . has_more_pages : <EOL> future . start_fetching_next_page ( ) <EOL> else : <EOL> event . set ( ) <EOL> def handle_error ( err ) : <EOL> event . set ( ) <EOL> self . fail ( err ) <EOL> future . add_callbacks ( callback = handle_page , callback_args = ( future , counter ) , errback = handle_error ) <EOL> event . wait ( ) <EOL> self . assertEqual ( next ( counter ) , <NUM_LIT:100> ) <EOL> future = self . session . execute_async ( SimpleStatement ( "<STR_LIT>" ) ) <EOL> event . clear ( ) <EOL> counter = count ( ) <EOL> future . add_callbacks ( callback = handle_page , callback_args = ( future , counter ) , errback = handle_error ) <EOL> event . wait ( ) <EOL> self . assertEqual ( next ( counter ) , <NUM_LIT:100> ) <EOL> future = self . session . execute_async ( prepared ) <EOL> event . clear ( ) <EOL> counter = count ( ) <EOL> future . add_callbacks ( callback = handle_page , callback_args = ( future , counter ) , errback = handle_error ) <EOL> event . wait ( ) <EOL> self . assertEqual ( next ( counter ) , <NUM_LIT:100> ) <EOL> def test_concurrent_with_paging ( self ) : <EOL> statements_and_params = zip ( cycle ( [ "<STR_LIT>" ] ) , <EOL> [ ( i , ) for i in range ( <NUM_LIT:100> ) ] ) <EOL> execute_concurrent ( self . session , list ( statements_and_params ) ) <EOL> prepared = self . session . prepare ( "<STR_LIT>" ) <EOL> for fetch_size in ( <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:7> , <NUM_LIT:10> , <NUM_LIT> , <NUM_LIT:100> , <NUM_LIT> , <NUM_LIT> ) : <EOL> self . session . default_fetch_size = fetch_size <EOL> results = execute_concurrent_with_args ( self . session , prepared , [ None ] * <NUM_LIT:10> ) <EOL> self . assertEqual ( <NUM_LIT:10> , len ( results ) ) <EOL> for ( success , result ) in results : <EOL> self . assertTrue ( success ) <EOL> self . assertEqual ( <NUM_LIT:100> , len ( list ( result ) ) ) <EOL> def test_fetch_size ( self ) : <EOL> """<STR_LIT>""" <EOL> statements_and_params = zip ( cycle ( [ "<STR_LIT>" ] ) , <EOL> [ ( i , ) for i in range ( <NUM_LIT:100> ) ] ) <EOL> execute_concurrent ( self . session , list ( statements_and_params ) ) <EOL> prepared = self . session . prepare ( "<STR_LIT>" ) <EOL> self . session . default_fetch_size = <NUM_LIT:10> <EOL> result = self . session . execute ( prepared , [ ] ) <EOL> self . assertTrue ( result . has_more_pages ) <EOL> self . session . default_fetch_size = <NUM_LIT> <EOL> result = self . session . execute ( prepared , [ ] ) <EOL> self . assertFalse ( result . has_more_pages ) <EOL> self . session . default_fetch_size = None <EOL> result = self . session . execute ( prepared , [ ] ) <EOL> self . assertFalse ( result . has_more_pages ) <EOL> self . session . default_fetch_size = <NUM_LIT:10> <EOL> prepared . fetch_size = <NUM_LIT> <EOL> result = self . session . execute ( prepared , [ ] ) <EOL> self . assertFalse ( result . has_more_pages ) <EOL> prepared . fetch_size = None <EOL> result = self . session . execute ( prepared , [ ] ) <EOL> self . assertFalse ( result . has_more_pages ) <EOL> prepared . fetch_size = <NUM_LIT:10> <EOL> result = self . session . execute ( prepared , [ ] ) <EOL> self . assertTrue ( result . has_more_pages ) <EOL> prepared . fetch_size = <NUM_LIT> <EOL> bound = prepared . bind ( [ ] ) <EOL> result = self . session . execute ( bound , [ ] ) <EOL> self . assertFalse ( result . has_more_pages ) <EOL> prepared . fetch_size = None <EOL> bound = prepared . bind ( [ ] ) <EOL> result = self . session . execute ( bound , [ ] ) <EOL> self . assertFalse ( result . has_more_pages ) <EOL> prepared . fetch_size = <NUM_LIT:10> <EOL> bound = prepared . bind ( [ ] ) <EOL> result = self . session . execute ( bound , [ ] ) <EOL> self . assertTrue ( result . has_more_pages ) <EOL> bound . fetch_size = <NUM_LIT> <EOL> result = self . session . execute ( bound , [ ] ) <EOL> self . assertFalse ( result . has_more_pages ) <EOL> bound . fetch_size = None <EOL> result = self . session . execute ( bound , [ ] ) <EOL> self . assertFalse ( result . has_more_pages ) <EOL> bound . fetch_size = <NUM_LIT:10> <EOL> result = self . session . execute ( bound , [ ] ) <EOL> self . assertTrue ( result . has_more_pages ) <EOL> s = SimpleStatement ( "<STR_LIT>" , fetch_size = None ) <EOL> result = self . session . execute ( s , [ ] ) <EOL> self . assertFalse ( result . has_more_pages ) <EOL> s = SimpleStatement ( "<STR_LIT>" ) <EOL> result = self . session . execute ( s , [ ] ) <EOL> self . assertTrue ( result . has_more_pages ) <EOL> s = SimpleStatement ( "<STR_LIT>" ) <EOL> s . fetch_size = None <EOL> result = self . session . execute ( s , [ ] ) <EOL> self . assertFalse ( result . has_more_pages ) </s>
