From 4371a4f2a85c3c68d292fd3dffbae6d405275040 Mon Sep 17 00:00:00 2001 From: Brunno Vanelli Date: Mon, 5 Feb 2024 20:25:47 +0100 Subject: [PATCH 1/3] refactor: Add unit testing and refactor code to follow Python standards. - Reformat file with community standards black and ruff, with basic config - Format all variable namings using PEP8. - Add unit tests for two of the auxiliary methods. - Add testing on CI using github actions and pytest - Improve code clarity by using f-strings, available since python 3.6 --- .github/workflows/test.yaml | 35 +++ .gitignore | 2 + .pre-commit-config.yaml | 23 ++ src/app/__init__.py | 0 src/app/modbus_server.py | 424 ++++++++++++++++++++---------------- tests/__init__.py | 0 tests/test_server.py | 31 +++ tests/test_utils.py | 7 + 8 files changed, 334 insertions(+), 188 deletions(-) create mode 100644 .github/workflows/test.yaml create mode 100644 src/app/__init__.py create mode 100644 tests/__init__.py create mode 100644 tests/test_server.py create mode 100644 tests/test_utils.py diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml new file mode 100644 index 0000000..0cc0408 --- /dev/null +++ b/.github/workflows/test.yaml @@ -0,0 +1,35 @@ +name: Run tests + +on: + pull_request: + push: + branches: [ main ] + +jobs: + build: + + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.9", "3.10", "3.11"] + + steps: + - uses: actions/checkout@v3 + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v4 + with: + python-version: ${{ matrix.python-version }} + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install ruff pytest + if [ -f requirements.txt ]; then pip install -r requirements.txt; fi + - name: Lint with ruff + run: | + # stop the build if there are Python syntax errors or undefined names + ruff --format=github --select=E9,F63,F7,F82 --target-version=py37 . + # default set of ruff rules with GitHub Annotations + ruff --format=github --target-version=py37 . + - name: Test with pytest + run: | + pytest diff --git a/.gitignore b/.gitignore index c96d7d7..773239e 100644 --- a/.gitignore +++ b/.gitignore @@ -63,6 +63,7 @@ local.properties # Reference: https://intellij-support.jetbrains.com/hc/en-us/articles/206544839 # User-specific stuff: +.idea/ .idea/**/workspace.xml .idea/**/tasks.xml .idea/dictionaries @@ -223,3 +224,4 @@ $RECYCLE.BIN/ *.lnk # End of https://www.gitignore.io/api/osx,java,linux,eclipse,windows,netbeans,java-web,intellij +__pycache__/ diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 827ea64..93314ae 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -14,3 +14,26 @@ repos: rev: v2.12.0 hooks: - id: hadolint-docker + - repo: 'https://github.com/charliermarsh/ruff-pre-commit' + rev: v0.2.0 + hooks: + - id: ruff + args: + - '--line-length=120' + - '--fix' + - '--exit-non-zero-on-fix' + - repo: 'https://github.com/pycqa/isort' + rev: 5.13.2 + hooks: + - id: isort + name: isort (python) + args: + - '--profile' + - black + - '--filter-files' + - repo: 'https://github.com/psf/black' + rev: 24.1.1 + hooks: + - id: black + args: + - '--line-length=120' diff --git a/src/app/__init__.py b/src/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/app/modbus_server.py b/src/app/modbus_server.py index e30b056..e2f3d02 100644 --- a/src/app/modbus_server.py +++ b/src/app/modbus_server.py @@ -6,109 +6,125 @@ Last modified by: Michael Oberdorf Last modified at: 2023-12-06 *************************************************************************** """ -import sys +import argparse +import json +import logging import os +import pathlib import socket -from pymodbus.server.sync import StartTcpServer -from pymodbus.server.sync import StartTlsServer +import sys +from typing import Literal, Optional + +from pymodbus.datastore import ( + ModbusSequentialDataBlock, + ModbusServerContext, + ModbusSlaveContext, + ModbusSparseDataBlock, +) from pymodbus.device import ModbusDeviceIdentification -from pymodbus.datastore import ModbusSequentialDataBlock, ModbusSparseDataBlock -from pymodbus.datastore import ModbusSlaveContext, ModbusServerContext -from pymodbus.transaction import ModbusRtuFramer, ModbusBinaryFramer -import logging -import argparse -import json +from pymodbus.server.sync import StartTcpServer, StartTlsServer # default configuration file path -default_config_file='/app/modbus_server.json' -VERSION='1.3.0' +default_config_file = "/app/modbus_server.json" +VERSION = "1.3.0" + +log = logging.getLogger() + + """ ############################################################################### # F U N C T I O N S ############################################################################### """ -def get_ip_address(): + + +def get_ip_address() -> str: """ get_ip_address is a small function that determines the IP address of the outbound ethernet interface @return: string, IP address """ - ipaddr = '' + ipaddr = "" try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) ipaddr = s.getsockname()[0] - except: + except Exception: pass - return(ipaddr) + return ipaddr -def run_server(listener_address: str = '0.0.0.0', listener_port: int = 5020, tls_cert: str = None, tls_key: str = None, zeroMode: bool = False, discreteInputs: dict = dict(), coils: dict = dict(), holdingRegisters: dict = dict(), inputRegisters: dict = dict()): + +def run_server( + listener_address: str = "0.0.0.0", + listener_port: int = 5020, + tls_cert: str = None, + tls_key: str = None, + zero_mode: bool = False, + discrete_inputs: Optional[dict] = None, + coils: Optional[dict] = None, + holding_registers: Optional[dict] = None, + input_registers: Optional[dict] = None, +): """ Run the modbus server(s) @param listener_address: string, IP address to bind the listener (default: '0.0.0.0') @param listener_port: integer, TCP port to bin the listener (default: 5020) @param tls_cert: boolean, path to certificate to start tcp server with TLS (default: None) @param tls_key: boolean, path to private key to start tcp server with TLS (default: None) - @param zeroMode: boolean, request to address(0-7) will map to the address (0-7) instead of (1-8) (default: False) - @param discreteInputs: dict(), initial addresses and their values (default: dict()) + @param zero_mode: boolean, request to address(0-7) will map to the address (0-7) instead of (1-8) (default: False) + @param discrete_inputs: dict(), initial addresses and their values (default: dict()) @param coils: dict(), initial addresses and their values (default: dict()) - @param holdingRegisters: dict(), initial addresses and their values (default: dict()) - @param inputRegisters: dict(), initial addresses and their values (default: dict()) + @param holding_registers: dict(), initial addresses and their values (default: dict()) + @param input_registers: dict(), initial addresses and their values (default: dict()) """ # initialize data store - log.debug('Initialize discrete input') - if isinstance(discreteInputs, dict) and len(discreteInputs) > 0: - #log.debug('using dictionary from configuration file:') - #log.debug(discreteInputs) - di = ModbusSparseDataBlock(discreteInputs) + log.debug("Initialize discrete input") + if isinstance(discrete_inputs, dict) and discrete_inputs: + # log.debug('using dictionary from configuration file:') + # log.debug(discreteInputs) + di = ModbusSparseDataBlock(discrete_inputs) else: - #log.debug('set all registers to 0xaa') - #di = ModbusSequentialDataBlock(0x00, [0xaa]*65536) - log.debug('set all registers to 0x00') + # log.debug('set all registers to 0xaa') + # di = ModbusSequentialDataBlock(0x00, [0xaa]*65536) + log.debug("set all registers to 0x00") di = ModbusSequentialDataBlock.create() - log.debug('Initialize coils') - if isinstance(coils, dict) and len(coils) > 0: - #log.debug('using dictionary from configuration file:') - #log.debug(coils) + log.debug("Initialize coils") + if coils: + # log.debug('using dictionary from configuration file:') + # log.debug(coils) co = ModbusSparseDataBlock(coils) else: - #log.debug('set all registers to 0xbb') - #co = ModbusSequentialDataBlock(0x00, [0xbb]*65536) - log.debug('set all registers to 0x00') + # log.debug('set all registers to 0xbb') + # co = ModbusSequentialDataBlock(0x00, [0xbb]*65536) + log.debug("set all registers to 0x00") co = ModbusSequentialDataBlock.create() - log.debug('Initialize holding registers') - if isinstance(holdingRegisters, dict) and len(holdingRegisters) > 0: - #log.debug('using dictionary from configuration file:') - #log.debug(holdingRegisters) - hr = ModbusSparseDataBlock(holdingRegisters) + log.debug("Initialize holding registers") + if isinstance(holding_registers, dict) and holding_registers: + # log.debug('using dictionary from configuration file:') + # log.debug(holdingRegisters) + hr = ModbusSparseDataBlock(holding_registers) else: - #log.debug('set all registers to 0xcc') - #hr = ModbusSequentialDataBlock(0x00, [0xcc]*65536) - log.debug('set all registers to 0x00') + # log.debug('set all registers to 0xcc') + # hr = ModbusSequentialDataBlock(0x00, [0xcc]*65536) + log.debug("set all registers to 0x00") hr = ModbusSequentialDataBlock.create() - log.debug('Initialize input registers') - if isinstance(inputRegisters, dict) and len(inputRegisters) > 0: - #log.debug('using dictionary from configuration file:') - #log.debug(inputRegisters) - ir = ModbusSparseDataBlock(inputRegisters) + log.debug("Initialize input registers") + if isinstance(input_registers, dict) and input_registers: + # log.debug('using dictionary from configuration file:') + # log.debug(inputRegisters) + ir = ModbusSparseDataBlock(input_registers) else: - #log.debug('set all registers to 0xdd') - #ir = ModbusSequentialDataBlock(0x00, [0xdd]*65536) - log.debug('set all registers to 0x00') + # log.debug('set all registers to 0xdd') + # ir = ModbusSequentialDataBlock(0x00, [0xdd]*65536) + log.debug("set all registers to 0x00") ir = ModbusSequentialDataBlock.create() - store = ModbusSlaveContext( - di=di, - co=co, - hr=hr, - ir=ir, - zero_mode=zeroMode - ) + store = ModbusSlaveContext(di=di, co=co, hr=hr, ir=ir, zero_mode=zero_mode) - log.debug('Define Modbus server context') + log.debug("Define Modbus server context") context = ModbusServerContext(slaves=store, single=True) # ----------------------------------------------------------------------- # @@ -116,89 +132,110 @@ def run_server(listener_address: str = '0.0.0.0', listener_port: int = 5020, tls # ----------------------------------------------------------------------- # # If you don't set this or any fields, they are defaulted to empty strings. # ----------------------------------------------------------------------- # - log.debug('Define Modbus server identity') + log.debug("Define Modbus server identity") identity = ModbusDeviceIdentification() - identity.VendorName = 'Pymodbus' - identity.ProductCode = 'PM' - identity.VendorUrl = 'http://github.com/riptideio/pymodbus/' - identity.ProductName = 'Pymodbus Server' - identity.ModelName = 'Pymodbus Server' - identity.MajorMinorRevision = '2.5.3' + identity.VendorName = "Pymodbus" + identity.ProductCode = "PM" + identity.VendorUrl = "http://github.com/riptideio/pymodbus/" + identity.ProductName = "Pymodbus Server" + identity.ModelName = "Pymodbus Server" + identity.MajorMinorRevision = "2.5.3" # ----------------------------------------------------------------------- # # run the server # ----------------------------------------------------------------------- # - startTLS=False - if tls_cert and tls_key and os.path.isfile(tls_cert) and os.path.isfile(tls_key): startTLS=True + start_tls = False + if tls_cert and tls_key and os.path.isfile(tls_cert) and os.path.isfile(tls_key): + start_tls = True - if startTLS: - log.info('Starting Modbus TCP server with TLS on ' + listener_address + ':' + str(listener_port)) - StartTlsServer(context, identity=identity, certfile=tls_cert, keyfile=tls_key, address=(listener_address, listener_port)) + if start_tls: + log.info(f"Starting Modbus TCP server with TLS on {listener_address}:{listener_port}") + StartTlsServer( + context, + identity=identity, + certfile=tls_cert, + keyfile=tls_key, + address=(listener_address, listener_port), + ) else: - log.info('Starting Modbus TCP server on ' + listener_address + ':' + str(listener_port)) + log.info(f"Starting Modbus TCP server on {listener_address}:{listener_port}") StartTcpServer(context, identity=identity, address=(listener_address, listener_port)) # TCP with different framer # StartTcpServer(context, identity=identity, framer=ModbusRtuFramer, address=(listener_address, listener_port)) -def prepareRegister(register: dict, initType: str, initializeUndefinedRegisters: bool = False) -> dict: +def prepare_register( + register: dict, + init_type: Literal["boolean", "word"], + initialize_undefined_registers: bool = False, +) -> dict: """ Function to prepare the register to have the correct data types @param register: dict(), the register dictionary, loaded from json file - @param initType: str(), how to initialize the register values 'boolean' or 'word' - @param initializeUndefinedRegisters: boolean, fill undefined registers with 0x00 (default: False) + @param init_type: str(), how to initialize the register values 'boolean' or 'word' + @param initialize_undefined_registers: boolean, fill undefined registers with 0x00 (default: False) @return: dict(), register with correct data types """ - outRegister=dict() + out_register = dict() if not isinstance(register, dict): - log.error('Unexpected input in function prepareRegister') - return(outRegister) - if len(register) == 0: return(outRegister) + log.error("Unexpected input in function prepareRegister") + return out_register + if len(register) == 0: + return out_register for key in register: - if isinstance(key, str): - keyOut = int(key, 0) - log.debug(' Transform register id: ' + str(key) + ' ('+ str(type(key)) + ') to: ' + str(keyOut) + ' (' + str(type(keyOut)) + ')') - else: keyOut = key - - val = register[key] - valOut = val - if initType == 'word' and isinstance(val, str) and str(val)[0:2] == '0x' and len(val) >= 3 and len(val) <= 6: - valOut = int(val, 16) - log.debug(' Transform value for register: ' + str(keyOut) + ' from: ' + str(val) + ' ('+ str(type(key)) + ') to: ' + str(valOut) + ' (' + str(type(valOut)) + ')') - elif initType == 'word' and isinstance(val, int) and val >=0 and val <= 65535: - valOut = val - log.debug(' Use value for register: {}: {}'.format(str(keyOut), str(valOut))) - elif initType == 'boolean': - if isinstance(val, bool): - valOut = val - log.debug(' Set register: ' + str(keyOut) + ' to: ' + str(valOut) + ' (' + str(type(valOut)) + ')') - elif isinstance(val, int): - if val == 0: - valOut = False - else: - valOut = True - log.debug(' Transform value for register: ' + str(keyOut) + ' from: ' + str(val) + ' ('+ str(type(key)) + ') to: ' + str(valOut) + ' (' + str(type(valOut)) + ')') - else: - log.error(' Malformed input or input is out of range for register: {} -> value is {} - skip this register initialization!'.format(str(keyOut), str(val))) - continue - outRegister[keyOut] = valOut - - if initializeUndefinedRegisters: - if initType == 'word': - log.debug(' Fill undefined registers with 0x00') - elif initType == 'boolean': - log.debug(' Fill undefined registers with False') + if isinstance(key, str): + key_out = int(key, 0) + log.debug(f" Transform register id: {key} ({type(key)}) to: {key_out} ({type(key_out)})") + else: + key_out = key + + val = register[key] + val_out = val + if init_type == "word" and isinstance(val, str) and str(val)[0:2] == "0x" and 3 <= len(val) <= 6: + val_out = int(val, 16) + log.debug( + f" Transform value for register: {key_out} from: {val} ({type(key)}) to: {val_out} ({type(val_out)})" + ) + elif init_type == "word" and isinstance(val, int) and 0 <= val <= 65535: + val_out = val + log.debug(" Use value for register: {}: {}".format(str(key_out), str(val_out))) + elif init_type == "boolean": + if isinstance(val, bool): + val_out = val + log.debug(f" Set register: {key_out} to: {val_out} ({type(val_out)})") + elif isinstance(val, int): + if val == 0: + val_out = False + else: + val_out = True + log.debug( + f" Transform value for register: {key_out} from: {val} ({type(key)}) to: " + f"{val_out} ({type(val_out)})" + ) + else: + log.error( + f" Malformed input or input is out of range for register: " + f"{key_out} -> value is {val} - skip this register initialization!" + ) + continue + out_register[key_out] = val_out + + if initialize_undefined_registers: + if init_type == "word": + log.debug(" Fill undefined registers with 0x00") + elif init_type == "boolean": + log.debug(" Fill undefined registers with False") for r in range(0, 65536, 1): - if r not in outRegister: - if initType == 'word': - #log.debug(' Initialize address: ' + str(r) + ' with 0') - outRegister[r] = 0 - elif initType == 'boolean': - #log.debug(' Initialize address: ' + str(r) + ' with False') - outRegister[r] = False + if r not in out_register: + if init_type == "word": + # log.debug(' Initialize address: ' + str(r) + ' with 0') + out_register[r] = 0 + elif init_type == "boolean": + # log.debug(' Initialize address: ' + str(r) + ' with False') + out_register[r] = False - return(outRegister) + return out_register """ @@ -206,72 +243,83 @@ def prepareRegister(register: dict, initType: str, initializeUndefinedRegisters: # M A I N ############################################################################### """ -# intialize variable -config_file=None - -# Parsing environment variables -if 'CONFIG_FILE' in os.environ: - if not os.path.isfile(os.environ['CONFIG_FILE']): - print('ERROR:', 'configuration file not exist:', os.environ['CONFIG_FILE']) - sys.exit(1) - else: - config_file=os.environ['CONFIG_FILE'] - - -# Parsing command line arguments -parser = argparse.ArgumentParser(description='Modbus TCP Server') -group = parser.add_argument_group() -group.add_argument('-f', '--config_file', help='The configuration file in json format (default: ' + default_config_file +')') -args = parser.parse_args() -if args.config_file: - if not os.path.isfile(args.config_file): - print('ERROR:', 'configuration file not exist:', args.config_file) - sys.exit(1) - else: - config_file=args.config_file - -# define default if no config file path is set -if not config_file: - config_file=default_config_file - -# read configuration file -with open(config_file, encoding='utf-8') as f: - CONFIG = json.load(f) - - -# Initialize logger -#FORMAT = ('%(asctime)-15s %(threadName)-15s %(levelname)-8s %(module)-15s:%(lineno)-8s %(message)s') -FORMAT = CONFIG['server']['logging']['format'] -logging.basicConfig(format=FORMAT) -log = logging.getLogger() -if CONFIG['server']['logging']['logLevel'].lower() == 'debug': log.setLevel(logging.DEBUG) -elif CONFIG['server']['logging']['logLevel'].lower() == 'info': log.setLevel(logging.INFO) -elif CONFIG['server']['logging']['logLevel'].lower() == 'warn': log.setLevel(logging.WARN) -elif CONFIG['server']['logging']['logLevel'].lower() == 'error': log.setLevel(logging.ERROR) -else: log.setLevel(logging.INFO) - - -# start the server -log.info('Starting Modbus TCP Server, v' + str(VERSION)) -log.debug('Loaded successfully the configuration file: {}'.format(config_file)) - -# be sure the data types within the dictionaries are correct (json will only allow strings as keys) -discreteInputs = prepareRegister(register = CONFIG['registers']['discreteInput'], initType='boolean', initializeUndefinedRegisters = CONFIG['registers']['initializeUndefinedRegisters']) -coils=prepareRegister(register = CONFIG['registers']['coils'], initType='boolean', initializeUndefinedRegisters = CONFIG['registers']['initializeUndefinedRegisters']) -holdingRegisters=prepareRegister(register = CONFIG['registers']['holdingRegister'], initType='word', initializeUndefinedRegisters = CONFIG['registers']['initializeUndefinedRegisters']) -inputRegisters=prepareRegister(register = CONFIG['registers']['inputRegister'], initType='word', initializeUndefinedRegisters = CONFIG['registers']['initializeUndefinedRegisters']) - -# try to get the interface IP address -localIPAddr = get_ip_address() -if localIPAddr != '': log.info('Outbund device IP address is: ' + localIPAddr) -run_server( - listener_address=CONFIG['server']['listenerAddress'], - listener_port=CONFIG['server']['listenerPort'], - tls_cert=CONFIG['server']['tlsParams']['privateKey'], - tls_key=CONFIG['server']['tlsParams']['certificate'], - zeroMode=CONFIG['registers']['zeroMode'], - discreteInputs=discreteInputs, - coils=coils, - holdingRegisters=holdingRegisters, - inputRegisters=inputRegisters +if __name__ == "__main__": + # Parsing command line arguments + parser = argparse.ArgumentParser(description="Modbus TCP Server") + group = parser.add_argument_group() + group.add_argument( + "-f", + "--config_file", + help=f"The configuration file in json format (default: {default_config_file})", + default=default_config_file, + ) + + args = parser.parse_args() + if "CONFIG_FILE" in os.environ: + config_file = os.environ["CONFIG_FILE"] + else: # will either use the command line argument or the default value + config_file = args.config_file + # check if file actually exists + if not pathlib.Path(config_file).is_file(): + print(f"ERROR: configuration file '{args.config_file}' does not exist.") + sys.exit(1) + + # read configuration file + with open(config_file, encoding="utf-8") as f: + CONFIG = json.load(f) + + # Initialize logger + if CONFIG["server"]["logging"]["logLevel"].lower() == "debug": + log.setLevel(logging.DEBUG) + elif CONFIG["server"]["logging"]["logLevel"].lower() == "info": + log.setLevel(logging.INFO) + elif CONFIG["server"]["logging"]["logLevel"].lower() == "warn": + log.setLevel(logging.WARN) + elif CONFIG["server"]["logging"]["logLevel"].lower() == "error": + log.setLevel(logging.ERROR) + else: + log.setLevel(logging.INFO) + FORMAT = CONFIG["server"]["logging"]["format"] + logging.basicConfig(format=FORMAT) + + # start the server + log.info(f"Starting Modbus TCP Server, v{VERSION}") + log.debug(f"Loaded successfully the configuration file: {config_file}") + + # be sure the data types within the dictionaries are correct (json will only allow strings as keys) + configured_discrete_inputs = prepare_register( + register=CONFIG["registers"]["discreteInput"], + init_type="boolean", + initialize_undefined_registers=CONFIG["registers"]["initializeUndefinedRegisters"], + ) + configured_coils = prepare_register( + register=CONFIG["registers"]["coils"], + init_type="boolean", + initialize_undefined_registers=CONFIG["registers"]["initializeUndefinedRegisters"], + ) + configured_holding_registers = prepare_register( + register=CONFIG["registers"]["holdingRegister"], + init_type="word", + initialize_undefined_registers=CONFIG["registers"]["initializeUndefinedRegisters"], + ) + configured_input_registers = prepare_register( + register=CONFIG["registers"]["inputRegister"], + init_type="word", + initialize_undefined_registers=CONFIG["registers"]["initializeUndefinedRegisters"], + ) + + # try to get the interface IP address + local_ip_addr = get_ip_address() + if local_ip_addr != "": + log.info(f"Outbound device IP address is: {local_ip_addr}") + run_server( + listener_address=CONFIG["server"]["listenerAddress"], + listener_port=CONFIG["server"]["listenerPort"], + tls_cert=CONFIG["server"]["tlsParams"]["privateKey"], + tls_key=CONFIG["server"]["tlsParams"]["certificate"], + zero_mode=CONFIG["registers"]["zeroMode"], + discrete_inputs=configured_discrete_inputs, + coils=configured_coils, + holding_registers=configured_holding_registers, + input_registers=configured_input_registers, ) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_server.py b/tests/test_server.py new file mode 100644 index 0000000..8f2e2e9 --- /dev/null +++ b/tests/test_server.py @@ -0,0 +1,31 @@ +# -*- coding: utf-8 -*- +from src.app.modbus_server import prepare_register + + +def test_prepare_register(): + register_example_data = { + "0": "0x84D9", + "1": "0x41ED", + "3": "0xC24C", + "5": "0xBF80", + "6": "0x0068", + "7": "0x006C", + "8": "0x0074", + "9": "0x0032", + } + register = prepare_register(register=register_example_data, init_type="word", initialize_undefined_registers=False) + assert len(register) == 8 + assert register[9] == 0x0032 + + # full register should contain all entries as normal registered, but all other memory addresses set to 0 + full_register = prepare_register( + register=register_example_data, init_type="word", initialize_undefined_registers=True + ) + for key in register: + assert register[key] == full_register[key] + assert full_register[10] == 0 + assert len(full_register) == 65536 + + +# def test_server(): +# run_server() diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 0000000..cacc5f2 --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,7 @@ +# -*- coding: utf-8 -*- +from src.app.modbus_server import get_ip_address + + +def test_ip_address(): + ip = get_ip_address() + assert isinstance(ip, str) From d4d45977c4a518a789d4100128867549f1369b5f Mon Sep 17 00:00:00 2001 From: Brunno Vanelli Date: Tue, 6 Feb 2024 21:37:20 +0100 Subject: [PATCH 2/3] fix: Fix test pipeline and apply suggestions. --- .github/workflows/test.yaml | 6 +++--- src/app/modbus_server.py | 10 ++++------ 2 files changed, 7 insertions(+), 9 deletions(-) diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index 0cc0408..d5a4121 100644 --- a/.github/workflows/test.yaml +++ b/.github/workflows/test.yaml @@ -23,13 +23,13 @@ jobs: run: | python -m pip install --upgrade pip pip install ruff pytest - if [ -f requirements.txt ]; then pip install -r requirements.txt; fi + if [ -f requirements.txt ]; then cd src/ && pip install -r requirements.txt; fi - name: Lint with ruff run: | # stop the build if there are Python syntax errors or undefined names - ruff --format=github --select=E9,F63,F7,F82 --target-version=py37 . + ruff --select=E9,F63,F7,F82 --target-version=py37 . # default set of ruff rules with GitHub Annotations - ruff --format=github --target-version=py37 . + ruff --target-version=py37 . - name: Test with pytest run: | pytest diff --git a/src/app/modbus_server.py b/src/app/modbus_server.py index e2f3d02..6b5ee5a 100644 --- a/src/app/modbus_server.py +++ b/src/app/modbus_server.py @@ -10,7 +10,6 @@ import json import logging import os -import pathlib import socket import sys from typing import Literal, Optional @@ -90,7 +89,7 @@ def run_server( di = ModbusSequentialDataBlock.create() log.debug("Initialize coils") - if coils: + if isinstance(coils, dict) and coils: # log.debug('using dictionary from configuration file:') # log.debug(coils) co = ModbusSparseDataBlock(coils) @@ -260,8 +259,8 @@ def prepare_register( else: # will either use the command line argument or the default value config_file = args.config_file # check if file actually exists - if not pathlib.Path(config_file).is_file(): - print(f"ERROR: configuration file '{args.config_file}' does not exist.") + if not os.path.isfile(config_file): + print(f"ERROR: configuration file '{config_file}' does not exist.") sys.exit(1) # read configuration file @@ -279,8 +278,7 @@ def prepare_register( log.setLevel(logging.ERROR) else: log.setLevel(logging.INFO) - FORMAT = CONFIG["server"]["logging"]["format"] - logging.basicConfig(format=FORMAT) + logging.basicConfig(format=CONFIG["server"]["logging"]["format"]) # start the server log.info(f"Starting Modbus TCP Server, v{VERSION}") From 1f9ae1e0f4ed1b784c3161052b11d429bf2fea17 Mon Sep 17 00:00:00 2001 From: Brunno Vanelli Date: Thu, 15 Feb 2024 19:44:37 +0100 Subject: [PATCH 3/3] fix: Fix wrong conditional on test pipeline. --- .github/workflows/test.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index d5a4121..8287c6e 100644 --- a/.github/workflows/test.yaml +++ b/.github/workflows/test.yaml @@ -23,7 +23,7 @@ jobs: run: | python -m pip install --upgrade pip pip install ruff pytest - if [ -f requirements.txt ]; then cd src/ && pip install -r requirements.txt; fi + cd src/ && pip install -r requirements.txt - name: Lint with ruff run: | # stop the build if there are Python syntax errors or undefined names