<s> import sqlalchemy as sa <EOL> from sqlalchemy . ext . declarative import declarative_base <EOL> from crate . client . sqlalchemy . types import Object <EOL> from crate . client . cursor import Cursor <EOL> from mock import patch , MagicMock <EOL> from unittest import TestCase <EOL> fake_cursor = MagicMock ( name = '<STR_LIT>' ) <EOL> FakeCursor = MagicMock ( name = '<STR_LIT>' , spec = Cursor ) <EOL> FakeCursor . return_value = fake_cursor <EOL> @ patch ( '<STR_LIT>' , FakeCursor ) <EOL> class CreateTableTest ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . engine = sa . create_engine ( '<STR_LIT>' ) <EOL> self . Base = declarative_base ( bind = self . engine ) <EOL> def test_create_table_with_basic_types ( self ) : <EOL> class User ( self . Base ) : <EOL> __tablename__ = '<STR_LIT>' <EOL> string_col = sa . Column ( sa . String , primary_key = True ) <EOL> unicode_col = sa . Column ( sa . Unicode ) <EOL> text_col = sa . Column ( sa . Text ) <EOL> int_col = sa . Column ( sa . Integer ) <EOL> long_col = sa . Column ( sa . BigInteger ) <EOL> bool_col = sa . Column ( sa . Boolean ) <EOL> short_col = sa . Column ( sa . SmallInteger ) <EOL> ts_col = sa . Column ( sa . DateTime ) <EOL> float_col = sa . Column ( sa . Float ) <EOL> self . Base . metadata . create_all ( ) <EOL> fake_cursor . execute . assert_called_with ( <EOL> ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> ( ) ) <EOL> def test_with_obj_column ( self ) : <EOL> class DummyTable ( self . Base ) : <EOL> __tablename__ = '<STR_LIT>' <EOL> pk = sa . Column ( sa . String , primary_key = True ) <EOL> obj_col = sa . Column ( Object ) <EOL> self . Base . metadata . create_all ( ) <EOL> fake_cursor . execute . assert_called_with ( <EOL> ( '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> ( ) ) <EOL> def test_with_clustered_by ( self ) : <EOL> class DummyTable ( self . Base ) : <EOL> __tablename__ = '<STR_LIT:t>' <EOL> __table_args__ = { <EOL> '<STR_LIT>' : '<STR_LIT:p>' <EOL> } <EOL> pk = sa . Column ( sa . String , primary_key = True ) <EOL> p = sa . Column ( sa . String ) <EOL> self . Base . metadata . create_all ( ) <EOL> fake_cursor . execute . assert_called_with ( <EOL> ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> ( ) ) <EOL> def test_with_partitioned_by ( self ) : <EOL> class DummyTable ( self . Base ) : <EOL> __tablename__ = '<STR_LIT:t>' <EOL> __table_args__ = { <EOL> '<STR_LIT>' : '<STR_LIT:p>' , <EOL> '<STR_LIT>' : <NUM_LIT:1> <EOL> } <EOL> pk = sa . Column ( sa . String , primary_key = True ) <EOL> p = sa . Column ( sa . String ) <EOL> self . Base . metadata . create_all ( ) <EOL> fake_cursor . execute . assert_called_with ( <EOL> ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> ( ) ) <EOL> def test_with_number_of_shards_and_replicas ( self ) : <EOL> class DummyTable ( self . Base ) : <EOL> __tablename__ = '<STR_LIT:t>' <EOL> __table_args__ = { <EOL> '<STR_LIT>' : '<STR_LIT:2>' , <EOL> '<STR_LIT>' : <NUM_LIT:3> <EOL> } <EOL> pk = sa . Column ( sa . String , primary_key = True ) <EOL> self . Base . metadata . create_all ( ) <EOL> fake_cursor . execute . assert_called_with ( <EOL> ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> ( ) ) <EOL> def test_with_clustered_by_and_number_of_shards ( self ) : <EOL> class DummyTable ( self . Base ) : <EOL> __tablename__ = '<STR_LIT:t>' <EOL> __table_args__ = { <EOL> '<STR_LIT>' : '<STR_LIT:p>' , <EOL> '<STR_LIT>' : <NUM_LIT:3> <EOL> } <EOL> pk = sa . Column ( sa . String , primary_key = True ) <EOL> p = sa . Column ( sa . String , primary_key = True ) <EOL> self . Base . metadata . create_all ( ) <EOL> fake_cursor . execute . assert_called_with ( <EOL> ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> ( ) ) </s>
