• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号

Python testing.assert_raises函数代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Python中sqlalchemy.testing.assert_raises函数的典型用法代码示例。如果您正苦于以下问题:Python assert_raises函数的具体用法?Python assert_raises怎么用?Python assert_raises使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。



在下文中一共展示了assert_raises函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: test_can_use_session_in_outer_rollback_hook

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

        sess = Session()

        assertions = []
        @event.listens_for(sess, "after_soft_rollback")
        def do_something(session, previous_transaction):
            if session.is_active:
                assertions.append('name' not in u.__dict__)
                assertions.append(u.name == 'u1')

        u = User(name='u1', id=1)
        sess.add(u)
        sess.commit()

        u2 = User(name='u1', id=1)
        sess.add(u2)
        assert_raises(
            sa.orm.exc.FlushError,
            sess.commit
        )
        sess.rollback()
        eq_(assertions, [True, True])
开发者ID:MVReddy,项目名称:sqlalchemy,代码行数:25,代码来源:test_events.py


示例2: test_foreignkey_missing_insert

    def test_foreignkey_missing_insert(self):
        Table("t1", self.metadata, Column("id", Integer, primary_key=True))
        t2 = Table(
            "t2",
            self.metadata,
            Column("id", Integer, ForeignKey("t1.id"), primary_key=True),
        )
        self.metadata.create_all()

        # want to ensure that "null value in column "id" violates not-
        # null constraint" is raised (IntegrityError on psycoopg2, but
        # ProgrammingError on pg8000), and not "ProgrammingError:
        # (ProgrammingError) relationship "t2_id_seq" does not exist".
        # the latter corresponds to autoincrement behavior, which is not
        # the case here due to the foreign key.

        for eng in [
            engines.testing_engine(options={"implicit_returning": False}),
            engines.testing_engine(options={"implicit_returning": True}),
        ]:
            with expect_warnings(
                ".*has no Python-side or server-side default.*"
            ):
                assert_raises(
                    (exc.IntegrityError, exc.ProgrammingError),
                    eng.execute,
                    t2.insert(),
                )
开发者ID:BY-jk,项目名称:sqlalchemy,代码行数:28,代码来源:test_query.py


示例3: test_noninherited_warning

    def test_noninherited_warning(self):
        A, B, b_table, a_table, Dest, dest_table = (
            self.classes.A,
            self.classes.B,
            self.tables.b_table,
            self.tables.a_table,
            self.classes.Dest,
            self.tables.dest_table,
        )

        mapper(A, a_table, properties={"some_dest": relationship(Dest)})
        mapper(B, b_table, inherits=A, concrete=True)
        mapper(Dest, dest_table)
        b = B()
        dest = Dest()
        assert_raises(AttributeError, setattr, b, "some_dest", dest)
        clear_mappers()

        mapper(A, a_table, properties={"a_id": a_table.c.id})
        mapper(B, b_table, inherits=A, concrete=True)
        mapper(Dest, dest_table)
        b = B()
        assert_raises(AttributeError, setattr, b, "a_id", 3)
        clear_mappers()

        mapper(A, a_table, properties={"a_id": a_table.c.id})
        mapper(B, b_table, inherits=A, concrete=True)
        mapper(Dest, dest_table)
开发者ID:BY-jk,项目名称:sqlalchemy,代码行数:28,代码来源:test_concrete.py


示例4: test_notcallable

    def test_notcallable(self):
        class Point(object):
            pass
        table = self.point_map(Point)
        alias = aliased(Point)

        assert_raises(TypeError, alias)
开发者ID:aburan28,项目名称:sqlalchemy,代码行数:7,代码来源:test_utils.py


示例5: test_clauseelement

    def test_clauseelement(self):
        metadata = MetaData()
        table = Table('test_table', metadata,
            Column('foo', Integer))
        metadata.create_all(bind=testing.db)
        try:
            for elem in [
                table.select,
                lambda **kwargs: sa.func.current_timestamp(**kwargs).select(),
               # func.current_timestamp().select,
                lambda **kwargs:text("select * from test_table", **kwargs)
            ]:
                for bind in (
                    testing.db,
                    testing.db.connect()
                ):
                    try:
                        e = elem(bind=bind)
                        assert e.bind is bind
                        e.execute().close()
                    finally:
                        if isinstance(bind, engine.Connection):
                            bind.close()

                e = elem()
                assert e.bind is None
                assert_raises(
                    exc.UnboundExecutionError,
                    e.execute
                )
        finally:
            if isinstance(bind, engine.Connection):
                bind.close()
            metadata.drop_all(bind=testing.db)
