本文整理汇总了Python中sqlalchemy.testing.expect_warnings函数的典型用法代码示例。如果您正苦于以下问题:Python expect_warnings函数的具体用法?Python expect_warnings怎么用?Python expect_warnings使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了expect_warnings函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_unsupported_casts
def test_unsupported_casts(self):
t = sql.table('t', sql.column('col'))
m = mysql
specs = [
(m.MSBit, "t.col"),
(FLOAT, "t.col"),
(Float, "t.col"),
(m.MSFloat, "t.col"),
(m.MSDouble, "t.col"),
(m.MSReal, "t.col"),
(m.MSYear, "t.col"),
(m.MSYear(2), "t.col"),
(Boolean, "t.col"),
(BOOLEAN, "t.col"),
(m.MSEnum, "t.col"),
(m.MSEnum("1", "2"), "t.col"),
(m.MSSet, "t.col"),
(m.MSSet("1", "2"), "t.col"),
]
for type_, expected in specs:
with expect_warnings(
"Datatype .* does not support CAST on MySQL;"
):
self.assert_compile(cast(t.c.col, type_), expected)
开发者ID:robin900,项目名称:sqlalchemy,代码行数:31,代码来源:test_compiler.py
示例2: test_report_primary_error_when_rollback_fails
def test_report_primary_error_when_rollback_fails(self):
User, users = self.classes.User, self.tables.users
mapper(User, users)
session = Session(testing.db)
with expect_warnings(".*during handling of a previous exception.*"):
session.begin_nested()
savepoint = session.\
connection()._Connection__transaction._savepoint
# force the savepoint to disappear
session.connection().dialect.do_release_savepoint(
session.connection(), savepoint
)
# now do a broken flush
session.add_all([User(id=1), User(id=1)])
assert_raises_message(
sa_exc.DBAPIError,
"ROLLBACK TO SAVEPOINT ",
session.flush
)
开发者ID:LynYang,项目名称:sqlalchemy,代码行数:25,代码来源:test_transaction.py
示例3: 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
示例4: test_reconnect_on_reentrant_plus_closewresult
def test_reconnect_on_reentrant_plus_closewresult(self):
conn = self.db.connect(close_with_result=True)
self.dbapi.shutdown("rollback")
# raises error
with expect_warnings(
"An exception has occurred during handling .*"
"something broke on execute but we didn't lose the connection",
py2konly=True,
):
assert_raises_message(
tsa.exc.DBAPIError,
"Lost the DB connection on rollback",
conn.execute,
select([1]),
)
assert conn.closed
assert conn.invalidated
assert_raises_message(
tsa.exc.StatementError,
"This Connection is closed",
conn.execute,
select([1]),
)
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:27,代码来源:test_reconnect.py
示例5: 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
示例6: test_unsupported_cast_literal_bind
def test_unsupported_cast_literal_bind(self):
expr = cast(column("foo", Integer) + 5, Float)
with expect_warnings("Datatype FLOAT does not support CAST on MySQL;"):
self.assert_compile(expr, "(foo + 5)", literal_binds=True)
dialect = mysql.MySQLDialect()
dialect.server_version_info = (3, 9, 8)
with expect_warnings("Current MySQL version does not support CAST"):
eq_(
str(
expr.compile(
dialect=dialect, compile_kwargs={"literal_binds": True}
)
),
"(foo + 5)",
)
开发者ID:monetate,项目名称:sqlalchemy,代码行数:17,代码来源:test_compiler.py
示例7: test_cast_grouped_expression_pre_4
def test_cast_grouped_expression_pre_4(self):
dialect = mysql.dialect()
dialect.server_version_info = (3, 2, 3)
with expect_warnings("Current MySQL version does not support CAST;"):
self.assert_compile(
cast(sql.column('x') + sql.column('y'), Integer),
"(x + y)",
dialect=dialect
)
开发者ID:robin900,项目名称:sqlalchemy,代码行数:9,代码来源:test_compiler.py
示例8: test_execution_options_ignored_mid_transaction
def test_execution_options_ignored_mid_transaction(self):
bind = mock.Mock()
conn = mock.Mock(engine=bind)
bind.contextual_connect = mock.Mock(return_value=conn)
sess = Session(bind=bind)
sess.execute("select 1")
with expect_warnings(
"Connection is already established for the "
"given bind; execution_options ignored"):
sess.connection(execution_options={'isolation_level': 'FOO'})
开发者ID:LynYang,项目名称:sqlalchemy,代码行数:10,代码来源:test_transaction.py
示例9: test_not_supported
def test_not_supported(self):
dialect, connection = self._fixture(None)
with expect_warnings("Could not fetch transaction isolation level"):
assert_raises_message(
NotImplementedError,
"Can't fetch isolation",
dialect.get_isolation_level,
connection,
)
开发者ID:monetate,项目名称:sqlalchemy,代码行数:10,代码来源:test_engine.py
示例10: test_no_cast_pre_4
def test_no_cast_pre_4(self):
self.assert_compile(
cast(Column("foo", Integer), String), "CAST(foo AS CHAR)"
)
dialect = mysql.dialect()
dialect.server_version_info = (3, 2, 3)
with expect_warnings("Current MySQL version does not support CAST;"):
self.assert_compile(
cast(Column("foo", Integer), String), "foo", dialect=dialect
)
开发者ID:monetate,项目名称:sqlalchemy,代码行数:10,代码来源:test_compiler.py
示例11: _test_warning
def _test_warning(self, stmt, offending_clause, expected):
with expect_warnings(
"Can't resolve label reference %r;" % offending_clause
):
self.assert_compile(stmt, expected)
assert_raises_message(
exc.SAWarning,
"Can't resolve label reference %r; converting to text"
% offending_clause,
stmt.compile,
)
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:11,代码来源:test_text.py
示例12: test_anticipate_no_pk_lower_case_table
def test_anticipate_no_pk_lower_case_table(self):
t = table(
"t",
Column("id", Integer, primary_key=True, autoincrement=False),
Column("notpk", String(10), nullable=True),
)
with expect_warnings(
"Column 't.id' is marked as a member.*" "may not store NULL.$"
):
self.assert_compile(
t.insert(), "INSERT INTO t () VALUES ()", params={}
)
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:12,代码来源:test_insert.py
示例13: test_warn_on_unannotated_matched_column
def test_warn_on_unannotated_matched_column(self):
User = self.classes.User
compiler = evaluator.EvaluatorCompiler(User)
with expect_warnings(
r"Evaluating non-mapped column expression 'othername' "
"onto ORM instances; this is a deprecated use case."):
meth = compiler.process(User.name == Column('othername', String))
u1 = User(id=5)
meth(u1)
开发者ID:cpcloud,项目名称:sqlalchemy,代码行数:12,代码来源:test_evaluator.py
示例14: _test
def _test(self, fn, arg, offending_clause, expected):
with expect_warnings("Textual "):
stmt = fn(arg)
self.assert_compile(stmt, expected)
assert_raises_message(
exc.SAWarning,
r"Textual (?:SQL|column|SQL FROM) expression %(stmt)r should be "
r"explicitly declared (?:with|as) text\(%(stmt)r\)"
% {"stmt": util.ellipses_string(offending_clause)},
fn,
arg,
)
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:13,代码来源:test_text.py
示例15: test_anticipate_no_pk_non_composite_pk
def test_anticipate_no_pk_non_composite_pk(self):
t = Table(
"t",
MetaData(),
Column("x", Integer, primary_key=True, autoincrement=False),
Column("q", Integer),
)
with expect_warnings(
"Column 't.x' is marked as a member.*" "may not store NULL.$"
):
self.assert_compile(
t.insert(), "INSERT INTO t (q) VALUES (:q)", params={"q": 5}
)
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:13,代码来源:test_insert.py
示例16: test_ensure_is_disconnect_gets_connection
def test_ensure_is_disconnect_gets_connection(self):
def is_disconnect(e, conn, cursor):
# connection is still present
assert conn.connection is not None
# the error usually occurs on connection.cursor(),
# though MySQLdb we get a non-working cursor.
# assert cursor is None
self.engine.dialect.is_disconnect = is_disconnect
conn = self.engine.connect()
self.engine.test_shutdown()
with expect_warnings(
"An exception has occurred during handling .*", py2konly=True
):
assert_raises(tsa.exc.DBAPIError, conn.execute, select([1]))
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:15,代码来源:test_reconnect.py
示例17: test_anticipate_no_pk_composite_pk
def test_anticipate_no_pk_composite_pk(self):
t = Table(
"t",
MetaData(),
Column("x", Integer, primary_key=True),
Column("y", Integer, primary_key=True),
)
with expect_warnings(
"Column 't.y' is marked as a member.*"
"Note that as of SQLAlchemy 1.1,"
):
self.assert_compile(
t.insert(), "INSERT INTO t (x) VALUES (:x)", params={"x": 5}
)
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:15,代码来源:test_insert.py
示例18: test_anticipate_no_pk_composite_pk_prefetch
def test_anticipate_no_pk_composite_pk_prefetch(self):
t = Table(
't', MetaData(), Column('x', Integer, primary_key=True),
Column('y', Integer, primary_key=True)
)
d = postgresql.dialect()
d.implicit_returning = False
with expect_warnings(
"Column 't.y' is marked as a member.*"
"Note that as of SQLAlchemy 1.1,"
):
self.assert_compile(
t.insert(),
"INSERT INTO t (x) VALUES (%(x)s)",
params={'x': 5},
dialect=d
)
开发者ID:robin900,项目名称:sqlalchemy,代码行数:17,代码来源:test_insert.py
示例19: test_no_show_variables
def test_no_show_variables(self):
from sqlalchemy.testing import mock
engine = engines.testing_engine()
def my_execute(self, statement, *args, **kw):
if statement.startswith("SHOW VARIABLES"):
statement = "SELECT 1 FROM DUAL WHERE 1=0"
return real_exec(self, statement, *args, **kw)
real_exec = engine._connection_cls._execute_text
with mock.patch.object(
engine._connection_cls, "_execute_text", my_execute):
with expect_warnings(
"Could not retrieve SQL_MODE; please ensure the "
"MySQL user has permissions to SHOW VARIABLES"
):
engine.connect()
开发者ID:cpcloud,项目名称:sqlalchemy,代码行数:17,代码来源:test_dialect.py
示例20: test_anticipate_no_pk_non_composite_pk_implicit_returning
def test_anticipate_no_pk_non_composite_pk_implicit_returning(self):
t = Table(
"t",
MetaData(),
Column("x", Integer, primary_key=True, autoincrement=False),
Column("q", Integer),
)
d = postgresql.dialect()
d.implicit_returning = True
with expect_warnings(
"Column 't.x' is marked as a member.*" "may not store NULL.$"
):
self.assert_compile(
t.insert(),
"INSERT INTO t (q) VALUES (%(q)s)",
params={"q": 5},
dialect=d,
)
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:18,代码来源:test_insert.py
注:本文中的sqlalchemy.testing.expect_warnings函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论