test_metadata.py

from contextlib import contextmanager
import pickle

import sqlalchemy as tsa
from sqlalchemy import ARRAY
from sqlalchemy import BigInteger
from sqlalchemy import bindparam
from sqlalchemy import BLANK_SCHEMA
from sqlalchemy import Boolean
from sqlalchemy import CheckConstraint
from sqlalchemy import Column
from sqlalchemy import column
from sqlalchemy import ColumnDefault
from sqlalchemy import Computed
from sqlalchemy import desc
from sqlalchemy import Enum
from sqlalchemy import event
from sqlalchemy import exc
from sqlalchemy import Float
from sqlalchemy import ForeignKey
from sqlalchemy import ForeignKeyConstraint
from sqlalchemy import func
from sqlalchemy import Identity
from sqlalchemy import Index
from sqlalchemy import insert_sentinel
from sqlalchemy import Integer
from sqlalchemy import MetaData
from sqlalchemy import Numeric
from sqlalchemy import PrimaryKeyConstraint
from sqlalchemy import schema
from sqlalchemy import select
from sqlalchemy import Sequence
from sqlalchemy import String
from sqlalchemy import Table
from sqlalchemy import table
from sqlalchemy import testing
from sqlalchemy import text
from sqlalchemy import TypeDecorator
from sqlalchemy import types as sqltypes
from sqlalchemy import Unicode
from sqlalchemy import UniqueConstraint
from sqlalchemy.engine import default
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.ext.compiler import deregister
from sqlalchemy.schema import AddConstraint
from sqlalchemy.schema import CreateIndex
from sqlalchemy.schema import DefaultClause
from sqlalchemy.schema import DropIndex
from sqlalchemy.sql import naming
from sqlalchemy.sql import operators
from sqlalchemy.sql.base import _NONE_NAME
from sqlalchemy.sql.elements import literal_column
from sqlalchemy.sql.schema import _InsertSentinelColumnDefault
from sqlalchemy.sql.schema import RETAIN_SCHEMA
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import ComparesTables
from sqlalchemy.testing import emits_warning
from sqlalchemy.testing import eq_
from sqlalchemy.testing import eq_ignore_whitespace
from sqlalchemy.testing import expect_raises_message
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
from sqlalchemy.testing import is_false
from sqlalchemy.testing import is_true
from sqlalchemy.testing import mock
from sqlalchemy.testing import Variation
from sqlalchemy.testing.assertions import expect_warnings


