<s> from datetime import datetime <EOL> from cassandra . cqlengine import columns <EOL> from cassandra . cqlengine import functions <EOL> from cassandra . cqlengine import query <EOL> from cassandra . cqlengine . management import sync_table , drop_table <EOL> from cassandra . cqlengine . models import Model <EOL> from cassandra . cqlengine . named import NamedTable <EOL> from cassandra . cqlengine . operators import EqualsOperator <EOL> from cassandra . cqlengine . statements import WhereClause <EOL> from tests . integration . cqlengine import DEFAULT_KEYSPACE <EOL> from tests . integration . cqlengine . base import BaseCassEngTestCase <EOL> from tests . integration . cqlengine import execute_count <EOL> class TestQuerySetOperation ( BaseCassEngTestCase ) : <EOL> def test_maxtimeuuid_function ( self ) : <EOL> """<STR_LIT>""" <EOL> now = datetime . now ( ) <EOL> where = WhereClause ( '<STR_LIT:time>' , EqualsOperator ( ) , functions . MaxTimeUUID ( now ) ) <EOL> where . set_context_id ( <NUM_LIT:5> ) <EOL> self . assertEqual ( str ( where ) , '<STR_LIT>' ) <EOL> ctx = { } <EOL> where . update_context ( ctx ) <EOL> self . assertEqual ( ctx , { '<STR_LIT:5>' : columns . DateTime ( ) . to_database ( now ) } ) <EOL> def test_mintimeuuid_function ( self ) : <EOL> """<STR_LIT>""" <EOL> now = datetime . now ( ) <EOL> where = WhereClause ( '<STR_LIT:time>' , EqualsOperator ( ) , functions . MinTimeUUID ( now ) ) <EOL> where . set_context_id ( <NUM_LIT:5> ) <EOL> self . assertEqual ( str ( where ) , '<STR_LIT>' ) <EOL> ctx = { } <EOL> where . update_context ( ctx ) <EOL> self . assertEqual ( ctx , { '<STR_LIT:5>' : columns . DateTime ( ) . to_database ( now ) } ) <EOL> class TokenTestModel ( Model ) : <EOL> __table_name__ = "<STR_LIT>" <EOL> key = columns . Integer ( primary_key = True ) <EOL> val = columns . Integer ( ) <EOL> class TestTokenFunction ( BaseCassEngTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestTokenFunction , self ) . setUp ( ) <EOL> sync_table ( TokenTestModel ) <EOL> def tearDown ( self ) : <EOL> super ( TestTokenFunction , self ) . tearDown ( ) <EOL> drop_table ( TokenTestModel ) <EOL> @ execute_count ( <NUM_LIT> ) <EOL> def test_token_function ( self ) : <EOL> """<STR_LIT>""" <EOL> assert TokenTestModel . objects ( ) . count ( ) == <NUM_LIT:0> <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> TokenTestModel . create ( key = i , val = i ) <EOL> assert TokenTestModel . objects ( ) . count ( ) == <NUM_LIT:10> <EOL> seen_keys = set ( ) <EOL> last_token = None <EOL> for instance in TokenTestModel . objects ( ) . limit ( <NUM_LIT:5> ) : <EOL> last_token = instance . key <EOL> seen_keys . add ( last_token ) <EOL> assert len ( seen_keys ) == <NUM_LIT:5> <EOL> for instance in TokenTestModel . objects ( pk__token__gt = functions . Token ( last_token ) ) : <EOL> seen_keys . add ( instance . key ) <EOL> assert len ( seen_keys ) == <NUM_LIT:10> <EOL> assert all ( [ i in seen_keys for i in range ( <NUM_LIT:10> ) ] ) <EOL> def test_compound_pk_token_function ( self ) : <EOL> class TestModel ( Model ) : <EOL> p1 = columns . Text ( partition_key = True ) <EOL> p2 = columns . Text ( partition_key = True ) <EOL> func = functions . Token ( '<STR_LIT:a>' , '<STR_LIT:b>' ) <EOL> q = TestModel . objects . filter ( pk__token__gt = func ) <EOL> where = q . _where [ <NUM_LIT:0> ] <EOL> where . set_context_id ( <NUM_LIT:1> ) <EOL> self . assertEqual ( str ( where ) , '<STR_LIT>' . format ( <NUM_LIT:1> , <NUM_LIT:2> ) ) <EOL> str ( q . _select_query ( ) ) <EOL> func = functions . Token ( ( '<STR_LIT:a>' , '<STR_LIT:b>' ) ) <EOL> q = TestModel . objects . filter ( pk__token__gt = func ) <EOL> where = q . _where [ <NUM_LIT:0> ] <EOL> where . set_context_id ( <NUM_LIT:1> ) <EOL> self . assertEqual ( str ( where ) , '<STR_LIT>' . format ( <NUM_LIT:1> , <NUM_LIT:2> ) ) <EOL> str ( q . _select_query ( ) ) <EOL> self . assertRaises ( query . QueryException , TestModel . objects . filter , pk__token__gt = <NUM_LIT:10> ) <EOL> func = functions . Token ( '<STR_LIT:a>' , '<STR_LIT:b>' ) <EOL> self . assertRaises ( query . QueryException , TestModel . objects . filter , p1__gt = func ) <EOL> func = functions . Token ( '<STR_LIT:a>' ) <EOL> self . assertRaises ( query . QueryException , TestModel . objects . filter , pk__token__gt = func ) <EOL> @ execute_count ( <NUM_LIT:7> ) <EOL> def test_named_table_pk_token_function ( self ) : <EOL> """<STR_LIT>""" <EOL> for i in range ( <NUM_LIT:5> ) : <EOL> TokenTestModel . create ( key = i , val = i ) <EOL> named = NamedTable ( DEFAULT_KEYSPACE , TokenTestModel . __table_name__ ) <EOL> query = named . objects . all ( ) . limit ( <NUM_LIT:1> ) <EOL> first_page = list ( query ) <EOL> last = first_page [ - <NUM_LIT:1> ] <EOL> self . assertTrue ( len ( first_page ) is <NUM_LIT:1> ) <EOL> next_page = list ( query . filter ( pk__token__gt = functions . Token ( last . key ) ) ) <EOL> self . assertTrue ( len ( next_page ) is <NUM_LIT:1> ) </s>
