(self)
| 2217 | |
| 2218 | @testing.emits_warning("The garbage collector is trying to clean up") |
| 2219 | def test_execute_events(self): |
| 2220 | stmts = [] |
| 2221 | cursor_stmts = [] |
| 2222 | |
| 2223 | def execute( |
| 2224 | conn, clauseelement, multiparams, params, execution_options |
| 2225 | ): |
| 2226 | stmts.append((str(clauseelement), params, multiparams)) |
| 2227 | |
| 2228 | def cursor_execute( |
| 2229 | conn, cursor, statement, parameters, context, executemany |
| 2230 | ): |
| 2231 | cursor_stmts.append((str(statement), parameters, None)) |
| 2232 | |
| 2233 | # TODO: this test is kind of a mess |
| 2234 | |
| 2235 | for engine in [ |
| 2236 | engines.testing_engine(), |
| 2237 | engines.testing_engine().connect(), |
| 2238 | ]: |
| 2239 | event.listen(engine, "before_execute", execute) |
| 2240 | event.listen(engine, "before_cursor_execute", cursor_execute) |
| 2241 | m = MetaData() |
| 2242 | t1 = Table( |
| 2243 | "t1", |
| 2244 | m, |
| 2245 | Column("c1", Integer, primary_key=True), |
| 2246 | Column( |
| 2247 | "c2", |
| 2248 | String(50), |
| 2249 | default=func.lower("Foo"), |
| 2250 | primary_key=True, |
| 2251 | ), |
| 2252 | implicit_returning=False, |
| 2253 | ) |
| 2254 | |
| 2255 | if isinstance(engine, Connection): |
| 2256 | ctx = None |
| 2257 | conn = engine |
| 2258 | else: |
| 2259 | ctx = conn = engine.connect() |
| 2260 | |
| 2261 | trans = conn.begin() |
| 2262 | try: |
| 2263 | m.create_all(conn, checkfirst=False) |
| 2264 | try: |
| 2265 | conn.execute(t1.insert(), dict(c1=5, c2="some data")) |
| 2266 | conn.execute(t1.insert(), dict(c1=6)) |
| 2267 | eq_( |
| 2268 | conn.execute(text("select * from t1")).fetchall(), |
| 2269 | [(5, "some data"), (6, "foo")], |
| 2270 | ) |
| 2271 | finally: |
| 2272 | m.drop_all(conn) |
| 2273 | trans.commit() |
| 2274 | finally: |
| 2275 | if ctx: |
| 2276 | ctx.close() |
nothing calls this directly
no test coverage detected