(
self,
read=False,
nowait=False,
of=None,
key_share=None,
assert_q_of=None,
assert_sel_of=None,
)
| 17 | cls.mapper_registry.map_imperatively(User, users) |
| 18 | |
| 19 | def _assert( |
| 20 | self, |
| 21 | read=False, |
| 22 | nowait=False, |
| 23 | of=None, |
| 24 | key_share=None, |
| 25 | assert_q_of=None, |
| 26 | assert_sel_of=None, |
| 27 | ): |
| 28 | User = self.classes.User |
| 29 | s = fixture_session() |
| 30 | q = s.query(User).with_for_update( |
| 31 | read=read, nowait=nowait, of=of, key_share=key_share |
| 32 | ) |
| 33 | sel = q._compile_state().statement |
| 34 | |
| 35 | assert q._for_update_arg.read is read |
| 36 | assert sel._for_update_arg.read is read |
| 37 | |
| 38 | assert q._for_update_arg.nowait is nowait |
| 39 | assert sel._for_update_arg.nowait is nowait |
| 40 | |
| 41 | assert q._for_update_arg.key_share is key_share |
| 42 | assert sel._for_update_arg.key_share is key_share |
| 43 | |
| 44 | eq_(q._for_update_arg.of, assert_q_of) |
| 45 | eq_(sel._for_update_arg.of, assert_sel_of) |
| 46 | |
| 47 | def test_key_share(self): |
| 48 | self._assert(key_share=True) |
no test coverage detected