class MetaDataTest(fixtures.TestBase, ComparesTables):
    def test_metadata_contains(self):
        metadata = MetaData()
        t1 = Table("t1", metadata, Column("x", Integer))
        t2 = Table("t2", metadata, Column("x", Integer), schema="foo")
        t3 = Table("t2", MetaData(), Column("x", Integer))
        t4 = Table("t1", MetaData(), Column("x", Integer), schema="foo")

        assert "t1" in metadata
        assert "foo.t2" in metadata
        assert "t2" not in metadata
        assert "foo.t1" not in metadata
        assert t1 in metadata
        assert t2 in metadata
        assert t3 not in metadata
        assert t4 not in metadata

    def test_uninitialized_column_copy(self):
        for col in [
            Column("foo", String(), nullable=False),
            Column("baz", String(), unique=True),
            Column(Integer(), primary_key=True),
            Column(
                "bar",
                Integer(),
                Sequence("foo_seq"),
                primary_key=True,
                key="bar",
                autoincrement="ignore_fk",
            ),
            Column(Integer(), ForeignKey("bat.blah"), doc="this is a col"),
            Column(
                "bar",
                Integer(),
                ForeignKey("bat.blah"),
                primary_key=True,
                comment="this is a comment",
                key="bar",
            ),
            Column("bar", Integer(), info={"foo": "bar"}),
        ]:
            c2 = col._copy()
            for attr in (
                "name",
                "type",
                "nullable",
                "primary_key",
                "key",
                "unique",
                "info",
                "doc",
                "autoincrement",
            ):
                eq_(getattr(col, attr), getattr(c2, attr))
            eq_(len(col.foreign_keys), len(c2.foreign_keys))
            if col.default:
                eq_(c2.default.name, "foo_seq")
            for a1, a2 in zip(col.foreign_keys, c2.foreign_keys):
                assert a1 is not a2
                eq_(a2._colspec, "bat.blah")

    def test_col_subclass_copy(self):
        class MyColumn(schema.Column):
            def __init__(self, *args, **kw):
                self.widget = kw.pop("widget", None)
                super().__init__(*args, **kw)

            def _copy(self, *arg, **kw):
                c = super()._copy(*arg, **kw)
                c.widget = self.widget
                return c

        c1 = MyColumn("foo", Integer, widget="x")
        c2 = c1._copy()
        assert isinstance(c2, MyColumn)
        eq_(c2.widget, "x")

    def test_uninitialized_column_copy_events(self):
        msgs = []

        def write(c, t):
            msgs.append("attach %s.%s" % (t.name, c.name))

        c1 = Column("foo", String())
        m = MetaData()
        for i in range(3):
            cx = c1._copy()
            # as of 0.7, these events no longer copy.  its expected
            # that listeners will be re-established from the
            # natural construction of things.
            cx._on_table_attach(write)
            Table("foo%d" % i, m, cx)
        eq_(msgs, ["attach foo0.foo", "attach foo1.foo", "attach foo2.foo"])

    def test_schema_collection_add(self):
        metadata = MetaData()

        Table("t1", metadata, Column("x", Integer), schema="foo")
        Table("t2", metadata, Column("x", Integer), schema="bar")
        Table("t3", metadata, Column("x", Integer))

        eq_(metadata._schemas, {"foo", "bar"})
        eq_(len(metadata.tables), 3)

    def test_schema_collection_remove(self):
        metadata = MetaData()

        t1 = Table("t1", metadata, Column("x", Integer), schema="foo")
        Table("t2", metadata, Column("x", Integer), schema="bar")
        t3 = Table("t3", metadata, Column("x", Integer), schema="bar")

        metadata.remove(t3)
        eq_(metadata._schemas, {"foo", "bar"})
        eq_(len(metadata.tables), 2)

        metadata.remove(t1)
        eq_(metadata._schemas, {"bar"})
        eq_(len(metadata.tables), 1)

    def test_schema_collection_remove_all(self):
        metadata = MetaData()

        Table("t1", metadata, Column("x", Integer), schema="foo")
        Table("t2", metadata, Column("x", Integer), schema="bar")

        metadata.clear()
        eq_(metadata._schemas, set())
        eq_(len(metadata.tables), 0)

    def test_metadata_tables_immutable(self):
        # this use case was added due to #1917.
        metadata = MetaData()

        Table("t1", metadata, Column("x", Integer))
        assert "t1" in metadata.tables

        assert_raises(TypeError, lambda: metadata.tables.pop("t1"))

    @testing.provide_metadata
    def test_dupe_tables(self):
        metadata = self.metadata
        Table(
            "table1",
            metadata,
            Column("col1", Integer, primary_key=True),
            Column("col2", String(20)),
        )

        metadata.create_all(testing.db)
        Table("table1", metadata, autoload_with=testing.db)

        def go():
            Table(
                "table1",
                metadata,
                Column("col1", Integer, primary_key=True),
                Column("col2", String(20)),
            )

        assert_raises_message(
            tsa.exc.InvalidRequestError,
            "Table 'table1' is already defined for this "
            "MetaData instance.  Specify 'extend_existing=True' "
            "to redefine options and columns on an existing "
            "Table object.",
            go,
        )

    def test_fk_copy(self):
        c1 = Column("foo", Integer)
        c2 = Column("bar", Integer)
        m = MetaData()
        t1 = Table("t", m, c1, c2)

        kw = dict(
            onupdate="X",
            ondelete="Y",
            use_alter=True,
            name="f1",
            deferrable="Z",
            initially="Q",
            link_to_name=True,
            comment="foo",
        )

        fk1 = ForeignKey(c1, **kw)
        fk2 = ForeignKeyConstraint((c1,), (c2,), **kw)

        t1.append_constraint(fk2)
        fk1c = fk1._copy()
        fk2c = fk2._copy()

        for k in kw:
            eq_(getattr(fk1c, k), kw[k])
            eq_(getattr(fk2c, k), kw[k])

    def test_check_constraint_copy(self):
        def r(x):
            return x

        c = CheckConstraint(
            "foo bar",
            name="name",
            initially=True,
            deferrable=True,
            comment="foo",
            _create_rule=r,
        )
        c2 = c._copy()
        eq_(c2.name, "name")
        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):
        m = MetaData()
        a = Table("a", m, Column("id", Integer, primary_key=True))

        aid = Column("a_id", ForeignKey("a.id"))
        b = Table("b", m, aid)
        b.append_column(aid)

        assert b.c.a_id.references(a.c.id)
        eq_(len(b.constraints), 2)

    def test_fk_construct(self):
        c1 = Column("foo", Integer)
        c2 = Column("bar", Integer)
        m = MetaData()
        t1 = Table("t", m, c1, c2)
        fk1 = ForeignKeyConstraint(("foo",), ("bar",), table=t1)
        assert fk1 in t1.constraints

    def test_fk_constraint_col_collection_w_table(self):
        c1 = Column("foo", Integer)
        c2 = Column("bar", Integer)
        m = MetaData()
        t1 = Table("t", m, c1, c2)
        fk1 = ForeignKeyConstraint(("foo",), ("bar",), table=t1)
        eq_(dict(fk1.columns), {"foo": c1})

    def test_fk_constraint_col_collection_no_table(self):
        fk1 = ForeignKeyConstraint(("foo", "bat"), ("bar", "hoho"))
        eq_(dict(fk1.columns), {})
        eq_(fk1.column_keys, ["foo", "bat"])
        eq_(fk1._col_description, "foo, bat")
        eq_(fk1._elements, {"foo": fk1.elements[0], "bat": fk1.elements[1]})

    def test_fk_constraint_col_collection_no_table_real_cols(self):
        c1 = Column("foo", Integer)
        c2 = Column("bar", Integer)
        fk1 = ForeignKeyConstraint((c1,), (c2,))
        eq_(dict(fk1.columns), {})
        eq_(fk1.column_keys, ["foo"])
        eq_(fk1._col_description, "foo")
        eq_(fk1._elements, {"foo": fk1.elements[0]})

    def test_fk_constraint_col_collection_added_to_table(self):
        c1 = Column("foo", Integer)
        m = MetaData()
        fk1 = ForeignKeyConstraint(("foo",), ("bar",))
        Table("t", m, c1, fk1)
        eq_(dict(fk1.columns), {"foo": c1})
        eq_(fk1._elements, {"foo": fk1.elements[0]})

    def test_fk_constraint_col_collection_via_fk(self):
        fk = ForeignKey("bar")
        c1 = Column("foo", Integer, fk)
        m = MetaData()
        t1 = Table("t", m, c1)
        fk1 = fk.constraint
        eq_(fk1.column_keys, ["foo"])
        assert fk1 in t1.constraints
        eq_(fk1.column_keys, ["foo"])
        eq_(dict(fk1.columns), {"foo": c1})
        eq_(fk1._elements, {"foo": fk})

    def test_fk_no_such_parent_col_error(self):
        meta = MetaData()
        a = Table("a", meta, Column("a", Integer))
        Table("b", meta, Column("b", Integer))

        def go():
            a.append_constraint(ForeignKeyConstraint(["x"], ["b.b"]))

        assert_raises_message(
            exc.ArgumentError,
            "Can't create ForeignKeyConstraint on "
            "table 'a': no column named 'x' is present.",
            go,
        )

    def test_fk_given_non_col(self):
        not_a_col = bindparam("x")
        assert_raises_message(
            exc.ArgumentError,
            "String column name or Column object for DDL foreign "
            "key constraint expected, got .*Bind",
            ForeignKey,
            not_a_col,
        )

    def test_fk_given_non_col_clauseelem(self):
        class Foo:
            def __clause_element__(self):
                return bindparam("x")

        assert_raises_message(
            exc.ArgumentError,
            "String column name or Column object for DDL foreign "
            "key constraint expected, got .*Foo",
            ForeignKey,
            Foo(),
        )

    def test_fk_given_col_non_table(self):
        t = Table("t", MetaData(), Column("x", Integer))
        xa = t.alias().c.x
        assert_raises_message(
            exc.ArgumentError,
            "ForeignKey received Column not bound to a Table, got: .*Alias",
            ForeignKey,
            xa,
        )

    def test_fk_given_col_non_table_clauseelem(self):
        t = Table("t", MetaData(), Column("x", Integer))

        class Foo:
            def __clause_element__(self):
                return t.alias().c.x

        assert_raises_message(
            exc.ArgumentError,
            "ForeignKey received Column not bound to a Table, got: .*Alias",
            ForeignKey,
            Foo(),
        )

    def test_fk_no_such_target_col_error_upfront(self):
        meta = MetaData()
        a = Table("a", meta, Column("a", Integer))
        Table("b", meta, Column("b", Integer))

        a.append_constraint(ForeignKeyConstraint(["a"], ["b.x"]))

        assert_raises_message(
            exc.NoReferencedColumnError,
            "Could not initialize target column for ForeignKey 'b.x' on "
            "table 'a': table 'b' has no column named 'x'",
            getattr,
            list(a.foreign_keys)[0],
            "column",
        )

    def test_fk_no_such_target_col_error_delayed(self):
        meta = MetaData()
        a = Table("a", meta, Column("a", Integer))
        a.append_constraint(ForeignKeyConstraint(["a"], ["b.x"]))

        Table("b", meta, Column("b", Integer))

        assert_raises_message(
            exc.NoReferencedColumnError,
            "Could not initialize target column for ForeignKey 'b.x' on "
            "table 'a': table 'b' has no column named 'x'",
            getattr,
            list(a.foreign_keys)[0],
            "column",
        )

    def test_fk_mismatched_local_remote_cols(self):
        assert_raises_message(
            exc.ArgumentError,
            "ForeignKeyConstraint number of constrained columns must "
            "match the number of referenced columns.",
            ForeignKeyConstraint,
            ["a"],
            ["b.a", "b.b"],
        )

        assert_raises_message(
            exc.ArgumentError,
            "ForeignKeyConstraint number of constrained columns "
            "must match the number of referenced columns.",
            ForeignKeyConstraint,
            ["a", "b"],
            ["b.a"],
        )

        assert_raises_message(
            exc.ArgumentError,
            "ForeignKeyConstraint with duplicate source column "
            "references are not supported.",
            ForeignKeyConstraint,
            ["a", "a"],
            ["b.a", "b.b"],
        )

    def test_pickle_metadata_sequence_restated(self):
        m1 = MetaData()
        Table(
            "a",
            m1,
            Column("id", Integer, primary_key=True),
            Column("x", Integer, Sequence("x_seq")),
        )

        m2 = pickle.loads(pickle.dumps(m1))

        s2 = Sequence("x_seq")
        t2 = Table(
            "a",
            m2,
            Column("id", Integer, primary_key=True),
            Column("x", Integer, s2),
            extend_existing=True,
        )

        assert m2._sequences["x_seq"] is t2.c.x.default
        assert m2._sequences["x_seq"] is s2

    def test_sequence_restated_replaced(self):
        """Test restatement of Sequence replaces."""

        m1 = MetaData()
        s1 = Sequence("x_seq")
        t = Table("a", m1, Column("x", Integer, s1))
        assert m1._sequences["x_seq"] is s1

        s2 = Sequence("x_seq")
        Table("a", m1, Column("x", Integer, s2), extend_existing=True)
        assert t.c.x.default is s2
        assert m1._sequences["x_seq"] is s2

    def test_sequence_attach_to_table(self):
        m1 = MetaData()
        s1 = Sequence("s")
        Table("a", m1, Column("x", Integer, s1))
        assert s1.metadata is m1

    def test_sequence_attach_to_existing_table(self):
        m1 = MetaData()
        s1 = Sequence("s")
        t = Table("a", m1, Column("x", Integer))
        t.c.x._init_items(s1)
        assert s1.metadata is m1

    def test_pickle_metadata_sequence_implicit(self):
        m1 = MetaData()
        Table(
            "a",
            m1,
            Column("id", Integer, primary_key=True),
            Column("x", Integer, Sequence("x_seq")),
        )

        m2 = pickle.loads(pickle.dumps(m1))

        t2 = Table("a", m2, extend_existing=True)

        eq_(m2._sequences, {"x_seq": t2.c.x.default})

    def test_pickle_metadata_schema(self):
        m1 = MetaData()
        Table(
            "a",
            m1,
            Column("id", Integer, primary_key=True),
            Column("x", Integer, Sequence("x_seq")),
            schema="y",
        )

        m2 = pickle.loads(pickle.dumps(m1))

        Table("a", m2, schema="y", extend_existing=True)

        eq_(m2._schemas, m1._schemas)

    def test_metadata_schema_arg(self):
        m1 = MetaData(schema="sch1")
        m2 = MetaData(schema="sch1", quote_schema=True)
        m3 = MetaData(schema="sch1", quote_schema=False)
        m4 = MetaData()

        for (
            i,
            (
                name,
                metadata,
                schema_,
                quote_schema,
                exp_schema,
                exp_quote_schema,
            ),
        ) in enumerate(
            [
                ("t1", m1, None, None, "sch1", None),
                ("t2", m1, "sch2", None, "sch2", None),
                ("t3", m1, "sch2", True, "sch2", True),
                ("t4", m1, "sch1", None, "sch1", None),
                ("t5", m1, BLANK_SCHEMA, None, None, None),
                ("t1", m2, None, None, "sch1", True),
                ("t2", m2, "sch2", None, "sch2", None),
                ("t3", m2, "sch2", True, "sch2", True),
                ("t4", m2, "sch1", None, "sch1", None),
                ("t1", m3, None, None, "sch1", False),
                ("t2", m3, "sch2", None, "sch2", None),
                ("t3", m3, "sch2", True, "sch2", True),
                ("t4", m3, "sch1", None, "sch1", None),
                ("t1", m4, None, None, None, None),
                ("t2", m4, "sch2", None, "sch2", None),
                ("t3", m4, "sch2", True, "sch2", True),
                ("t4", m4, "sch1", None, "sch1", None),
                ("t5", m4, BLANK_SCHEMA, None, None, None),
            ]
        ):
            kw = {}
            if schema_ is not None:
                kw["schema"] = schema_
            if quote_schema is not None:
                kw["quote_schema"] = quote_schema
            t = Table(name, metadata, **kw)
            eq_(t.schema, exp_schema, "test %d, table schema" % i)
            eq_(
                t.schema.quote if t.schema is not None else None,
                exp_quote_schema,
                "test %d, table quote_schema" % i,
            )
            seq = Sequence(name, metadata=metadata, **kw)
            eq_(seq.schema, exp_schema, "test %d, seq schema" % i)
            eq_(
                seq.schema.quote if seq.schema is not None else None,
                exp_quote_schema,
                "test %d, seq quote_schema" % i,
            )

    def test_manual_dependencies(self):
        meta = MetaData()
        a = Table("a", meta, Column("foo", Integer))
        b = Table("b", meta, Column("foo", Integer))
        c = Table("c", meta, Column("foo", Integer))
        d = Table("d", meta, Column("foo", Integer))
        e = Table("e", meta, Column("foo", Integer))

        e.add_is_dependent_on(c)
        a.add_is_dependent_on(b)
        b.add_is_dependent_on(d)
        e.add_is_dependent_on(b)
        c.add_is_dependent_on(a)
        eq_(meta.sorted_tables, [d, b, a, c, e])

    def test_sort_by_parameter_order(self):
        meta = MetaData()
        a = Table("a", meta, Column("foo", Integer))
        b = Table("b", meta, Column("foo", Integer))
        c = Table("c", meta, Column("foo", Integer))
        d = Table("d", meta, Column("foo", Integer))
        e = Table("e", meta, Column("foo", Integer))

        e.add_is_dependent_on(c)
        a.add_is_dependent_on(b)
        eq_(meta.sorted_tables, [b, c, d, a, e])

    def test_fks_sort_by_parameter_order(self):
        meta = MetaData()
        a = Table("a", meta, Column("foo", Integer, ForeignKey("b.foo")))
        b = Table("b", meta, Column("foo", Integer))
        c = Table("c", meta, Column("foo", Integer))
        d = Table("d", meta, Column("foo", Integer))
        e = Table("e", meta, Column("foo", Integer, ForeignKey("c.foo")))

        eq_(meta.sorted_tables, [b, c, d, a, e])

    def test_cycles_fks_warning_one(self):
        meta = MetaData()
        a = Table("a", meta, Column("foo", Integer, ForeignKey("b.foo")))
        b = Table("b", meta, Column("foo", Integer, ForeignKey("d.foo")))
        c = Table("c", meta, Column("foo", Integer, ForeignKey("b.foo")))
        d = Table("d", meta, Column("foo", Integer, ForeignKey("c.foo")))
        e = Table("e", meta, Column("foo", Integer))

        with testing.expect_warnings(
            "Cannot correctly sort tables; there are unresolvable cycles "
            'between tables "b, c, d", which is usually caused by mutually '
            "dependent foreign key constraints.  "
            "Foreign key constraints involving these tables will not be "
            "considered"
        ):
            eq_(meta.sorted_tables, [b, c, d, e, a])

    def test_cycles_fks_warning_two(self):
        meta = MetaData()
        a = Table("a", meta, Column("foo", Integer, ForeignKey("b.foo")))
        b = Table("b", meta, Column("foo", Integer, ForeignKey("a.foo")))
        c = Table("c", meta, Column("foo", Integer, ForeignKey("e.foo")))
        d = Table("d", meta, Column("foo", Integer))
        e = Table("e", meta, Column("foo", Integer, ForeignKey("d.foo")))

        with testing.expect_warnings(
            "Cannot correctly sort tables; there are unresolvable cycles "
            'between tables "a, b", which is usually caused by mutually '
            "dependent foreign key constraints.  "
            "Foreign key constraints involving these tables will not be "
            "considered"
        ):
            eq_(meta.sorted_tables, [a, b, d, e, c])

    def test_cycles_fks_fks_delivered_separately(self):
        meta = MetaData()
        a = Table("a", meta, Column("foo", Integer, ForeignKey("b.foo")))
        b = Table("b", meta, Column("foo", Integer, ForeignKey("a.foo")))
        c = Table("c", meta, Column("foo", Integer, ForeignKey("e.foo")))
        d = Table("d", meta, Column("foo", Integer))
        e = Table("e", meta, Column("foo", Integer, ForeignKey("d.foo")))

        results = schema.sort_tables_and_constraints(
            sorted(meta.tables.values(), key=lambda t: t.key)
        )
        results[-1] = (None, set(results[-1][-1]))
        eq_(
            results,
            [
                (a, set()),
                (b, set()),
                (d, {fk.constraint for fk in d.foreign_keys}),
                (e, {fk.constraint for fk in e.foreign_keys}),
                (c, {fk.constraint for fk in c.foreign_keys}),
                (
                    None,
                    {fk.constraint for fk in a.foreign_keys}.union(
                        fk.constraint for fk in b.foreign_keys
                    ),
                ),
            ],
        )

    def test_cycles_fks_usealter(self):
        meta = MetaData()
        a = Table("a", meta, Column("foo", Integer, ForeignKey("b.foo")))
        b = Table(
            "b",
            meta,
            Column("foo", Integer, ForeignKey("d.foo", use_alter=True)),
        )
        c = Table("c", meta, Column("foo", Integer, ForeignKey("b.foo")))
        d = Table("d", meta, Column("foo", Integer, ForeignKey("c.foo")))
        e = Table("e", meta, Column("foo", Integer))

        eq_(meta.sorted_tables, [b, e, a, c, d])

    def test_nonexistent(self):
        assert_raises(
            tsa.exc.NoSuchTableError,
            Table,
            "fake_table",
            MetaData(),
            autoload_with=testing.db,
        )

    def test_assorted_repr(self):
        t1 = Table("foo", MetaData(), Column("x", Integer))
        i1 = Index("bar", t1.c.x)
        ck = schema.CheckConstraint("x > y", name="someconstraint")

        for const, exp in (
            (Sequence("my_seq"), "Sequence('my_seq')"),
            (Sequence("my_seq", start=5), "Sequence('my_seq', start=5)"),
            (Column("foo", Integer), "Column('foo', Integer(), table=None)"),
            (
                Column(
                    "foo",
                    Integer,
                    primary_key=True,
                    nullable=False,
                    onupdate=1,
                    default=42,
                    server_default="42",
                    comment="foo",
                ),
                "Column('foo', Integer(), table=None, primary_key=True, "
                f"nullable=False, onupdate={ColumnDefault(1)}, default="
                f"{ColumnDefault(42)}, server_default={DefaultClause('42')}, "
                "comment='foo')",
            ),
            (
                Column(
                    "foo",
                    Integer,
                    primary_key=True,
                    nullable=False,
                    onupdate=1,
                    insert_default=42,
                    server_default="42",
                    comment="foo",
                ),
                "Column('foo', Integer(), table=None, primary_key=True, "
                f"nullable=False, onupdate={ColumnDefault(1)}, default="
                f"{ColumnDefault(42)}, server_default={DefaultClause('42')}, "
                "comment='foo')",
            ),
            (
                Table("bar", MetaData(), Column("x", String)),
                "Table('bar', MetaData(), "
                "Column('x', String(), table=<bar>), schema=None)",
            ),
            (
                schema.DefaultGenerator(for_update=True),
                "DefaultGenerator(for_update=True)",
            ),
            (schema.Index("bar", "c"), "Index('bar', 'c')"),
            (i1, "Index('bar', Column('x', Integer(), table=<foo>))"),
            (schema.FetchedValue(), "FetchedValue()"),
            (
                ck,
                "CheckConstraint("
                "%s"
                ", name='someconstraint')" % repr(ck.sqltext),
            ),
            (
                ColumnDefault(("foo", "bar")),
                "ScalarElementColumnDefault(('foo', 'bar'))",
            ),
        ):
            eq_(repr(const), exp)

    @testing.variation("kind", ["engine", "conn", "something"])
    def test_metadata_bind(self, connection, kind):
        with expect_raises_message(
            exc.ArgumentError,
            "expected schema argument to be a string, got",
        ):
            if kind.engine:
                MetaData(connection.engine)
            elif kind.conn:
                MetaData(connection)
            else:
                MetaData(42)  # type: ignore


