From 8dcf876fe9a06f3360b8d260459cdff050b2aa00 Mon Sep 17 00:00:00 2001 From: Federico Caselli Date: Mon, 13 Apr 2020 12:16:21 +0200 Subject: 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 --- lib/sqlalchemy/testing/fixtures.py | 4 + lib/sqlalchemy/testing/suite/test_sequence.py | 130 +++++++++++++++++--------- 2 files changed, 89 insertions(+), 45 deletions(-) (limited to 'lib/sqlalchemy/testing') diff --git a/lib/sqlalchemy/testing/fixtures.py b/lib/sqlalchemy/testing/fixtures.py index 041daf35e..931fa6b77 100644 --- a/lib/sqlalchemy/testing/fixtures.py +++ b/lib/sqlalchemy/testing/fixtures.py @@ -135,6 +135,7 @@ class TablesTest(TestBase): metadata = None tables = None other = None + sequences = None @classmethod def setup_class(cls): @@ -153,6 +154,7 @@ class TablesTest(TestBase): cls.other = adict() cls.tables = adict() + cls.sequences = adict() cls.bind = cls.setup_bind() cls.metadata = sa.MetaData() @@ -172,6 +174,7 @@ class TablesTest(TestBase): if cls.run_create_tables == "once": cls.metadata.create_all(cls.bind) cls.tables.update(cls.metadata.tables) + cls.sequences.update(cls.metadata._sequences) def _setup_each_tables(self): if self.run_define_tables == "each": @@ -179,6 +182,7 @@ class TablesTest(TestBase): if self.run_create_tables == "each": self.metadata.create_all(self.bind) self.tables.update(self.metadata.tables) + self.sequences.update(self.metadata._sequences) elif self.run_create_tables == "each": self.metadata.create_all(self.bind) 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(), [], + ) -- cgit v1.2.1