Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_read_check_length(self):
# Write data shorter than what should be read
ads.write(
self.endpoint,
value=1,
index_group=constants.INDEXGROUP_DATA,
index_offset=1,
plc_datatype=constants.PLCTYPE_USINT,
)
with self.assertRaises(RuntimeError):
# Since the length is checked, this must give an error
ads.read(
self.endpoint,
index_group=constants.INDEXGROUP_DATA,
index_offset=1,
plc_datatype=constants.PLCTYPE_UINT,
check_length=True,
)
# If the length is not checked, no error should be raised
value = ads.read(
self.endpoint,
index_group=constants.INDEXGROUP_DATA,
index_offset=1,
plc_datatype=constants.PLCTYPE_UINT,
check_length=False,
)
self.assertEqual(value, 1)
def test_read_check_length(self):
# Write data shorter than what should be read
ads.write(
self.endpoint,
value=1,
index_group=constants.INDEXGROUP_DATA,
index_offset=1,
plc_datatype=constants.PLCTYPE_USINT,
)
with self.assertRaises(RuntimeError):
# Since the length is checked, this must give an error
ads.read(
self.endpoint,
index_group=constants.INDEXGROUP_DATA,
index_offset=1,
plc_datatype=constants.PLCTYPE_UINT,
check_length=True,
)
def test_write_float(self):
value = 123.456
ads.write(
self.endpoint,
index_group=constants.INDEXGROUP_DATA,
index_offset=1,
value=value,
plc_datatype=constants.PLCTYPE_REAL,
)
# Retrieve list of received requests from server
requests = self.test_server.request_history
# Assert that the server received a request
self.assertEqual(len(requests), 1)
# Assert that the server received the correct command
self.assert_command_id(requests[0], constants.ADSCOMMAND_WRITE)
def tearDownClass(cls):
cls.test_server.stop()
# wait a bit for server to shutdown
time.sleep(1)
ads.close_port()
if platform_is_linux():
ads.delete_route(cls.endpoint)
cls.test_server = AdsTestServer(AdvancedHandler(), logging=False)
cls.test_server.start()
# Endpoint AMS Address
cls.endpoint = AmsAddr(TEST_SERVER_AMS_NET_ID, TEST_SERVER_AMS_PORT)
# Open AMS Port
ads.open_port()
# wait a bit otherwise error might occur
time.sleep(1)
# NOTE: On a Windows machine, this route needs to be configured
# within the router service for the tests to work.
if platform_is_linux():
ads.add_route(cls.endpoint, TEST_SERVER_IP_ADDRESS)
def tearDownClass(cls):
cls.test_server.stop()
# wait a bit for server to shutdown
time.sleep(1)
ads.close_port()
if platform_is_linux():
ads.delete_route(cls.endpoint)
def setUpClass(cls):
# Start dummy ADS Endpoint
cls.test_server = AdsTestServer(AdvancedHandler(), logging=False)
cls.test_server.start()
# Endpoint AMS Address
cls.endpoint = AmsAddr(TEST_SERVER_AMS_NET_ID, TEST_SERVER_AMS_PORT)
# Open AMS Port
ads.open_port()
# wait a bit otherwise error might occur
time.sleep(1)
# NOTE: On a Windows machine, this route needs to be configured
# within the router service for the tests to work.
if platform_is_linux():
ads.add_route(cls.endpoint, TEST_SERVER_IP_ADDRESS)
def setUpClass(cls):
# Start dummy ADS Endpoint
cls.test_server = AdsTestServer(logging=False)
cls.test_server.start()
# Endpoint AMS Address
cls.endpoint = AmsAddr(TEST_SERVER_AMS_NET_ID, TEST_SERVER_AMS_PORT)
# Open AMS Port
ads.open_port()
# wait a bit otherwise error might occur
time.sleep(1)
# NOTE: On a Windows machine, this route needs to be configured
# within the router service for the tests to work.
if platform_is_linux():
ads.add_route(cls.endpoint, TEST_SERVER_IP_ADDRESS)
netid = pyads.get_local_address().netid
self.assertEqual(netid, "0.0.0.0.1.5")
# Change netid by String
pyads.set_local_address("0.0.0.0.1.6")
netid = pyads.get_local_address().netid
self.assertEqual(netid, "0.0.0.0.1.6")
# Change netid by Struct
pyads.set_local_address(org_adr.netIdStruct())
netid = pyads.get_local_address().netid
self.assertEqual(netid, org_netid)
# Check raised error on short netid
with self.assertRaises(ValueError):
pyads.ads.set_local_address("1.2.3")
# Check raised error on invalid netid
with self.assertRaises(ValueError):
pyads.set_local_address("1.2.3.a")
# Check wrong netid datatype
with self.assertRaises(AssertionError):
pyads.set_local_address(123)