开发者ID:23andMe,项目名称:sqlalchemy,代码行数:34,代码来源:test_bind.py


示例6: test_collection

    def test_collection(self):
        users, addresses, Address = (self.tables.users,
                                self.tables.addresses,
                                self.classes.Address)

        canary = Mock()
        class User(fixtures.ComparableEntity):
            @validates('addresses')
            def validate_address(self, key, ad):
                canary(key, ad)
                assert '@' in ad.email_address
                return ad

        mapper(User, users, properties={
                'addresses': relationship(Address)}
        )
        mapper(Address, addresses)
        sess = Session()
        u1 = User(name='edward')
        a0 = Address(email_address='noemail')
        assert_raises(AssertionError, u1.addresses.append, a0)
        a1 = Address(id=15, email_address='[email protected]')
        u1.addresses.append(a1)
        eq_(canary.mock_calls, [call('addresses', a0), call('addresses', a1)])
        sess.add(u1)
        sess.commit()

        eq_(
            sess.query(User).filter_by(name='edward').one(),
            User(name='edward', addresses=[Address(email_address='[email protected]')])
        )
开发者ID:robin900,项目名称:sqlalchemy,代码行数:31,代码来源:test_validators.py


示例7: test_versioncheck_for_update

    def test_versioncheck_for_update(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)
        s2.refresh(f1s2, lockmode='update')
        f1s2.value='f1 new value'

        assert_raises(
            exc.DBAPIError,
            s1.refresh, f1s1, lockmode='update_nowait'
        )
        s1.rollback()

        s2.commit()
        s1.refresh(f1s1, lockmode='update_nowait')
        assert f1s1.version_id == f1s2.version_id
开发者ID:Daniel-B-Smith,项目名称:sqlalchemy,代码行数:25,代码来源:test_versioning.py


示例8: test_dunder_sub

    def test_dunder_sub(self):
        _, _, twin1, twin2, _, _ = self._create_sets()

        # basic set math
        set1 = util.IdentitySet([1, 2, 3])
        set2 = util.IdentitySet([2, 3, 4])
        eq_(set1 - set2, util.IdentitySet([1]))
        eq_(set2 - set1, util.IdentitySet([4]))

        # empty sets
        empty = util.IdentitySet([])
        eq_(empty - empty, empty)

        # the same sets
        eq_(twin1 - twin2, empty)
        eq_(twin2 - twin1, empty)

        # totally different sets
        unique1 = util.IdentitySet([1])
        unique2 = util.IdentitySet([2])
        eq_(unique1 - unique2, util.IdentitySet([1]))
        eq_(unique2 - unique1, util.IdentitySet([2]))

        # not an IdentitySet
        def should_raise():
            not_an_identity_set = object()
            unique1 - not_an_identity_set
        assert_raises(TypeError, should_raise)
开发者ID:MVReddy,项目名称:sqlalchemy,代码行数:28,代码来源:test_utils.py


示例9: test_symmetric_difference

    def test_symmetric_difference(self):
        _, _, twin1, twin2, _, _ = self._create_sets()

        # basic set math
        set1 = util.IdentitySet([1, 2, 3])
        set2 = util.IdentitySet([2, 3, 4])
        eq_(set1.symmetric_difference(set2), util.IdentitySet([1, 4]))
        eq_(set2.symmetric_difference(set1), util.IdentitySet([1, 4]))

        # empty sets
        empty = util.IdentitySet([])
        eq_(empty.symmetric_difference(empty), empty)

        # the same sets
        eq_(twin1.symmetric_difference(twin2), empty)
        eq_(twin2.symmetric_difference(twin1), empty)

        # totally different sets
        unique1 = util.IdentitySet([1])
        unique2 = util.IdentitySet([2])
        eq_(unique1.symmetric_difference(unique2), util.IdentitySet([1, 2]))
        eq_(unique2.symmetric_difference(unique1), util.IdentitySet([1, 2]))

        # not an IdentitySet
        not_an_identity_set = object()
        assert_raises(
            TypeError, unique1.symmetric_difference, not_an_identity_set)
开发者ID:MVReddy,项目名称:sqlalchemy,代码行数:27,代码来源:test_utils.py


