honor --erp5_sql_connection_string for --save / --load (adhoc)

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

product/ERP5Type/tests/ERP5TypeTestCase.py

@@ -25,6 +25,7 @@ Globals.get_request = get_request
 
 from Testing import ZopeTestCase
 from Testing.ZopeTestCase.PortalTestCase import PortalTestCase, user_name
+from Products.ERP5Type.tests.utils import getMySQLArguments
 from Products.CMFCore.utils import getToolByName
 from Products.ERP5Type.Utils import getLocalPropertySheetList, \
                                     removeLocalPropertySheet, \
@@ -137,21 +138,17 @@ from Products.ERP5Type.InitGenerator import initializeProductDocumentRegistry
 initializeProductDocumentRegistry()
 
 from AccessControl.SecurityManagement import newSecurityManager, noSecurityManager
-from AccessControl.User import User
 
 from Acquisition import aq_base
 import time
 import md5
 import traceback
-import sys
 import os
 from cStringIO import StringIO
 from urllib import urlretrieve
 from glob import glob
 import pysvn
 
-from Products.ERP5.ERP5Site import ERP5Site
-
 portal_name = 'erp5_portal'
 
 # we keep a reference to all sites for wich setup failed the first time, to
@@ -188,7 +185,8 @@ class ERP5TypeTestCase(PortalTestCase):
 
     def getRevision(self):
       try:
-        return pysvn.Client().info('%s/Products/ERP5' % os.environ['INSTANCE_HOME']).revision.number
+        return pysvn.Client().info('%s/Products/ERP5'
+                    % os.environ['INSTANCE_HOME']).revision.number
       except:
         return None
 
@@ -602,7 +600,8 @@ def setupERP5Site( business_template_list=(),
             if not quiet:
               ZopeTestCase._print('Dumping MySQL database ... ')
             instance_home = os.environ['INSTANCE_HOME']
-            os.system('mysqldump -u test test > %s/dump.sql' % instance_home)
+            os.system('mysqldump %s > %s/dump.sql' % (
+                      getMySQLArguments(), instance_home))
             if not quiet:
               ZopeTestCase._print('Dumping static files ... ')
             for dir in ('Constraint', 'Document', 'PropertySheet'):

product/ERP5Type/tests/custom_zodb.py

@@ -2,6 +2,7 @@ import ZODB
 import os
 from ZODB.DemoStorage import DemoStorage
 from ZODB.FileStorage import FileStorage
+from Products.ERP5Type.tests.utils import getMySQLArguments
 
 instance_home = os.environ.get('INSTANCE_HOME')
 data_fs_path = os.environ.get('erp5_tests_data_fs_path')
@@ -17,7 +18,7 @@ elif os.environ.get('erp5_load_data_fs'):
   else:
     Storage = DemoStorage(base=FileStorage(new_data_fs_path), quota=(1<<20))
   print("Restoring MySQL database ... ")
-  os.system("mysql -u test test < %s/dump.sql" % instance_home)
+  os.system("mysql %s < %s/dump.sql" % (getMySQLArguments(), instance_home))
   print("Restoring static files ... ")
   for dir in ('Constraint', 'Document', 'PropertySheet'):
     if os.path.exists('%s/%s.bak' % (instance_home, dir)):

product/ERP5Type/tests/utils.py

@@ -29,6 +29,8 @@
 """Utility functions and classes for unit testing
 """
 
+import os
+
 import Products.ERP5Type
 from Products.MailHost.MailHost import MailHost
 
@@ -108,6 +110,18 @@ def _recreateMemcachedTool(portal):
   portal.manage_delObjects(['portal_memcached'])
   portal._setObject('portal_memcached', MemcachedTool.MemcachedTool())
 
+# test runner shared functions
+def getMySQLArguments():
+  """Returns arguments to pass to mysql by heuristically converting the
+  connection string.
+  Only simple cases are coverred for now.
+  """
+  connection_string = os.environ.get('erp5_sql_connection_string')
+  if not connection_string:
+    return '-u test test'
+  db, user = connection_string.split()
+  return '-u %s %s' % (user, db)
+
 # decorators
 class reindex(object):
   """Decorator to commit transaction and flush activities after the method is