(self, statement_type)
| 299 | |
| 300 | @testing.combinations("update_mappings", "update_stmt") |
| 301 | def test_bulk_update(self, statement_type): |
| 302 | User = self.classes.User |
| 303 | |
| 304 | s = fixture_session(expire_on_commit=False) |
| 305 | objects = [User(name="u1"), User(name="u2"), User(name="u3")] |
| 306 | s.add_all(objects) |
| 307 | s.commit() |
| 308 | |
| 309 | s = fixture_session() |
| 310 | data = [ |
| 311 | {"id": 1, "name": "u1new"}, |
| 312 | {"id": 2, "name": "u2"}, |
| 313 | {"id": 3, "name": "u3new"}, |
| 314 | ] |
| 315 | |
| 316 | if statement_type == "update_mappings": |
| 317 | with self.sql_execution_asserter() as asserter: |
| 318 | s.bulk_update_mappings(User, data) |
| 319 | elif statement_type == "update_stmt": |
| 320 | with self.sql_execution_asserter() as asserter: |
| 321 | s.execute(update(User), data) |
| 322 | |
| 323 | asserter.assert_( |
| 324 | CompiledSQL( |
| 325 | "UPDATE users SET name=:name WHERE users.id = :users_id", |
| 326 | [ |
| 327 | {"users_id": 1, "name": "u1new"}, |
| 328 | {"users_id": 2, "name": "u2"}, |
| 329 | {"users_id": 3, "name": "u3new"}, |
| 330 | ], |
| 331 | ) |
| 332 | ) |
| 333 | |
| 334 | def test_bulk_insert(self): |
| 335 | (User,) = self.classes("User") |
nothing calls this directly
no test coverage detected