示例10: test_row_case_sensitive

    def test_row_case_sensitive(self):
        row = testing.db.execute(
            select([
                literal_column("1").label("case_insensitive"),
                literal_column("2").label("CaseSensitive")
            ])
        ).first()

        eq_(list(row.keys()), ["case_insensitive", "CaseSensitive"])

        in_("case_insensitive", row._keymap)
        in_("CaseSensitive", row._keymap)
        not_in_("casesensitive", row._keymap)

        eq_(row["case_insensitive"], 1)
        eq_(row["CaseSensitive"], 2)

        assert_raises(
            KeyError,
            lambda: row["Case_insensitive"]
        )
        assert_raises(
            KeyError,
            lambda: row["casesensitive"]
        )
开发者ID:mattastica,项目名称:sqlalchemy,代码行数:25,代码来源:test_resultset.py


示例11: test_dunder_or

    def test_dunder_or(self):
        super_, sub_, twin1, twin2, _, _ = self._create_sets()

        # basic set math
        eq_(sub_ | super_, super_)
        eq_(super_ | sub_, super_)

        # the same sets
        eq_(twin1 | twin2, twin1)
        eq_(twin2 | twin1, twin1)

        # empty sets
        empty = util.IdentitySet([])
        eq_(empty | empty, empty)

        # totally different sets
        unique1 = util.IdentitySet([1])
        unique2 = util.IdentitySet([2])
        eq_(unique1 | unique2, util.IdentitySet([1, 2]))

        # not an IdentitySet
        def should_raise():
            not_an_identity_set = object()
            return unique1 | not_an_identity_set
        assert_raises(TypeError, should_raise)
开发者ID:MVReddy,项目名称:sqlalchemy,代码行数:25,代码来源:test_utils.py


示例12: test_savepoint_lost_still_runs

    def test_savepoint_lost_still_runs(self):
        User = self.classes.User
        s = self.session(bind=self.bind)
        trans = s.begin_nested()
        s.connection()
        u1 = User(name='ed')
        s.add(u1)

        # kill off the transaction
        nested_trans = trans._connections[self.bind][1]
        nested_trans._do_commit()

        is_(s.transaction, trans)
        assert_raises(
            sa_exc.DBAPIError,
            s.rollback
        )

        assert u1 not in s.new

        is_(trans._state, _session.CLOSED)
        is_not_(s.transaction, trans)
        is_(s.transaction._state, _session.ACTIVE)

        is_(s.transaction.nested, False)

        is_(s.transaction._parent, None)
开发者ID:FluxIX,项目名称:sqlalchemy,代码行数:27,代码来源:test_transaction.py


示例13: test_no_selects

    def test_no_selects(self):
        Subset, common = self.classes.Subset, self.tables.common

        subset_select = select([common.c.id, common.c.data])
        assert_raises(
            sa.exc.InvalidRequestError, mapper, Subset, subset_select
        )
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:7,代码来源:test_selectable.py


示例14: test_accounting_commit_fails_delete

    def test_accounting_commit_fails_delete(self):
        User = self.classes.User
        sess = create_session(autocommit=True)

        fail = False

        def fail_fn(*arg, **kw):
            if fail:
                raise Exception("commit fails")

        event.listen(sess, "after_flush_postexec", fail_fn)
        u1 = User(name='ed')
        sess.add(u1)
        sess.flush()

        sess.delete(u1)
        fail = True
        assert_raises(
            Exception,
            sess.flush
        )
        fail = False

        assert u1 in sess
        assert u1 not in sess.deleted
        sess.delete(u1)
        sess.flush()
        assert u1 not in sess
        eq_(
            sess.query(User.name).order_by(User.name).all(),
            []
        )
开发者ID:LynYang,项目名称:sqlalchemy,代码行数:32,代码来源:test_transaction.py


示例15: test_dunder_xor

    def test_dunder_xor(self):
        _, _, twin1, twin2, _, _ = self._create_sets()

        # basic set math
        set1 = util.IdentitySet([1, 2, 3])
        set2 = util.IdentitySet([2, 3, 4])
        eq_(set1 ^ set2, util.IdentitySet([1, 4]))
        eq_(set2 ^ set1, util.IdentitySet([1, 4]))

        # empty sets
        empty = util.IdentitySet([])
        eq_(empty ^ empty, empty)

        # the same sets
        eq_(twin1 ^ twin2, empty)
        eq_(twin2 ^ twin1, empty)

        # totally different sets
        unique1 = util.IdentitySet([1])
        unique2 = util.IdentitySet([2])
        eq_(unique1 ^ unique2, util.IdentitySet([1, 2]))
        eq_(unique2 ^ unique1, util.IdentitySet([1, 2]))

        # not an IdentitySet
        def should_raise():
            not_an_identity_set = object()
            return unique1 ^ not_an_identity_set
        assert_raises(TypeError, should_raise)
