| 2439 | |
| 2440 | @testing.combinations(True, False, argnames="close") |
| 2441 | def test_close_parameter(self, testing_engine, close): |
| 2442 | eng = testing_engine( |
| 2443 | options=dict( |
| 2444 | pool_size=1, |
| 2445 | max_overflow=0, |
| 2446 | poolclass=( |
| 2447 | QueuePool |
| 2448 | if not testing.db.dialect.is_async |
| 2449 | else AsyncAdaptedQueuePool |
| 2450 | ), |
| 2451 | ) |
| 2452 | ) |
| 2453 | |
| 2454 | conn = eng.connect() |
| 2455 | dbapi_conn_one = conn.connection.dbapi_connection |
| 2456 | conn.close() |
| 2457 | |
| 2458 | eng_copy = copy.copy(eng) |
| 2459 | eng_copy.dispose(close=close) |
| 2460 | |
| 2461 | copy_conn = eng_copy.connect() |
| 2462 | dbapi_conn_two = copy_conn.connection.dbapi_connection |
| 2463 | |
| 2464 | is_not(dbapi_conn_one, dbapi_conn_two) |
| 2465 | |
| 2466 | conn = eng.connect() |
| 2467 | if close: |
| 2468 | is_not(dbapi_conn_one, conn.connection.dbapi_connection) |
| 2469 | else: |
| 2470 | is_(dbapi_conn_one, conn.connection.dbapi_connection) |
| 2471 | |
| 2472 | conn.close() |
| 2473 | copy_conn.close() |
| 2474 | |
| 2475 | def test_retval_flag(self): |
| 2476 | canary = [] |