def test_tcp_garbage(named_port):
with create_socket("10.53.0.7", named_port) as sock:
msg = create_msg("a.example.", "A")
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
- (response, rtime) = dns.query.receive_tcp(sock, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
+ dns.query.receive_tcp(sock, timeout())
wire = msg.to_wire()
assert len(wire) > 0
with pytest.raises(EOFError):
try:
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
- (response, rtime) = dns.query.receive_tcp(sock, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
+ dns.query.receive_tcp(sock, timeout())
except ConnectionError as e:
raise EOFError from e
def test_tcp_garbage_response(named_port):
with create_socket("10.53.0.7", named_port) as sock:
msg = create_msg("a.example.", "A")
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
- (response, rtime) = dns.query.receive_tcp(sock, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
+ dns.query.receive_tcp(sock, timeout())
wire = msg.to_wire()
assert len(wire) > 0
# the connection to be terminated
rmsg = dns.message.make_response(msg)
- (sbytes, stime) = dns.query.send_tcp(sock, rmsg, timeout())
+ dns.query.send_tcp(sock, rmsg, timeout())
with pytest.raises(EOFError):
try:
- (response, rtime) = dns.query.receive_tcp(sock, timeout())
+ dns.query.receive_tcp(sock, timeout())
except ConnectionError as e:
raise EOFError from e
def test_close_wait(named_port):
with create_socket("10.53.0.7", named_port) as sock:
msg = create_msg("a.example.", "A")
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
- (response, rtime) = dns.query.receive_tcp(sock, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
+ dns.query.receive_tcp(sock, timeout())
msg = dns.message.make_query("a.example.", "A", use_edns=0, payload=1232)
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
# Shutdown the socket, but ignore the other side closing the socket
# first because we sent DNS message with EDNS0
# available for the query below and it will time out.
with create_socket("10.53.0.7", named_port) as sock:
msg = create_msg("a.example.", "A")
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
- (response, rtime) = dns.query.receive_tcp(sock, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
+ dns.query.receive_tcp(sock, timeout())
# GL #4273
msg.additional.append(
dns.rrset.from_text(dns.name.root, 0, 1, "URI", "0 0 " + "b" * 65503)
)
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
- (response, rtime) = dns.query.receive_tcp(sock, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
+ dns.query.receive_tcp(sock, timeout())
# Now check that the server is alive and well
msg = create_msg("a.example.", "A")
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
- (response, rtime) = dns.query.receive_tcp(sock, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
+ dns.query.receive_tcp(sock, timeout())
with pytest.raises(EOFError):
try:
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
- (response, rtime) = dns.query.receive_tcp(sock, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
+ dns.query.receive_tcp(sock, timeout())
except ConnectionError as e:
raise EOFError from e
time.sleep(1)
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
- (response, rtime) = dns.query.receive_tcp(sock, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
+ dns.query.receive_tcp(sock, timeout())
time.sleep(2)
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
- (response, rtime) = dns.query.receive_tcp(sock, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
+ dns.query.receive_tcp(sock, timeout())
time.sleep(6)
with pytest.raises(EOFError):
try:
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
- (response, rtime) = dns.query.receive_tcp(sock, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
+ dns.query.receive_tcp(sock, timeout())
except ConnectionError as e:
raise EOFError from e
time.sleep(1)
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
- (response, rtime) = dns.query.receive_tcp(sock, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
+ dns.query.receive_tcp(sock, timeout())
time.sleep(2)
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
- (response, rtime) = dns.query.receive_tcp(sock, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
+ dns.query.receive_tcp(sock, timeout())
time.sleep(6)
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
- (response, rtime) = dns.query.receive_tcp(sock, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
+ dns.query.receive_tcp(sock, timeout())
def test_pipelining_timeout(named_port):
time.sleep(1)
# Send and receive 25 DNS queries
- for n in range(25):
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
- for n in range(25):
- (response, rtime) = dns.query.receive_tcp(sock, timeout())
+ for _ in range(25):
+ dns.query.send_tcp(sock, msg, timeout())
+ for _ in range(25):
+ dns.query.receive_tcp(sock, timeout())
time.sleep(3)
# Send and receive 25 DNS queries
- for n in range(25):
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
- for n in range(25):
- (response, rtime) = dns.query.receive_tcp(sock, timeout())
+ for _ in range(25):
+ dns.query.send_tcp(sock, msg, timeout())
+ for _ in range(25):
+ dns.query.receive_tcp(sock, timeout())
time.sleep(6)
with pytest.raises(EOFError):
try:
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
- (response, rtime) = dns.query.receive_tcp(sock, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
+ dns.query.receive_tcp(sock, timeout())
except ConnectionError as e:
raise EOFError from e
name = dns.name.from_text("example.")
msg = create_msg("example.", "AXFR")
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
# Receive the initial DNS message with SOA
- (response, rtime) = dns.query.receive_tcp(
- sock, timeout(), one_rr_per_rrset=True
- )
+ (response, _) = dns.query.receive_tcp(sock, timeout(), one_rr_per_rrset=True)
soa = response.get_rrset(
dns.message.ANSWER, name, dns.rdataclass.IN, dns.rdatatype.SOA
)
# Pull DNS message from wire until the second SOA is received
while True:
- (response, rtime) = dns.query.receive_tcp(
+ (response, _) = dns.query.receive_tcp(
sock, timeout(), one_rr_per_rrset=True
)
soa = response.get_rrset(
# Send and receive single large RDATA over TCP
msg = create_msg("large.example.", "TXT")
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
- (response, rtime) = dns.query.receive_tcp(sock, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
+ dns.query.receive_tcp(sock, timeout())
# Send and receive 28 large (~32k) DNS queries that should
# fill the default maximum 208k TCP send buffer
- for n in range(28):
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
+ for _ in range(28):
+ dns.query.send_tcp(sock, msg, timeout())
# configure idle interval is 5 seconds, sleep 6 to make sure we are
# above the interval
with pytest.raises(EOFError):
try:
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
- (response, rtime) = dns.query.receive_tcp(sock, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
+ dns.query.receive_tcp(sock, timeout())
except ConnectionError as e:
raise EOFError from e
name = dns.name.from_text("example.")
msg = create_msg("example.", "AXFR")
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
# Receive the initial DNS message with SOA
- (response, rtime) = dns.query.receive_tcp(
- sock, timeout(), one_rr_per_rrset=True
- )
+ (response, _) = dns.query.receive_tcp(sock, timeout(), one_rr_per_rrset=True)
soa = response.get_rrset(
dns.message.ANSWER, name, dns.rdataclass.IN, dns.rdatatype.SOA
)
with pytest.raises(ConnectionResetError):
# Process queued TCP messages
while True:
- (response, rtime) = dns.query.receive_tcp(
+ (response, _) = dns.query.receive_tcp(
sock, timeout(), one_rr_per_rrset=True
)
soa = response.get_rrset(
name = dns.name.from_text("example.")
msg = create_msg("example.", "AXFR")
- (sbytes, stime) = dns.query.send_tcp(sock, msg, timeout())
+ dns.query.send_tcp(sock, msg, timeout())
# Receive the initial DNS message with SOA
- (response, rtime) = dns.query.receive_tcp(
- sock, timeout(), one_rr_per_rrset=True
- )
+ (response, _) = dns.query.receive_tcp(sock, timeout(), one_rr_per_rrset=True)
soa = response.get_rrset(
dns.message.ANSWER, name, dns.rdataclass.IN, dns.rdatatype.SOA
)
with pytest.raises(EOFError):
while True:
time.sleep(1)
- (response, rtime) = dns.query.receive_tcp(
+ (response, _) = dns.query.receive_tcp(
sock, timeout(), one_rr_per_rrset=True
)
soa = response.get_rrset(