(self)
| 1617 | session.commit() |
| 1618 | |
| 1619 | def test_data_loaded(self): |
| 1620 | User, Task = self.classes("User", "Task") |
| 1621 | session = fixture_session() |
| 1622 | |
| 1623 | all_tasks = session.query(Task).all() # noqa: F841 |
| 1624 | all_users = session.query(User).all() # noqa: F841 |
| 1625 | |
| 1626 | # expire all objects |
| 1627 | session.expire_all() |
| 1628 | |
| 1629 | # now load w/ the special paths. User.phone needs to be |
| 1630 | # undeferred |
| 1631 | tasks = ( |
| 1632 | session.query(Task) |
| 1633 | .options( |
| 1634 | joinedload(Task.managed_by).load_only(User.name), |
| 1635 | joinedload(Task.created_by).load_only(User.name, User.phone), |
| 1636 | ) |
| 1637 | .all() |
| 1638 | ) |
| 1639 | |
| 1640 | session.close() |
| 1641 | for task in tasks: |
| 1642 | if task.name == "t1": |
| 1643 | # for User u2, created_by path includes User.phone |
| 1644 | eq_(task.created_by.phone, "p2") |
| 1645 | |
| 1646 | # for User u3, managed_by path does not |
| 1647 | assert "phone" not in task.managed_by.__dict__ |
| 1648 | elif task.name == "t2": |
| 1649 | # User u1 was loaded by both created_by and managed_by |
| 1650 | # path, so 'phone' should be unconditionally populated |
| 1651 | is_(task.created_by, task.managed_by) |
| 1652 | eq_(task.created_by.phone, "p1") |
| 1653 | eq_(task.managed_by.phone, "p1") |
| 1654 | else: |
| 1655 | assert False |
| 1656 | |
| 1657 | |
| 1658 | class SelfReferentialMultiPathTest(testing.fixtures.DeclarativeMappedTest): |
nothing calls this directly
no test coverage detected