class ToMetaDataTest(fixtures.TestBase, AssertsCompiledSQL, ComparesTables):
    @testing.requires.check_constraints
    def test_copy(self):
        # TODO: modernize this test for 2.0

        from sqlalchemy.testing.schema import Table

        meta = MetaData()

        table = Table(
            "mytable",
            meta,
            Column("myid", Integer, Sequence("foo_id_seq"), primary_key=True),
            Column("name", String(40), nullable=True),
            Column(
                "foo",
                String(40),
                nullable=False,
                server_default="x",
                server_onupdate="q",
            ),
            Column(
                "bar", String(40), nullable=False, default="y", onupdate="z"
            ),
            Column(
                "description", String(30), CheckConstraint("description='hi'")
            ),
            UniqueConstraint("name"),
            test_needs_fk=True,
        )

        table2 = Table(
            "othertable",
            meta,
            Column("id", Integer, Sequence("foo_seq"), primary_key=True),
            Column("myid", Integer, ForeignKey("mytable.myid")),
            test_needs_fk=True,
        )

        table3 = Table(
            "has_comments",
            meta,
            Column("foo", Integer, comment="some column"),
            comment="table comment",
        )

        def test_to_metadata():
            meta2 = MetaData()
            table_c = table.to_metadata(meta2)
            table2_c = table2.to_metadata(meta2)
            table3_c = table3.to_metadata(meta2)
            return (table_c, table2_c, table3_c)

        def test_pickle():
            meta.bind = testing.db
            meta2 = pickle.loads(pickle.dumps(meta))
            pickle.loads(pickle.dumps(meta2))
            return (
                meta2.tables["mytable"],
                meta2.tables["othertable"],
                meta2.tables["has_comments"],
            )

        def test_pickle_via_reflect():
            # this is the most common use case, pickling the results of a
            # database reflection
            meta2 = MetaData()
            t1 = Table("mytable", meta2, autoload_with=testing.db)
            Table("othertable", meta2, autoload_with=testing.db)
            Table("has_comments", meta2, autoload_with=testing.db)
            meta3 = pickle.loads(pickle.dumps(meta2))
            assert meta3.tables["mytable"] is not t1

            return (
                meta3.tables["mytable"],
                meta3.tables["othertable"],
                meta3.tables["has_comments"],
            )

        meta.create_all(testing.db)
        try:
            for test, has_constraints, reflect in (
                (test_to_metadata, True, False),
                (test_pickle, True, False),
                (test_pickle_via_reflect, False, True),
            ):
                table_c, table2_c, table3_c = test()
                self.assert_tables_equal(table, table_c)
                self.assert_tables_equal(table2, table2_c)
                assert table is not table_c
                assert table.primary_key is not table_c.primary_key
                assert (
                    list(table2_c.c.myid.foreign_keys)[0].column
                    is table_c.c.myid
                )
                assert (
                    list(table2_c.c.myid.foreign_keys)[0].column
                    is not table.c.myid
                )
                assert "x" in str(table_c.c.foo.server_default.arg)
                if not reflect:
                    assert isinstance(table_c.c.myid.default, Sequence)
                    assert str(table_c.c.foo.server_onupdate.arg) == "q"
                    assert str(table_c.c.bar.default.arg) == "y"
                    assert (
                        getattr(
                            table_c.c.bar.onupdate.arg,
                            "arg",
                            table_c.c.bar.onupdate.arg,
                        )
                        == "z"
                    )
                    assert isinstance(table2_c.c.id.default, Sequence)

                # constraints don't get reflected for any dialect right
                # now

                if has_constraints:
                    for c in table_c.c.description.constraints:
                        if isinstance(c, CheckConstraint):
                            break
                    else:
                        assert False
                    assert str(c.sqltext) == "description='hi'"
                    for c in table_c.constraints:
                        if isinstance(c, UniqueConstraint):
                            break
                    else:
                        assert False
                    assert c.columns.contains_column(table_c.c.name)
                    assert not c.columns.contains_column(table.c.name)

                if testing.requires.comment_reflection.enabled:
                    eq_(table3_c.comment, "table comment")
                    eq_(table3_c.c.foo.comment, "some column")

        finally:
            meta.drop_all(testing.db)

    def test_col_key_fk_parent(self):
        # test #2643
        m1 = MetaData()
        a = Table("a", m1, Column("x", Integer))
        b = Table("b", m1, Column("x", Integer, ForeignKey("a.x"), key="y"))
        assert b.c.y.references(a.c.x)

        m2 = MetaData()
        b2 = b.to_metadata(m2)
        a2 = a.to_metadata(m2)
        assert b2.c.y.references(a2.c.x)

    def test_fk_w_no_colname(self):
        """test a ForeignKey that refers to table name only.  the column
        name is assumed to be the same col name on parent table.

        this is a little used feature from long ago that nonetheless is
        still in the code.

        The feature was found to be not working but is repaired for
        SQLAlchemy 2.0.

        """
        m1 = MetaData()
        a = Table("a", m1, Column("x", Integer))
        b = Table("b", m1, Column("x", Integer, ForeignKey("a")))
        assert b.c.x.references(a.c.x)

        m2 = MetaData()
        b2 = b.to_metadata(m2)
        a2 = a.to_metadata(m2)
        assert b2.c.x.references(a2.c.x)

    def test_fk_w_no_colname_name_missing(self):
        """test a ForeignKey that refers to table name only.  the column
        name is assumed to be the same col name on parent table.

        this is a little used feature from long ago that nonetheless is
        still in the code.

        """
        m1 = MetaData()
        a = Table("a", m1, Column("x", Integer))
        b = Table("b", m1, Column("y", Integer, ForeignKey("a")))

        with expect_raises_message(
            exc.NoReferencedColumnError,
            "Could not initialize target column for ForeignKey 'a' on "