summaryrefslogtreecommitdiff
path: root/lib/sqlalchemy/testing
diff options
context:
space:
mode:
authorFederico Caselli <cfederico87@gmail.com>2020-04-13 12:16:21 +0200
committerFederico Caselli <cfederico87@gmail.com>2020-06-03 20:53:47 +0200
commit8dcf876fe9a06f3360b8d260459cdff050b2aa00 (patch)
treeab6a0493ef1c1b7b9d869df3e3b36d9ee557c318 /lib/sqlalchemy/testing
parent7f0cb933f2b1979a8d781855618b7fd3bf280037 (diff)
downloadsqlalchemy-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.py4
-rw-r--r--lib/sqlalchemy/testing/suite/test_sequence.py130
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(), [],
+ )