summaryrefslogtreecommitdiff
path: root/lib/sqlalchemy/testing/suite/test_sequence.py
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/suite/test_sequence.py
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/suite/test_sequence.py')
-rw-r--r--lib/sqlalchemy/testing/suite/test_sequence.py130
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(), [],
+ )