diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..cc49654 --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,31 @@ +# Pre-commit hooks to run tests and ensure code is cleaned. +# See https://pre-commit.com for more information +repos: +- repo: local + hooks: + - id: pylint + name: pylint + entry: pylint + language: system + types: [python] + require_serial: true +- repo: https://github.com/PyCQA/flake8 + rev: 6.0.0 + hooks: + - id: flake8 + args: ['--max-line-length=100'] +- repo: https://github.com/asottile/pyupgrade + rev: v3.3.1 + hooks: + - id: pyupgrade + args: ['--keep-percent-format', '--py37-plus'] +- repo: https://github.com/psf/black + rev: 22.12.0 + hooks: + - id: black + args: ['--target-version', 'py37', '--line-length', '100'] +- repo: https://github.com/PyCQA/isort + rev: 5.11.4 + hooks: + - id: isort + args: ['--profile', 'black', '--line-length', '100'] diff --git a/.woodpecker.yml b/.woodpecker.yml index 37013ed..a8d17d2 100644 --- a/.woodpecker.yml +++ b/.woodpecker.yml @@ -4,19 +4,12 @@ clone: tags: true pipeline: - test-pylint: - group: test - image: debian + tests: + image: brenard/python-pre-commit:latest commands: - DEBIAN_FRONTEND=noninteractive apt-get -qq update < /dev/null > /dev/null - - DEBIAN_FRONTEND=noninteractive apt-get -qq -y install --no-install-recommends python3-requests pylint3 < /dev/null > /dev/null - - python3 -m pylint check_esphome_devices - - test-flake8: - group: test - image: pipelinecomponents/flake8 - commands: - - flake8 check_esphome_devices + - DEBIAN_FRONTEND=noninteractive apt-get -qq -y install --no-install-recommends python3-requests < /dev/null > /dev/null + - pre-commit run --all-files build: image: brenard/debian-python-deb diff --git a/check_esphome_devices b/check_esphome_devices index fea4537..9e96c0f 100755 --- a/check_esphome_devices +++ b/check_esphome_devices @@ -20,182 +20,174 @@ Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. """ import argparse -import sys -import time import logging import re +import sys +import time import requests - # nagios exit code -STATUS = { - 'OK': 0, - 'WARNING': 1, - 'CRITICAL': 2, - 'UNKNOWN': 3 -} +STATUS = {"OK": 0, "WARNING": 1, "CRITICAL": 2, "UNKNOWN": 3} -DEFAULT_HOST = 'http://127.0.0.1:6052' +DEFAULT_HOST = "http://127.0.0.1:6052" DEFAULT_RETRY_COUNT = 4 DEFAULT_RETRY_DELAY = 1 parser = argparse.ArgumentParser() -parser.add_argument( - '-d', '--debug', - action="store_true", - dest="debug", - default=False -) +parser.add_argument("-d", "--debug", action="store_true", dest="debug", default=False) parser.add_argument( - '-H', '--host', + "-H", + "--host", action="store", dest="host", - help=f'ESPHome dashboard URL (default: {DEFAULT_HOST})', + help=f"ESPHome dashboard URL (default: {DEFAULT_HOST})", type=str, - default=DEFAULT_HOST + default=DEFAULT_HOST, ) parser.add_argument( - '-r', '--retry', + "-r", + "--retry", action="store", dest="retry_count", - help=( - 'Number of retry to retreive device status ' - f'(default: {DEFAULT_RETRY_COUNT})'), + help=("Number of retry to retreive device status " f"(default: {DEFAULT_RETRY_COUNT})"), type=int, - default=DEFAULT_RETRY_COUNT + default=DEFAULT_RETRY_COUNT, ) parser.add_argument( - '-D', '--delay', + "-D", + "--delay", action="store", dest="retry_delay", help=( - 'Delay in second between two retry to retreive device status ' - f'(default: {DEFAULT_RETRY_DELAY}s)'), + "Delay in second between two retry to retreive device status " + f"(default: {DEFAULT_RETRY_DELAY}s)" + ), type=int, - default=DEFAULT_RETRY_DELAY + default=DEFAULT_RETRY_DELAY, ) def exclude_pattern(value): - """ Check and compile exclusion pattern parameter """ + """Check and compile exclusion pattern parameter""" return re.compile(value) parser.add_argument( - '-x', '--exclude', + "-x", + "--exclude", action="append", dest="exclude", - help='Regex exclude pattern(s)', + help="Regex exclude pattern(s)", type=exclude_pattern, - default=[] + default=[], ) options = parser.parse_args() logging.basicConfig( level=logging.DEBUG if options.debug else logging.INFO, - format='%(asctime)s - %(levelname)s - %(message)s') + format="%(asctime)s - %(levelname)s - %(message)s", +) -if options.host[-1] == '/': +if options.host[-1] == "/": options.host = options.host[-1] def is_excluded(name): - """ Check if device is excluded """ + """Check if device is excluded""" for pattern in options.exclude: if pattern.search(name): - logging.debug('Device %s is excluded', name) + logging.debug("Device %s is excluded", name) return True - logging.debug('Device %s is not excluded', name) + logging.debug("Device %s is not excluded", name) return False -r = requests.get(f'{options.host}/devices') +r = requests.get(f"{options.host}/devices") devices_data = r.json() -logging.debug('Devices data: %s (%s)', devices_data, type(devices_data)) +logging.debug("Devices data: %s (%s)", devices_data, type(devices_data)) if not devices_data: - print('UNKNOWN - Fail to retreive devices using ESPHome Dashboard API') - sys.exit(STATUS['UNKNOWN']) + print("UNKNOWN - Fail to retreive devices using ESPHome Dashboard API") + sys.exit(STATUS["UNKNOWN"]) COUNT = 0 while COUNT < options.retry_count: - r = requests.get(f'{options.host}/ping') + r = requests.get(f"{options.host}/ping") COUNT += 1 ping_data = r.json() - logging.debug('Ping data: %s (%s)', ping_data, type(ping_data)) + logging.debug("Ping data: %s (%s)", ping_data, type(ping_data)) if ping_data: UNREACHABLE = False for dev in ping_data: - if ( - not ping_data[dev] - and not is_excluded(dev.replace('.yaml', '')) - ): + if not ping_data[dev] and not is_excluded(dev.replace(".yaml", "")): UNREACHABLE = True break if not UNREACHABLE: break - logging.debug('Wait %d seconds before retry...', options.retry_delay) + logging.debug("Wait %d seconds before retry...", options.retry_delay) time.sleep(options.retry_delay) if not ping_data: - print( - 'UNKNOWN - Fail to retreive devices status ' - 'using ESPHome Dashboard API') - sys.exit(STATUS['UNKNOWN']) + print("UNKNOWN - Fail to retreive devices status " "using ESPHome Dashboard API") + sys.exit(STATUS["UNKNOWN"]) UPDATE_AVAILABLE = 0 UNREACHABLE_DEVICES = 0 NO_PING_DATA = 0 errors = [] devices = {} -for dev in devices_data['configured']: - devices[dev['name']] = dev - logging.debug('Device %s: %s', dev['name'], dev) - if is_excluded(dev['name']): +for dev in devices_data["configured"]: + devices[dev["name"]] = dev + logging.debug("Device %s: %s", dev["name"], dev) + if is_excluded(dev["name"]): continue - if dev['deployed_version'] != dev['current_version']: + if dev["deployed_version"] != dev["current_version"]: UPDATE_AVAILABLE += 1 errors.append( f'Update available for device {dev["name"]} ' - f'({dev["deployed_version"]} => {dev["current_version"]})') + f'({dev["deployed_version"]} => {dev["current_version"]})' + ) - if dev['configuration'] not in ping_data: + if dev["configuration"] not in ping_data: NO_PING_DATA += 1 - errors.append( - f'No ping data found for device {dev["name"]} ' - f'({dev["configuration"]})') - elif not ping_data[dev['configuration']]: + errors.append(f'No ping data found for device {dev["name"]} ' f'({dev["configuration"]})') + elif not ping_data[dev["configuration"]]: UNREACHABLE_DEVICES += 1 errors.append(f'Device {dev["name"]} is unreachable') if not errors: - print(f'OK - no problem detected on the {len(devices)} devices') - EXIT_STATUS = STATUS['OK'] + print(f"OK - no problem detected on the {len(devices)} devices") + EXIT_STATUS = STATUS["OK"] else: msg = [] if UNREACHABLE_DEVICES: - msg.append(f'{UNREACHABLE_DEVICES} unreachable devices') + msg.append(f"{UNREACHABLE_DEVICES} unreachable devices") if NO_PING_DATA: - msg.append(f'{NO_PING_DATA} missing ping device status') + msg.append(f"{NO_PING_DATA} missing ping device status") if UPDATE_AVAILABLE: - msg.append(f'{UPDATE_AVAILABLE} update available') + msg.append(f"{UPDATE_AVAILABLE} update available") print(f'WARNING - {", ".join(msg)}') - print("\n".join([f'- {error}' for error in errors])) - EXIT_STATUS = STATUS['WARNING'] + print("\n".join([f"- {error}" for error in errors])) + EXIT_STATUS = STATUS["WARNING"] -print("\nDevices:\n" + "\n".join([ - ( - f'- {name} (version = ' - f'{dev["deployed_version"] if dev["deployed_version"] else "unknown"}' - f', address = {dev["address"] if dev["address"] else "unknown"})' +print( + "\nDevices:\n" + + "\n".join( + [ + ( + f"- {name} (version = " + f'{dev["deployed_version"] if dev["deployed_version"] else "unknown"}' + f', address = {dev["address"] if dev["address"] else "unknown"})' + ) + for name, dev in devices.items() + ] ) - for name, dev in devices.items() -])) +) sys.exit(EXIT_STATUS)