<commit_before>"""Tests for the clx.py file.

There are quite a few methods in the CIPDriver which are difficult to
read or test due to both code clarity and complexity issues. As well as
there being no way to control the execution of many of the private
methods through the public API. This has lead to testing of quite a few
private API methods to achieve an acceptable test coverage.
"""
from logging import exception
from pycomm3.const import SUCCESS
from pycomm3.packets.responses import RegisterSessionResponsePacket, ResponsePacket, GenericConnectedResponsePacket
from pycomm3.packets.requests import GenericConnectedRequestPacket, GenericUnconnectedRequestPacket, RegisterSessionRequestPacket, RequestPacket, UnRegisterSessionRequestPacket
import socket
from unittest import mock

import pytest
from pycomm3.cip_base import CIPDriver
from pycomm3.exceptions import CommError, DataError, PycommError
from pycomm3.socket_ import Socket
from pycomm3.tag import Tag

CONNECT_PATH = '192.168.1.100/1'
DEFAULT_PORT = CIPDriver(CONNECT_PATH,
    init_info = False,
    init_tags = False
)._cfg['port']

def test_cip_get_module_info_raises_data_error_if_response_falsy():
    with mock.patch.object(CIPDriver, 'generic_message') as mock_generic_message:
        mock_generic_message.return_value = False
        with pytest.raises(DataError):
            driver = CIPDriver(CONNECT_PATH)
            driver.get_module_info(1)

        assert mock_generic_message.called

def test_get_module_info_returns_expected_identity_dict():
    EXPECTED_DICT = {
        'vendor': 'Rockwell Automation/Allen-Bradley', # uint 1 for Rockwell Automation/Allen-Bradley
        'product_type': 'Residual Gas Analyzer', # uint 0x1E for 'Residual Gas Analyzer'
        'product_code': 1, # uint
        'version_major': 1, # int
        'version_minor': 1, # int
        'revision': '1.1', # int, int
        'serial': '00000001', # udint
        'device_type': 'This is a test',
        'status': '0000000100000001', # uint
        'state': 'Major Unrecoverable Fault', # uint 5 Major Unrecoverable Fault
    }
    RESPONSE_BYTES = b"\x01\x00\x1E\x00\x01\x00\x01\x01\x01\x01\x01\x00\x00\x00" + \
        len(EXPECTED_DICT['device_type']).to_bytes(1, 'little') + \
        bytes(EXPECTED_DICT['device_type'], 'ascii') + int(5).to_bytes(4, 'little')<commit_msg>Remove Unused Default Port Value from CIPDriver tests<commit_after>"""Tests for the clx.py file.

There are quite a few methods in the CIPDriver which are difficult to
read or test due to both code clarity and complexity issues. As well as
there being no way to control the execution of many of the private
methods through the public API. This has lead to testing of quite a few
private API methods to achieve an acceptable test coverage.
"""
from logging import exception
from pycomm3.const import SUCCESS
from pycomm3.packets.responses import RegisterSessionResponsePacket, ResponsePacket, GenericConnectedResponsePacket
from pycomm3.packets.requests import GenericConnectedRequestPacket, GenericUnconnectedRequestPacket, RegisterSessionRequestPacket, RequestPacket, UnRegisterSessionRequestPacket
import socket
from unittest import mock

import pytest
from pycomm3.cip_base import CIPDriver
from pycomm3.exceptions import CommError, DataError, PycommError
from pycomm3.socket_ import Socket
from pycomm3.tag import Tag

CONNECT_PATH = '192.168.1.100/1'

def test_cip_get_module_info_raises_data_error_if_response_falsy():
    with mock.patch.object(CIPDriver, 'generic_message') as mock_generic_message:
        mock_generic_message.return_value = False
        with pytest.raises(DataError):
            driver = CIPDriver(CONNECT_PATH)
            driver.get_module_info(1)

        assert mock_generic_message.called

def test_get_module_info_returns_expected_identity_dict():
    EXPECTED_DICT = {
        'vendor': 'Rockwell Automation/Allen-Bradley', # uint 1 for Rockwell Automation/Allen-Bradley
        'product_type': 'Residual Gas Analyzer', # uint 0x1E for 'Residual Gas Analyzer'
        'product_code': 1, # uint
        'version_major': 1, # int
        'version_minor': 1, # int
        'revision': '1.1', # int, int
        'serial': '00000001', # udint
        'device_type': 'This is a test',
        'status': '0000000100000001', # uint
        'state': 'Major Unrecoverable Fault', # uint 5 Major Unrecoverable Fault
    }
    RESPONSE_BYTES = b"\x01\x00\x1E\x00\x01\x00\x01\x01\x01\x01\x01\x00\x00\x00" + \
        len(EXPECTED_DICT['device_type']).to_bytes(1, 'little') + \
        bytes(EXPECTED_DICT['device_type'], 'ascii') + int(5).to_bytes(4, 'little')
{'fraction_non_alphanumeric': 0.08475675675675676, 'fraction_numerical': 0.03545945945945946, 'mean_word_length': 4.215332581736189, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 6, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6177192', 'n_tokens_mistral': 1514, 'n_tokens_neox': 1420, 'n_words': 405}