<s> from copy import deepcopy <EOL> from cassandra import ConsistencyLevel <EOL> from cassandra . auth import PlainTextAuthProvider <EOL> from cassandra . cluster import Cluster <EOL> from mock import patch , Mock <EOL> from unittest import TestCase <EOL> from django_cassandra_engine . connection import CassandraConnection , Cursor <EOL> from django_cassandra_engine . utils import get_cassandra_connection <EOL> class CassandraConnectionTestCase ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . cassandra_connection = get_cassandra_connection ( ) <EOL> self . connection = self . cassandra_connection . connection <EOL> self . cassandra_connection . connect ( ) <EOL> def test_cursor ( self ) : <EOL> self . assertIsInstance ( self . connection . cursor ( ) , Cursor ) <EOL> self . assertEqual ( self . connection . cursor ( ) . connection , self . connection ) <EOL> def test_connected_to_db ( self ) : <EOL> from cassandra . cqlengine import connection as cql_connection <EOL> self . assertIsInstance ( cql_connection . cluster , Cluster ) <EOL> self . assertIsNotNone ( cql_connection . session ) <EOL> def test_session_property ( self ) : <EOL> from cassandra . cqlengine import connection as cql_connection <EOL> self . assertEqual ( self . connection . session , cql_connection . session ) <EOL> def test_cluster_property ( self ) : <EOL> from cassandra . cqlengine import connection as cql_connection <EOL> self . assertEqual ( self . connection . cluster , cql_connection . cluster ) <EOL> def test_connection_options ( self ) : <EOL> connection_options = self . cassandra_connection . settings_dict [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . assertEqual ( <EOL> self . connection . connection_options , connection_options ) <EOL> @ patch ( "<STR_LIT>" ) <EOL> def test_connection_setup_called_first_time_with_proper_options ( <EOL> self , connection_mock ) : <EOL> settings = self . cassandra_connection . settings_dict <EOL> connection_mock . get_session . return_value = None <EOL> connection = CassandraConnection ( ** settings ) <EOL> connection_mock . setup . assert_called_once_with ( <EOL> connection . hosts , connection . keyspace , <EOL> ** settings [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> @ patch ( "<STR_LIT>" ) <EOL> def test_connection_setup_called_second_time_session_not_none ( <EOL> self , connection_mock ) : <EOL> settings = self . cassandra_connection . settings_dict <EOL> session_mock = Mock ( ) <EOL> session_mock . is_shutdown = False <EOL> connection_mock . get_session . return_value = session_mock <EOL> CassandraConnection ( ** settings ) <EOL> self . assertFalse ( connection_mock . setup . called ) <EOL> @ patch ( "<STR_LIT>" ) <EOL> def test_connection_setup_called_second_time_session_is_shutdown ( <EOL> self , connection_mock ) : <EOL> settings = self . cassandra_connection . settings_dict <EOL> session_mock = Mock ( ) <EOL> session_mock . is_shutdown = True <EOL> connection_mock . get_session . return_value = session_mock <EOL> CassandraConnection ( ** settings ) <EOL> self . assertTrue ( connection_mock . setup . called ) <EOL> def test_connection_auth_provider_added_to_connection_options ( self ) : <EOL> settings = self . cassandra_connection . settings_dict <EOL> settings [ '<STR_LIT>' ] = '<STR_LIT:user>' <EOL> settings [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> connection = CassandraConnection ( ** settings ) <EOL> self . assertIsInstance ( connection . connection_options [ '<STR_LIT>' ] , <EOL> PlainTextAuthProvider ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_connection_auth_provider_not_changed ( self , setup_mock ) : <EOL> settings = deepcopy ( self . cassandra_connection . settings_dict ) <EOL> settings [ '<STR_LIT>' ] = '<STR_LIT:user>' <EOL> settings [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> settings [ '<STR_LIT>' ] [ '<STR_LIT>' ] = { } <EOL> settings [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> connection = CassandraConnection ( ** settings ) <EOL> self . assertEqual ( connection . connection_options [ '<STR_LIT>' ] , <EOL> settings [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> setup_mock . assert_called_once ( ) <EOL> def test_connection_session_options_default_timeout ( self ) : <EOL> session_opts = self . cassandra_connection . settings_dict [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . assertEqual ( self . connection . session_options , session_opts ) <EOL> self . assertEqual ( self . connection . session . default_timeout , <EOL> session_opts . get ( '<STR_LIT>' ) ) <EOL> def test_connection_session_default_consistency ( self ) : <EOL> settings = self . cassandra_connection . settings_dict <EOL> settings [ '<STR_LIT>' ] [ '<STR_LIT>' ] = { <EOL> '<STR_LIT>' : ConsistencyLevel . ALL <EOL> } <EOL> connection = CassandraConnection ( ** settings ) <EOL> self . assertEqual ( connection . session . default_consistency_level , <EOL> ConsistencyLevel . ALL ) <EOL> def test_raw_cql_cursor_queries ( self ) : <EOL> cursor = self . connection . cursor ( ) <EOL> self . assertEqual ( <EOL> cursor . execute ( "<STR_LIT>" ) [ <NUM_LIT:0> ] [ '<STR_LIT:count>' ] , <NUM_LIT:0> <EOL> ) <EOL> cursor . execute ( "<STR_LIT>" ) <EOL> self . assertEqual ( <EOL> cursor . execute ( "<STR_LIT>" ) [ <NUM_LIT:0> ] [ '<STR_LIT:count>' ] , <NUM_LIT:1> <EOL> ) </s>
