(self)
| 2273 | |
| 2274 | @testing.emits_warning("The garbage collector is trying to clean up") |
| 2275 | def test_execute_events(self): |
| 2276 | stmts = [] |
| 2277 | cursor_stmts = [] |
| 2278 | |
| 2279 | def execute( |
| 2280 | conn, clauseelement, multiparams, params, execution_options |
| 2281 | ): |
| 2282 | stmts.append((str(clauseelement), params, multiparams)) |
| 2283 | |
| 2284 | def cursor_execute( |
| 2285 | conn, cursor, statement, parameters, context, executemany |
| 2286 | ): |
| 2287 | cursor_stmts.append((str(statement), parameters, None)) |
| 2288 | |
| 2289 | # TODO: this test is kind of a mess |
| 2290 | |
| 2291 | for engine in [ |
| 2292 | engines.testing_engine(), |
| 2293 | engines.testing_engine().connect(), |
| 2294 | ]: |
| 2295 | event.listen(engine, "before_execute", execute) |
| 2296 | event.listen(engine, "before_cursor_execute", cursor_execute) |
| 2297 | m = MetaData() |
| 2298 | t1 = Table( |
| 2299 | "t1", |
| 2300 | m, |
| 2301 | Column("c1", Integer, primary_key=True), |
| 2302 | Column( |
| 2303 | "c2", |
| 2304 | String(50), |
| 2305 | default=func.lower("Foo"), |
| 2306 | primary_key=True, |
| 2307 | ), |
| 2308 | implicit_returning=False, |
| 2309 | ) |
| 2310 | |
| 2311 | if isinstance(engine, Connection): |
| 2312 | ctx = None |
| 2313 | conn = engine |
| 2314 | else: |
| 2315 | ctx = conn = engine.connect() |
| 2316 | |
| 2317 | trans = conn.begin() |
| 2318 | try: |
| 2319 | m.create_all(conn, checkfirst=False) |
| 2320 | try: |
| 2321 | conn.execute(t1.insert(), dict(c1=5, c2="some data")) |
| 2322 | conn.execute(t1.insert(), dict(c1=6)) |
| 2323 | eq_( |
| 2324 | conn.execute(text("select * from t1")).fetchall(), |
| 2325 | [(5, "some data"), (6, "foo")], |
| 2326 | ) |
| 2327 | finally: |
| 2328 | m.drop_all(conn) |
| 2329 | trans.commit() |
| 2330 | finally: |
| 2331 | if ctx: |
| 2332 | ctx.close() |
nothing calls this directly
no test coverage detected