diff options
author | Mike Bayer <mike_mp@zzzcomputing.com> | 2023-01-05 10:34:37 -0500 |
---|---|---|
committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2023-01-09 09:51:03 -0500 |
commit | 3aeb30ea104ca6cbe6972f7ec64233cadbaccd82 (patch) | |
tree | 72a21cad47aaad28e11428b646883e18df4903e7 /test/dialect/oracle/test_reflection.py | |
parent | aafded2fe6f955d23ab974574978f2a6f96234b9 (diff) | |
download | sqlalchemy-3aeb30ea104ca6cbe6972f7ec64233cadbaccd82.tar.gz |
warn and skip for FKs that refer to invisible cols for Oracle
Supported use case for foreign key constraints where the local column is
marked as "invisible". The errors normally generated when a
:class:`.ForeignKeyConstraint` is created that check for the target column
are disabled when reflecting, and the constraint is skipped with a warning
in the same way which already occurs for an :class:`.Index` with a similar
issue.
tests are added for indexes, unique constraints, and primary key
constraints, which were already working; indexes and uniques warn,
primary keys don't which we would assume is because we never see those
PK columns in the first place.
Constraints now raise an informative ConstraintColumnNotFoundError
in the general case for strings in the "pending colargs" collection
not being resolvable.
Fixes: #9059
Change-Id: I400cf0bff6abba0e0c75f38b07617be1a8ec3453
Diffstat (limited to 'test/dialect/oracle/test_reflection.py')
-rw-r--r-- | test/dialect/oracle/test_reflection.py | 170 |
1 files changed, 161 insertions, 9 deletions
diff --git a/test/dialect/oracle/test_reflection.py b/test/dialect/oracle/test_reflection.py index afe9ad7aa..0ac991d2a 100644 --- a/test/dialect/oracle/test_reflection.py +++ b/test/dialect/oracle/test_reflection.py @@ -37,6 +37,7 @@ from sqlalchemy.testing import AssertsCompiledSQL from sqlalchemy.testing import config from sqlalchemy.testing import eq_ from sqlalchemy.testing import expect_raises +from sqlalchemy.testing import expect_warnings from sqlalchemy.testing import fixtures from sqlalchemy.testing import is_ from sqlalchemy.testing import is_true @@ -345,20 +346,24 @@ class MultiSchemaTest(fixtures.TestBase, AssertsCompiledSQL): ) -class ConstraintTest(fixtures.TablesTest): +class ConstraintTest(AssertsCompiledSQL, fixtures.TestBase): __only_on__ = "oracle" __backend__ = True - run_deletes = None - @classmethod - def define_tables(cls, metadata): - Table("foo", metadata, Column("id", Integer, primary_key=True)) + @testing.fixture + def plain_foo_table(self, metadata, connection): + foo = Table("foo", metadata, Column("id", Integer, primary_key=True)) + foo.create(connection) + return foo + + def test_oracle_has_no_on_update_cascade( + self, metadata, connection, plain_foo_table + ): - def test_oracle_has_no_on_update_cascade(self, connection): bar = Table( "bar", - self.tables_test_metadata, + metadata, Column("id", Integer, primary_key=True), Column( "foo_id", Integer, ForeignKey("foo.id", onupdate="CASCADE") @@ -368,14 +373,17 @@ class ConstraintTest(fixtures.TablesTest): bat = Table( "bat", - self.tables_test_metadata, + metadata, Column("id", Integer, primary_key=True), Column("foo_id", Integer), ForeignKeyConstraint(["foo_id"], ["foo.id"], onupdate="CASCADE"), ) assert_warns(exc.SAWarning, bat.create, connection) - def test_reflect_check_include_all(self, connection): + def test_reflect_check_include_all( + self, metadata, connection, plain_foo_table + ): + insp = inspect(connection) eq_(insp.get_check_constraints("foo"), []) eq_( @@ -386,6 +394,150 @@ class ConstraintTest(fixtures.TablesTest): ['"ID" IS NOT NULL'], ) + @testing.fixture + def invisible_fk_fixture(self, metadata, connection): + Table("table_b", metadata, Column("id", Integer, primary_key=True)) + Table( + "table_a", + metadata, + Column("id", Integer, primary_key=True), + Column("a_col1", Integer), + ) + metadata.create_all(connection) + + connection.exec_driver_sql( + "alter table table_a modify (a_col1 invisible)" + ) + + connection.exec_driver_sql( + "alter table table_a add constraint FK_table_a_a_col1 " + "foreign key(a_col1) references table_b" + ) + + @testing.fixture + def invisible_index_fixture(self, metadata, connection): + Table( + "table_a", + metadata, + Column("id", Integer, primary_key=True), + Column("a_col1", Integer), + Index("idx_col1", "a_col1"), + ) + metadata.create_all(connection) + + connection.exec_driver_sql( + "alter table table_a modify (a_col1 invisible)" + ) + + @testing.fixture + def invisible_uq_fixture(self, metadata, connection): + Table( + "table_a", + metadata, + Column("id", Integer, primary_key=True), + Column("a_col1", Integer), + UniqueConstraint("a_col1", name="uq_col1"), + ) + metadata.create_all(connection) + + connection.exec_driver_sql( + "alter table table_a modify (a_col1 invisible)" + ) + + @testing.fixture + def invisible_pk_fixture(self, metadata, connection): + Table( + "table_a", + metadata, + Column("id", Integer, primary_key=True), + Column("a_col1", Integer), + ) + Table( + "table_b", + metadata, + Column("comp_id1", Integer, primary_key=True), + Column("comp_id2", Integer, primary_key=True), + Column("a_col1", Integer), + ) + metadata.create_all(connection) + + connection.exec_driver_sql("alter table table_a modify (id invisible)") + connection.exec_driver_sql( + "alter table table_b modify (comp_id2 invisible)" + ) + + def test_no_resolve_fks_w_invisible( + self, connection, invisible_fk_fixture + ): + metadata_reflect = MetaData() + + with expect_warnings( + r"On reflected table table_a, skipping reflection of foreign key " + r"constraint fk_table_a_a_col1; one or more subject columns " + r"within name\(s\) a_col1 are not present in the table" + ): + metadata_reflect.reflect(connection) + + ta = metadata_reflect.tables["table_a"] + tb = metadata_reflect.tables["table_b"] + self.assert_compile( + select(ta, tb), + "SELECT table_a.id, table_b.id AS id_1 FROM table_a, table_b", + ) + + def test_no_resolve_idx_w_invisible( + self, connection, invisible_index_fixture + ): + metadata_reflect = MetaData() + + with expect_warnings( + r"index key 'a_col1' was not located in columns " + r"for table 'table_a'" + ): + metadata_reflect.reflect(connection) + + ta = metadata_reflect.tables["table_a"] + self.assert_compile( + select(ta), + "SELECT table_a.id FROM table_a", + ) + + def test_no_resolve_uq_w_invisible(self, connection, invisible_uq_fixture): + metadata_reflect = MetaData() + + with expect_warnings( + r"index key 'a_col1' was not located in columns " + r"for table 'table_a'" + ): + metadata_reflect.reflect(connection) + + ta = metadata_reflect.tables["table_a"] + self.assert_compile( + select(ta), + "SELECT table_a.id FROM table_a", + ) + + def test_no_resolve_pk_w_invisible(self, connection, invisible_pk_fixture): + metadata_reflect = MetaData() + + metadata_reflect.reflect(connection) + + # single col pk fully invisible + ta = metadata_reflect.tables["table_a"] + eq_(list(ta.primary_key), []) + self.assert_compile( + select(ta), + "SELECT table_a.a_col1 FROM table_a", + ) + + # composite pk one col invisible + tb = metadata_reflect.tables["table_b"] + eq_(list(tb.primary_key), [tb.c.comp_id1]) + self.assert_compile( + select(tb), + "SELECT table_b.comp_id1, table_b.a_col1 FROM table_b", + ) + class SystemTableTablenamesTest(fixtures.TestBase): __only_on__ = "oracle" |