From 209e12c50c040c425a0aa7edc1a02136f9d25d62 Mon Sep 17 00:00:00 2001 From: Christiansalles Date: Mon, 14 Sep 2026 11:33:03 -0300 Subject: [PATCH] Fix fragmented TCP response framing --- .../commands/ExecutionAndOnline.py | 19 +- src/CognexNativePy/commands/FileAndJob.py | 28 +- src/CognexNativePy/commands/Image.py | 11 +- .../commands/SettingsAndCellsValues.py | 28 +- src/CognexNativePy/utils.py | 205 ++++++++++++- tests/test_response_framing.py | 278 ++++++++++++++++++ 6 files changed, 525 insertions(+), 44 deletions(-) create mode 100644 tests/test_response_framing.py diff --git a/src/CognexNativePy/commands/ExecutionAndOnline.py b/src/CognexNativePy/commands/ExecutionAndOnline.py index cc35a9f..a7e366a 100644 --- a/src/CognexNativePy/commands/ExecutionAndOnline.py +++ b/src/CognexNativePy/commands/ExecutionAndOnline.py @@ -1,7 +1,12 @@ import socket from CognexNativePy.CognexCommandError import CognexCommandError -from CognexNativePy.utils import send_command, receive_data +from CognexNativePy.utils import ( + _receive_exact_lines, + _receive_sized_response, + _receive_status_response, + send_command, +) class ExecutionAndOnline: @@ -32,7 +37,7 @@ def set_online(self, mode: int) -> None: else: command = f"SO{mode}" send_command(self.socket, command) - status_code = receive_data(self.socket)[0] + status_code = _receive_status_response(self.socket)[0] status_messages = { "0": "Unrecognized command.", "-1": "The value given for Int is either out of range, or is not a valid integer.", @@ -62,7 +67,7 @@ def get_online(self) -> int: """ command = "GO" send_command(self.socket, command) - online_state = int(receive_data(self.socket)[0]) + online_state = int(_receive_exact_lines(self.socket, 1)[0]) if online_state in [0, 1]: return int(online_state) @@ -110,7 +115,7 @@ def set_event(self, event_code: int) -> tuple: command = f"SE{event_code}" send_command(self.socket, command) - data_received = receive_data(self.socket) + data_received = _receive_sized_response(self.socket) status_code, result = data_received[0], data_received[1] status_messages = { @@ -159,7 +164,7 @@ def set_event_and_wait(self, event_code: int) -> None: command = f"SW{event_code}" send_command(self.socket, command) - status_code = receive_data(self.socket)[0] + status_code = _receive_status_response(self.socket)[0] status_messages = { "0": "Unrecognized command.", @@ -187,7 +192,7 @@ def reset_system(self) -> None: """ command = "RT" send_command(self.socket, command) - status_code = receive_data(self.socket)[0] + status_code = _receive_status_response(self.socket)[0] status_messages = { "-6": "User does not have Full Access to execute the command. For more information, see Cognex Documentation.", @@ -225,7 +230,7 @@ def send_message(self, message: str, event_code: int = None) -> None: command += f'{event_code}' send_command(self.socket, command) - status_code = receive_data(self.socket)[0] + status_code = _receive_status_response(self.socket)[0] status_messages = { "0": "Unrecognized command.", diff --git a/src/CognexNativePy/commands/FileAndJob.py b/src/CognexNativePy/commands/FileAndJob.py index d56b7bd..806d410 100644 --- a/src/CognexNativePy/commands/FileAndJob.py +++ b/src/CognexNativePy/commands/FileAndJob.py @@ -1,6 +1,10 @@ from CognexNativePy.CognexCommandError import CognexCommandError -from CognexNativePy.utils import (format_data, receive_data, - receive_data_from_socket, send_command) +from CognexNativePy.utils import ( + _receive_status_response, + format_data, + receive_data_from_socket, + send_command, +) class FileAndJob: @@ -34,7 +38,7 @@ def load_file(self, filename: str) -> None: """ command = f"LF{filename}" send_command(self.socket, command) - status_code = receive_data(self.socket)[0] + status_code = _receive_status_response(self.socket)[0] status_messages = { "0": "Unrecognized command.", "-1": "The filename is missing.", @@ -79,7 +83,7 @@ def store_file(self, filename: str) -> None: else: command = f"TF{filename}" send_command(self.socket, command) - status_code = receive_data(self.socket)[0] + status_code = _receive_status_response(self.socket)[0] status_messages = { "0": "Unrecognized command.", @@ -166,7 +170,7 @@ def write_file(self, filename: str, size: int, data: bytes, checksum: str) -> No send_command(self.socket, f"{format_data(data)}") send_command(self.socket, f"{checksum}") - status_code = receive_data(self.socket)[0] + status_code = _receive_status_response(self.socket)[0] status_messages = { "0": "Unrecognized command.", @@ -210,7 +214,7 @@ def delete_file(self, filename: str) -> None: raise ValueError("The filename must have a .JOB or .CXD extension.") command = f"DF{filename}" send_command(self.socket, command) - status_code = receive_data(self.socket)[0] + status_code = _receive_status_response(self.socket)[0] status_messages = { "0": "Unrecognized command.", @@ -248,7 +252,7 @@ def get_file(self) -> str: """ command = "GF" send_command(self.socket, command) - data_received = receive_data(self.socket) + data_received = _receive_status_response(self.socket, successful_value_lines=1) status_code, filename = data_received[0], data_received[1] status_messages = { "0": "Unrecognized command.", @@ -287,7 +291,7 @@ def set_job(self, job_id: int) -> None: command = f"SJ{job_id}" send_command(self.socket, command) - status_code = receive_data(self.socket)[0] + status_code = _receive_status_response(self.socket)[0] status_messages = { "0": "Unrecognized command.", @@ -331,7 +335,7 @@ def store_job(self, job_id: int, job_name: str) -> None: command = f"TJ{job_id}{job_name}" send_command(self.socket, command) - data_received = receive_data(self.socket) + data_received = _receive_status_response(self.socket) status_code = data_received[0] status_messages = { @@ -416,7 +420,7 @@ def write_job(self, job_id: int, job_name: str, job_size: int, job_data: bytes, send_command(self.socket, f"{job_size}") send_command(self.socket, f"{format_data(job_data)}") send_command(self.socket, f"{job_checksum}") - status_code = receive_data(self.socket)[0] + status_code = _receive_status_response(self.socket)[0] status_messages = { "0": "Unrecognized command.", @@ -454,7 +458,7 @@ def delete_job(self, job_id: int) -> None: command = f"DJ{job_id}" send_command(self.socket, command) - status_code = receive_data(self.socket)[0] + status_code = _receive_status_response(self.socket)[0] status_messages = { "0": "Unrecognized command.", @@ -491,7 +495,7 @@ def get_job(self) -> int: """ command = "GJ" send_command(self.socket, command) - data_received = receive_data(self.socket) + data_received = _receive_status_response(self.socket, successful_value_lines=1) status_code = data_received[0] status_messages = { diff --git a/src/CognexNativePy/commands/Image.py b/src/CognexNativePy/commands/Image.py index 0262ea6..a595364 100644 --- a/src/CognexNativePy/commands/Image.py +++ b/src/CognexNativePy/commands/Image.py @@ -1,7 +1,12 @@ import socket from CognexNativePy.CognexCommandError import CognexCommandError -from CognexNativePy.utils import receive_data_from_socket, send_command, format_data, receive_data +from CognexNativePy.utils import ( + _receive_status_response, + format_data, + receive_data_from_socket, + send_command, +) class Image: @@ -94,7 +99,7 @@ def write_bmp(self, image_size: int, image_data: bytes, image_checksum: str) -> send_command(self.socket, f"{image_size}") send_command(self.socket, f"{format_data(image_data)}") send_command(self.socket, f"{image_checksum}") - status_code = receive_data(self.socket)[0] + status_code = _receive_status_response(self.socket)[0] status_messages = { "0": "Unrecognized command.", @@ -132,7 +137,7 @@ def write_image(self, image_size: int, image_data: bytes, image_checksum: str) - send_command(self.socket, f"{format_data(image_data)}") send_command(self.socket, f"{image_checksum}") - status_code = receive_data(self.socket)[0] + status_code = _receive_status_response(self.socket)[0] status_messages = { "0": "Unrecognized command.", diff --git a/src/CognexNativePy/commands/SettingsAndCellsValues.py b/src/CognexNativePy/commands/SettingsAndCellsValues.py index e557d80..1211a08 100644 --- a/src/CognexNativePy/commands/SettingsAndCellsValues.py +++ b/src/CognexNativePy/commands/SettingsAndCellsValues.py @@ -3,7 +3,13 @@ from typing import Union from CognexNativePy.CognexCommandError import CognexCommandError -from CognexNativePy.utils import receive_data, send_command, receive_data_from_socket, format_data +from CognexNativePy.utils import ( + _receive_exact_lines, + _receive_status_response, + format_data, + receive_data_from_socket, + send_command, +) class SettingsAndCellsValues: @@ -41,7 +47,7 @@ def get_value(self, cell_or_tag: str, row: int = None) -> str: command = f'GV{cell_or_tag}' send_command(self.socket, command) - data_received = receive_data(self.socket) + data_received = _receive_status_response(self.socket, successful_value_lines=1) status_code = data_received[0] status_messages = { "0": "Unrecognized command.", @@ -90,7 +96,7 @@ def set_integer_value(self, cell_or_tag: str, row_or_value: int, value_if_cell: command = f'SI{cell_or_tag} {row_or_value}' send_command(self.socket, command) - data_received = receive_data(self.socket) + data_received = _receive_status_response(self.socket) status_code = data_received[0] status_messages = { "0": "Unrecognized command.", @@ -143,7 +149,7 @@ def set_float_value(self, cell_or_tag: str, row_or_value: int, value_if_cell: fl command = f'SF{cell_or_tag} {row_or_value}' send_command(self.socket, command) - data_received = receive_data(self.socket) + data_received = _receive_status_response(self.socket) status_code = data_received[0] status_messages = { "0": "Unrecognized command.", @@ -222,7 +228,7 @@ def set_region( col_offset_or_high} {high_or_wide} {wide_or_angle} {angle_or_curve}' send_command(self.socket, command) - data_received = receive_data(self.socket) + data_received = _receive_status_response(self.socket) status_code = data_received[0] status_messages = { "0": "Unrecognized command.", @@ -267,7 +273,7 @@ def set_string(self, column: str, row: int, string_value: str) -> None: formatted_row = f"{row:03d}" command = f"SS{column}{formatted_row}{string_value}" send_command(self.socket, command) - data_received = receive_data(self.socket) + data_received = _receive_status_response(self.socket) status_code = data_received[0] status_messages = { "0": "Unrecognized command.", @@ -296,7 +302,7 @@ def get_info(self) -> dict: """ command = "GI" send_command(self.socket, command) - data_received = receive_data(self.socket) + data_received = _receive_status_response(self.socket, successful_value_lines=5) status_code = data_received[0] status_messages = { "0": "Unrecognized command.", @@ -378,7 +384,7 @@ def write_settings(self, size: int, settings: str, checksum: str) -> None: send_command(self.socket, f"{size}") send_command(self.socket, f"{format_data(settings)}") send_command(self.socket, f"{checksum}") - data_received = receive_data(self.socket) + data_received = _receive_status_response(self.socket) status_code = data_received[0] status_messages = { "0": "Unrecognized command.", @@ -407,7 +413,7 @@ def store_settings(self) -> None: """ command = "TS" send_command(self.socket, command) - data_received = receive_data(self.socket) + data_received = _receive_status_response(self.socket) status_code = data_received[0] status_messages = { "0": "Unrecognized command.", @@ -441,7 +447,7 @@ def set_ip_address_lock(self, lock: int) -> None: command = f"SL{lock}" send_command(self.socket, command) - data_received = receive_data(self.socket) + data_received = _receive_status_response(self.socket) status_code = data_received[0] status_messages = { "0": "Unrecognized command.", @@ -469,5 +475,5 @@ def get_ip_address_lock(self) -> int: """ command = "GL" send_command(self.socket, command) - data_received = receive_data(self.socket) + data_received = _receive_exact_lines(self.socket, 1) return int(data_received[0]) diff --git a/src/CognexNativePy/utils.py b/src/CognexNativePy/utils.py index 7726fdf..9ba1a3a 100644 --- a/src/CognexNativePy/utils.py +++ b/src/CognexNativePy/utils.py @@ -1,10 +1,183 @@ +import select import socket import textwrap +import time +import weakref from CognexNativePy.CognexCommandError import CognexCommandError PORT = 23 DEBUG = False +_LINE_END = b'\r\n' +_RECEIVE_TIMEOUT_SECONDS = 30.0 +_RECV_CHUNK_BYTES = 4096 +_MAX_BUFFERED_BYTES = 64 * 1024 +_READERS = weakref.WeakKeyDictionary() + + +class _SocketReader: + """Buffers a socket byte stream until complete protocol frames exist.""" + + def __init__(self, connection: socket.socket): + self._connection = weakref.ref(connection) + self.buffer = bytearray() + self.invalid = False + + @property + def connection(self) -> socket.socket: + connection = self._connection() + if connection is None: + raise CognexCommandError("Socket session is no longer usable.") + return connection + + def _fail(self, message: str): + self.invalid = True + self.buffer.clear() + try: + self.connection.close() + finally: + raise CognexCommandError(message) + + def _receive_more(self, deadline: float) -> None: + if self.invalid: + raise CognexCommandError("Socket session is no longer usable.") + + remaining = deadline - time.monotonic() + if remaining <= 0: + self._fail("Timed out while receiving a complete response.") + + previous_timeout = None + timeout_supported = hasattr(self.connection, 'gettimeout') and hasattr(self.connection, 'settimeout') + if timeout_supported: + previous_timeout = self.connection.gettimeout() + self.connection.settimeout(remaining) + + try: + chunk = self.connection.recv(_RECV_CHUNK_BYTES) + except (socket.timeout, TimeoutError): + self._fail("Timed out while receiving a complete response.") + except OSError: + self._fail("Socket error while receiving a response.") + finally: + if timeout_supported and not self.invalid: + self.connection.settimeout(previous_timeout) + + if not chunk: + self._fail("Connection closed before a complete response was received.") + if len(self.buffer) + len(chunk) > _MAX_BUFFERED_BYTES: + self._fail("Response exceeded the maximum buffered size.") + self.buffer.extend(chunk) + + def read_line(self, deadline: float) -> str: + while True: + end = self.buffer.find(_LINE_END) + if end >= 0: + line = bytes(self.buffer[:end]) + del self.buffer[:end + len(_LINE_END)] + try: + return line.decode('ascii') + except UnicodeDecodeError: + self._fail("Response contained non-ASCII data.") + self._receive_more(deadline) + + def read_bytes(self, size: int, deadline: float) -> bytes: + while len(self.buffer) < size: + self._receive_more(deadline) + data = bytes(self.buffer[:size]) + del self.buffer[:size] + return data + + def read_prompt(self, prompt: bytes, deadline: float) -> None: + while True: + if self.buffer == prompt: + self.buffer.clear() + return + if self.buffer and not prompt.startswith(self.buffer): + self._fail("Unexpected login prompt received.") + self._receive_more(deadline) + + def finish_response(self) -> None: + if self.buffer: + self._fail("Unexpected data remained after the response.") + + def ensure_can_send(self) -> None: + if self.invalid: + raise CognexCommandError("Socket session is no longer usable.") + if self.buffer: + self._fail("Unexpected data was pending before a command.") + try: + readable, _, _ = select.select([self.connection], [], [], 0) + except (OSError, TypeError, ValueError): + return + if not readable: + return + try: + pending = self.connection.recv(1, socket.MSG_PEEK) + except (BlockingIOError, socket.timeout): + return + except OSError: + self._fail("Socket error while checking pending data.") + if pending: + self._fail("Unexpected data was pending before a command.") + self._fail("Connection was closed before a command was sent.") + + +def _get_reader(connection: socket.socket) -> _SocketReader: + reader = _READERS.get(connection) + if reader is None: + reader = _SocketReader(connection) + _READERS[connection] = reader + return reader + + +def _deadline() -> float: + return time.monotonic() + _RECEIVE_TIMEOUT_SECONDS + + +def _receive_exact_lines(connection: socket.socket, line_count: int) -> list: + reader = _get_reader(connection) + deadline = _deadline() + lines = [reader.read_line(deadline) for _ in range(line_count)] + reader.finish_response() + return lines + + +def _receive_status_response(connection: socket.socket, successful_value_lines: int = 0) -> list: + reader = _get_reader(connection) + deadline = _deadline() + status = reader.read_line(deadline) + response = [status] + if status == "1": + response.extend(reader.read_line(deadline) for _ in range(successful_value_lines)) + else: + response.extend('' for _ in range(successful_value_lines)) + reader.finish_response() + return response + + +def _receive_sized_response(connection: socket.socket) -> list: + reader = _get_reader(connection) + deadline = _deadline() + status = reader.read_line(deadline) + if status != "1": + reader.finish_response() + return [status, ''] + + try: + payload_size = int(reader.read_line(deadline)) + except ValueError: + reader._fail("Response contained an invalid payload size.") + if payload_size < 0: + reader._fail("Response contained an invalid payload size.") + payload = reader.read_bytes(payload_size, deadline) + if reader.read_bytes(len(_LINE_END), deadline) != _LINE_END: + reader._fail("Response payload was not terminated by CRLF.") + try: + decoded_payload = payload.decode('ascii') + except UnicodeDecodeError: + reader._fail("Response contained non-ASCII data.") + reader.finish_response() + return [status, decoded_payload] def send_command(socket: socket.socket, string_command: str): @@ -18,6 +191,7 @@ def send_command(socket: socket.socket, string_command: str): Returns: None """ + _get_reader(socket).ensure_can_send() if DEBUG: with open('out.txt', 'a') as f: f.write(string_command+'\n') @@ -35,8 +209,12 @@ def receive_data(socket: socket.socket) -> list: Returns: list: The received data as a list of strings. """ - data = socket.recv(4096) - string_data = data.decode('ascii').split('\r\n') + reader = _get_reader(socket) + deadline = _deadline() + string_data = [reader.read_line(deadline)] + while reader.buffer.find(_LINE_END) >= 0: + string_data.append(reader.read_line(deadline)) + string_data.append('') if DEBUG: with open('in.txt', 'a') as f: f.write("\n".join(string_data)) @@ -58,14 +236,17 @@ def open_socket(host_adress: str) -> socket.socket: """ s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(_RECEIVE_TIMEOUT_SECONDS) try: s.connect((host_adress, PORT)) - data_received = receive_data(s)[0] + data_received = _get_reader(s).read_line(_deadline()) if not (data_received.startswith('Welcome')): + close_socket(s) raise CognexCommandError(f'Error logging in, expected "Welcome [...]", received "{data_received}"') else: return s except socket.timeout: + close_socket(s) raise CognexCommandError(f'Connection attempt to {host_adress} timed out') @@ -79,6 +260,7 @@ def close_socket(socket: socket.socket) -> None: Returns: None """ + _READERS.pop(socket, None) socket.close() @@ -94,14 +276,15 @@ def login_to_cognex_system(socket: socket.socket, user: str, password: str): Returns: None """ - expected_responses = ['User: ', 'Password: ', 'User Logged In'] - commands = [user, password, None] - for expected_response, command in zip(expected_responses, commands): - if receive_data(socket)[0] == expected_response: - if command is not None: - send_command(socket, command) - else: - raise CognexCommandError(f'Error logging in, expected "{expected_response}"') + reader = _get_reader(socket) + deadline = _deadline() + reader.read_prompt(b'User: ', deadline) + send_command(socket, user) + reader.read_prompt(b'Password: ', deadline) + send_command(socket, password) + if reader.read_line(deadline) != 'User Logged In': + reader._fail('Error logging in, expected "User Logged In"') + reader.finish_response() def format_data(data: bytes): diff --git a/tests/test_response_framing.py b/tests/test_response_framing.py new file mode 100644 index 0000000..3c39cbb --- /dev/null +++ b/tests/test_response_framing.py @@ -0,0 +1,278 @@ +import inspect +import socket +import sys +import unittest +from collections import deque +from pathlib import Path +from unittest.mock import patch + + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'src')) + +from CognexNativePy.CognexCommandError import CognexCommandError +from CognexNativePy.commands.ExecutionAndOnline import ExecutionAndOnline +from CognexNativePy.commands.FileAndJob import FileAndJob +from CognexNativePy.commands.SettingsAndCellsValues import SettingsAndCellsValues +from CognexNativePy.utils import ( + _MAX_BUFFERED_BYTES, + _receive_status_response, + login_to_cognex_system, + open_socket, + receive_data, + send_command, +) + + +class FakeSocket: + def __init__(self, chunks=(), connect_error=None): + self.chunks = deque(chunks) + self.connect_error = connect_error + self.connected_to = None + self.sent = [] + self.timeout = None + self.timeout_history = [] + self.closed = False + + def connect(self, address): + self.connected_to = address + if self.connect_error is not None: + raise self.connect_error + + def sendall(self, data): + self.sent.append(data) + + def recv(self, size, flags=0): + if not self.chunks: + raise socket.timeout() + chunk = self.chunks.popleft() + if isinstance(chunk, BaseException): + raise chunk + return chunk + + def gettimeout(self): + return self.timeout + + def settimeout(self, timeout): + self.timeout = timeout + self.timeout_history.append(timeout) + + def close(self): + self.closed = True + + +class ReceiveDataTests(unittest.TestCase): + def test_receive_data_waits_for_crlf_split_between_bytes(self): + connection = FakeSocket((b'1\r', b'\n')) + + self.assertEqual(receive_data(connection), ['1', '']) + self.assertFalse(connection.closed) + + def test_receive_data_preserves_legacy_list_shape_for_coalesced_lines(self): + connection = FakeSocket((b'1\r\n1.000\r\n',)) + + self.assertEqual(receive_data(connection), ['1', '1.000', '']) + + def test_timeout_invalidates_the_socket(self): + connection = FakeSocket((socket.timeout(),)) + + with self.assertRaisesRegex(CognexCommandError, 'Timed out'): + receive_data(connection) + + self.assertTrue(connection.closed) + + with self.assertRaisesRegex(CognexCommandError, 'no longer usable'): + receive_data(connection) + + def test_eof_in_the_middle_of_a_line_invalidates_the_socket(self): + connection = FakeSocket((b'1\r', b'')) + + with self.assertRaisesRegex(CognexCommandError, 'closed before a complete response'): + receive_data(connection) + + self.assertTrue(connection.closed) + + def test_non_ascii_data_invalidates_the_socket(self): + connection = FakeSocket((b'\xff\r\n',)) + + with self.assertRaisesRegex(CognexCommandError, 'non-ASCII'): + receive_data(connection) + + self.assertTrue(connection.closed) + + def test_oversized_unterminated_response_invalidates_the_socket(self): + connection = FakeSocket((b'x' * (_MAX_BUFFERED_BYTES + 1),)) + + with self.assertRaisesRegex(CognexCommandError, 'maximum buffered size'): + receive_data(connection) + + self.assertTrue(connection.closed) + + def test_extra_data_after_a_complete_response_invalidates_the_socket(self): + connection = FakeSocket((b'1\r\nunexpected',)) + + with self.assertRaisesRegex(CognexCommandError, 'remained after the response'): + _receive_status_response(connection) + + self.assertTrue(connection.closed) + + def test_buffered_partial_data_prevents_the_next_command(self): + connection = FakeSocket((b'1\r\npartial',)) + self.assertEqual(receive_data(connection), ['1', '']) + + with self.assertRaisesRegex(CognexCommandError, 'pending before a command'): + send_command(connection, 'GO') + + self.assertTrue(connection.closed) + + def test_pending_kernel_data_prevents_a_new_command(self): + client, server = socket.socketpair() + self.addCleanup(server.close) + server.sendall(b'stale') + + with self.assertRaisesRegex(CognexCommandError, 'pending before a command'): + send_command(client, 'GO') + + self.assertEqual(client.fileno(), -1) + + +class LoginFramingTests(unittest.TestCase): + def test_fragmented_and_coalesced_login_prompts(self): + connection = FakeSocket(( + b'Welcome to In-Sight(TM) IS3808 Session 1\r\nUs', + b'er: ', + b'Pass', + b'word: ', + b'User Logged In\r', + b'\n', + )) + + with patch('CognexNativePy.utils.socket.socket', return_value=connection): + opened = open_socket('camera.example') + login_to_cognex_system(opened, 'user', 'password') + + self.assertIs(opened, connection) + self.assertEqual(connection.connected_to, ('camera.example', 23)) + self.assertEqual(connection.sent, [b'user\r\n', b'password\r\n']) + self.assertFalse(connection.closed) + + def test_connection_timeout_closes_the_socket(self): + connection = FakeSocket(connect_error=socket.timeout()) + + with patch('CognexNativePy.utils.socket.socket', return_value=connection): + with self.assertRaisesRegex(CognexCommandError, 'Connection attempt .* timed out'): + open_socket('camera.example') + + self.assertTrue(connection.closed) + + +class CommandFramingTests(unittest.TestCase): + def test_get_value_reads_fragmented_status_and_value(self): + connection = FakeSocket((b'1\r\n', b'1.000\r\n')) + settings = SettingsAndCellsValues(connection) + + self.assertEqual(settings.get_value('ExampleTag'), '1.000') + self.assertEqual(connection.sent, [b'GVExampleTag\r\n']) + + def test_get_value_handles_crlf_fragmentation_and_next_command(self): + connection = FakeSocket(( + b'1\r', b'\n1.000\r', b'\n', + b'1\r\n2.000\r\n', + )) + settings = SettingsAndCellsValues(connection) + + self.assertEqual(settings.get_value('FirstTag'), '1.000') + self.assertEqual(settings.get_value('SecondTag'), '2.000') + self.assertFalse(connection.closed) + + def test_get_value_keeps_an_explicit_empty_value(self): + connection = FakeSocket((b'1\r\n\r\n',)) + + self.assertEqual(SettingsAndCellsValues(connection).get_value('ExampleTag'), '') + + def test_get_value_error_does_not_wait_for_a_value(self): + connection = FakeSocket((b'-1\r\n',)) + + with self.assertRaisesRegex(CognexCommandError, 'cell ID or symbolic tag is invalid'): + SettingsAndCellsValues(connection).get_value('ExampleTag') + + self.assertFalse(connection.closed) + + def test_get_file_and_get_job_read_their_second_line(self): + file_socket = FakeSocket((b'1\r\n', b'Example.job\r\n')) + job_socket = FakeSocket((b'1\r', b'\n42\r\n')) + + self.assertEqual(FileAndJob(file_socket).get_file(), 'Example.job') + self.assertEqual(FileAndJob(job_socket).get_job(), 42) + + def test_get_info_reads_all_five_information_lines(self): + connection = FakeSocket(( + b'1\r\nSerial Number: 123\r', + b'\nApplication Version: 1.2\r\nMonitor Version: 3.4\r\n', + b'MAC Address: 00-00-00-00-00-00\r\nDate of Build: Today\r\n', + )) + + result = SettingsAndCellsValues(connection).get_info() + + self.assertEqual(result['Serial Number'], '123') + self.assertEqual(result['Application Version'], '1.2') + self.assertEqual(result['Monitor Version'], '3.4') + self.assertEqual(result['MAC Address'], '00-00-00-00-00-00') + self.assertEqual(result['Date of Build'], 'Today') + + def test_set_event_reads_a_fragmented_sized_payload(self): + connection = FakeSocket(( + b'1\r\n8\r\n', + b'1\r', + b'\n', + )) + + result = ExecutionAndOnline(connection).set_event(0) + + self.assertEqual(result, ('1', '1')) + + def test_set_event_preserves_command_errors(self): + connection = FakeSocket((b'0\r\n',)) + + with self.assertRaisesRegex(CognexCommandError, 'Unrecognized command'): + ExecutionAndOnline(connection).set_event(0) + + self.assertFalse(connection.closed) + + def test_set_event_rejects_a_negative_payload_size(self): + connection = FakeSocket((b'1\r\n-1\r\n',)) + + with self.assertRaisesRegex(CognexCommandError, 'invalid payload size'): + ExecutionAndOnline(connection).set_event(0) + + self.assertTrue(connection.closed) + + def test_one_deadline_is_shared_by_a_fragmented_response(self): + connection = FakeSocket((b'1\r\n', b'value\r\n')) + + with patch('CognexNativePy.utils.time.monotonic', side_effect=(100.0, 101.0, 105.0)): + response = _receive_status_response(connection, successful_value_lines=1) + + self.assertEqual(response, ['1', 'value']) + self.assertEqual( + [timeout for timeout in connection.timeout_history if timeout is not None], + [29.0, 25.0], + ) + + def test_single_line_status_command_accepts_fragmented_crlf(self): + connection = FakeSocket((b'1\r', b'\n')) + + self.assertIsNone(SettingsAndCellsValues(connection).set_integer_value('Tag', 10)) + + +class PublicApiTests(unittest.TestCase): + def test_public_function_signatures_are_unchanged(self): + self.assertEqual(list(inspect.signature(receive_data).parameters), ['socket']) + self.assertEqual(list(inspect.signature(open_socket).parameters), ['host_adress']) + self.assertEqual( + list(inspect.signature(login_to_cognex_system).parameters), + ['socket', 'user', 'password'], + ) + + +if __name__ == '__main__': + unittest.main()