]> git.ipfire.org Git - thirdparty/knot-resolver.git/commitdiff
pytests: import test_oob
authorTomas Krizek <tomas.krizek@nic.cz>
Mon, 19 Nov 2018 17:35:15 +0000 (18:35 +0100)
committerTomas Krizek <tomas.krizek@nic.cz>
Tue, 4 Dec 2018 16:13:42 +0000 (17:13 +0100)
tests/pytests/test_conn_mgmt.py

index 28730f3b0bcef74d2d60e2802798685920b7c4bc..b9762bee6b66a85edfcbe2bd5c091a3f5fbe5535 100644 (file)
@@ -1,5 +1,6 @@
 """TCP Connection Management tests"""
 
+import socket
 import struct
 import time
 
@@ -120,6 +121,28 @@ def test_ignore_jumbo_message(kresd_sock):
     assert answer.id == msgid2
 
 
+@pytest.mark.parametrize('sock_func_name', [
+    'ip_tcp_socket',
+    'ip6_tcp_socket',
+])
+def test_oob(kresd, sock_func_name):
+    """TCP out-of-band (urgent) data must not crash resolver."""
+    make_sock = getattr(kresd, sock_func_name)
+    sock = make_sock()
+    msg_buff, msgid = utils.get_msgbuff()
+    sock.sendall(msg_buff, socket.MSG_OOB)
+
+    try:
+        msg_answer = utils.receive_parse_answer(sock)
+        assert msg_answer.id == msgid
+    except ConnectionError:
+        pass  # TODO kresd responds with TCP RST, this should be fixed
+
+    # check kresd is alive
+    sock2 = make_sock()
+    utils.ping_alive(sock2)
+
+
 def flood_buffer(msgcount):
     flood_buff = bytes()
     msgbuff, _ = utils.get_msgbuff()