(conn, test_frame1, request)
| 1208 | |
| 1209 | @pytest.mark.parametrize("conn", sqlalchemy_connectable) |
| 1210 | def test_to_sql_callable(conn, test_frame1, request): |
| 1211 | conn = request.getfixturevalue(conn) |
| 1212 | |
| 1213 | check = [] # used to double check function below is really being used |
| 1214 | |
| 1215 | def sample(pd_table, conn, keys, data_iter): |
| 1216 | check.append(1) |
| 1217 | data = [dict(zip(keys, row)) for row in data_iter] |
| 1218 | conn.execute(pd_table.table.insert(), data) |
| 1219 | |
| 1220 | with pandasSQL_builder(conn, need_transaction=True) as pandasSQL: |
| 1221 | pandasSQL.to_sql(test_frame1, "test_frame", method=sample) |
| 1222 | assert pandasSQL.has_table("test_frame") |
| 1223 | assert check == [1] |
| 1224 | assert count_rows(conn, "test_frame") == len(test_frame1) |
| 1225 | |
| 1226 | |
| 1227 | @pytest.mark.parametrize("conn", all_connectable_types) |
nothing calls this directly
no test coverage detected