开发者ID:MVReddy,项目名称:sqlalchemy,代码行数:28,代码来源:test_utils.py


示例16: test_rollback_recover

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

        mapper(User, users)

        session = sessionmaker()()

        u1, u2, u3 = User(name='u1'), User(name='u2'), User(name='u3')

        session.add_all([u1, u2, u3])

        session.commit()

        session.delete(u2)
        u4 = User(name='u2')
        session.add(u4)
        session.flush()

        u5 = User(name='u3')
        session.add(u5)
        assert_raises(orm_exc.FlushError, session.flush)

        assert u5 not in session
        assert u2 not in session.deleted

        session.rollback()
开发者ID:LynYang,项目名称:sqlalchemy,代码行数:26,代码来源:test_transaction.py


示例17: test_invalidate_conn_interrupt_nodisconnect_workaround

    def test_invalidate_conn_interrupt_nodisconnect_workaround(self):
        # test [ticket:3803] workaround for no disconnect on keyboard interrupt

        @event.listens_for(self.db, "handle_error")
        def cancel_disconnect(ctx):
            ctx.is_disconnect = False

        pool = self.db.pool

        conn = self.db.connect()
        self.dbapi.shutdown("interrupt_dont_break")

        def go():
            with conn.begin():
                conn.execute(select([1]))

        assert_raises(
            MockExitIsh,
            go
        )

        assert not conn.invalidated

        eq_(pool._invalidate_time, 0)  # pool not invalidated

        conn.execute(select([1]))
        assert not conn.invalidated
开发者ID:rlugojr,项目名称:sqlalchemy,代码行数:27,代码来源:test_reconnect.py


示例18: test_scalar

    def test_scalar(self):
        users = self.tables.users
        canary = Mock()

        class User(fixtures.ComparableEntity):
            @validates('name')
            def validate_name(self, key, name):
                canary(key, name)
                ne_(name, 'fred')
                return name + ' modified'

        mapper(User, users)
        sess = Session()
        u1 = User(name='ed')
        eq_(u1.name, 'ed modified')
        assert_raises(AssertionError, setattr, u1, "name", "fred")
        eq_(u1.name, 'ed modified')
        eq_(canary.mock_calls, [call('name', 'ed'), call('name', 'fred')])

        sess.add(u1)
        sess.commit()

        eq_(
            sess.query(User).filter_by(name='ed modified').one(),
            User(name='ed')
        )
开发者ID:robin900,项目名称:sqlalchemy,代码行数:26,代码来源:test_validators.py


示例19: test_conn_reusable

    def test_conn_reusable(self):
        conn = db.connect()

        conn.execute(select([1]))

        assert len(dbapi.connections) == 1

        dbapi.shutdown()

        assert_raises(
            tsa.exc.DBAPIError,
            conn.execute, select([1])
        )

        assert not conn.closed
        assert conn.invalidated

        # ensure all connections closed (pool was recycled)
        gc_collect()
        assert len(dbapi.connections) == 0

        # test reconnects
        conn.execute(select([1]))
        assert not conn.invalidated
        assert len(dbapi.connections) == 1
开发者ID:pshken,项目名称:sqlalchemy,代码行数:25,代码来源:test_reconnect.py


示例20: test_w_mapper_versioning

    def test_w_mapper_versioning(self):
        class SomeClass(Versioned, self.Base, ComparableEntity):
            __tablename__ = "sometable"

            id = Column(Integer, primary_key=True)
            name = Column(String(50))

        SomeClass.__mapper__.version_id_col = SomeClass.__table__.c.version

        self.create_tables()
        sess = self.session
        sc = SomeClass(name="sc1")
        sess.add(sc)
        sess.commit()

        s2 = Session(sess.bind)
        sc2 = s2.query(SomeClass).first()
        sc2.name = "sc1modified"

        sc.name = "sc1modified_again"
        sess.commit()

        eq_(sc.version, 2)

        assert_raises(orm_exc.StaleDataError, s2.flush)
开发者ID:23andMe,项目名称:sqlalchemy,代码行数:25,代码来源:test_versioning.py



注:本文中的sqlalchemy.testing.assert_raises函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Python testing.assert_raises_message函数代码示例发布时间:2022-05-27
下一篇:
Python testing.against函数代码示例发布时间:2022-05-27
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap