(self, connection, filters)
| 161 | @testing.requires.insert_executemany_returning |
| 162 | @testing.variation("filters", ["unique", "sliced", "plain"]) |
| 163 | def test_splice_horizontally(self, connection, filters): |
| 164 | users = self.tables.users |
| 165 | addresses = self.tables.addresses |
| 166 | |
| 167 | if filters.unique: |
| 168 | r1 = connection.execute( |
| 169 | users.insert().returning(users.c.user_name), |
| 170 | [ |
| 171 | dict(user_id=1, user_name="john"), |
| 172 | dict(user_id=2, user_name="john"), |
| 173 | ], |
| 174 | ) |
| 175 | r2 = connection.execute( |
| 176 | addresses.insert().returning( |
| 177 | addresses.c.address, |
| 178 | ), |
| 179 | [ |
| 180 | dict(address_id=1, user_id=1, address="foo@bar.com"), |
| 181 | dict(address_id=2, user_id=2, address="foo@bar.com"), |
| 182 | ], |
| 183 | ) |
| 184 | else: |
| 185 | r1 = connection.execute( |
| 186 | users.insert().returning(users.c.user_name, users.c.user_id), |
| 187 | [ |
| 188 | dict(user_id=1, user_name="john"), |
| 189 | dict(user_id=2, user_name="jack"), |
| 190 | ], |
| 191 | ) |
| 192 | r2 = connection.execute( |
| 193 | addresses.insert().returning( |
| 194 | addresses.c.address_id, |
| 195 | addresses.c.address, |
| 196 | addresses.c.user_id, |
| 197 | ), |
| 198 | [ |
| 199 | dict(address_id=1, user_id=1, address="foo@bar.com"), |
| 200 | dict(address_id=2, user_id=2, address="bar@bat.com"), |
| 201 | ], |
| 202 | ) |
| 203 | |
| 204 | if filters.sliced: |
| 205 | r1 = r1.columns(users.c.user_name) |
| 206 | r2 = r2.columns(addresses.c.address, addresses.c.user_id) |
| 207 | elif filters.unique: |
| 208 | r1 = r1.unique() |
| 209 | r2 = r2.unique() |
| 210 | |
| 211 | rows = r1.splice_horizontally(r2).all() |
| 212 | |
| 213 | if filters.sliced: |
| 214 | eq_( |
| 215 | rows, |
| 216 | [ |
| 217 | ("john", "foo@bar.com", 1), |
| 218 | ("jack", "bar@bat.com", 2), |
| 219 | ], |
| 220 | ) |
nothing calls this directly
no test coverage detected