From 1ce9e1e5520ddd0d5ae11e5848bf43f2b515b591 Mon Sep 17 00:00:00 2001 From: Eddy Pedroni Date: Tue, 24 May 2022 22:11:24 +0200 Subject: Added mock oscilloscope server instead of mock socket --- lab_control/sds1000xe.py | 4 +-- lab_control/test/mock_sds1000xe_server.py | 43 +++++++++++++++++++++++++++++++ lab_control/test/mocksocket.py | 11 -------- lab_control/test/sds1000xe_test.py | 32 ++++++++--------------- 4 files changed, 56 insertions(+), 34 deletions(-) create mode 100644 lab_control/test/mock_sds1000xe_server.py delete mode 100644 lab_control/test/mocksocket.py (limited to 'lab_control') diff --git a/lab_control/sds1000xe.py b/lab_control/sds1000xe.py index 0ad996f..fff4f24 100644 --- a/lab_control/sds1000xe.py +++ b/lab_control/sds1000xe.py @@ -1,6 +1,6 @@ import socket class SDS1000XE: - def __init__(self, ip, port=5025, socket=socket.socket()): - self.socket = socket + def __init__(self, ip, port=5025): + self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.socket.connect((ip, port)) diff --git a/lab_control/test/mock_sds1000xe_server.py b/lab_control/test/mock_sds1000xe_server.py new file mode 100644 index 0000000..7da90c9 --- /dev/null +++ b/lab_control/test/mock_sds1000xe_server.py @@ -0,0 +1,43 @@ +import socket +import threading + +class MockSDS1000XEServer: + IP = "0.0.0.0" + PORT = 5025 + + MOTD = "Hello".encode() + + def __init__(self): + self.serverSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.serverSocket.bind((MockSDS1000XEServer.IP, MockSDS1000XEServer.PORT)) + self.serverSocket.listen(1) + print("server socket is bound") + + self.main = threading.Thread(target=self._mainLoop) + self.main.start() + + def _mainLoop(self) -> None: + clientSocket, _ = self.serverSocket.accept() + clientSocket.settimeout(0.1) + clientSocket.send(MockSDS1000XEServer.MOTD) + + try: + while True: + request = clientSocket.recv(4096).decode() + if request != "": + print(f"** Received request: {request}") + response = self._handleRequest(request) + clientSocket.send(response.encode()) + except Exception as e: + print("Exception caught in main server loop") + print(e) + + def _handleRequest(self, request: str) -> str: + return "NOP" + + def stop(self): + print("Closing server socket") + self.serverSocket.close() + print("joining main") + self.main.join() + print("stopped") diff --git a/lab_control/test/mocksocket.py b/lab_control/test/mocksocket.py deleted file mode 100644 index 7352846..0000000 --- a/lab_control/test/mocksocket.py +++ /dev/null @@ -1,11 +0,0 @@ -class MockSocket: - def __init__(self): - self.connected = False - self.ip = None - self.port = None - - def connect(self, address: tuple[str, int]) -> None: - self.connected = True - self.ip = address[0] - self.port = address[1] - diff --git a/lab_control/test/sds1000xe_test.py b/lab_control/test/sds1000xe_test.py index 1bc1888..7cbee70 100644 --- a/lab_control/test/sds1000xe_test.py +++ b/lab_control/test/sds1000xe_test.py @@ -1,34 +1,24 @@ import pytest from lab_control.sds1000xe import SDS1000XE -from lab_control.test.mocksocket import MockSocket +from lab_control.mock_sds1000xe_server import MockSDS1000XEServer testIP = "127.0.0.1" -testPort = 4000 -defaultPort = 5025 @pytest.fixture -def mockSocket(): - return MockSocket() +def mockServer(): + s = MockSDS1000XEServer() + yield s + s.stop() @pytest.fixture -def uut(mockSocket): - return SDS1000XE(testIP, port=testPort, socket=mockSocket) +def uut(mockServer): + return SDS1000XE(testIP) def test_defaults(): - mockSocket = MockSocket() - assert not mockSocket.connected - assert mockSocket.ip == None - assert mockSocket.port == None - - uut = SDS1000XE(testIP, socket=mockSocket) - assert mockSocket.connected - assert mockSocket.ip == testIP - assert mockSocket.port == defaultPort - -def test_differentPort(uut, mockSocket): - assert mockSocket.connected - assert mockSocket.ip == testIP - assert mockSocket.port == testPort + pass + +def test_differentPort(uut, mockServer): + assert False -- cgit v1.2.3