本文整理汇总了Python中sqlalchemy.testing.assert_raises_message函数的典型用法代码示例。如果您正苦于以下问题:Python assert_raises_message函数的具体用法?Python assert_raises_message怎么用?Python assert_raises_message使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了assert_raises_message函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_no_table_needs_pl
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
)
开发者ID:FluxIX,项目名称:sqlalchemy,代码行数:7,代码来源:test_selectable.py
示例2: test_bitwise_required_for_empty
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", ''
)
开发者ID:gencer,项目名称:sqlalchemy,代码行数:7,代码来源:test_types.py
示例3: test_with_transaction
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
开发者ID:Attsun1031,项目名称:sqlalchemy,代码行数:26,代码来源:test_reconnect.py
示例4: test_unknown_legacy_lock_mode
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'
)
开发者ID:FluxIX,项目名称:sqlalchemy,代码行数:7,代码来源:test_lockmode.py
示例5: test_auto_detach_on_gc_session
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
开发者ID:DeepakAlevoor,项目名称:sqlalchemy,代码行数:28,代码来源:test_session.py
示例6: test_child_row_switch_two
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
开发者ID:Daniel-B-Smith,项目名称:sqlalchemy,代码行数:35,代码来源:test_versioning.py
示例7: test_lobs_without_convert_many_rows
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()
开发者ID:eoghanmurray,项目名称:sqlalchemy,代码行数:27,代码来源:test_types.py
示例8: _assert_unorderable_types
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_)
开发者ID:MVReddy,项目名称:sqlalchemy,代码行数:7,代码来源:test_utils.py
示例9: test_limit_offset_for_update
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,
)
开发者ID:BY-jk,项目名称:sqlalchemy,代码行数:34,代码来源:test_dialect.py
示例10: test_overlapping_attribute_error
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,
)
开发者ID:BY-jk,项目名称:sqlalchemy,代码行数:32,代码来源:test_manytomany.py
示例11: test_skip_not_describable
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,
)
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:30,代码来源:test_reflection.py
示例12: test_missing_bind_posn
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)
)
开发者ID:chundi,项目名称:sqlalchemy,代码行数:7,代码来源:test_text.py
示例13: test_missing_bind_kw
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
)
开发者ID:chundi,项目名称:sqlalchemy,代码行数:7,代码来源:test_text.py
示例14: test_with_transaction
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
开发者ID:MVReddy,项目名称:sqlalchemy,代码行数:35,代码来源:test_reconnect.py
示例15: test_validator_multi_warning
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
)
开发者ID:robin900,项目名称:sqlalchemy,代码行数:34,代码来源:test_validators.py
示例16: test_versioncheck
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)
开发者ID:Daniel-B-Smith,项目名称:sqlalchemy,代码行数:33,代码来源:test_versioning.py
示例17: _assert_raises_ambig_join
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
)
开发者ID:BY-jk,项目名称:sqlalchemy,代码行数:30,代码来源:test_rel_fn.py
示例18: test_concurrent_mod_err_noexpire_on_commit
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
)
开发者ID:Daniel-B-Smith,项目名称:sqlalchemy,代码行数:25,代码来源:test_versioning.py
示例19: test_identity_conflict
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)
)
开发者ID:CyberCollins,项目名称:sqlalchemy,代码行数:25,代码来源:test_session.py
示例20: test_varchar_raise
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()
)
开发者ID:23andMe,项目名称:sqlalchemy,代码行数:28,代码来源:test_compiler.py
注:本文中的sqlalchemy.testing.assert_raises_message函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论