text: import datetime
  import sys
  import time
  import unittest2
  import pymysql
  from pymysql.tests import base
  from pymysql._compat import text_type
  
  
  class TempUser:
      def __init__(self, c, user, db, auth=None, authdata=None, password=None):
          self._c = c
          self._user = user
          self._db = db
          create = "CREATE USER " + user
          if password is not None:
              create += " IDENTIFIED BY '%s'" % password
          elif auth is not None:
              create += " IDENTIFIED WITH %s" % auth
              if authdata is not None:
                  create += " AS '%s'" % authdata
          try:
              c.execute(create)
              self._created = True
          except pymysql.err.InternalError:
              # already exists - TODO need to check the same plugin applies
              self._created = False
          try:
              c.execute("GRANT SELECT ON %s.* TO %s" % (db, user))
              self._grant = True
          except pymysql.err.InternalError:
              self._grant = False
  
      def __enter__(self):
          return self
  
      def __exit__(self, exc_type, exc_value, traceback):
          if self._grant:
              self._c.execute("REVOKE SELECT ON %s.* FROM %s" % (self._db, self._user))
          if self._created:
              self._c.execute("DROP USER %s" % self._user)
  
  
  class TestAuthentication(base.PyMySQLTestCase):
      socket_auth = False
      socket_found = False
      two_questions_found = False
      three_attempts_found = False
      pam_found = False
      mysql_old_password_found = False
      sha256_password_found = False
  
      import os
      osuser = os.environ.get('USER')
  
      # socket auth requires the current user and for the connection to be a socket
      # rest do grants @localhost due to incomplete logic - TODO change to @% then
      db = base.PyMySQLTestCase.databases[0].copy()
  
      socket_auth = db.get('unix_socket') is not None \
                    and db.get('host') in ('localhost', '127.0.0.1')
  
      cur = pymysql.connect(**db).cursor()
      del db['user']
      cur.execute("SHOW PLUGINS")
      for r in cur:
          if (r[1], r[2]) != (u'ACTIVE', u'AUTHENTICATION'):
              continue
          if r[3] == u'auth_socket.so':
              socket_plugin_name = r[0]
              socket_found = True
          elif r[3] == u'dialog_examples.so':
              if r[0] == 'two_questions':
                  two_questions_found = True
              elif r[0] == 'three_attempts':
                  three_attempts_found = True
          elif r[0] == u'pam':
              pam_found = True
              pam_plugin_name = r[3].split('.')[0]
              if pam_plugin_name == 'auth_pam':
                  pam_plugin_name = 'pam'
              # MySQL: authentication_pam
              # https://dev.mysql.com/doc/refman/5.5/en/pam-authentication-plugin.html
  
              # MariaDB: pam
              # https://mariadb.com/kb/en/mariadb/pam-authentication-plugin/
  
              # Names differ but functionality is close
          elif r[0] == u'mysql_old_password':
              mysql_old_password_found = True
          elif r[0] == u'sha256_password':
              sha256_password_found = True
          # else:
          #    print("plugin: %r" % r[0])
  
      def test_plugin(self):
          # Bit of an assumption that the current user is a native password
          self.assertEqual('mysql_native_password', self.connections[0]._auth_plugin_name)
  
      @unittest2.skipUnless(socket_auth, "connection to unix_socket required")
      @unittest2.skipIf(socket_found, "socket plugin already installed")
      def testSocketAuthInstallPlugin(self):
          # needs plugin. lets install it.
          cur = self.connections[0].cursor()
          try:
              cur.execute("install plugin auth_socket soname 'auth_socket.so'")
              TestAuthentication.socket_found = True
              self.socket_plugin_name = 'auth_socket'
              self.realtestSocketAuth()
          except pymysql.err.InternalError:
              try:
                  cur.execute("install soname 'auth_socket'")
                  TestAuthentication.socket_found = True
                  self.socket_plugin_name = 'unix_socket'
                  self.realtestSocketAuth()
              except pymysql.err.InternalError:
                  TestAuthentication.socket_found = False
                  raise unittest2.SkipTest('we couldn\'t install the socket plugin')
          finally:
              if TestAuthentication.socket_found:
                  cur.execute("uninstall plugin %s" % self.socket_plugin_name)
  
      @unittest2.skipUnless(socket_auth, "connection to unix_socket required")
      @unittest2.skipUnless(socket_found, "no socket plugin")
      def testSocketAuth(self):
          self.realtestSocketAuth()
  
      def realtestSocketAuth(self):
          with TempUser(self.connections[0].cursor(), TestAuthentication.osuser + '@localhost',
                        self.databases[0]['db'], self.socket_plugin_name) as u:
              c = pymysql.connect(user=TestAuthentication.osuser, **self.db)
  
      class Dialog(object):
          fail = False
  
          def __init__(self, con):
              self.fail = TestAuthentication.Dialog.fail
              pass
  
          def prompt(self, echo, prompt):
              if self.fail:
                  self.fail = False
                  return b'bad guess at a password'
              return self.m.get(prompt)
  
      class DialogHandler(object):
  
          def __init__(self, con):
              self.con = con
  
          def authenticate(self, pkt):
              while True:
                  flag = pkt.read_uint8()
                  echo = (flag & 0x06) == 0x02
                  last = (flag & 0x01) == 0x01
                  prompt = pkt.read_all()
  
                  if prompt == b'Password, please:':
                      self.con.write_packet(b'stillnotverysecret\0')
                  else:
                      self.con.write_packet(b'no idea what to do with this prompt\0')
                  pkt = self.con._read_packet()
                  pkt.check_error()
                  if pkt.is_ok_packet() or last:
                      break
              return pkt
  
      class DefectiveHandler(object):
          def __init__(self, con):
              self.con = con
  
      @unittest2.skipUnless(socket_auth, "connection to unix_socket required")
      @unittest2.skipIf(two_questions_found, "two_questions plugin already installed")
      def testDialogAuthTwoQuestionsInstallPlugin(self):
          # needs plugin. lets install it.
          cur = self.connections[0].cursor()
          try:
              cur.execute("install plugin two_questions soname 'dialog_examples.so'")
              TestAuthentication.two_questions_found = True
              self.realTestDialogAuthTwoQuestions()
          except pymysql.err.InternalError:
              raise unittest2.SkipTest('we couldn\'t install the two_questions plugin')
          finally:
              if TestAuthentication.two_questions_found:
                  cur.execute("uninstall plugin two_questions")
  
      @unittest2.skipUnless(socket_auth, "connection to unix_socket required")
      @unittest2.skipUnless(two_questions_found, "no two questions auth plugin")
      def testDialogAuthTwoQuestions(self):
          self.realTestDialogAuthTwoQuestions()
  
      def realTestDialogAuthTwoQuestions(self):
          TestAuthentication.Dialog.fail = False
          TestAuthentication.Dialog.m = {b'Password, please:': b'notverysecret',
                                         b'Are you sure ?': b'yes, of course'}
          with TempUser(self.connections[0].cursor(), 'pymysql_2q@localhost',
                        self.databases[0]['db'], 'two_questions', 'notverysecret') as u:
              with self.assertRaises(pymysql.err.OperationalError):
                  pymysql.connect(user='pymysql_2q', **self.db)
              pymysql.connect(user='pymysql_2q', auth_plugin_map={b'dialog': TestAuthentication.Dialog}, **self.db)
  
      @unittest2.skipUnless(socket_auth, "connection to unix_socket required")
      @unittest2.skipIf(three_attempts_found, "three_attempts plugin already installed")
      def testDialogAuthThreeAttemptsQuestionsInstallPlugin(self):
          # needs plugin. lets install it.
          cur = self.connections[0].cursor()
          try:
              cur.execute("install plugin three_attempts soname 'dialog_examples.so'")
              TestAuthentication.three_attempts_found = True
              self.realTestDialogAuthThreeAttempts()
          except pymysql.err.InternalError:
              raise unittest2.SkipTest('we couldn\'t install the three_attempts plugin')
          finally:
              if TestAuthentication.three_attempts_found:
                  cur.execute("uninstall plugin three_attempts")
  
      @unittest2.skipUnless(socket_auth, "connection to unix_socket required")
      @unittest2.skipUnless(three_attempts_found, "no three attempts plugin")
      def testDialogAuthThreeAttempts(self):
          self.realTestDialogAuthThreeAttempts()
  
      def realTestDialogAuthThreeAttempts(self):
          TestAuthentication.Dialog.m = {b'Password, please:': b'stillnotverysecret'}
          TestAuthentication.Dialog.fail = True  # fail just once. We've got three attempts after all
          with TempUser(self.connections[0].cursor(), 'pymysql_3a@localhost',
                        self.databases[0]['db'], 'three_attempts', 'stillnotverysecret') as u:
              pymysql.connect(user='pymysql_3a', auth_plugin_map={b'dialog': TestAuthentication.Dialog}, **self.db)
              pymysql.connect(user='pymysql_3a', auth_plugin_map={b'dialog': TestAuthentication.DialogHandler}, **self.db)
              with self.assertRaises(pymysql.err.OperationalError):
                  pymysql.connect(user='pymysql_3a', auth_plugin_map={b'dialog': object}, **self.db)
  
              with self.assertRaises(pymysql.err.OperationalError):
                  pymysql.connect(user='pymysql_3a', auth_plugin_map={b'dialog': TestAuthentication.DefectiveHandler},
                                  **self.db)
              with self.assertRaises(pymysql.err.OperationalError):
                  pymysql.connect(user='pymysql_3a', auth_plugin_map={b'notdialogplugin': TestAuthentication.Dialog},
                                  **self.db)
              TestAuthentication.Dialog.m = {b'Password, please:': b'I do not know'}
              with self.assertRaises(pymysql.err.OperationalError):
                  pymysql.connect(user='pymysql_3a', auth_plugin_map={b'dialog': TestAuthentication.Dialog}, **self.db)
              TestAuthentication.Dialog.m = {b'Password, please:': None}
              with self.assertRaises(pymysql.err.OperationalError):
                  pymysql.connect(user='pymysql_3a', auth_plugin_map={b'dialog': TestAuthentication.Dialog}, **self.db)
  
      @unittest2.skipUnless(socket_auth, "connection to unix_socket required")
      @unittest2.skipIf(pam_found, "pam plugin already installed")
      @unittest2.skipIf(os.environ.get('PASSWORD') is None, "PASSWORD env var required")
      @unittest2.skipIf(os.environ.get('PAMSERVICE') is None, "PAMSERVICE env var required")
      def testPamAuthInstallPlugin(self):
          # needs plugin. lets install it.
          cur = self.connections[0].cursor()
          try:
              cur.execute("install plugin pam soname 'auth_pam.so'")
              TestAuthentication.pam_found = True
              self.realTestPamAuth()
          except pymysql.err.InternalError:
              raise unittest2.SkipTest('we couldn\'t install the auth_pam plugin')
          finally:
              if TestAuthentication.pam_found:
                  cur.execute("uninstall plugin pam")
  
      @unittest2.skipUnless(socket_auth, "connection to unix_socket required")
      @unittest2.skipUnless(pam_found, "no pam plugin")
      @unittest2.skipIf(os.environ.get('PASSWORD') is None, "PASSWORD env var required")
      @unittest2.skipIf(os.environ.get('PAMSERVICE') is None, "PAMSERVICE env var required")
      def testPamAuth(self):
          self.realTestPamAuth()
  
      def realTestPamAuth(self):
          db = self.db.copy()
          import os
          db['password'] = os.environ.get('PASSWORD')
          cur = self.connections[0].cursor()
          try:
              cur.execute('show grants for ' + TestAuthentication.osuser + '@localhost')
              grants = cur.fetchone()[0]
              cur.execute('drop user ' + TestAuthentication.osuser + '@localhost')
          except pymysql.OperationalError as e:
              # assuming the user doesn't exist which is ok too
              self.assertEqual(1045, e.args[0])
              grants = None
          with TempUser(cur, TestAuthentication.osuser + '@localhost',
                        self.databases[0]['db'], 'pam', os.environ.get('PAMSERVICE')) as u:
              try:
                  c = pymysql.connect(user=TestAuthentication.osuser, **db)
                  db['password'] = 'very bad guess at password'
                  with self.assertRaises(pymysql.err.OperationalError):
                      pymysql.connect(user=TestAuthentication.osuser,
                                      auth_plugin_map={b'mysql_cleartext_password': TestAuthentication.DefectiveHandler},
                                      **self.db)
              except pymysql.OperationalError as e:
                  self.assertEqual(1045, e.args[0])
                  # we had 'bad guess at password' work with pam. Well at least we get a permission denied here
                  with self.assertRaises(pymysql.err.OperationalError):
                      pymysql.connect(user=TestAuthentication.osuser,
                                      auth_plugin_map={b'mysql_cleartext_password': TestAuthentication.DefectiveHandler},
                                      **self.db)
          if grants:
              # recreate the user
              cur.execute(grants)
  
      # select old_password("crummy p\tassword");
      # | old_password("crummy p\tassword") |
      # | 2a01785203b08770                  |
      @unittest2.skipUnless(socket_auth, "connection to unix_socket required")
      @unittest2.skipUnless(mysql_old_password_found, "no mysql_old_password plugin")
      def testMySQLOldPasswordAuth(self):
          if self.mysql_server_is(self.connections[0], (5, 7, 0)):
              raise unittest2.SkipTest('Old passwords aren\'t supported in 5.7')
          # pymysql.err.OperationalError: (1045, "Access denied for user 'old_pass_user'@'localhost' (using password: YES)")
          # from login in MySQL-5.6
          if self.mysql_server_is(self.connections[0], (5, 6, 0)):
              raise unittest2.SkipTest('Old passwords don\'t authenticate in 5.6')
          db = self.db.copy()
          db['password'] = "crummy p\tassword"
          with self.connections[0] as c:
              # deprecated in 5.6
              if sys.version_info[0:2] >= (3, 2) and self.mysql_server_is(self.connections[0], (5, 6, 0)):
                  with self.assertWarns(pymysql.err.Warning) as cm:
                      c.execute("SELECT OLD_PASSWORD('%s')" % db['password'])
              else:
                  c.execute("SELECT OLD_PASSWORD('%s')" % db['password'])
              v = c.fetchone()[0]
              self.assertEqual(v, '2a01785203b08770')
              # only works in MariaDB and MySQL-5.6 - can't separate out by version
              # if self.mysql_server_is(self.connections[0], (5, 5, 0)):
              #    with TempUser(c, 'old_pass_user@localhost',
              #                  self.databases[0]['db'], 'mysql_old_password', '2a01785203b08770') as u:
              #        cur = pymysql.connect(user='old_pass_user', **db).cursor()
              #        cur.execute("SELECT VERSION()")
              c.execute("SELECT @@secure_auth")
              secure_auth_setting = c.fetchone()[0]
              c.execute('set old_passwords=1')
              # pymysql.err.Warning: 'pre-4.1 password hash' is deprecated and will be removed in a future release. Please use post-4.1 password hash instead
              if sys.version_info[0:2] >= (3, 2) and self.mysql_server_is(self.connections[0], (5, 6, 0)):
                  with self.assertWarns(pymysql.err.Warning) as cm:
                      c.execute('set global secure_auth=0')
              else:
                  c.execute('set global secure_auth=0')
              with TempUser(c, 'old_pass_user@localhost',
                            self.databases[0]['db'], password=db['password']) as u:
                  cur = pymysql.connect(user='old_pass_user', **db).cursor()
                  cur.execute("SELECT VERSION()")
              c.execute('set global secure_auth=%r' % secure_auth_setting)
  
      @unittest2.skipUnless(socket_auth, "connection to unix_socket required")
      @unittest2.skipUnless(sha256_password_found, "no sha256 password authentication plugin found")
      def testAuthSHA256(self):
          c = self.connections[0].cursor()
          with TempUser(c, 'pymysql_sha256@localhost',
                        self.databases[0]['db'], 'sha256_password') as u:
              if self.mysql_server_is(self.connections[0], (5, 7, 0)):
                  c.execute("SET PASSWORD FOR 'pymysql_sha256'@'localhost' ='Sh@256Pa33'")
              else:
                  c.execute('SET old_passwords = 2')
                  c.execute("SET PASSWORD FOR 'pymysql_sha256'@'localhost' = PASSWORD('Sh@256Pa33')")
              db = self.db.copy()
              db['password'] = "Sh@256Pa33"
              # not implemented yet so thows error
              with self.assertRaises(pymysql.err.OperationalError):
                  pymysql.connect(user='pymysql_256', **db)
  
  
  class TestConnection(base.PyMySQLTestCase):
  
      def test_utf8mb4(self):
          """This test requires MySQL >= 5.5"""
          arg = self.databases[0].copy()
          arg['charset'] = 'utf8mb4'
          conn = pymysql.connect(**arg)
  
      def test_largedata(self):
          """Large query and response (>=16MB)"""
          cur = self.connections[0].cursor()
          cur.execute("SELECT @@max_allowed_packet")
          if cur.fetchone()[0] < 16 * 1024 * 1024 + 10:
              print("Set max_allowed_packet to bigger than 17MB")
              return
          t = 'a' * (16 * 1024 * 1024)
          cur.execute("SELECT '" + t + "'")
          assert cur.fetchone()[0] == t
  
      def test_autocommit(self):
          con = self.connections[0]
          self.assertFalse(con.get_autocommit())
  
          cur = con.cursor()
          cur.execute("SET AUTOCOMMIT=1")
          self.assertTrue(con.get_autocommit())
  
          con.autocommit(False)
          self.assertFalse(con.get_autocommit())
          cur.execute("SELECT @@AUTOCOMMIT")
          self.assertEqual(cur.fetchone()[0], 0)
  
      def test_select_db(self):
          con = self.connections[0]
          current_db = self.databases[0]['db']
          other_db = self.databases[1]['db']
  
          cur = con.cursor()
          cur.execute('SELECT database()')
          self.assertEqual(cur.fetchone()[0], current_db)
  
          con.select_db(other_db)
          cur.execute('SELECT database()')
          self.assertEqual(cur.fetchone()[0], other_db)
  
      def test_connection_gone_away(self):
          """
          http://dev.mysql.com/doc/refman/5.0/en/gone-away.html
          http://dev.mysql.com/doc/refman/5.0/en/error-messages-client.html#error_cr_server_gone_error
          """
          con = self.connections[0]
          cur = con.cursor()
          cur.execute("SET wait_timeout=1")
          time.sleep(2)
          with self.assertRaises(pymysql.OperationalError) as cm:
              cur.execute("SELECT 1+1")
          # error occures while reading, not writing because of socket buffer.
          # self.assertEqual(cm.exception.args[0], 2006)
          self.assertIn(cm.exception.args[0], (2006, 2013))
  
      def test_init_command(self):
          conn = pymysql.connect(
              init_command='SELECT "bar"; SELECT "baz"',
              **self.databases[0]
          )
          c = conn.cursor()
          c.execute('select "foobar";')
          self.assertEqual(('foobar',), c.fetchone())
          conn.close()
          with self.assertRaises(pymysql.err.Error):
              conn.ping(reconnect=False)
  
      def test_read_default_group(self):
          conn = pymysql.connect(
              read_default_group='client',
              **self.databases[0]
          )
          self.assertTrue(conn.open)
  
      def test_context(self):
          with self.assertRaises(ValueError):
              c = pymysql.connect(**self.databases[0])
              with c as cur:
                  cur.execute('create table test ( a int )')
                  c.begin()
                  cur.execute('insert into test values ((1))')
                  raise ValueError('pseudo abort')
                  c.commit()
          c = pymysql.connect(**self.databases[0])
          with c as cur:
              cur.execute('select count(*) from test')
              self.assertEqual(0, cur.fetchone()[0])
              cur.execute('insert into test values ((1))')
          with c as cur:
              cur.execute('select count(*) from test')
              self.assertEqual(1, cur.fetchone()[0])
              cur.execute('drop table test')
  
      def test_set_charset(self):
          c = pymysql.connect(**self.databases[0])
          c.set_charset('utf8')
          # TODO validate setting here
  
      def test_defer_connect(self):
          import socket
          for db in self.databases:
              d = db.copy()
              try:
                  sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
                  sock.connect(d['unix_socket'])
              except KeyError:
                  sock = socket.create_connection(
                      (d.get('host', 'localhost'), d.get('port', 3306)))
              for k in ['unix_socket', 'host', 'port']:
                  try:
                      del d[k]
                  except KeyError:
                      pass
  
              c = pymysql.connect(defer_connect=True, **d)
              self.assertFalse(c.open)
              c.connect(sock)
              c.close()
              sock.close()
  
      @unittest2.skipUnless(sys.version_info[0:2] >= (3, 2), "required py-3.2")
      def test_no_delay_warning(self):
          current_db = self.databases[0].copy()
          current_db['no_delay'] = True
          with self.assertWarns(DeprecationWarning) as cm:
              conn = pymysql.connect(**current_db)
  
  
  # A custom type and function to escape it
  class Foo(object):
      value = "bar"
  
  
  def escape_foo(x, d):
      return x.value
  
  
  class TestEscape(base.PyMySQLTestCase):
      def test_escape_string(self):
          con = self.connections[0]
          cur = con.cursor()
  
          self.assertEqual(con.escape("foo'bar"), "'foo\\'bar'")
          # added NO_AUTO_CREATE_USER as not including it in 5.7 generates warnings
          cur.execute("SET sql_mode='NO_BACKSLASH_ESCAPES,NO_AUTO_CREATE_USER'")
          self.assertEqual(con.escape("foo'bar"), "'foo''bar'")
  
      def test_escape_builtin_encoders(self):
          con = self.connections[0]
          cur = con.cursor()
  
          val = datetime.datetime(2012, 3, 4, 5, 6)
          self.assertEqual(con.escape(val, con.encoders), "'2012-03-04 05:06:00'")
  
      def test_escape_custom_object(self):
          con = self.connections[0]
          cur = con.cursor()
  
          mapping = {Foo: escape_foo}
          self.assertEqual(con.escape(Foo(), mapping), "bar")
  
      def test_escape_fallback_encoder(self):
          con = self.connections[0]
          cur = con.cursor()
  
          class Custom(str):
              pass
  
          mapping = {text_type: pymysql.escape_string}
          self.assertEqual(con.escape(Custom('foobar'), mapping), "'foobar'")
  
      def test_escape_no_default(self):
          con = self.connections[0]
          cur = con.cursor()
  
          self.assertRaises(TypeError, con.escape, 42, {})
  
      def test_escape_dict_value(self):
          con = self.connections[0]
          cur = con.cursor()
  
          mapping = con.encoders.copy()
          mapping[Foo] = escape_foo
          self.assertEqual(con.escape({'foo': Foo()}, mapping), {'foo': "bar"})
  
      def test_escape_list_item(self):
          con = self.connections[0]
          cur = con.cursor()
  
          mapping = con.encoders.copy()
          mapping[Foo] = escape_foo
          self.assertEqual(con.escape([Foo()], mapping), "(bar)")
  
      def test_previous_cursor_not_closed(self):
          con = self.connections[0]
          cur1 = con.cursor()
          cur1.execute("SELECT 1; SELECT 2")
          cur2 = con.cursor()
          cur2.execute("SELECT 3")
          self.assertEqual(cur2.fetchone()[0], 3)
  
      def test_commit_during_multi_result(self):
          con = self.connections[0]
          cur = con.cursor()
          cur.execute("SELECT 1; SELECT 2")
          con.commit()
          cur.execute("SELECT 3")
          self.assertEqual(cur.fetchone()[0], 3)
  