| 1159 | |
| 1160 | @testing.requires.subqueries |
| 1161 | def test_union(self, connection): |
| 1162 | t1, t2, t3 = self.tables("t1", "t2", "t3") |
| 1163 | s1, s2 = ( |
| 1164 | select(t1.c.col3.label("col3"), t1.c.col4.label("col4")).where( |
| 1165 | t1.c.col2.in_(["t1col2r1", "t1col2r2"]), |
| 1166 | ), |
| 1167 | select(t2.c.col3.label("col3"), t2.c.col4.label("col4")).where( |
| 1168 | t2.c.col2.in_(["t2col2r2", "t2col2r3"]), |
| 1169 | ), |
| 1170 | ) |
| 1171 | u = union(s1, s2) |
| 1172 | |
| 1173 | wanted = [ |
| 1174 | ("aaa", "aaa"), |
| 1175 | ("bbb", "bbb"), |
| 1176 | ("bbb", "ccc"), |
| 1177 | ("ccc", "aaa"), |
| 1178 | ] |
| 1179 | found1 = self._fetchall_sorted(connection.execute(u)) |
| 1180 | eq_(found1, wanted) |
| 1181 | |
| 1182 | found2 = self._fetchall_sorted( |
| 1183 | connection.execute(u.alias("bar").select()) |
| 1184 | ) |
| 1185 | eq_(found2, wanted) |
| 1186 | |
| 1187 | def test_union_ordered(self, connection): |
| 1188 | t1, t2, t3 = self.tables("t1", "t2", "t3") |