Test transaction handling
(self)
| 280 | session.close() |
| 281 | |
| 282 | def test_transactions(self): |
| 283 | """Test transaction handling""" |
| 284 | session = self.Session() |
| 285 | try: |
| 286 | # Test rollback |
| 287 | user = User(username="rollback_test", email="rollback@example.com") |
| 288 | session.add(user) |
| 289 | session.flush() # Get ID without committing |
| 290 | user_id = user.id |
| 291 | |
| 292 | session.rollback() |
| 293 | |
| 294 | # User should not exist after rollback |
| 295 | result = session.query(User).filter_by(id=user_id).first() |
| 296 | assert result is None |
| 297 | |
| 298 | # Test commit |
| 299 | user = User(username="commit_test", email="commit@example.com") |
| 300 | session.add(user) |
| 301 | session.commit() |
| 302 | |
| 303 | # User should exist after commit |
| 304 | result = session.query(User).filter_by(username="commit_test").first() |
| 305 | assert result is not None |
| 306 | |
| 307 | finally: |
| 308 | session.close() |
| 309 | |
| 310 | def test_bulk_operations(self): |
| 311 | """Test bulk insert and update operations""" |