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()