pytests: import test_long_lived (test2)

parent 7e91e347
......@@ -41,13 +41,6 @@ def make_ssl_context():
return context
def ping_alive(sock):
buff, msgid = utils.get_msgbuff()
sock.sendall(buff)
answer = utils.receive_parse_answer(sock)
return answer.id == msgid
class Kresd(ContextDecorator):
def __init__(self, workdir, port, tls_port, ip=None, ip6=None):
if ip is None and ip6 is None:
......@@ -110,11 +103,11 @@ class Kresd(ContextDecorator):
def all_ports_alive(self):
alive = True
if self.ip:
alive &= ping_alive(self.ip_tcp_socket())
alive &= ping_alive(self.ip_tls_socket())
alive &= utils.ping_alive(self.ip_tcp_socket())
alive &= utils.ping_alive(self.ip_tls_socket())
if self.ip6:
alive &= ping_alive(self.ip6_tcp_socket())
alive &= ping_alive(self.ip6_tls_socket())
alive &= utils.ping_alive(self.ip6_tcp_socket())
alive &= utils.ping_alive(self.ip6_tls_socket())
return alive
def _wait_for_tcp_port(self, delay=0.1, max_attempts=20):
......@@ -127,7 +120,7 @@ class Kresd(ContextDecorator):
time.sleep(delay)
continue
else:
return ping_alive(sock)
return utils.ping_alive(sock)
finally:
sock.close()
raise RuntimeError("Kresd didn't start in time")
......
"""TCP Connection Management tests"""
import time
import utils
......@@ -34,3 +36,18 @@ def test_pipelining(kresd_sock):
msg_answer = utils.receive_parse_answer(kresd_sock)
assert msg_answer.id == msgid1
def test_long_lived(kresd_sock):
"""
Test establishes a TCP connection a sends several queries over it. They are sent
seqeuntially, each with a delay, which in total exceeds maximum timeout.
Expected: kresd must not close the connection
"""
utils.ping_alive(kresd_sock)
end_time = time.time() + utils.MAX_TIMEOUT
while time.time() < end_time:
time.sleep(3)
utils.ping_alive(kresd_sock)
......@@ -21,11 +21,6 @@ import pytest
import utils
# default net.tcp_in_idle is 10s, TCP_DEFER_ACCEPT 3s, some extra for
# Python handling / edge cases
MAX_TIMEOUT = 16
def send_incorrect_repeatedly(sock, buff, delay=1):
"""Utility function to keep sending the buffer until MAX_TIMEOUT is reached.
......@@ -34,7 +29,7 @@ def send_incorrect_repeatedly(sock, buff, delay=1):
If the connection remains open, test is failed.
"""
end_time = time.time() + MAX_TIMEOUT
end_time = time.time() + utils.MAX_TIMEOUT
with pytest.raises(BrokenPipeError, message="kresd didn't close connection"):
while time.time() < end_time:
......
......@@ -5,6 +5,11 @@ import dns
import dns.message
# default net.tcp_in_idle is 10s, TCP_DEFER_ACCEPT 3s, some extra for
# Python handling / edge cases
MAX_TIMEOUT = 16
def receive_answer(sock):
answer_total_len = 0
data = sock.recv(2)
......@@ -67,3 +72,10 @@ def get_garbage(length):
def get_prefixed_garbage(length):
data = get_garbage(length)
return prepare_buffer(data)
def ping_alive(sock):
buff, msgid = get_msgbuff()
sock.sendall(buff)
answer = receive_parse_answer(sock)
return answer.id == msgid
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment