diff options
Diffstat (limited to 'lib/sqlalchemy/testing/suite/test_sequence.py')
| -rw-r--r-- | lib/sqlalchemy/testing/suite/test_sequence.py | 130 |
1 files changed, 85 insertions, 45 deletions
diff --git a/lib/sqlalchemy/testing/suite/test_sequence.py b/lib/sqlalchemy/testing/suite/test_sequence.py index dda447c0d..55e8e8406 100644 --- a/lib/sqlalchemy/testing/suite/test_sequence.py +++ b/lib/sqlalchemy/testing/suite/test_sequence.py @@ -1,12 +1,13 @@ from .. import config from .. import fixtures from ..assertions import eq_ +from ..assertions import is_true from ..config import requirements from ..schema import Column from ..schema import Table +from ... import inspect from ... import Integer from ... import MetaData -from ... import schema from ... import Sequence from ... import String from ... import testing @@ -88,69 +89,108 @@ class SequenceCompilerTest(testing.AssertsCompiledSQL, fixtures.TestBase): ) -class HasSequenceTest(fixtures.TestBase): +class HasSequenceTest(fixtures.TablesTest): + run_deletes = None + __requires__ = ("sequences",) __backend__ = True - def test_has_sequence(self, connection): - s1 = Sequence("user_id_seq") - connection.execute(schema.CreateSequence(s1)) - try: - eq_( - connection.dialect.has_sequence(connection, "user_id_seq"), - True, + @classmethod + def define_tables(cls, metadata): + Sequence("user_id_seq", metadata=metadata) + Sequence("other_seq", metadata=metadata) + if testing.requires.schemas.enabled: + Sequence( + "user_id_seq", schema=config.test_schema, metadata=metadata + ) + Sequence( + "schema_seq", schema=config.test_schema, metadata=metadata ) - finally: - connection.execute(schema.DropSequence(s1)) + Table( + "user_id_table", metadata, Column("id", Integer, primary_key=True), + ) + + def test_has_sequence(self, connection): + eq_( + inspect(connection).has_sequence("user_id_seq"), True, + ) + + def test_has_sequence_other_object(self, connection): + eq_( + inspect(connection).has_sequence("user_id_table"), False, + ) @testing.requires.schemas def test_has_sequence_schema(self, connection): - s1 = Sequence("user_id_seq", schema=config.test_schema) - connection.execute(schema.CreateSequence(s1)) - try: - eq_( - connection.dialect.has_sequence( - connection, "user_id_seq", schema=config.test_schema - ), - True, - ) - finally: - connection.execute(schema.DropSequence(s1)) + eq_( + inspect(connection).has_sequence( + "user_id_seq", schema=config.test_schema + ), + True, + ) def test_has_sequence_neg(self, connection): - eq_(connection.dialect.has_sequence(connection, "user_id_seq"), False) + eq_( + inspect(connection).has_sequence("some_sequence"), False, + ) @testing.requires.schemas def test_has_sequence_schemas_neg(self, connection): eq_( - connection.dialect.has_sequence( - connection, "user_id_seq", schema=config.test_schema + inspect(connection).has_sequence( + "some_sequence", schema=config.test_schema ), False, ) @testing.requires.schemas def test_has_sequence_default_not_in_remote(self, connection): - s1 = Sequence("user_id_seq") - connection.execute(schema.CreateSequence(s1)) - try: - eq_( - connection.dialect.has_sequence( - connection, "user_id_seq", schema=config.test_schema - ), - False, - ) - finally: - connection.execute(schema.DropSequence(s1)) + eq_( + inspect(connection).has_sequence( + "other_sequence", schema=config.test_schema + ), + False, + ) @testing.requires.schemas def test_has_sequence_remote_not_in_default(self, connection): - s1 = Sequence("user_id_seq", schema=config.test_schema) - connection.execute(schema.CreateSequence(s1)) - try: - eq_( - connection.dialect.has_sequence(connection, "user_id_seq"), - False, - ) - finally: - connection.execute(schema.DropSequence(s1)) + eq_( + inspect(connection).has_sequence("schema_seq"), False, + ) + + def test_get_sequence_names(self, connection): + exp = {"other_seq", "user_id_seq"} + + res = set(inspect(connection).get_sequence_names()) + is_true(res.intersection(exp) == exp) + is_true("schema_seq" not in res) + + @testing.requires.schemas + def test_get_sequence_names_no_sequence_schema(self, connection): + eq_( + inspect(connection).get_sequence_names( + schema=config.test_schema_2 + ), + [], + ) + + @testing.requires.schemas + def test_get_sequence_names_sequences_schema(self, connection): + eq_( + sorted( + inspect(connection).get_sequence_names( + schema=config.test_schema + ) + ), + ["schema_seq", "user_id_seq"], + ) + + +class HasSequenceTestEmpty(fixtures.TestBase): + __requires__ = ("sequences",) + __backend__ = True + + def test_get_sequence_names_no_sequence(self, connection): + eq_( + inspect(connection).get_sequence_names(), [], + ) |
