From: Martin Willi <martin@strongswan.org>
Date: Fri, 4 Nov 2022 12:40:32 +0000 (+0100)
Subject: vici: Add simple Python transport layer tests, including timeout
X-Git-Tag: 5.9.9rc1~13^2~2
X-Git-Url: http://git.ipfire.org/?a=commitdiff_plain;h=aa0da01fde3dc609cb39c548d31aa9865b21418f;p=thirdparty%2Fstrongswan.git

vici: Add simple Python transport layer tests, including timeout
---

diff --git a/src/libcharon/plugins/vici/python/test/test_protocol.py b/src/libcharon/plugins/vici/python/test/test_protocol.py
index 1924884a11..1f71630686 100644
--- a/src/libcharon/plugins/vici/python/test/test_protocol.py
+++ b/src/libcharon/plugins/vici/python/test/test_protocol.py
@@ -1,6 +1,7 @@
 import pytest
+import socket
 
-from vici.protocol import Packet, Message, FiniteStream
+from vici.protocol import Packet, Message, FiniteStream, Transport
 from vici.exception import DeserializationException
 
 
@@ -142,3 +143,26 @@ class TestMessage(object):
         assert deserialized_message["key1"] == b"value1"
         assert deserialized_section["sub-section"]["key2"] == b"value2"
         assert deserialized_section["list1"] == [b"item1", b"item2"]
+
+
+class TestTransport(object):
+
+    def interconnect(self):
+        c, s = socket.socketpair(socket.AF_UNIX)
+        return Transport(c), Transport(s)
+
+    def test_sendrecv(self):
+        c, s = self.interconnect()
+        c.send(b"foo")
+        assert s.receive() == b"foo"
+        s.send(b"foobarbaz")
+        s.send(b"")
+        assert c.receive() == b"foobarbaz"
+        assert c.receive() == b""
+
+    def test_timeout(self):
+        c, s = self.interconnect()
+        c.send(b"foo")
+        assert s.receive(timeout=1) == b"foo"
+        with pytest.raises(socket.timeout):
+            s.receive(timeout=0.1)