import sqlalchemy as sa
from sqlalchemy import exc as sa_exc
from sqlalchemy import types as sql_types
-from sqlalchemy import schema
from sqlalchemy import inspect
from sqlalchemy import MetaData, Integer, String
from sqlalchemy.engine.reflection import Inspector
assert not config.db.dialect.has_table(conn, "nonexistent_table")
-class HasSequenceTest(fixtures.TestBase):
- __requires__ = 'sequences',
-
- def test_has_sequence(self):
- metadata = MetaData()
- Table('users', metadata, Column('user_id', sa.Integer,
- sa.Sequence('user_id_seq'), primary_key=True),
- Column('user_name', sa.String(40)))
- metadata.create_all(bind=testing.db)
- try:
- eq_(testing.db.dialect.has_sequence(testing.db,
- 'user_id_seq'), True)
- finally:
- metadata.drop_all(bind=testing.db)
- eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq'),
- False)
-
- @testing.requires.schemas
- def test_has_sequence_schema(self):
- test_schema = 'test_schema'
- s1 = sa.Sequence('user_id_seq', schema=test_schema)
- s2 = sa.Sequence('user_id_seq')
- testing.db.execute(schema.CreateSequence(s1))
- testing.db.execute(schema.CreateSequence(s2))
- eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq',
- schema=test_schema), True)
- eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq'),
- True)
- testing.db.execute(schema.DropSequence(s1))
- eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq',
- schema=test_schema), False)
- eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq'),
- True)
- testing.db.execute(schema.DropSequence(s2))
- eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq',
- schema=test_schema), False)
- eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq'),
- False)
-
class ComponentReflectionTest(fixtures.TablesTest):
run_inserts = run_deletes = None
self._test_get_table_oid('users', schema='test_schema')
-__all__ = ('ComponentReflectionTest', 'HasSequenceTest', 'HasTableTest')
+__all__ = ('ComponentReflectionTest', 'HasTableTest')
from .. import fixtures, config
from ..config import requirements
from ..assertions import eq_
+from ... import testing
-from sqlalchemy import Integer, String, Sequence
+from ... import Integer, String, Sequence, schema
from ..schema import Table, Column
(1, "some data")
)
+
+class HasSequenceTest(fixtures.TestBase):
+ __requires__ = 'sequences',
+
+ def test_has_sequence(self):
+ s1 = Sequence('user_id_seq')
+ testing.db.execute(schema.CreateSequence(s1))
+ try:
+ eq_(testing.db.dialect.has_sequence(testing.db,
+ 'user_id_seq'), True)
+ finally:
+ testing.db.execute(schema.DropSequence(s1))
+
+ @testing.requires.schemas
+ def test_has_sequence_schema(self):
+ s1 = Sequence('user_id_seq', schema="test_schema")
+ testing.db.execute(schema.CreateSequence(s1))
+ try:
+ eq_(testing.db.dialect.has_sequence(testing.db,
+ 'user_id_seq', schema="test_schema"), True)
+ finally:
+ testing.db.execute(schema.DropSequence(s1))
+
+ def test_has_sequence_neg(self):
+ eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq'),
+ False)
+
+ @testing.requires.schemas
+ def test_has_sequence_schemas_neg(self):
+ eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq',
+ schema="test_schema"),
+ False)
+
+ @testing.requires.schemas
+ def test_has_sequence_default_not_in_remote(self):
+ s1 = Sequence('user_id_seq')
+ testing.db.execute(schema.CreateSequence(s1))
+ try:
+ eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq',
+ schema="test_schema"),
+ False)
+ finally:
+ testing.db.execute(schema.DropSequence(s1))
+
+ @testing.requires.schemas
+ def test_has_sequence_remote_not_in_default(self):
+ s1 = Sequence('user_id_seq', schema="test_schema")
+ testing.db.execute(schema.CreateSequence(s1))
+ try:
+ eq_(testing.db.dialect.has_sequence(testing.db, 'user_id_seq'),
+ False)
+ finally:
+ testing.db.execute(schema.DropSequence(s1))
+
+