(self, user_address_fixture)
| 1007 | return sess.scalars(collection.select()).all() |
| 1008 | |
| 1009 | def test_persistence(self, user_address_fixture): |
| 1010 | addresses = self.tables.addresses |
| 1011 | User, Address = user_address_fixture() |
| 1012 | |
| 1013 | sess = fixture_session() |
| 1014 | u1 = User(name="jack") |
| 1015 | a1 = Address(email_address="foo") |
| 1016 | sess.add_all([u1, a1]) |
| 1017 | sess.flush() |
| 1018 | |
| 1019 | eq_( |
| 1020 | sess.connection().scalar( |
| 1021 | select(func.count(cast(1, Integer))).where( |
| 1022 | addresses.c.user_id != None |
| 1023 | ) |
| 1024 | ), # noqa |
| 1025 | 0, |
| 1026 | ) |
| 1027 | u1 = sess.get(User, u1.id) |
| 1028 | u1.addresses.add(a1) |
| 1029 | sess.flush() |
| 1030 | |
| 1031 | eq_( |
| 1032 | sess.connection() |
| 1033 | .execute( |
| 1034 | select(addresses).where(addresses.c.user_id != None) # noqa |
| 1035 | ) |
| 1036 | .fetchall(), |
| 1037 | [(a1.id, u1.id, "foo")], |
| 1038 | ) |
| 1039 | |
| 1040 | u1.addresses.remove(a1) |
| 1041 | sess.flush() |
| 1042 | eq_( |
| 1043 | sess.connection().scalar( |
| 1044 | select(func.count(cast(1, Integer))).where( |
| 1045 | addresses.c.user_id != None |
| 1046 | ) |
| 1047 | ), # noqa |
| 1048 | 0, |
| 1049 | ) |
| 1050 | |
| 1051 | u1.addresses.add(a1) |
| 1052 | sess.flush() |
| 1053 | eq_( |
| 1054 | sess.connection() |
| 1055 | .execute( |
| 1056 | select(addresses).where(addresses.c.user_id != None) # noqa |
| 1057 | ) |
| 1058 | .fetchall(), |
| 1059 | [(a1.id, u1.id, "foo")], |
| 1060 | ) |
| 1061 | |
| 1062 | a2 = Address(email_address="bar") |
| 1063 | u1.addresses.remove(a1) |
| 1064 | u1.addresses.add(a2) |
| 1065 | sess.flush() |
| 1066 | eq_( |
nothing calls this directly
no test coverage detected