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 | |
| 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')
| -rw-r--r-- | lib/sqlalchemy/testing/fixtures.py | 4 | ||||
| -rw-r--r-- | lib/sqlalchemy/testing/suite/test_sequence.py | 130 |
2 files changed, 89 insertions, 45 deletions
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(), [], + ) |
