[tests] added tests for socket connections

Author: MrClock8163

.github/workflows/run-tests.yml

@@ -8,6 +8,7 @@ env:
   GEOCOMPY_TEST_PORT_CLIENT: /tmp/geocompy_pty_client
   GEOCOMPY_TEST_PORT_SERVER: /tmp/geocompy_pty_server
   GEOCOMPY_TEST_PORT_FAULTY: /tmp/geocompy_pty_faulty
+  GEOCOMPY_TEST_TCPPORT_SERVER: 12345
 
 jobs:
   testing:
@@ -30,6 +31,7 @@ jobs:
         run: |
           socat pty,raw,echo=0,link=$GEOCOMPY_TEST_PORT_CLIENT pty,raw,echo=0,link=$GEOCOMPY_TEST_PORT_SERVER &
           python tests/com_echo_server.py &
+          python tests/tcp_echo_server.py &
 
       - name: Test with pytest
         run: |

tests/tcp_echo_server.py

@@ -0,0 +1,43 @@
+from os import environ
+
+import socket
+
+
+portname = environ.get("GEOCOMPY_TEST_TCPPORT_SERVER", "")
+if portname == "":
+    raise ValueError(
+        "Echo server serial port name must be set in "
+        "'GEOCOMPY_TEST_TCPPORT_SERVER' environment variable"
+    )
+
+
+def echo_server(port: int) -> None:
+    with socket.socket(
+        socket.AF_INET,
+        socket.SOCK_STREAM,
+        socket.IPPROTO_TCP
+    ) as soc:
+        soc.bind(("127.0.0.1", port))
+        while True:
+            soc.listen(1)
+            print(f"Listening on {port}")
+            server, address = soc.accept()
+            print(f"Connected to {address}")
+
+            with server:
+                while True:
+                    try:
+                        data = server.recv(1024)
+                        if data == b"":
+                            break
+
+                        data = data.strip(b"\r\n")
+                        if data != b"":
+                            server.send(data + b"\r\n")
+
+                    except Exception:
+                        break
+
+
+if __name__ == "__main__":
+    echo_server(int(portname))

tests/test_communication.py

@@ -1,15 +1,28 @@
 from os import environ
+import socket
+
 import pytest
 from serial import Serial
 from geocompy.communication import (
     get_dummy_logger,
     open_serial,
+    open_socket,
     SerialConnection,
+    SocketConnection,
     crc16_bitwise,
     crc16_bytewise
 )
 
 
+@pytest.fixture
+def sock() -> socket.socket:
+    return socket.socket(
+        socket.AF_INET,
+        socket.SOCK_STREAM,
+        socket.IPPROTO_TCP
+    )
+
+
 portname = environ.get("GEOCOMPY_TEST_PORT_CLIENT", "")
 if portname == "":  # pragma: no coverage
     raise ValueError(
@@ -24,6 +37,13 @@
         "'GEOCOMPY_TEST_PORT_FAULTY' environment variable"
     )
 
+tcpport = environ.get("GEOCOMPY_TEST_TCPPORT_SERVER", "")
+if faultyportname == "":  # pragma: no coverage
+    raise ValueError(
+        "Echo server tcp port must be set in "
+        "'GEOCOMPY_TEST_TCPPORT_SERVER' environment variable"
+    )
+
 
 class TestDummyLogger:
     def test_get_dummy_logger(self) -> None:
@@ -32,6 +52,80 @@ def test_get_dummy_logger(self) -> None:
         assert len(log.handlers) == 1
 
 
+class TestSocketConnection:
+    def test_init(self, sock: socket.socket) -> None:
+        sock.connect(("127.0.0.1", int(tcpport)))
+        with SocketConnection(sock) as client:
+            assert client.is_open()
+
+        client.close()
+        assert not client.is_open()
+
+    def test_open_socket(self) -> None:
+        with open_socket("127.0.0.1", int(tcpport), "tcp") as soc:
+            assert soc.is_open()
+
+        with pytest.raises(Exception):
+            open_socket("127.0.0.1", int(tcpport), "rfcomm", timeout=1)
+
+        with pytest.raises(ValueError):
+            open_socket(
+                "127.0.0.1",
+                int(tcpport),
+                "mistake",  # type: ignore[arg-type]
+                timeout=1
+            )
+
+    def test_messaging(self) -> None:
+        with open_socket(
+            "127.0.0.1",
+            int(tcpport),
+            "tcp"
+        ) as soc:
+            soc.is_open()
+            request = "Test"
+            assert soc.exchange(request) == request
+
+            soc.send("ascii")
+            assert soc.receive() == "ascii"
+
+            assert soc.exchange_binary(b"00\r\n") == b"00"
+
+            soc.reset()
+
+        with pytest.raises(ConnectionError):
+            soc.send("closed")
+
+        with pytest.raises(ConnectionError):
+            soc.receive()
+
+        with open_socket(
+            "127.0.0.1",
+            int(tcpport),
+            "tcp",
+            sync_after_timeout=True
+        ) as soc:
+            soc.send("msg1")
+            soc.send("msg2")
+            soc.send("msg3")
+            soc._timeout_counter = 3
+            assert soc.exchange("recovered") == "recovered"
+
+            with soc.timeout_override(1):
+                with pytest.raises(TimeoutError):
+                    soc.receive()
+
+                assert soc._timeout_counter == 1
+
+                with pytest.raises(TimeoutError):
+                    soc.receive()
+
+                assert soc._timeout_counter == 2
+
+        with pytest.raises(ConnectionError):
+            soc.receive()
+
+
 class TestSerialConnection:
     def test_init(self) -> None:
         port = Serial(portname)