<s> import datetime <EOL> import decimal <EOL> import time <EOL> import sys <EOL> import unittest2 <EOL> import pymysql <EOL> from pymysql . tests import base <EOL> class TestConnection ( base . PyMySQLTestCase ) : <EOL> def test_utf8mb4 ( self ) : <EOL> """<STR_LIT>""" <EOL> arg = self . databases [ <NUM_LIT:0> ] . copy ( ) <EOL> arg [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> conn = pymysql . connect ( ** arg ) <EOL> def test_largedata ( self ) : <EOL> """<STR_LIT>""" <EOL> cur = self . connections [ <NUM_LIT:0> ] . cursor ( ) <EOL> cur . execute ( "<STR_LIT>" ) <EOL> if cur . fetchone ( ) [ <NUM_LIT:0> ] < <NUM_LIT:16> * <NUM_LIT> * <NUM_LIT> + <NUM_LIT:10> : <EOL> print ( "<STR_LIT>" ) <EOL> return <EOL> t = '<STR_LIT:a>' * ( <NUM_LIT:16> * <NUM_LIT> * <NUM_LIT> ) <EOL> cur . execute ( "<STR_LIT>" + t + "<STR_LIT:'>" ) <EOL> assert cur . fetchone ( ) [ <NUM_LIT:0> ] == t <EOL> def test_autocommit ( self ) : <EOL> con = self . connections [ <NUM_LIT:0> ] <EOL> self . assertFalse ( con . get_autocommit ( ) ) <EOL> cur = con . cursor ( ) <EOL> cur . execute ( "<STR_LIT>" ) <EOL> self . assertTrue ( con . get_autocommit ( ) ) <EOL> con . autocommit ( False ) <EOL> self . assertFalse ( con . get_autocommit ( ) ) <EOL> cur . execute ( "<STR_LIT>" ) <EOL> self . assertEqual ( cur . fetchone ( ) [ <NUM_LIT:0> ] , <NUM_LIT:0> ) <EOL> def test_select_db ( self ) : <EOL> con = self . connections [ <NUM_LIT:0> ] <EOL> current_db = self . databases [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> other_db = self . databases [ <NUM_LIT:1> ] [ '<STR_LIT>' ] <EOL> cur = con . cursor ( ) <EOL> cur . execute ( '<STR_LIT>' ) <EOL> self . assertEqual ( cur . fetchone ( ) [ <NUM_LIT:0> ] , current_db ) <EOL> con . select_db ( other_db ) <EOL> cur . execute ( '<STR_LIT>' ) <EOL> self . assertEqual ( cur . fetchone ( ) [ <NUM_LIT:0> ] , other_db ) <EOL> def test_connection_gone_away ( self ) : <EOL> """<STR_LIT>""" <EOL> con = self . connections [ <NUM_LIT:0> ] <EOL> cur = con . cursor ( ) <EOL> cur . execute ( "<STR_LIT>" ) <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> with self . assertRaises ( pymysql . OperationalError ) as cm : <EOL> cur . execute ( "<STR_LIT>" ) <EOL> self . assertIn ( cm . exception . args [ <NUM_LIT:0> ] , ( <NUM_LIT> , <NUM_LIT> ) ) <EOL> def test_init_command ( self ) : <EOL> conn = pymysql . connect ( <EOL> init_command = '<STR_LIT>' , <EOL> ** self . databases [ <NUM_LIT:0> ] <EOL> ) <EOL> c = conn . cursor ( ) <EOL> c . execute ( '<STR_LIT>' ) <EOL> self . assertEqual ( ( '<STR_LIT>' , ) , c . fetchone ( ) ) <EOL> conn . close ( ) <EOL> with self . assertRaises ( pymysql . err . Error ) : <EOL> conn . ping ( reconnect = False ) <EOL> def test_read_default_group ( self ) : <EOL> conn = pymysql . connect ( <EOL> read_default_group = '<STR_LIT>' , <EOL> ** self . databases [ <NUM_LIT:0> ] <EOL> ) <EOL> self . assertTrue ( conn . open ) <EOL> def test_context ( self ) : <EOL> with self . assertRaises ( ValueError ) : <EOL> c = pymysql . connect ( ** self . databases [ <NUM_LIT:0> ] ) <EOL> with c as cur : <EOL> cur . execute ( '<STR_LIT>' ) <EOL> c . begin ( ) <EOL> cur . execute ( '<STR_LIT>' ) <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> c . commit ( ) <EOL> c = pymysql . connect ( ** self . databases [ <NUM_LIT:0> ] ) <EOL> with c as cur : <EOL> cur . execute ( '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:0> , cur . fetchone ( ) [ <NUM_LIT:0> ] ) <EOL> cur . execute ( '<STR_LIT>' ) <EOL> with c as cur : <EOL> cur . execute ( '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , cur . fetchone ( ) [ <NUM_LIT:0> ] ) <EOL> cur . execute ( '<STR_LIT>' ) <EOL> def test_set_charset ( self ) : <EOL> c = pymysql . connect ( ** self . databases [ <NUM_LIT:0> ] ) <EOL> c . set_charset ( '<STR_LIT:utf8>' ) <EOL> def test_defer_connect ( self ) : <EOL> import socket <EOL> for db in self . databases : <EOL> d = db . copy ( ) <EOL> try : <EOL> sock = socket . socket ( socket . AF_UNIX , socket . SOCK_STREAM ) <EOL> sock . connect ( d [ '<STR_LIT>' ] ) <EOL> except KeyError : <EOL> sock = socket . create_connection ( <EOL> ( d . get ( '<STR_LIT:host>' , '<STR_LIT:localhost>' ) , d . get ( '<STR_LIT:port>' , <NUM_LIT> ) ) ) <EOL> for k in [ '<STR_LIT>' , '<STR_LIT:host>' , '<STR_LIT:port>' ] : <EOL> try : <EOL> del d [ k ] <EOL> except KeyError : <EOL> pass <EOL> c = pymysql . connect ( defer_connect = True , ** d ) <EOL> self . assertFalse ( c . open ) <EOL> c . connect ( sock ) <EOL> c . close ( ) <EOL> @ unittest2 . skipUnless ( sys . version_info [ <NUM_LIT:0> : <NUM_LIT:2> ] >= ( <NUM_LIT:3> , <NUM_LIT:2> ) , "<STR_LIT>" ) <EOL> def test_no_delay_warning ( self ) : <EOL> current_db = self . databases [ <NUM_LIT:0> ] . copy ( ) <EOL> current_db [ '<STR_LIT>' ] = True <EOL> with self . assertWarns ( DeprecationWarning ) as cm : <EOL> conn = pymysql . connect ( ** current_db ) <EOL> class Foo ( object ) : <EOL> value = "<STR_LIT:bar>" <EOL> def escape_foo ( x , d ) : <EOL> return x . value <EOL> class TestEscape ( base . PyMySQLTestCase ) : <EOL> def test_escape_string ( self ) : <EOL> con = self . connections [ <NUM_LIT:0> ] <EOL> cur = con . cursor ( ) <EOL> self . assertEqual ( con . escape ( "<STR_LIT>" ) , "<STR_LIT>" ) <EOL> cur . execute ( "<STR_LIT>" ) <EOL> self . assertEqual ( con . escape ( "<STR_LIT>" ) , "<STR_LIT>" ) <EOL> def test_escape_builtin_encoders ( self ) : <EOL> con = self . connections [ <NUM_LIT:0> ] <EOL> cur = con . cursor ( ) <EOL> val = datetime . datetime ( <NUM_LIT> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:6> ) <EOL> self . assertEqual ( con . escape ( val , con . encoders ) , "<STR_LIT>" ) <EOL> def test_escape_custom_object ( self ) : <EOL> con = self . connections [ <NUM_LIT:0> ] <EOL> cur = con . cursor ( ) <EOL> mapping = { Foo : escape_foo } <EOL> self . assertEqual ( con . escape ( Foo ( ) , mapping ) , "<STR_LIT:bar>" ) <EOL> def test_escape_fallback_encoder ( self ) : <EOL> con = self . connections [ <NUM_LIT:0> ] <EOL> cur = con . cursor ( ) <EOL> class Custom ( str ) : <EOL> pass <EOL> mapping = { pymysql . text_type : pymysql . escape_string } <EOL> self . assertEqual ( con . escape ( Custom ( '<STR_LIT>' ) , mapping ) , "<STR_LIT>" ) <EOL> def test_escape_no_default ( self ) : <EOL> con = self . connections [ <NUM_LIT:0> ] <EOL> cur = con . cursor ( ) <EOL> self . assertRaises ( TypeError , con . escape , <NUM_LIT> , { } ) <EOL> def test_escape_dict_value ( self ) : <EOL> con = self . connections [ <NUM_LIT:0> ] <EOL> cur = con . cursor ( ) <EOL> mapping = con . encoders . copy ( ) <EOL> mapping [ Foo ] = escape_foo <EOL> self . assertEqual ( con . escape ( { '<STR_LIT:foo>' : Foo ( ) } , mapping ) , { '<STR_LIT:foo>' : "<STR_LIT:bar>" } ) <EOL> def test_escape_list_item ( self ) : <EOL> con = self . connections [ <NUM_LIT:0> ] <EOL> cur = con . cursor ( ) <EOL> mapping = con . encoders . copy ( ) <EOL> mapping [ Foo ] = escape_foo <EOL> self . assertEqual ( con . escape ( [ Foo ( ) ] , mapping ) , "<STR_LIT>" ) </s>
