summaryrefslogtreecommitdiff
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/dialect/postgresql/test_reflection.py104
-rw-r--r--test/dialect/test_sqlite.py2
-rw-r--r--test/perf/many_table_reflection.py2
-rw-r--r--test/requirements.py4
-rw-r--r--test/sql/test_metadata.py32
5 files changed, 134 insertions, 10 deletions
diff --git a/test/dialect/postgresql/test_reflection.py b/test/dialect/postgresql/test_reflection.py
index 99bc14d78..6bcd7a87c 100644
--- a/test/dialect/postgresql/test_reflection.py
+++ b/test/dialect/postgresql/test_reflection.py
@@ -32,6 +32,7 @@ from sqlalchemy.dialects.postgresql import TSRANGE
from sqlalchemy.engine import ObjectKind
from sqlalchemy.engine import ObjectScope
from sqlalchemy.schema import CreateIndex
+from sqlalchemy.sql import ddl as sa_ddl
from sqlalchemy.sql.schema import CheckConstraint
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import fixtures
@@ -1496,6 +1497,7 @@ class ReflectionTest(
"initially": "DEFERRED",
"match": "FULL",
},
+ "comment": None,
},
"company_industry_id_fkey": {
"name": "company_industry_id_fkey",
@@ -1504,6 +1506,7 @@ class ReflectionTest(
"referred_table": "industry",
"referred_schema": None,
"options": {"onupdate": "CASCADE", "ondelete": "CASCADE"},
+ "comment": None,
},
}
metadata.create_all(connection)
@@ -1928,7 +1931,7 @@ class ReflectionTest(
)
def test_reflect_check_warning(self):
- rows = [("foo", "some name", "NOTCHECK foobar")]
+ rows = [("foo", "some name", "NOTCHECK foobar", None)]
conn = mock.Mock(
execute=lambda *arg, **kw: mock.MagicMock(
fetchall=lambda: rows, __iter__=lambda self: iter(rows)
@@ -1941,10 +1944,20 @@ class ReflectionTest(
def test_reflect_extra_newlines(self):
rows = [
- ("foo", "some name", "CHECK (\n(a \nIS\n NOT\n\n NULL\n)\n)"),
- ("foo", "some other name", "CHECK ((b\nIS\nNOT\nNULL))"),
- ("foo", "some CRLF name", "CHECK ((c\r\n\r\nIS\r\nNOT\r\nNULL))"),
- ("foo", "some name", "CHECK (c != 'hi\nim a name\n')"),
+ (
+ "foo",
+ "some name",
+ "CHECK (\n(a \nIS\n NOT\n\n NULL\n)\n)",
+ None,
+ ),
+ ("foo", "some other name", "CHECK ((b\nIS\nNOT\nNULL))", None),
+ (
+ "foo",
+ "some CRLF name",
+ "CHECK ((c\r\n\r\nIS\r\nNOT\r\nNULL))",
+ None,
+ ),
+ ("foo", "some name", "CHECK (c != 'hi\nim a name\n')", None),
]
conn = mock.Mock(
execute=lambda *arg, **kw: mock.MagicMock(
@@ -1960,18 +1973,30 @@ class ReflectionTest(
{
"name": "some name",
"sqltext": "a \nIS\n NOT\n\n NULL\n",
+ "comment": None,
+ },
+ {
+ "name": "some other name",
+ "sqltext": "b\nIS\nNOT\nNULL",
+ "comment": None,
},
- {"name": "some other name", "sqltext": "b\nIS\nNOT\nNULL"},
{
"name": "some CRLF name",
"sqltext": "c\r\n\r\nIS\r\nNOT\r\nNULL",
+ "comment": None,
+ },
+ {
+ "name": "some name",
+ "sqltext": "c != 'hi\nim a name\n'",
+ "comment": None,
},
- {"name": "some name", "sqltext": "c != 'hi\nim a name\n'"},
],
)
def test_reflect_with_not_valid_check_constraint(self):
- rows = [("foo", "some name", "CHECK ((a IS NOT NULL)) NOT VALID")]
+ rows = [
+ ("foo", "some name", "CHECK ((a IS NOT NULL)) NOT VALID", None)
+ ]
conn = mock.Mock(
execute=lambda *arg, **kw: mock.MagicMock(
fetchall=lambda: rows, __iter__=lambda self: iter(rows)
@@ -1987,6 +2012,7 @@ class ReflectionTest(
"name": "some name",
"sqltext": "a IS NOT NULL",
"dialect_options": {"not_valid": True},
+ "comment": None,
}
],
)
@@ -2069,6 +2095,68 @@ class ReflectionTest(
[["b"]],
)
+ def test_reflection_constraint_comments(self, connection, metadata):
+ t = Table(
+ "foo",
+ metadata,
+ Column("id", Integer),
+ Column("foo_id", ForeignKey("foo.id", name="fk_1")),
+ Column("foo_other_id", ForeignKey("foo.id", name="fk_2")),
+ CheckConstraint("id>0", name="ch_1"),
+ CheckConstraint("id<1000", name="ch_2"),
+ PrimaryKeyConstraint("id", name="foo_pk"),
+ UniqueConstraint("id", "foo_id", name="un_1"),
+ UniqueConstraint("id", "foo_other_id", name="un_2"),
+ )
+ metadata.create_all(connection)
+
+ def check(elements, exp):
+ elements = {c["name"]: c["comment"] for c in elements}
+ eq_(elements, exp)
+
+ def all_none():
+ insp = inspect(connection)
+ is_(insp.get_pk_constraint("foo")["comment"], None)
+ check(
+ insp.get_check_constraints("foo"), {"ch_1": None, "ch_2": None}
+ )
+ check(
+ insp.get_unique_constraints("foo"),
+ {"un_1": None, "un_2": None},
+ )
+ check(insp.get_foreign_keys("foo"), {"fk_1": None, "fk_2": None})
+
+ all_none()
+
+ c = next(c for c in t.constraints if c.name == "ch_1")
+ u = next(c for c in t.constraints if c.name == "un_1")
+ f = next(c for c in t.foreign_key_constraints if c.name == "fk_1")
+ p = t.primary_key
+ c.comment = "cc comment"
+ u.comment = "uc comment"
+ f.comment = "fc comment"
+ p.comment = "pk comment"
+ for cst in [c, u, f, p]:
+ connection.execute(sa_ddl.SetConstraintComment(cst))
+
+ insp = inspect(connection)
+ eq_(insp.get_pk_constraint("foo")["comment"], "pk comment")
+ check(
+ insp.get_check_constraints("foo"),
+ {"ch_1": "cc comment", "ch_2": None},
+ )
+ check(
+ insp.get_unique_constraints("foo"),
+ {"un_1": "uc comment", "un_2": None},
+ )
+ check(
+ insp.get_foreign_keys("foo"), {"fk_1": "fc comment", "fk_2": None}
+ )
+
+ for cst in [c, u, f, p]:
+ connection.execute(sa_ddl.DropConstraintComment(cst))
+ all_none()
+
class CustomTypeReflectionTest(fixtures.TestBase):
class CustomType:
diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py
index ed9d67612..aae6d41e9 100644
--- a/test/dialect/test_sqlite.py
+++ b/test/dialect/test_sqlite.py
@@ -2372,8 +2372,8 @@ class ConstraintReflectionTest(fixtures.TestBase):
eq_(
inspector.get_check_constraints("cp"),
[
- {"sqltext": "q > 1 AND q < 6", "name": None},
{"sqltext": "q == 1 OR (q > 2 AND q < 5)", "name": "cq"},
+ {"sqltext": "q > 1 AND q < 6", "name": None},
],
)
diff --git a/test/perf/many_table_reflection.py b/test/perf/many_table_reflection.py
index 8749df5c2..149237fae 100644
--- a/test/perf/many_table_reflection.py
+++ b/test/perf/many_table_reflection.py
@@ -100,7 +100,7 @@ def generate_meta(schema_name, table_number, min_cols, max_cols, dialect_name):
def log(fn):
@wraps(fn)
def wrap(*a, **kw):
- print("Running ", fn.__name__, "...", flush=True, end="")
+ print("Running", fn.__name__, "...", flush=True, end="")
try:
r = fn(*a, **kw)
except NotImplementedError:
diff --git a/test/requirements.py b/test/requirements.py
index db12990a3..17c11fc5d 100644
--- a/test/requirements.py
+++ b/test/requirements.py
@@ -171,6 +171,10 @@ class DefaultRequirements(SuiteRequirements):
return only_on(["postgresql", "mysql", "mariadb", "oracle"])
@property
+ def constraint_comment_reflection(self):
+ return only_on(["postgresql"])
+
+ @property
def unbounded_varchar(self):
"""Target database must support VARCHAR with no length"""
diff --git a/test/sql/test_metadata.py b/test/sql/test_metadata.py
index 1a070fcf5..33b6e130f 100644
--- a/test/sql/test_metadata.py
+++ b/test/sql/test_metadata.py
@@ -237,6 +237,7 @@ class MetaDataTest(fixtures.TestBase, ComparesTables):
deferrable="Z",
initially="Q",
link_to_name=True,
+ comment="foo",
)
fk1 = ForeignKey(c1, **kw)
@@ -259,6 +260,7 @@ class MetaDataTest(fixtures.TestBase, ComparesTables):
name="name",
initially=True,
deferrable=True,
+ comment="foo",
_create_rule=r,
)
c2 = c._copy()
@@ -266,6 +268,7 @@ class MetaDataTest(fixtures.TestBase, ComparesTables):
eq_(str(c2.sqltext), "foo bar")
eq_(c2.initially, True)
eq_(c2.deferrable, True)
+ eq_(c2.comment, "foo")
assert c2._create_rule is r
def test_col_replace_w_constraint(self):
@@ -3577,6 +3580,35 @@ class ConstraintTest(fixtures.TestBase):
for c in t3.constraints:
assert c.table is t3
+ def test_ColumnCollectionConstraint_copy(self):
+ m = MetaData()
+
+ t = Table("tbl", m, Column("a", Integer), Column("b", Integer))
+ t2 = Table("t2", m, Column("a", Integer), Column("b", Integer))
+
+ kw = {
+ "comment": "baz",
+ "name": "ccc",
+ "initially": "foo",
+ "deferrable": "bar",
+ }
+
+ UniqueConstraint(t.c.a, **kw)
+ CheckConstraint(t.c.a > 5, **kw)
+ ForeignKeyConstraint([t.c.a], [t2.c.a], **kw)
+ PrimaryKeyConstraint(t.c.a, **kw)
+
+ m2 = MetaData()
+
+ t3 = t.to_metadata(m2)
+
+ eq_(len(t3.constraints), 4)
+
+ for c in t3.constraints:
+ assert c.table is t3
+ for k, v in kw.items():
+ eq_(getattr(c, k), v)
+
def test_check_constraint_copy(self):
m = MetaData()
t = Table("tbl", m, Column("a", Integer), Column("b", Integer))
View Lorry Controller Status