From: Mike Bayer Date: Mon, 2 Oct 2017 20:50:09 +0000 (-0400) Subject: PG dialect test fixes X-Git-Tag: rel_1_2_0~66 X-Git-Url: http://git.ipfire.org/gitweb.cgi?a=commitdiff_plain;h=f846a789b78f977b4d7e8e16b412b07e77f2ab13;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git PG dialect test fixes Make sure we clear the plugin registry before testing that the "postgres" name raises. Also move non-backend tests out of MiscTest into a new suite. Change-Id: Icd1bb4745aa07f52d585fcf959f76fcd8bdc7f24 --- diff --git a/lib/sqlalchemy/util/langhelpers.py b/lib/sqlalchemy/util/langhelpers.py index 9ca19f1381..c90431d337 100644 --- a/lib/sqlalchemy/util/langhelpers.py +++ b/lib/sqlalchemy/util/langhelpers.py @@ -193,6 +193,9 @@ class PluginLoader(object): self.impls = {} self.auto_fn = auto_fn + def clear(self): + self.impls.clear() + def load(self, name): if name in self.impls: return self.impls[name]() diff --git a/test/dialect/postgresql/test_dialect.py b/test/dialect/postgresql/test_dialect.py index 27e0ac0659..e985718c77 100644 --- a/test/dialect/postgresql/test_dialect.py +++ b/test/dialect/postgresql/test_dialect.py @@ -20,33 +20,19 @@ from sqlalchemy.engine import url from sqlalchemy.testing import is_ from sqlalchemy.testing import expect_deprecated from ...engine import test_execute +from sqlalchemy import dialects -class MiscTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): +class DialectTest(fixtures.TestBase): + """python-side dialect tests. """ - __only_on__ = 'postgresql' - __backend__ = True - - @testing.provide_metadata - def test_date_reflection(self): - metadata = self.metadata - Table( - 'pgdate', metadata, Column('date1', DateTime(timezone=True)), - Column('date2', DateTime(timezone=False))) - metadata.create_all() - m2 = MetaData(testing.db) - t2 = Table('pgdate', m2, autoload=True) - assert t2.c.date1.type.timezone is True - assert t2.c.date2.type.timezone is False - - @testing.fails_on('+zxjdbc', - 'The JDBC driver handles the version parsing') def test_version_parsing(self): def mock_conn(res): return Mock( execute=Mock(return_value=Mock(scalar=Mock(return_value=res)))) + dialect = postgresql.dialect() for string, version in [ ( 'PostgreSQL 8.3.8 on i686-redhat-linux-gnu, compiled by ' @@ -70,9 +56,52 @@ class MiscTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): 'compiled by gcc (GCC) 4.8.5 20150623 ' '(Red Hat 4.8.5-11), 64-bit', (10,)) ]: - eq_(testing.db.dialect._get_server_version_info(mock_conn(string)), + eq_(dialect._get_server_version_info(mock_conn(string)), version) + def test_deprecated_dialect_name_still_loads(self): + dialects.registry.clear() + with expect_deprecated( + "The 'postgres' dialect name " + "has been renamed to 'postgresql'"): + dialect = url.URL("postgres").get_dialect() + is_(dialect, postgresql.dialect) + + @testing.requires.psycopg2_compatibility + def test_pg_dialect_use_native_unicode_from_config(self): + config = { + 'sqlalchemy.url': testing.db.url, + 'sqlalchemy.use_native_unicode': "false"} + + e = engine_from_config(config, _initialize=False) + eq_(e.dialect.use_native_unicode, False) + + config = { + 'sqlalchemy.url': testing.db.url, + 'sqlalchemy.use_native_unicode': "true"} + + e = engine_from_config(config, _initialize=False) + eq_(e.dialect.use_native_unicode, True) + + +class MiscBackendTest( + fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): + + __only_on__ = 'postgresql' + __backend__ = True + + @testing.provide_metadata + def test_date_reflection(self): + metadata = self.metadata + Table( + 'pgdate', metadata, Column('date1', DateTime(timezone=True)), + Column('date2', DateTime(timezone=False))) + metadata.create_all() + m2 = MetaData(testing.db) + t2 = Table('pgdate', m2, autoload=True) + assert t2.c.date1.type.timezone is True + assert t2.c.date2.type.timezone is False + @testing.requires.psycopg2_compatibility def test_psycopg2_version(self): v = testing.db.dialect.psycopg2_version @@ -92,13 +121,6 @@ class MiscTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): psycopg2.Error) assert isinstance(exception, exc.OperationalError) - def test_deprecated_dialect_name_still_loads(self): - with expect_deprecated( - "The 'postgres' dialect name " - "has been renamed to 'postgresql'"): - dialect = url.URL("postgres").get_dialect() - is_(dialect, postgresql.dialect) - # currently not passing with pg 9.3 that does not seem to generate # any notices here, would rather find a way to mock this @testing.requires.no_coverage @@ -142,22 +164,6 @@ class MiscTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): new_encoding = c.execute("show client_encoding").fetchone()[0] eq_(new_encoding, test_encoding) - @testing.requires.psycopg2_compatibility - def test_pg_dialect_use_native_unicode_from_config(self): - config = { - 'sqlalchemy.url': testing.db.url, - 'sqlalchemy.use_native_unicode': "false"} - - e = engine_from_config(config, _initialize=False) - eq_(e.dialect.use_native_unicode, False) - - config = { - 'sqlalchemy.url': testing.db.url, - 'sqlalchemy.use_native_unicode': "true"} - - e = engine_from_config(config, _initialize=False) - eq_(e.dialect.use_native_unicode, True) - @testing.requires.psycopg2_or_pg8000_compatibility @engines.close_open_connections def test_autocommit_isolation_level(self): @@ -186,45 +192,41 @@ class MiscTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): ).scalar() eq_(r, exp) + @testing.provide_metadata def test_checksfor_sequence(self): - meta1 = MetaData(testing.db) + meta1 = self.metadata seq = Sequence('fooseq') t = Table( 'mytable', meta1, Column('col1', Integer, seq) ) seq.drop() - try: - testing.db.execute('CREATE SEQUENCE fooseq') - t.create(checkfirst=True) - finally: - t.drop(checkfirst=True) + testing.db.execute('CREATE SEQUENCE fooseq') + t.create(checkfirst=True) + @testing.provide_metadata def test_schema_roundtrips(self): - meta = MetaData(testing.db) + meta = self.metadata users = Table( 'users', meta, Column( 'id', Integer, primary_key=True), Column( 'name', String(50)), schema='test_schema') users.create() - try: - users.insert().execute(id=1, name='name1') - users.insert().execute(id=2, name='name2') - users.insert().execute(id=3, name='name3') - users.insert().execute(id=4, name='name4') - eq_(users.select().where(users.c.name == 'name2') - .execute().fetchall(), [(2, 'name2')]) - eq_(users.select(use_labels=True).where( - users.c.name == 'name2').execute().fetchall(), [(2, 'name2')]) - users.delete().where(users.c.id == 3).execute() - eq_(users.select().where(users.c.name == 'name3') - .execute().fetchall(), []) - users.update().where(users.c.name == 'name4' - ).execute(name='newname') - eq_(users.select(use_labels=True).where( - users.c.id == 4).execute().fetchall(), [(4, 'newname')]) - finally: - users.drop() + users.insert().execute(id=1, name='name1') + users.insert().execute(id=2, name='name2') + users.insert().execute(id=3, name='name3') + users.insert().execute(id=4, name='name4') + eq_(users.select().where(users.c.name == 'name2') + .execute().fetchall(), [(2, 'name2')]) + eq_(users.select(use_labels=True).where( + users.c.name == 'name2').execute().fetchall(), [(2, 'name2')]) + users.delete().where(users.c.id == 3).execute() + eq_(users.select().where(users.c.name == 'name3') + .execute().fetchall(), []) + users.update().where(users.c.name == 'name4' + ).execute(name='newname') + eq_(users.select(use_labels=True).where( + users.c.id == 4).execute().fetchall(), [(4, 'newname')]) def test_quoted_name_bindparam_ok(self): from sqlalchemy.sql.elements import quoted_name