From 6f945d34af1650ced39da0735b09dc0b56f84452 Mon Sep 17 00:00:00 2001 From: Joris Van den Bossche Date: Thu, 13 Mar 2014 10:25:19 +0100 Subject: [PATCH] TST: skip sql tests if connection to server fails --- pandas/io/tests/test_sql.py | 107 +++++++++++++++++++----------------- 1 file changed, 57 insertions(+), 50 deletions(-) diff --git a/pandas/io/tests/test_sql.py b/pandas/io/tests/test_sql.py index 46b5758904315..9ecb605def400 100644 --- a/pandas/io/tests/test_sql.py +++ b/pandas/io/tests/test_sql.py @@ -705,11 +705,14 @@ def setUp(self): try: import pymysql self.driver = pymysql - except ImportError: - raise nose.SkipTest + raise nose.SkipTest('pymysql not installed') + + try: + self.conn = self.connect() + except self.driver.err.OperationalError: + raise nose.SkipTest("Can't connect to MySQL server") - self.conn = self.connect() self.pandasSQL = sql.PandasSQLLegacy(self.conn, 'mysql') self._load_iris_data() @@ -725,53 +728,55 @@ def tearDown(self): class TestMySQLAlchemy(_TestSQLAlchemy): - flavor = 'mysql' - - def connect(self): - return sqlalchemy.create_engine( - 'mysql+{driver}://root@localhost/pandas_nosetest'.format(driver=self.driver)) + flavor = 'mysql' - def setUp(self): - if not SQLALCHEMY_INSTALLED: - raise nose.SkipTest('SQLAlchemy not installed') + def connect(self): + return sqlalchemy.create_engine( + 'mysql+{driver}://root@localhost/pandas_nosetest'.format(driver=self.driver)) - try: - import pymysql - self.driver = 'pymysql' + def setUp(self): + if not SQLALCHEMY_INSTALLED: + raise nose.SkipTest('SQLAlchemy not installed') - except ImportError: - raise nose.SkipTest + try: + import pymysql + self.driver = 'pymysql' + except ImportError: + raise nose.SkipTest('pymysql not installed') + try: self.conn = self.connect() self.pandasSQL = sql.PandasSQLAlchemy(self.conn) + except sqlalchemy.exc.OperationalError: + raise nose.SkipTest("Can't connect to MySQL server") - self._load_iris_data() - self._load_raw_sql() + self._load_iris_data() + self._load_raw_sql() - self._load_test1_data() + self._load_test1_data() + + def tearDown(self): + c = self.conn.execute('SHOW TABLES') + for table in c.fetchall(): + self.conn.execute('DROP TABLE %s' % table[0]) - def tearDown(self): - c = self.conn.execute('SHOW TABLES') - for table in c.fetchall(): - self.conn.execute('DROP TABLE %s' % table[0]) + def test_default_type_conversion(self): + df = sql.read_table("types_test_data", self.conn) - def test_default_type_conversion(self): - df = sql.read_table("types_test_data", self.conn) - - self.assertTrue(issubclass(df.FloatCol.dtype.type, np.floating), - "FloatCol loaded with incorrect type") - self.assertTrue(issubclass(df.IntCol.dtype.type, np.integer), - "IntCol loaded with incorrect type") - # MySQL has no real BOOL type (it's an alias for TINYINT) - self.assertTrue(issubclass(df.BoolCol.dtype.type, np.integer), - "BoolCol loaded with incorrect type") - - # Int column with NA values stays as float - self.assertTrue(issubclass(df.IntColWithNull.dtype.type, np.floating), - "IntColWithNull loaded with incorrect type") - # Bool column with NA = int column with NA values => becomes float - self.assertTrue(issubclass(df.BoolColWithNull.dtype.type, np.floating), - "BoolColWithNull loaded with incorrect type") + self.assertTrue(issubclass(df.FloatCol.dtype.type, np.floating), + "FloatCol loaded with incorrect type") + self.assertTrue(issubclass(df.IntCol.dtype.type, np.integer), + "IntCol loaded with incorrect type") + # MySQL has no real BOOL type (it's an alias for TINYINT) + self.assertTrue(issubclass(df.BoolCol.dtype.type, np.integer), + "BoolCol loaded with incorrect type") + + # Int column with NA values stays as float + self.assertTrue(issubclass(df.IntColWithNull.dtype.type, np.floating), + "IntColWithNull loaded with incorrect type") + # Bool column with NA = int column with NA values => becomes float + self.assertTrue(issubclass(df.BoolColWithNull.dtype.type, np.floating), + "BoolColWithNull loaded with incorrect type") class TestPostgreSQLAlchemy(_TestSQLAlchemy): @@ -780,26 +785,28 @@ class TestPostgreSQLAlchemy(_TestSQLAlchemy): def connect(self): return sqlalchemy.create_engine( 'postgresql+{driver}://postgres@localhost/pandas_nosetest'.format(driver=self.driver)) - + def setUp(self): if not SQLALCHEMY_INSTALLED: raise nose.SkipTest('SQLAlchemy not installed') - + try: import psycopg2 self.driver = 'psycopg2' - except ImportError: - raise nose.SkipTest - - self.conn = self.connect() - self.pandasSQL = sql.PandasSQLAlchemy(self.conn) - + raise nose.SkipTest('psycopg2 not installed') + + try: + self.conn = self.connect() + self.pandasSQL = sql.PandasSQLAlchemy(self.conn) + except sqlalchemy.exc.OperationalError: + raise nose.SkipTest("Can't connect to PostgreSQL server") + self._load_iris_data() self._load_raw_sql() - + self._load_test1_data() - + def tearDown(self): c = self.conn.execute( "SELECT table_name FROM information_schema.tables"