(self)
| 168 | @unittest.skipUnless(have_ssl, "No SSL support") |
| 169 | @tests.util.retry_on_timeout |
| 170 | def testQueryTLSWithSocket(self): |
| 171 | for address in query_addresses: |
| 172 | with socket.socket( |
| 173 | dns.inet.af_for_address(address), socket.SOCK_STREAM |
| 174 | ) as base_s: |
| 175 | ll = dns.inet.low_level_address_tuple((address, 853)) |
| 176 | base_s.settimeout(2) |
| 177 | base_s.connect(ll) |
| 178 | ctx = ssl.create_default_context() |
| 179 | ctx.minimum_version = ssl.TLSVersion.TLSv1_2 |
| 180 | with ctx.wrap_socket( |
| 181 | base_s, server_hostname="dns.google" |
| 182 | ) as s: # lgtm[py/insecure-protocol] |
| 183 | s.setblocking(0) |
| 184 | qname = dns.name.from_text("dns.google.") |
| 185 | q = dns.message.make_query(qname, dns.rdatatype.A) |
| 186 | response = dns.query.tls(q, None, sock=s, timeout=2) |
| 187 | rrs = response.get_rrset( |
| 188 | response.answer, qname, dns.rdataclass.IN, dns.rdatatype.A |
| 189 | ) |
| 190 | self.assertTrue(rrs is not None) |
| 191 | seen = set([rdata.address for rdata in rrs]) |
| 192 | self.assertTrue("8.8.8.8" in seen) |
| 193 | self.assertTrue("8.8.4.4" in seen) |
| 194 | |
| 195 | @unittest.skipUnless(have_ssl, "No SSL support") |
| 196 | @tests.util.retry_on_timeout |
nothing calls this directly
no test coverage detected