a42ba8a8c6
* make timestamp handling stateless * adapt tests for stateless timestamp handling * initial version * add more type annotations * add more type annotations * fix Generator annotation for ha_proxy_confs * fix names of issue branches * add more type annotations * don't use depricated varn anymore * don't mark all test as async * fix imports * fix solarman unit tests - fake Mqtt class * print image build time during proxy start * update changelog * fix pytest collect warning * cleanup msg_get_time handler * addapt unit test * label debug images with debug * dump droped packages * fix warnings * add systemtest with invalid start byte * update changelog * update changelog * add exposed ports and healthcheck * add wget for healthcheck * add aiohttp * use config validation for healthcheck * add http server for healthcheck * calculate msg prossesing time * add healthy check methods * fix typo * log ConfigErr with DEBUG level * Update async_stream.py - check if processing time is < 5 sec * add a close handler to release internal resources * call modbus close hanlder on a close call * add exception handling for forward handler * update changelog * isolate Modbus fix * cleanup * update changelog * add heaithy handler * log unrelease references * add healtcheck * complete exposed port list * add wget for healtcheck * add aiohttp * use Enum class for State * calc processing time for healthcheck * add HTTP server for healthcheck * cleanup * Update CHANGELOG.md * updat changelog * add docstrings to state enum * set new state State.received * add healthy method * log healthcheck infos with DEBUG level * update changelog * S allius/issue100 (#101) * detect dead connections - disconnect connection on Msg receive timeout - improve connection trace (add connection id) * update changelog * fix merge conflict * fix unittests * S allius/issue108 (#109) * add more data types * adapt unittests * improve test coverage * fix linter warning * update changelog * S allius/issue102 (#110) * hotfix: don't send two MODBUS commands together * fix unit tests * remove read loop * optional sleep between msg read and sending rsp * wait after read 0.5s before sending a response * add pending state * fix state definitions * determine the connection timeout by the conn state * avoid sending MODBUS cmds in the inverter's reporting phase * update changelog * S allius/issue111 (#112) Synchronize regular MODBUS commands with the status of the inverter to prevent the inverter from crashing due to unexpected packets. * inital checkin * remove crontab entry for regular MODBUS cmds * add timer for regular MODBUS polling * fix Stop method call for already stopped timer * optimize MB_START_TIMEOUT value * cleanup * update changelog * fix buildx warnings * fix timer cleanup * fix Config.class_init() - return error string or None - release Schema structure after building thr config * add quit flag to docker push * fix timout calculation * rename python to debugpy * add asyncio log * cleanup shutdown - stop webserver on shutdown - enable asyncio debug mode for debug versions * update changelog * update changelog * fix exception in MODBUS timeout callback * update changelog
381 lines
12 KiB
Python
381 lines
12 KiB
Python
# test_with_pytest.py
|
|
import pytest
|
|
import asyncio
|
|
from app.src.modbus import Modbus
|
|
from app.src.infos import Infos, Register
|
|
|
|
pytest_plugins = ('pytest_asyncio',)
|
|
# pytestmark = pytest.mark.asyncio(scope="module")
|
|
|
|
class ModbusTestHelper(Modbus):
|
|
def __init__(self):
|
|
super().__init__(self.send_cb)
|
|
self.db = Infos()
|
|
self.pdu = None
|
|
self.send_calls = 0
|
|
self.recv_responses = 0
|
|
def send_cb(self, pdu: bytearray, log_lvl: int, state: str):
|
|
self.pdu = pdu
|
|
self.send_calls += 1
|
|
def resp_handler(self):
|
|
self.recv_responses += 1
|
|
|
|
def test_modbus_crc():
|
|
'''Check CRC-16 calculation'''
|
|
mb = Modbus(None)
|
|
assert 0x0b02 == mb._Modbus__calc_crc(b'\x01\x06\x20\x08\x00\x04')
|
|
assert 0 == mb._Modbus__calc_crc(b'\x01\x06\x20\x08\x00\x04\x02\x0b')
|
|
assert mb._Modbus__check_crc(b'\x01\x06\x20\x08\x00\x04\x02\x0b')
|
|
|
|
assert 0xc803 == mb._Modbus__calc_crc(b'\x01\x06\x20\x08\x00\x00')
|
|
assert 0 == mb._Modbus__calc_crc(b'\x01\x06\x20\x08\x00\x00\x03\xc8')
|
|
assert mb._Modbus__check_crc(b'\x01\x06\x20\x08\x00\x00\x03\xc8')
|
|
|
|
assert 0x5c75 == mb._Modbus__calc_crc(b'\x01\x03\x08\x01\x2c\x00\x2c\x02\x2c\x2c\x46')
|
|
|
|
def test_build_modbus_pdu():
|
|
'''Check building and sending a MODBUS RTU'''
|
|
mb = ModbusTestHelper()
|
|
mb.build_msg(1,6,0x2000,0x12)
|
|
assert mb.pdu == b'\x01\x06\x20\x00\x00\x12\x02\x07'
|
|
assert mb._Modbus__check_crc(mb.pdu)
|
|
assert mb.last_addr == 1
|
|
assert mb.last_fcode == 6
|
|
assert mb.last_reg == 0x2000
|
|
assert mb.last_len == 18
|
|
assert mb.err == 0
|
|
|
|
def test_recv_req():
|
|
'''Receive a valid request, which must transmitted'''
|
|
mb = ModbusTestHelper()
|
|
assert mb.recv_req(b'\x01\x06\x20\x00\x00\x12\x02\x07')
|
|
assert mb.last_fcode == 6
|
|
assert mb.last_reg == 0x2000
|
|
assert mb.last_len == 0x12
|
|
assert mb.err == 0
|
|
|
|
def test_recv_req_crc_err():
|
|
'''Receive a request with invalid CRC, which must be dropped'''
|
|
mb = ModbusTestHelper()
|
|
assert not mb.recv_req(b'\x01\x06\x20\x00\x00\x12\x02\x08')
|
|
assert mb.pdu == None
|
|
assert mb.last_fcode == 0
|
|
assert mb.last_reg == 0
|
|
assert mb.last_len == 0
|
|
assert mb.err == 1
|
|
|
|
def test_recv_resp_crc_err():
|
|
'''Receive a response with invalid CRC, which must be dropped'''
|
|
mb = ModbusTestHelper()
|
|
# simulate a transmitted request
|
|
mb.req_pend = True
|
|
mb.last_addr = 1
|
|
mb.last_fcode = 3
|
|
mb.last_reg == 0x300e
|
|
mb.last_len == 2
|
|
# check matching response, but with CRC error
|
|
call = 0
|
|
for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x04\x01\x2c\x00\x46\xbb\xf3', 'test'):
|
|
call += 1
|
|
assert mb.err == 1
|
|
assert 0 == call
|
|
assert mb.req_pend == True
|
|
# cleanup queue
|
|
mb._Modbus__stop_timer()
|
|
assert not mb.req_pend
|
|
|
|
def test_recv_resp_invalid_addr():
|
|
'''Receive a response with wrong server addr, which must be dropped'''
|
|
mb = ModbusTestHelper()
|
|
mb.req_pend = True
|
|
# simulate a transmitted request
|
|
mb.last_addr = 1
|
|
mb.last_fcode = 3
|
|
mb.last_reg == 0x300e
|
|
mb.last_len == 2
|
|
|
|
# check not matching response, with wrong server addr
|
|
call = 0
|
|
for key, update in mb.recv_resp(mb.db, b'\x02\x03\x04\x01\x2c\x00\x46\x88\xf4', 'test'):
|
|
call += 1
|
|
assert mb.err == 2
|
|
assert 0 == call
|
|
assert mb.req_pend == True
|
|
assert mb.que.qsize() == 0
|
|
|
|
# cleanup queue
|
|
mb._Modbus__stop_timer()
|
|
assert not mb.req_pend
|
|
|
|
def test_recv_recv_fcode():
|
|
'''Receive a response with wrong function code, which must be dropped'''
|
|
mb = ModbusTestHelper()
|
|
mb.build_msg(1,4,0x300e,2)
|
|
assert mb.que.qsize() == 0
|
|
assert mb.req_pend
|
|
|
|
# check not matching response, with wrong function code
|
|
call = 0
|
|
for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x04\x01\x2c\x00\x46\xbb\xf4', 'test'):
|
|
call += 1
|
|
|
|
assert mb.err == 3
|
|
assert 0 == call
|
|
assert mb.req_pend == True
|
|
assert mb.que.qsize() == 0
|
|
|
|
# cleanup queue
|
|
mb._Modbus__stop_timer()
|
|
assert not mb.req_pend
|
|
|
|
def test_recv_resp_len():
|
|
'''Receive a response with wrong data length, which must be dropped'''
|
|
mb = ModbusTestHelper()
|
|
mb.build_msg(1,3,0x300e,3)
|
|
assert mb.que.qsize() == 0
|
|
assert mb.req_pend
|
|
assert mb.last_len == 3
|
|
|
|
# check not matching response, with wrong data length
|
|
call = 0
|
|
for key, update, _ in mb.recv_resp(mb.db, b'\x01\x03\x04\x01\x2c\x00\x46\xbb\xf4', 'test'):
|
|
call += 1
|
|
|
|
assert mb.err == 4
|
|
assert 0 == call
|
|
assert mb.req_pend == True
|
|
assert mb.que.qsize() == 0
|
|
|
|
# cleanup queue
|
|
mb._Modbus__stop_timer()
|
|
assert not mb.req_pend
|
|
|
|
def test_recv_unexpect_resp():
|
|
'''Receive a response when we havb't sent a request'''
|
|
mb = ModbusTestHelper()
|
|
assert not mb.req_pend
|
|
|
|
# check unexpected response, which must be dropped
|
|
call = 0
|
|
for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x04\x01\x2c\x00\x46\xbb\xf4', 'test'):
|
|
call += 1
|
|
|
|
assert mb.err == 5
|
|
assert 0 == call
|
|
assert mb.req_pend == False
|
|
assert mb.que.qsize() == 0
|
|
|
|
def test_parse_resp():
|
|
'''Receive matching response and parse the values'''
|
|
mb = ModbusTestHelper()
|
|
mb.build_msg(1,3,0x3007,6)
|
|
assert mb.que.qsize() == 0
|
|
assert mb.req_pend
|
|
|
|
call = 0
|
|
exp_result = ['V0.0.212', 4.4, 0.7, 0.7, 30]
|
|
for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x0c\x01\x2c\x00\x2c\x00\x2c\x00\x46\x00\x46\x00\x46\x32\xc8', 'test'):
|
|
if key == 'grid':
|
|
assert update == True
|
|
elif key == 'inverter':
|
|
assert update == True
|
|
elif key == 'env':
|
|
assert update == True
|
|
else:
|
|
assert False
|
|
assert exp_result[call] == val
|
|
call += 1
|
|
assert 0 == mb.err
|
|
assert 5 == call
|
|
assert mb.que.qsize() == 0
|
|
assert not mb.req_pend
|
|
|
|
def test_queue():
|
|
mb = ModbusTestHelper()
|
|
mb.build_msg(1,3,0x3022,4)
|
|
assert mb.que.qsize() == 0
|
|
assert mb.req_pend
|
|
|
|
assert mb.send_calls == 1
|
|
assert mb.pdu == b'\x01\x030"\x00\x04\xeb\x03'
|
|
mb.pdu = None
|
|
assert mb.send_calls == 1
|
|
assert mb.pdu == None
|
|
|
|
assert mb.que.qsize() == 0
|
|
|
|
# cleanup queue
|
|
mb._Modbus__stop_timer()
|
|
assert not mb.req_pend
|
|
|
|
def test_queue2():
|
|
'''Check queue handling for build_msg() calls'''
|
|
mb = ModbusTestHelper()
|
|
mb.build_msg(1,3,0x3007,6)
|
|
mb.build_msg(1,6,0x2008,4)
|
|
assert mb.que.qsize() == 1
|
|
assert mb.req_pend
|
|
mb.build_msg(1,3,0x3007,6)
|
|
assert mb.que.qsize() == 2
|
|
assert mb.req_pend
|
|
|
|
assert mb.send_calls == 1
|
|
assert mb.pdu == b'\x01\x030\x07\x00\x06{\t'
|
|
call = 0
|
|
exp_result = ['V0.0.212', 4.4, 0.7, 0.7, 30]
|
|
for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x0c\x01\x2c\x00\x2c\x00\x2c\x00\x46\x00\x46\x00\x46\x32\xc8', 'test'):
|
|
if key == 'grid':
|
|
assert update == True
|
|
elif key == 'inverter':
|
|
assert update == True
|
|
elif key == 'env':
|
|
assert update == True
|
|
else:
|
|
assert False
|
|
assert exp_result[call] == val
|
|
call += 1
|
|
assert 0 == mb.err
|
|
assert 5 == call
|
|
|
|
assert mb.que.qsize() == 1
|
|
assert mb.send_calls == 2
|
|
assert mb.pdu == b'\x01\x06\x20\x08\x00\x04\x02\x0b'
|
|
|
|
for key, update, val in mb.recv_resp(mb.db, b'\x01\x06\x20\x08\x00\x04\x02\x0b', 'test'):
|
|
pass
|
|
|
|
assert mb.que.qsize() == 0
|
|
assert mb.send_calls == 3
|
|
assert mb.pdu == b'\x01\x030\x07\x00\x06{\t'
|
|
call = 0
|
|
for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x0c\x01\x2c\x00\x2c\x00\x2c\x00\x46\x00\x46\x00\x46\x32\xc8', 'test'):
|
|
call += 1
|
|
assert 0 == mb.err
|
|
assert 5 == call
|
|
|
|
assert mb.que.qsize() == 0
|
|
assert not mb.req_pend
|
|
|
|
def test_queue3():
|
|
'''Check queue handling for recv_req() calls'''
|
|
mb = ModbusTestHelper()
|
|
assert mb.recv_req(b'\x01\x03\x30\x07\x00\x06{\t', mb.resp_handler)
|
|
assert mb.recv_req(b'\x01\x06\x20\x08\x00\x04\x02\x0b', mb.resp_handler)
|
|
assert mb.que.qsize() == 1
|
|
assert mb.req_pend
|
|
assert mb.recv_req(b'\x01\x03\x30\x07\x00\x06{\t')
|
|
assert mb.que.qsize() == 2
|
|
assert mb.req_pend
|
|
|
|
assert mb.send_calls == 1
|
|
assert mb.pdu == b'\x01\x030\x07\x00\x06{\t'
|
|
assert mb.recv_responses == 0
|
|
|
|
call = 0
|
|
exp_result = ['V0.0.212', 4.4, 0.7, 0.7, 30]
|
|
for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x0c\x01\x2c\x00\x2c\x00\x2c\x00\x46\x00\x46\x00\x46\x32\xc8', 'test'):
|
|
if key == 'grid':
|
|
assert update == True
|
|
elif key == 'inverter':
|
|
assert update == True
|
|
elif key == 'env':
|
|
assert update == True
|
|
else:
|
|
assert False
|
|
assert exp_result[call] == val
|
|
call += 1
|
|
assert 0 == mb.err
|
|
assert 5 == call
|
|
assert mb.recv_responses == 1
|
|
|
|
assert mb.que.qsize() == 1
|
|
assert mb.send_calls == 2
|
|
assert mb.pdu == b'\x01\x06\x20\x08\x00\x04\x02\x0b'
|
|
|
|
for key, update, val in mb.recv_resp(mb.db, b'\x01\x06\x20\x08\x00\x04\x02\x0b', 'test'):
|
|
pass
|
|
assert 0 == mb.err
|
|
assert mb.recv_responses == 2
|
|
|
|
assert mb.que.qsize() == 0
|
|
assert mb.send_calls == 3
|
|
assert mb.pdu == b'\x01\x030\x07\x00\x06{\t'
|
|
call = 0
|
|
for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x0c\x01\x2c\x00\x2c\x00\x2c\x00\x46\x00\x46\x00\x46\x32\xc8', 'test'):
|
|
call += 1
|
|
assert 0 == mb.err
|
|
assert mb.recv_responses == 2
|
|
assert 5 == call
|
|
|
|
|
|
assert mb.que.qsize() == 0
|
|
assert not mb.req_pend
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_timeout():
|
|
'''Test MODBUS response timeout and RTU retransmitting'''
|
|
assert asyncio.get_running_loop()
|
|
mb = ModbusTestHelper()
|
|
mb.max_retries = 2
|
|
mb.timeout = 0.1 # 100ms timeout for fast testing, expect a time resolution of at least 10ms
|
|
assert asyncio.get_running_loop() == mb.loop
|
|
mb.build_msg(1,3,0x3007,6)
|
|
mb.build_msg(1,6,0x2008,4)
|
|
|
|
assert mb.que.qsize() == 1
|
|
assert mb.req_pend
|
|
assert mb.retry_cnt == 0
|
|
assert mb.send_calls == 1
|
|
assert mb.pdu == b'\x01\x030\x07\x00\x06{\t'
|
|
|
|
mb.pdu = None
|
|
await asyncio.sleep(0.11) # wait for first timeout and retransmittion
|
|
assert mb.que.qsize() == 1
|
|
assert mb.req_pend
|
|
assert mb.retry_cnt == 1
|
|
assert mb.send_calls == 2
|
|
assert mb.pdu == b'\x01\x030\x07\x00\x06{\t'
|
|
|
|
mb.pdu = None
|
|
await asyncio.sleep(0.11) # wait for second timeout and retransmittion
|
|
assert mb.que.qsize() == 1
|
|
assert mb.req_pend
|
|
assert mb.retry_cnt == 2
|
|
assert mb.send_calls == 3
|
|
assert mb.pdu == b'\x01\x030\x07\x00\x06{\t'
|
|
|
|
mb.pdu = None
|
|
await asyncio.sleep(0.11) # wait for third timeout and next pdu
|
|
assert mb.que.qsize() == 0
|
|
assert mb.req_pend
|
|
assert mb.retry_cnt == 0
|
|
assert mb.send_calls == 4
|
|
assert mb.pdu == b'\x01\x06\x20\x08\x00\x04\x02\x0b'
|
|
|
|
mb.max_retries = 0 # next pdu without retranmsission
|
|
await asyncio.sleep(0.11) # wait for fourth timout
|
|
assert mb.que.qsize() == 0
|
|
assert not mb.req_pend
|
|
assert mb.retry_cnt == 0
|
|
assert mb.send_calls == 4
|
|
|
|
# assert mb.counter == {}
|
|
|
|
def test_recv_unknown_data():
|
|
'''Receive a response with an unknwon register'''
|
|
mb = ModbusTestHelper()
|
|
assert 0x9000 not in mb.map
|
|
mb.map[0x9000] = {'reg': Register.TEST_REG1, 'fmt': '!H', 'ratio': 1}
|
|
|
|
mb.build_msg(1,3,0x9000,2)
|
|
|
|
# check matching response, but with CRC error
|
|
call = 0
|
|
for key, update, val in mb.recv_resp(mb.db, b'\x01\x03\x04\x01\x2c\x00\x46\xbb\xf4', 'test'):
|
|
call += 1
|
|
assert mb.err == 0
|
|
assert 0 == call
|
|
assert not mb.req_pend
|
|
|
|
del mb.map[0x9000]
|