Test various join operations
(self)
| 240 | session.close() |
| 241 | |
| 242 | def test_joins(self): |
| 243 | """Test various join operations""" |
| 244 | session = self.Session() |
| 245 | try: |
| 246 | # Create test data |
| 247 | user1 = User(username="author1", email="author1@example.com") |
| 248 | user2 = User(username="author2", email="author2@example.com") |
| 249 | |
| 250 | post1 = Post(title="Post 1", author=user1, published=True) |
| 251 | post2 = Post(title="Post 2", author=user1, published=False) |
| 252 | post3 = Post(title="Post 3", author=user2, published=True) |
| 253 | |
| 254 | comment1 = Comment(post=post1, user=user2, content="Nice post!") |
| 255 | comment2 = Comment(post=post1, user=user1, content="Thanks!") |
| 256 | |
| 257 | session.add_all([user1, user2, post1, post2, post3, comment1, comment2]) |
| 258 | session.commit() |
| 259 | |
| 260 | # Test INNER JOIN |
| 261 | result = session.query(User, Post).join(Post).filter(Post.published == True).all() |
| 262 | assert len(result) == 2 |
| 263 | |
| 264 | # Test LEFT JOIN |
| 265 | result = session.query(User).outerjoin(Post).filter( |
| 266 | or_(Post.id == None, Post.published == True) |
| 267 | ).distinct().all() |
| 268 | # We should have all users since we're doing LEFT JOIN |
| 269 | # But we filter to only users with no posts OR published posts |
| 270 | # So we expect user1 and user2 |
| 271 | assert len(result) >= 2 # At least the two authors |
| 272 | |
| 273 | # Test multiple joins |
| 274 | result = session.query(Comment).join(Post).join(User).filter( |
| 275 | User.username == "author1" |
| 276 | ).all() |
| 277 | assert len(result) == 2 |
| 278 | |
| 279 | finally: |
| 280 | session.close() |
| 281 | |
| 282 | def test_transactions(self): |
| 283 | """Test transaction handling""" |