diff options
author | Federico Caselli <cfederico87@gmail.com> | 2020-04-13 12:16:21 +0200 |
---|---|---|
committer | Federico Caselli <cfederico87@gmail.com> | 2020-06-03 20:53:47 +0200 |
commit | 8dcf876fe9a06f3360b8d260459cdff050b2aa00 (patch) | |
tree | ab6a0493ef1c1b7b9d869df3e3b36d9ee557c318 /lib/sqlalchemy/testing/suite/test_sequence.py | |
parent | 7f0cb933f2b1979a8d781855618b7fd3bf280037 (diff) | |
download | sqlalchemy-8dcf876fe9a06f3360b8d260459cdff050b2aa00.tar.gz |
Added reflection method :meth:`.Inspector.get_sequence_names`
Added new reflection method :meth:`.Inspector.get_sequence_names` which
returns all the sequences defined. Support for this method has been added
to the backend that support :class:`.Sequence`: PostgreSql, Oracle,
MSSQL and MariaDB >= 10.3.
Fixes: #2056
Change-Id: I0949696a39aa28c849edf2504779241f7443778a
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(), [], + ) |