def test_no_table_needs_pl(self):
        Subset = self.classes.Subset

        selectable = select([column("x"), column("y"), column("z")]).alias()
        assert_raises_message(
            sa.exc.ArgumentError, "could not assemble any primary key columns", mapper, Subset, selectable
        )

 def test_bitwise_required_for_empty(self):
     assert_raises_message(
         exc.ArgumentError,
         "Can't use the blank value '' in a SET without setting "
         "retrieve_as_bitwise=True",
         mysql.SET, "a", "b", ''
     )

 def test_with_transaction(self):
     conn = self.engine.connect()
     trans = conn.begin()
     eq_(conn.execute(select([1])).scalar(), 1)
     assert not conn.closed
     self.engine.test_shutdown()
     _assert_invalidated(conn.execute, select([1]))
     assert not conn.closed
     assert conn.invalidated
     assert trans.is_active
     assert_raises_message(
         tsa.exc.StatementError,
         "Can't reconnect until invalid transaction is rolled back",
         conn.execute, select([1]))
     assert trans.is_active
     assert_raises_message(
         tsa.exc.InvalidRequestError,
         "Can't reconnect until invalid transaction is rolled back",
         trans.commit
     )
     assert trans.is_active
     trans.rollback()
     assert not trans.is_active
     assert conn.invalidated
     eq_(conn.execute(select([1])).scalar(), 1)
     assert not conn.invalidated

 def test_unknown_legacy_lock_mode(self):
     User = self.classes.User
     sess = Session()
     assert_raises_message(
         exc.ArgumentError, "Unknown with_lockmode argument: 'unknown_mode'",
         sess.query(User.id).with_lockmode, 'unknown_mode'
     )

    def test_auto_detach_on_gc_session(self):
        users, User = self.tables.users, self.classes.User

        mapper(User, users)

        sess = Session()

        u1 = User(name='u1')
        sess.add(u1)
        sess.commit()

        # can't add u1 to Session,
        # already belongs to u2
        s2 = Session()
        assert_raises_message(
            sa.exc.InvalidRequestError,
            r".*is already attached to session",
            s2.add, u1
        )

        # garbage collect sess
        del sess
        gc_collect()

        # s2 lets it in now despite u1 having
        # session_key
        s2.add(u1)
        assert u1 in s2

    def test_child_row_switch_two(self):
        P = self.classes.P

        Session = sessionmaker()

        # TODO: not sure this test is
        # testing exactly what its looking for

        sess1 = Session()
        sess1.add(P(id='P1', data='P version 1'))
        sess1.commit()
        sess1.close()

        p1 = sess1.query(P).first()

        sess2 = Session()
        p2 = sess2.query(P).first()

        sess1.delete(p1)
        sess1.commit()

        # this can be removed and it still passes
        sess1.add(P(id='P1', data='P version 2'))
        sess1.commit()

        p2.data = 'P overwritten by concurrent tx'
        if testing.db.dialect.supports_sane_rowcount:
            assert_raises_message(
                orm.exc.StaleDataError,
                r"UPDATE statement on table 'p' expected to update "
                r"1 row\(s\); 0 were matched.",
                sess2.commit
            )
        else:
            sess2.commit

    def test_lobs_without_convert_many_rows(self):
        engine = testing_engine(
            options=dict(auto_convert_lobs=False, arraysize=1))
        result = engine.execute(
            "select id, data, bindata from z_test order by id")
        results = result.fetchall()

        def go():
            eq_(
                [
                    dict(
                        id=row["id"],
                        data=row["data"].read(),
                        bindata=row["bindata"].read()
                    ) for row in results
                ],
                self.data)
        # this comes from cx_Oracle because these are raw
        # cx_Oracle.Variable objects
        if testing.requires.oracle5x.enabled:
            assert_raises_message(
                testing.db.dialect.dbapi.ProgrammingError,
                "LOB variable no longer valid after subsequent fetch",
                go
            )
        else:
            go()

 def _assert_unorderable_types(self, callable_):
     if util.py3k:
         assert_raises_message(
             TypeError, 'unorderable types', callable_)
     else:
         assert_raises_message(
             TypeError, 'cannot compare sets using cmp()', callable_)

    def test_limit_offset_for_update(self):
        metadata = self.metadata
        # oracle can't actually do the ROWNUM thing with FOR UPDATE
        # very well.

        t = Table(
            "t1",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("data", Integer),
        )
        metadata.create_all()

        t.insert().execute(
            {"id": 1, "data": 1},
            {"id": 2, "data": 7},
            {"id": 3, "data": 12},
            {"id": 4, "data": 15},
            {"id": 5, "data": 32},
        )

        # here, we can't use ORDER BY.
        eq_(
            t.select().with_for_update().limit(2).execute().fetchall(),
            [(1, 1), (2, 7)],
        )

        # here, its impossible.  But we'd prefer it to raise ORA-02014
        # instead of issuing a syntax error.
        assert_raises_message(
            exc.DatabaseError,
            "ORA-02014",
            t.select().with_for_update().limit(2).offset(3).execute,
        )

    def test_overlapping_attribute_error(self):
        place, Transition, place_input, Place, transition = (
            self.tables.place,
            self.classes.Transition,
            self.tables.place_input,
            self.classes.Place,
            self.tables.transition,
        )

        mapper(
            Place,
            place,
            properties={
                "transitions": relationship(
                    Transition, secondary=place_input, backref="places"
                )
            },
        )
        mapper(
            Transition,
            transition,
            properties={
                "places": relationship(
                    Place, secondary=place_input, backref="transitions"
                )
            },
        )
        assert_raises_message(
            sa.exc.ArgumentError,
            "property of that name exists",
            sa.orm.configure_mappers,
        )

    def test_skip_not_describable(self):
        @event.listens_for(self.metadata, "before_drop")
        def cleanup(*arg, **kw):
            with testing.db.connect() as conn:
                conn.execute("DROP TABLE IF EXISTS test_t1")
                conn.execute("DROP TABLE IF EXISTS test_t2")
                conn.execute("DROP VIEW IF EXISTS test_v")

        with testing.db.connect() as conn:
            conn.execute("CREATE TABLE test_t1 (id INTEGER)")
            conn.execute("CREATE TABLE test_t2 (id INTEGER)")
            conn.execute("CREATE VIEW test_v AS SELECT id FROM test_t1")
            conn.execute("DROP TABLE test_t1")

            m = MetaData()
            with expect_warnings(
                "Skipping .* Table or view named .?test_v.? could not be "
                "reflected: .* references invalid table"
            ):
                m.reflect(views=True, bind=conn)
            eq_(m.tables["test_t2"].name, "test_t2")

            assert_raises_message(
                exc.UnreflectableTableError,
                "references invalid table",
                Table,
                "test_v",
                MetaData(),
                autoload_with=conn,
            )

 def test_missing_bind_posn(self):
     assert_raises_message(
         exc.ArgumentError,
         "This text\(\) construct doesn't define a bound parameter named 'bar'",
         text(":foo").bindparams,
         bindparam('foo', value=5), bindparam('bar', value=7)
     )

 def test_missing_bind_kw(self):
     assert_raises_message(
         exc.ArgumentError,
         "This text\(\) construct doesn't define a bound parameter named 'bar'",
         text(":foo").bindparams,
         foo=5, bar=7
     )

 def test_with_transaction(self):
     conn = engine.connect()
     trans = conn.begin()
     eq_(conn.execute(select([1])).scalar(), 1)
     assert not conn.closed
     engine.test_shutdown()
     try:
         conn.execute(select([1]))
         assert False
     except tsa.exc.DBAPIError as e:
         if not e.connection_invalidated:
             raise
     assert not conn.closed
     assert conn.invalidated
     assert trans.is_active
     assert_raises_message(
         tsa.exc.StatementError,
         "Can't reconnect until invalid transaction is "\
             "rolled back",
         conn.execute, select([1])
     )
     assert trans.is_active
     try:
         trans.commit()
         assert False
     except tsa.exc.InvalidRequestError as e:
         assert str(e) \
             == "Can't reconnect until invalid transaction is "\
             "rolled back"
     assert trans.is_active
     trans.rollback()
     assert not trans.is_active
     assert conn.invalidated
     eq_(conn.execute(select([1])).scalar(), 1)
     assert not conn.invalidated

    def test_validator_multi_warning(self):
        users = self.tables.users

        class Foo(object):
            @validates("name")
            def validate_one(self, key, value):
                pass

            @validates("name")
            def validate_two(self, key, value):
                pass

        assert_raises_message(
            exc.InvalidRequestError,
            "A validation function for mapped attribute "
            "'name' on mapper Mapper|Foo|users already exists",
            mapper, Foo, users
        )

        class Bar(object):
            @validates("id")
            def validate_three(self, key, value):
                return value + 10

            @validates("id", "name")
            def validate_four(self, key, value):
                return value + "foo"

        assert_raises_message(
            exc.InvalidRequestError,
            "A validation function for mapped attribute "
            "'name' on mapper Mapper|Bar|users already exists",
            mapper, Bar, users
        )

    def test_versioncheck(self):
        """query.with_lockmode performs a 'version check' on an already loaded instance"""

        Foo = self.classes.Foo


        s1 = self._fixture()
        f1s1 = Foo(value='f1 value')
        s1.add(f1s1)
        s1.commit()

        s2 = create_session(autocommit=False)
        f1s2 = s2.query(Foo).get(f1s1.id)
        f1s2.value='f1 new value'
        s2.commit()

        # load, version is wrong
        assert_raises_message(
                sa.orm.exc.StaleDataError,
                r"Instance .* has version id '\d+' which does not "
                r"match database-loaded version id '\d+'",
                s1.query(Foo).with_lockmode('read').get, f1s1.id
            )

        # reload it - this expires the old version first
        s1.refresh(f1s1, lockmode='read')

        # now assert version OK
        s1.query(Foo).with_lockmode('read').get(f1s1.id)

        # assert brand new load is OK too
        s1.close()
        s1.query(Foo).with_lockmode('read').get(f1s1.id)

 def _assert_raises_ambig_join(
     self, fn, relname, secondary_arg, *arg, **kw
 ):
     if secondary_arg is not None:
         assert_raises_message(
             exc.AmbiguousForeignKeysError,
             "Could not determine join condition between "
             "parent/child tables on relationship %s - "
             "there are multiple foreign key paths linking the "
             "tables via secondary table '%s'.  "
             "Specify the 'foreign_keys' argument, providing a list "
             "of those columns which should be counted as "
             "containing a foreign key reference from the "
             "secondary table to each of the parent and child tables."
             % (relname, secondary_arg),
             fn,
             *arg,
             **kw
         )
     else:
         assert_raises_message(
             exc.AmbiguousForeignKeysError,
             "Could not determine join condition between "
             "parent/child tables on relationship %s - "
             "there are no foreign keys linking these tables.  "
             % (relname,),
             fn,
             *arg,
             **kw
         )

    def test_concurrent_mod_err_noexpire_on_commit(self):
        sess = self._fixture(expire_on_commit=False)

        f1 = self.classes.Foo(value='f1')
        sess.add(f1)
        sess.commit()

        # here, we're not expired overall, so no load occurs and we
        # stay without a version id, unless we've emitted
        # a SELECT for it within the flush.
        f1.value

        s2 = Session(expire_on_commit=False)
        f2 = s2.query(self.classes.Foo).first()
        f2.value = 'f2'
        s2.commit()

        f1.value = 'f3'

        assert_raises_message(
            orm.exc.StaleDataError,
            r"UPDATE statement on table 'version_table' expected to "
            r"update 1 row\(s\); 0 were matched.",
            sess.commit
        )

    def test_identity_conflict(self):
        users, User = self.tables.users, self.classes.User

        mapper(User, users)
        for s in (
            create_session(),
            create_session(weak_identity_map=False),
        ):
            users.delete().execute()
            u1 = User(name="ed")
            s.add(u1)
            s.flush()
            s.expunge(u1)
            u2 = s.query(User).first()
            s.expunge(u2)
            s.identity_map.add(sa.orm.attributes.instance_state(u1))

            assert_raises_message(
                sa.exc.InvalidRequestError,
                "Can't attach instance <User.*?>; another instance "
                "with key .*? is already "
                "present in this session.",
                s.identity_map.add,
                sa.orm.attributes.instance_state(u2)
            )

    def test_varchar_raise(self):
        for type_ in (
            String,
            VARCHAR,
            String(),
            VARCHAR(),
            NVARCHAR(),
            Unicode,
            Unicode(),
        ):
            type_ = sqltypes.to_instance(type_)
            assert_raises_message(
                exc.CompileError,
                "VARCHAR requires a length on dialect mysql",
                type_.compile,
                dialect=mysql.dialect()
            )

            t1 = Table('sometable', MetaData(),
                Column('somecolumn', type_)
            )
            assert_raises_message(
                exc.CompileError,
                r"\(in table 'sometable', column 'somecolumn'\)\: "
                r"(?:N)?VARCHAR requires a length on dialect mysql",
                schema.CreateTable(t1).compile,
                dialect=mysql.dialect()
            )