Initial import of ZMySQLDDA unit test.

git-svn-id: https://svn.erp5.org/repos/public/erp5/trunk@13709 20353a03-c40f-0410-a6d1-a30d3c3de9de

product/ZMySQLDDA/tests/testDefferedConnection.py

+##############################################################################
+#
+# Copyright (c) 2006 Nexedi SARL and Contributors. All Rights Reserved.
+#                     Vincent Pelletier <vincent@nexedi.com>
+#
+# WARNING: This program as such is intended to be used by professional
+# programmers who take the whole responsability of assessing all potential
+# consequences resulting from its eventual inadequacies and bugs
+# End users who are looking for a ready-to-use solution with commercial
+# garantees and support are strongly adviced to contract a Free Software
+# Service Company
+#
+# This program is Free Software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either version 2
+# of the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
+#
+##############################################################################
+
+from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
+from AccessControl.SecurityManagement import newSecurityManager
+from _mysql_exceptions import OperationalError
+from Products.ZMySQLDA.db import hosed_connection
+from zLOG import LOG
+
+UNCONNECTED_STATE = 0
+CONNECTED_STATE = 1
+GLOBAL_DB_CONNECTED_FLAG = UNCONNECTED_STATE
+
+def fake_db_store_result(self, *args, **kw):
+  """
+    Mimic store_result to make sure it doesn't fail due to no executed queries.
+  """
+  return
+
+def fake_connection_forceReconnection(self):
+  """
+    Intercept reconnection symptom.
+  """
+  global GLOBAL_DB_CONNECTED_FLAG
+  GLOBAL_DB_CONNECTED_FLAG = CONNECTED_STATE
+  return self.original_forceReconnection()
+
+def fake_db_query(self, *args, **kw):
+  """
+    Mimic a failing query due to a disconnected socket from mysql server.
+  """
+  global GLOBAL_DB_CONNECTED_FLAG
+  if GLOBAL_DB_CONNECTED_FLAG == UNCONNECTED_STATE:
+    raise OperationalError, (hosed_connection[0], 'dummy exception')
+  return self.original_query(*args, **kw)
+
+class TestDefferedConnection(ERP5TypeTestCase):
+  """
+    Test MySQL Deffered Connection
+  """
+
+  def getBusinessTemplateList(self):
+    return tuple()
+
+  def getTitle(self):
+    return "Deffered Connection"
+
+  def setUp(self):
+    ERP5TypeTestCase.setUp(self)
+
+  def afterSetUp(self):
+    self.login()
+
+  def login(self):
+    uf = self.getPortal().acl_users
+    uf._doAddUser('vincent', '', ['Manager'], [])
+    user = uf.getUserById('vincent').__of__(uf)
+    newSecurityManager(None, user)
+
+  def monkeypatchConnection(self, connection):
+    """
+      Apply monkey patch on db and reset connection state to "unconnected".
+      Returns a tuple containing original functions.
+    """
+    mysql_class = connection._getConnection().__class__
+    mysql_class.original_query = mysql_class.query
+    mysql_class.query = fake_db_query
+    connection.__class__.original_forceReconnection = connection.__class__._forceReconnection
+    connection.__class__._forceReconnection = fake_connection_forceReconnection
+    GLOBAL_DB_CONNECTED_FLAG = UNCONNECTED_STATE
+
+  def unmonkeypatchConnection(self, connection):
+    """
+      Revert monkeypatching done on db.
+    """
+    connection.__class__._forceReconnection = connection.__class__.original_forceReconnection
+    delattr(connection.__class__, 'original_forceReconnection')
+    mysql_class = connection._getConnection().__class__
+    mysql_class.query = mysql_class.original_query
+    delattr(mysql_class, 'original_query')
+
+  def getDefferedConnection(self):
+    """
+      Return site's deffered connection object.
+    """
+    deffered = self.getPortal().erp5_sql_deferred_connection
+    deffered_connection = getattr(deffered, '_v_database_connection', None)
+    if getattr(deffered, '_v_database_connection', None) is None:
+      deffered.connect(deffered.connection_string)
+      deffered_connection = getattr(deffered, '_v_database_connection')
+    return deffered_connection
+
+  def test_00_basicReplaceQuery(self):
+    """
+      Check that a basic query succeeds.
+    """
+    connection = self.getDefferedConnection()
+    connection.query('REPLACE INTO `full_text` SET `uid`=0, `SearchableText`="dummy test"')
+    try:
+      get_transaction().commit()
+    except OperationalError:
+      self.fail()
+    except:
+      raise # Make sure the test is known to have failed, even if it's not
+            # the expected execution path.
+
+  def test_01_disconnectsCausesError(self):
+    """
+      Check that a disconnection from mysql causes classical
+      connection.db.query to fail.
+      This makes sure that disconnection-trick monkey patch does work.
+    """
+    connection = self.getDefferedConnection()
+    # Queue a query
+    connection.query('REPLACE INTO `full_text` SET `uid`=0, `SearchableText`="dummy test"')
+    # Replace dynamically the function used to send queries to mysql so it's
+    # dumber than the implemented one.
+    self.monkeypatchConnection(connection)
+    connection._query = connection._getConnection().query
+    try:
+      try:
+        get_transaction().commit()
+      except OperationalError, m:
+        if m[0] not in hosed_connection:
+          raise
+      except:
+        raise # Make sure the test is known to have failed, even if it's not
+              # the expected execution path.
+      else:
+        self.fail()
+    finally:
+      delattr(connection, '_query')
+      self.unmonkeypatchConnection(connection)
+
+  def test_02_disconnectionRobustness(self):
+    """
+      Check that if the connection gets closed before being used the
+      commit can happen without trouble.
+    """
+    connection = self.getDefferedConnection()
+    # Queue a query
+    connection.query('REPLACE INTO `full_text` SET `uid`=0, `SearchableText`="dummy test"')
+    # Artificially cause a connection close.
+    self.monkeypatchConnection(connection)
+    try:
+      try:
+        get_transaction().commit()
+      except OperationalError, m:
+        LOG('TestDefferedConnection', 0, 'OperationalError exception raised: %s' % (m, ))
+        self.fail()
+      except:
+        raise # Make sure the test is known to have failed, even if it's not
+              # the expected execution path.
+    finally:
+      self.unmonkeypatchConnection(connection)
+
+  def test_03_successiveTransactionsIsolation(self):
+    """
+      Check that multiple transactions (one after another) are correctly
+      isolated one from the other.
+    """
+    connection = self.getDefferedConnection()
+    # Queue a query
+    connection.query('REPLACE INTO `full_text` SET `uid`=0, `SearchableText`="dummy test"')
+    self.assertEqual(len(connection._getSQLStringList()), 1)
+    get_transaction().commit()
+    connection.query('REPLACE INTO `full_text` SET `uid`=0, `SearchableText`="dummy test"')
+    self.assertEqual(len(connection._getSQLStringList()), 1)
+
+if __name__ == '__main__':
+  unittest.main()