diff options
Diffstat (limited to 'test/dialect')
| -rw-r--r-- | test/dialect/mssql/test_engine.py | 20 | ||||
| -rw-r--r-- | test/dialect/mysql/test_reflection.py | 32 | ||||
| -rw-r--r-- | test/dialect/mysql/test_types.py | 6 | ||||
| -rw-r--r-- | test/dialect/postgresql/test_reflection.py | 180 | ||||
| -rw-r--r-- | test/dialect/test_oracle.py | 39 | ||||
| -rw-r--r-- | test/dialect/test_sqlite.py | 173 |
6 files changed, 391 insertions, 59 deletions
diff --git a/test/dialect/mssql/test_engine.py b/test/dialect/mssql/test_engine.py index 8ac9c6c16..4b4780d43 100644 --- a/test/dialect/mssql/test_engine.py +++ b/test/dialect/mssql/test_engine.py @@ -7,6 +7,8 @@ from sqlalchemy.engine import url from sqlalchemy.testing import fixtures from sqlalchemy import testing from sqlalchemy.testing import assert_raises_message, assert_warnings +from sqlalchemy.testing.mock import Mock + class ParseConnectTest(fixtures.TestBase): @@ -167,3 +169,21 @@ class ParseConnectTest(fixtures.TestBase): assert_raises_message(exc.SAWarning, 'Unrecognized server version info', engine.connect) + + +class VersionDetectionTest(fixtures.TestBase): + def test_pymssql_version(self): + dialect = pymssql.MSDialect_pymssql() + + for vers in [ + "Microsoft SQL Server Blah - 11.0.9216.62", + "Microsoft SQL Server (XYZ) - 11.0.9216.62 \n" + "Jul 18 2014 22:00:21 \nCopyright (c) Microsoft Corporation", + "Microsoft SQL Azure (RTM) - 11.0.9216.62 \n" + "Jul 18 2014 22:00:21 \nCopyright (c) Microsoft Corporation" + ]: + conn = Mock(scalar=Mock(return_value=vers)) + eq_( + dialect._get_server_version_info(conn), + (11, 0, 9216, 62) + )
\ No newline at end of file diff --git a/test/dialect/mysql/test_reflection.py b/test/dialect/mysql/test_reflection.py index bf35a2c6b..99733e397 100644 --- a/test/dialect/mysql/test_reflection.py +++ b/test/dialect/mysql/test_reflection.py @@ -283,6 +283,38 @@ class ReflectionTest(fixtures.TestBase, AssertsExecutionResults): view_names = dialect.get_view_names(connection, "information_schema") self.assert_('TABLES' in view_names) + @testing.provide_metadata + def test_reflection_with_unique_constraint(self): + insp = inspect(testing.db) + + meta = self.metadata + uc_table = Table('mysql_uc', meta, + Column('a', String(10)), + UniqueConstraint('a', name='uc_a')) + + uc_table.create() + + # MySQL converts unique constraints into unique indexes. + # separately we get both + indexes = dict((i['name'], i) for i in insp.get_indexes('mysql_uc')) + constraints = set(i['name'] + for i in insp.get_unique_constraints('mysql_uc')) + + self.assert_('uc_a' in indexes) + self.assert_(indexes['uc_a']['unique']) + self.assert_('uc_a' in constraints) + + # reflection here favors the unique index, as that's the + # more "official" MySQL construct + reflected = Table('mysql_uc', MetaData(testing.db), autoload=True) + + indexes = dict((i.name, i) for i in reflected.indexes) + constraints = set(uc.name for uc in reflected.constraints) + + self.assert_('uc_a' in indexes) + self.assert_(indexes['uc_a'].unique) + self.assert_('uc_a' not in constraints) + class RawReflectionTest(fixtures.TestBase): def setup(self): diff --git a/test/dialect/mysql/test_types.py b/test/dialect/mysql/test_types.py index 75dbe15e0..e65acc6db 100644 --- a/test/dialect/mysql/test_types.py +++ b/test/dialect/mysql/test_types.py @@ -154,10 +154,8 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): res ) - @testing.fails_if( - lambda: testing.against("mysql+mysqlconnector") - and not util.py3k, - "bug in mysqlconnector; http://bugs.mysql.com/bug.php?id=73266") + # fixed in mysql-connector as of 2.0.1, + # see http://bugs.mysql.com/bug.php?id=73266 @testing.provide_metadata def test_precision_float_roundtrip(self): t = Table('t', self.metadata, diff --git a/test/dialect/postgresql/test_reflection.py b/test/dialect/postgresql/test_reflection.py index bab41b0f7..8de71216e 100644 --- a/test/dialect/postgresql/test_reflection.py +++ b/test/dialect/postgresql/test_reflection.py @@ -7,14 +7,130 @@ from sqlalchemy.testing import fixtures from sqlalchemy import testing from sqlalchemy import inspect from sqlalchemy import Table, Column, MetaData, Integer, String, \ - PrimaryKeyConstraint, ForeignKey, join, Sequence + PrimaryKeyConstraint, ForeignKey, join, Sequence, UniqueConstraint, \ + Index from sqlalchemy import exc import sqlalchemy as sa from sqlalchemy.dialects.postgresql import base as postgresql -class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults): +class ForeignTableReflectionTest(fixtures.TablesTest, AssertsExecutionResults): + """Test reflection on foreign tables""" + + __requires__ = 'postgresql_test_dblink', + __only_on__ = 'postgresql >= 9.3' + __backend__ = True + + @classmethod + def define_tables(cls, metadata): + from sqlalchemy.testing import config + dblink = config.file_config.get( + 'sqla_testing', 'postgres_test_db_link') + + testtable = Table( + 'testtable', metadata, + Column('id', Integer, primary_key=True), + Column('data', String(30))) + + for ddl in [ + "CREATE SERVER test_server FOREIGN DATA WRAPPER postgres_fdw " + "OPTIONS (dbname 'test', host '%s')" % dblink, + "CREATE USER MAPPING FOR public \ + SERVER test_server options (user 'scott', password 'tiger')", + "CREATE FOREIGN TABLE test_foreigntable ( " + " id INT, " + " data VARCHAR(30) " + ") SERVER test_server OPTIONS (table_name 'testtable')", + ]: + sa.event.listen(metadata, "after_create", sa.DDL(ddl)) + + for ddl in [ + 'DROP FOREIGN TABLE test_foreigntable', + 'DROP USER MAPPING FOR public SERVER test_server', + "DROP SERVER test_server" + ]: + sa.event.listen(metadata, "before_drop", sa.DDL(ddl)) + + def test_foreign_table_is_reflected(self): + metadata = MetaData(testing.db) + table = Table('test_foreigntable', metadata, autoload=True) + eq_(set(table.columns.keys()), set(['id', 'data']), + "Columns of reflected foreign table didn't equal expected columns") + + def test_get_foreign_table_names(self): + inspector = inspect(testing.db) + with testing.db.connect() as conn: + ft_names = inspector.get_foreign_table_names() + eq_(ft_names, ['test_foreigntable']) + + def test_get_table_names_no_foreign(self): + inspector = inspect(testing.db) + with testing.db.connect() as conn: + names = inspector.get_table_names() + eq_(names, ['testtable']) + + +class MaterialiedViewReflectionTest( + fixtures.TablesTest, AssertsExecutionResults): + """Test reflection on materialized views""" + + __only_on__ = 'postgresql >= 9.3' + __backend__ = True + + @classmethod + def define_tables(cls, metadata): + testtable = Table( + 'testtable', metadata, + Column('id', Integer, primary_key=True), + Column('data', String(30))) + + # insert data before we create the view + @sa.event.listens_for(testtable, "after_create") + def insert_data(target, connection, **kw): + connection.execute( + target.insert(), + {"id": 89, "data": 'd1'} + ) + + materialized_view = sa.DDL( + "CREATE MATERIALIZED VIEW test_mview AS " + "SELECT * FROM testtable") + + plain_view = sa.DDL( + "CREATE VIEW test_regview AS " + "SELECT * FROM testtable") + + sa.event.listen(testtable, 'after_create', plain_view) + sa.event.listen(testtable, 'after_create', materialized_view) + sa.event.listen( + testtable, 'before_drop', + sa.DDL("DROP MATERIALIZED VIEW test_mview") + ) + sa.event.listen( + testtable, 'before_drop', + sa.DDL("DROP VIEW test_regview") + ) + def test_mview_is_reflected(self): + metadata = MetaData(testing.db) + table = Table('test_mview', metadata, autoload=True) + eq_(set(table.columns.keys()), set(['id', 'data']), + "Columns of reflected mview didn't equal expected columns") + + def test_mview_select(self): + metadata = MetaData(testing.db) + table = Table('test_mview', metadata, autoload=True) + eq_( + table.select().execute().fetchall(), + [(89, 'd1',)] + ) + + def test_get_view_names(self): + insp = inspect(testing.db) + eq_(set(insp.get_view_names()), set(['test_mview', 'test_regview'])) + + +class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults): """Test PostgreSQL domains""" __only_on__ = 'postgresql > 8.3' @@ -688,6 +804,66 @@ class ReflectionTest(fixtures.TestBase): 'labels': ['sad', 'ok', 'happy'] }]) + @testing.provide_metadata + def test_reflection_with_unique_constraint(self): + insp = inspect(testing.db) + + meta = self.metadata + uc_table = Table('pgsql_uc', meta, + Column('a', String(10)), + UniqueConstraint('a', name='uc_a')) + + uc_table.create() + + # PostgreSQL will create an implicit index for a unique + # constraint. Separately we get both + indexes = set(i['name'] for i in insp.get_indexes('pgsql_uc')) + constraints = set(i['name'] + for i in insp.get_unique_constraints('pgsql_uc')) + + self.assert_('uc_a' in indexes) + self.assert_('uc_a' in constraints) + + # reflection corrects for the dupe + reflected = Table('pgsql_uc', MetaData(testing.db), autoload=True) + + indexes = set(i.name for i in reflected.indexes) + constraints = set(uc.name for uc in reflected.constraints) + + self.assert_('uc_a' not in indexes) + self.assert_('uc_a' in constraints) + + @testing.provide_metadata + def test_reflect_unique_index(self): + insp = inspect(testing.db) + + meta = self.metadata + + # a unique index OTOH we are able to detect is an index + # and not a unique constraint + uc_table = Table('pgsql_uc', meta, + Column('a', String(10)), + Index('ix_a', 'a', unique=True)) + + uc_table.create() + + indexes = dict((i['name'], i) for i in insp.get_indexes('pgsql_uc')) + constraints = set(i['name'] + for i in insp.get_unique_constraints('pgsql_uc')) + + self.assert_('ix_a' in indexes) + assert indexes['ix_a']['unique'] + self.assert_('ix_a' not in constraints) + + reflected = Table('pgsql_uc', MetaData(testing.db), autoload=True) + + indexes = dict((i.name, i) for i in reflected.indexes) + constraints = set(uc.name for uc in reflected.constraints) + + self.assert_('ix_a' in indexes) + assert indexes['ix_a'].unique + self.assert_('ix_a' not in constraints) + class CustomTypeReflectionTest(fixtures.TestBase): diff --git a/test/dialect/test_oracle.py b/test/dialect/test_oracle.py index 187042036..a771c5d80 100644 --- a/test/dialect/test_oracle.py +++ b/test/dialect/test_oracle.py @@ -104,6 +104,28 @@ class QuotedBindRoundTripTest(fixtures.TestBase): (2, 2, 2) ) + def test_numeric_bind_round_trip(self): + eq_( + testing.db.scalar( + select([ + literal_column("2", type_=Integer()) + + bindparam("2_1", value=2)]) + ), + 4 + ) + + @testing.provide_metadata + def test_numeric_bind_in_crud(self): + t = Table( + "asfd", self.metadata, + Column("100K", Integer) + ) + t.create() + + testing.db.execute(t.insert(), {"100K": 10}) + eq_( + testing.db.scalar(t.select()), 10 + ) class CompileTest(fixtures.TestBase, AssertsCompiledSQL): __dialect__ = "oracle" #oracle.dialect() @@ -648,6 +670,23 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): "CREATE INDEX bar ON foo (x > 5)" ) + def test_table_options(self): + m = MetaData() + + t = Table( + 'foo', m, + Column('x', Integer), + prefixes=["GLOBAL TEMPORARY"], + oracle_on_commit="PRESERVE ROWS" + ) + + self.assert_compile( + schema.CreateTable(t), + "CREATE GLOBAL TEMPORARY TABLE " + "foo (x INTEGER) ON COMMIT PRESERVE ROWS" + ) + + class CompatFlagsTest(fixtures.TestBase, AssertsCompiledSQL): def _dialect(self, server_version, **kw): diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py index e77a03980..124208dbe 100644 --- a/test/dialect/test_sqlite.py +++ b/test/dialect/test_sqlite.py @@ -11,7 +11,7 @@ from sqlalchemy import Table, select, bindparam, Column,\ UniqueConstraint from sqlalchemy.types import Integer, String, Boolean, DateTime, Date, Time from sqlalchemy import types as sqltypes -from sqlalchemy import event +from sqlalchemy import event, inspect from sqlalchemy.util import u, ue from sqlalchemy import exc, sql, schema, pool, util from sqlalchemy.dialects.sqlite import base as sqlite, \ @@ -480,57 +480,6 @@ class DialectTest(fixtures.TestBase, AssertsExecutionResults): assert u('méil') in result.keys() assert ue('\u6e2c\u8a66') in result.keys() - def test_attached_as_schema(self): - cx = testing.db.connect() - try: - cx.execute('ATTACH DATABASE ":memory:" AS test_schema') - dialect = cx.dialect - assert dialect.get_table_names(cx, 'test_schema') == [] - meta = MetaData(cx) - Table('created', meta, Column('id', Integer), - schema='test_schema') - alt_master = Table('sqlite_master', meta, autoload=True, - schema='test_schema') - meta.create_all(cx) - eq_(dialect.get_table_names(cx, 'test_schema'), ['created']) - assert len(alt_master.c) > 0 - meta.clear() - reflected = Table('created', meta, autoload=True, - schema='test_schema') - assert len(reflected.c) == 1 - cx.execute(reflected.insert(), dict(id=1)) - r = cx.execute(reflected.select()).fetchall() - assert list(r) == [(1, )] - cx.execute(reflected.update(), dict(id=2)) - r = cx.execute(reflected.select()).fetchall() - assert list(r) == [(2, )] - cx.execute(reflected.delete(reflected.c.id == 2)) - r = cx.execute(reflected.select()).fetchall() - assert list(r) == [] - - # note that sqlite_master is cleared, above - - meta.drop_all() - assert dialect.get_table_names(cx, 'test_schema') == [] - finally: - cx.execute('DETACH DATABASE test_schema') - - @testing.exclude('sqlite', '<', (2, 6), 'no database support') - def test_temp_table_reflection(self): - cx = testing.db.connect() - try: - cx.execute('CREATE TEMPORARY TABLE tempy (id INT)') - assert 'tempy' in cx.dialect.get_table_names(cx, None) - meta = MetaData(cx) - tempy = Table('tempy', meta, autoload=True) - assert len(tempy.c) == 1 - meta.drop_all() - except: - try: - cx.execute('DROP TABLE tempy') - except exc.DBAPIError: - pass - raise def test_file_path_is_absolute(self): d = pysqlite_dialect.dialect() @@ -549,7 +498,6 @@ class DialectTest(fixtures.TestBase, AssertsExecutionResults): e = create_engine('sqlite+pysqlite:///foo.db') assert e.pool.__class__ is pool.NullPool - def test_dont_reflect_autoindex(self): meta = MetaData(testing.db) t = Table('foo', meta, Column('bar', String, primary_key=True)) @@ -575,6 +523,125 @@ class DialectTest(fixtures.TestBase, AssertsExecutionResults): finally: meta.drop_all() + def test_get_unique_constraints(self): + meta = MetaData(testing.db) + t1 = Table('foo', meta, Column('f', Integer), + UniqueConstraint('f', name='foo_f')) + t2 = Table('bar', meta, Column('b', Integer), + UniqueConstraint('b', name='bar_b'), + prefixes=['TEMPORARY']) + meta.create_all() + from sqlalchemy.engine.reflection import Inspector + try: + inspector = Inspector(testing.db) + eq_(inspector.get_unique_constraints('foo'), + [{'column_names': [u'f'], 'name': u'foo_f'}]) + eq_(inspector.get_unique_constraints('bar'), + [{'column_names': [u'b'], 'name': u'bar_b'}]) + finally: + meta.drop_all() + + +class AttachedMemoryDBTest(fixtures.TestBase): + __only_on__ = 'sqlite' + + dbname = None + + def setUp(self): + self.conn = conn = testing.db.connect() + if self.dbname is None: + dbname = ':memory:' + else: + dbname = self.dbname + conn.execute('ATTACH DATABASE "%s" AS test_schema' % dbname) + self.metadata = MetaData() + + def tearDown(self): + self.metadata.drop_all(self.conn) + self.conn.execute('DETACH DATABASE test_schema') + if self.dbname: + os.remove(self.dbname) + + def _fixture(self): + meta = self.metadata + ct = Table( + 'created', meta, + Column('id', Integer), + Column('name', String), + schema='test_schema') + + meta.create_all(self.conn) + return ct + + def test_no_tables(self): + insp = inspect(self.conn) + eq_(insp.get_table_names("test_schema"), []) + + def test_table_names_present(self): + self._fixture() + insp = inspect(self.conn) + eq_(insp.get_table_names("test_schema"), ["created"]) + + def test_table_names_system(self): + self._fixture() + insp = inspect(self.conn) + eq_(insp.get_table_names("test_schema"), ["created"]) + + def test_reflect_system_table(self): + meta = MetaData(self.conn) + alt_master = Table( + 'sqlite_master', meta, autoload=True, + autoload_with=self.conn, + schema='test_schema') + assert len(alt_master.c) > 0 + + def test_reflect_user_table(self): + self._fixture() + + m2 = MetaData() + c2 = Table('created', m2, autoload=True, autoload_with=self.conn) + eq_(len(c2.c), 2) + + def test_crud(self): + ct = self._fixture() + + self.conn.execute(ct.insert(), {'id': 1, 'name': 'foo'}) + eq_( + self.conn.execute(ct.select()).fetchall(), + [(1, 'foo')] + ) + + self.conn.execute(ct.update(), {'id': 2, 'name': 'bar'}) + eq_( + self.conn.execute(ct.select()).fetchall(), + [(2, 'bar')] + ) + self.conn.execute(ct.delete()) + eq_( + self.conn.execute(ct.select()).fetchall(), + [] + ) + + def test_col_targeting(self): + ct = self._fixture() + + self.conn.execute(ct.insert(), {'id': 1, 'name': 'foo'}) + row = self.conn.execute(ct.select()).first() + eq_(row['id'], 1) + eq_(row['name'], 'foo') + + def test_col_targeting_union(self): + ct = self._fixture() + + self.conn.execute(ct.insert(), {'id': 1, 'name': 'foo'}) + row = self.conn.execute(ct.select().union(ct.select())).first() + eq_(row['id'], 1) + eq_(row['name'], 'foo') + + +class AttachedFileDBTest(AttachedMemoryDBTest): + dbname = 'attached_db.db' + class SQLTest(fixtures.TestBase, AssertsCompiledSQL): |