| 1091 | |
| 1092 | @testing.requires.subqueries |
| 1093 | def test_union(self, connection): |
| 1094 | t1, t2, t3 = self.tables("t1", "t2", "t3") |
| 1095 | s1, s2 = ( |
| 1096 | select(t1.c.col3.label("col3"), t1.c.col4.label("col4")).where( |
| 1097 | t1.c.col2.in_(["t1col2r1", "t1col2r2"]), |
| 1098 | ), |
| 1099 | select(t2.c.col3.label("col3"), t2.c.col4.label("col4")).where( |
| 1100 | t2.c.col2.in_(["t2col2r2", "t2col2r3"]), |
| 1101 | ), |
| 1102 | ) |
| 1103 | u = union(s1, s2) |
| 1104 | |
| 1105 | wanted = [ |
| 1106 | ("aaa", "aaa"), |
| 1107 | ("bbb", "bbb"), |
| 1108 | ("bbb", "ccc"), |
| 1109 | ("ccc", "aaa"), |
| 1110 | ] |
| 1111 | found1 = self._fetchall_sorted(connection.execute(u)) |
| 1112 | eq_(found1, wanted) |
| 1113 | |
| 1114 | found2 = self._fetchall_sorted( |
| 1115 | connection.execute(u.alias("bar").select()) |
| 1116 | ) |
| 1117 | eq_(found2, wanted) |
| 1118 | |
| 1119 | def test_union_ordered(self, connection): |
| 1120 | t1, t2, t3 = self.tables("t1", "t2", "t3") |