diff options
| -rw-r--r-- | test/testlib/config.py | 38 | ||||
| -rw-r--r-- | test/testlib/engines.py | 17 |
2 files changed, 35 insertions, 20 deletions
diff --git a/test/testlib/config.py b/test/testlib/config.py index df2e81b0c..77cacc820 100644 --- a/test/testlib/config.py +++ b/test/testlib/config.py @@ -1,5 +1,5 @@ import testbase -import optparse, os, sys, re, ConfigParser, StringIO, time +import optparse, os, sys, re, ConfigParser, StringIO, time, warnings logging, require = None, None @@ -205,21 +205,29 @@ def _prep_testing_database(options, file_config): from testlib import engines from sqlalchemy import schema - # also create alt schemas etc. here? - e = engines.utf8_engine() - existing = e.table_names() - if existing: + try: + # also create alt schemas etc. here? + e = engines.utf8_engine() + existing = e.table_names() + if existing: + if not options.quiet: + print "Dropping existing tables in database: " + db_url + try: + print "Tables: %s" % ', '.join(existing) + except: + pass + print "Abort within 5 seconds..." + time.sleep(5) + md = schema.MetaData(e, reflect=True) + md.drop_all() + e.dispose() + except (KeyboardInterrupt, SystemExit): + raise + except Exception, e: if not options.quiet: - print "Dropping existing tables in database: " + db_url - try: - print "Tables: %s" % ', '.join(existing) - except: - pass - print "Abort within 5 seconds..." - time.sleep(5) - md = schema.MetaData(e, reflect=True) - md.drop_all() - e.dispose() + warnings.warn(RuntimeWarning( + "Error checking for existing tables in testing " + "database: %s" % e)) post_configure['prep_db'] = _prep_testing_database def _set_table_options(options, file_config): diff --git a/test/testlib/engines.py b/test/testlib/engines.py index da3374967..addef3f9e 100644 --- a/test/testlib/engines.py +++ b/test/testlib/engines.py @@ -24,10 +24,17 @@ def utf8_engine(url=None, options=None): from sqlalchemy.engine import url as engine_url if config.db.name == 'mysql': - url = url or config.db_url - url = engine_url.make_url(url) - url.query['charset'] = 'utf8' - url.query['use_unicode'] = '0' - url = str(url) + dbapi_ver = config.db.dialect.dbapi.version_info + if (dbapi_ver < (1, 2, 1) or + dbapi_ver in ((1, 2, 1, 'gamma', 1), (1, 2, 1, 'gamma', 2), + (1, 2, 1, 'gamma', 3), (1, 2, 1, 'gamma', 5))): + raise RuntimeError('Character set support unavailable with this ' + 'driver version: %s' % repr(dbapi_ver)) + else: + url = url or config.db_url + url = engine_url.make_url(url) + url.query['charset'] = 'utf8' + url.query['use_unicode'] = '0' + url = str(url) return testing_engine(url, options) |