(self)
| 197 | udp_socket.close() |
| 198 | |
| 199 | def __enter__(self): |
| 200 | (self.left, self.right) = socket.socketpair() |
| 201 | # We're making the sockets now so they can be sent to by the |
| 202 | # caller immediately (i.e. no race with the listener starting |
| 203 | # in the thread). |
| 204 | self.open_sockets(self.port, ConnectionType.UDP, ConnectionType.TCP) |
| 205 | self.open_sockets(self.dot_port, ConnectionType.DOQ, ConnectionType.DOT) |
| 206 | self.open_sockets(self.doh_port, ConnectionType.DOH3, ConnectionType.DOH) |
| 207 | for proto, sock in self.sockets.items(): |
| 208 | self.addresses[proto] = sock.getsockname() |
| 209 | if self.use_thread: |
| 210 | self.start() |
| 211 | return self |
| 212 | |
| 213 | def __exit__(self, ex_ty, ex_va, ex_tr): |
| 214 | if self.left: |
nothing calls this directly
no test coverage detected