Provides a mocking engine based on the current testing.db. This is normally used to test DDL generation flow as emitted by an Engine. It should not be used in other cases, as assert_compile() and assert_sql_execution() are much better choices with fewer moving parts.
(dialect_name=None)
| 393 | |
| 394 | |
| 395 | def mock_engine(dialect_name=None): |
| 396 | """Provides a mocking engine based on the current testing.db. |
| 397 | |
| 398 | This is normally used to test DDL generation flow as emitted |
| 399 | by an Engine. |
| 400 | |
| 401 | It should not be used in other cases, as assert_compile() and |
| 402 | assert_sql_execution() are much better choices with fewer |
| 403 | moving parts. |
| 404 | |
| 405 | """ |
| 406 | |
| 407 | from sqlalchemy import create_mock_engine |
| 408 | |
| 409 | if not dialect_name: |
| 410 | dialect_name = config.db.name |
| 411 | |
| 412 | buffer = [] |
| 413 | |
| 414 | def executor(sql, *a, **kw): |
| 415 | buffer.append(sql) |
| 416 | |
| 417 | def assert_sql(stmts): |
| 418 | recv = [re.sub(r"[\n\t]", "", str(s)) for s in buffer] |
| 419 | assert recv == stmts, recv |
| 420 | |
| 421 | def print_sql(): |
| 422 | d = engine.dialect |
| 423 | return "\n".join(str(s.compile(dialect=d)) for s in engine.mock) |
| 424 | |
| 425 | engine = create_mock_engine(dialect_name + "://", executor) |
| 426 | assert not hasattr(engine, "mock") |
| 427 | engine.mock = buffer |
| 428 | engine.assert_sql = assert_sql |
| 429 | engine.print_sql = print_sql |
| 430 | return engine |
| 431 | |
| 432 | |
| 433 | class DBAPIProxyCursor: |
nothing calls this directly
no test coverage detected