From 964045f53e49c89497812af4684b995ec332c160 Mon Sep 17 00:00:00 2001 From: Mathieu MD Date: Fri, 28 Mar 2025 09:09:28 +0100 Subject: [PATCH 01/28] Update README.md - Fix #108 - Enhance a few manual installation details --- README.md | 20 +++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index 959a2fb..4e8a90a 100644 --- a/README.md +++ b/README.md @@ -49,8 +49,14 @@ installed. You can also use the `requirements.txt` file for installation with pip. ``` +# Packages: pynetbox pyzabbix + +# Install them through requirements.txt from a venv: +virtualenv .venv +source .venv/bin/activate +.venv/bin/pip --require-virtualenv install -r requirements.txt ``` ### Config file @@ -67,25 +73,25 @@ cp config.py.example config.py Set the following environment variables: ``` -ZABBIX_HOST="https://zabbix.local" -ZABBIX_USER="username" -ZABBIX_PASS="Password" -NETBOX_HOST="https://netbox.local" -NETBOX_TOKEN="secrettoken" +export ZABBIX_HOST="https://zabbix.local" +export ZABBIX_USER="username" +export ZABBIX_PASS="Password" +export NETBOX_HOST="https://netbox.local" +export NETBOX_TOKEN="secrettoken" ``` Or, you can use a Zabbix API token to login instead of using a username and password. In that case `ZABBIX_USER` and `ZABBIX_PASS` will be ignored. ``` -ZABBIX_TOKEN=othersecrettoken +export ZABBIX_TOKEN=othersecrettoken ``` If you are using custom SSL certificates for NetBox and/or Zabbix, you can set the following environment variable to the path of your CA bundle file: ```bash -REQUEST_CA_BUNDLE=/path/to/your/ca-bundle.crt +export REQUESTS_CA_BUNDLE=/path/to/your/ca-bundle.crt ``` ### NetBox custom fields From 50c13c20cbe74686ea6dd03f31db2f7e9ad3dade Mon Sep 17 00:00:00 2001 From: Mathieu MD Date: Fri, 28 Mar 2025 09:11:14 +0100 Subject: [PATCH 02/28] Update README.md Use Bash syntax --- README.md | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index 4e8a90a..2b174a7 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ A script to create, update and delete Zabbix hosts using NetBox device objects. To pull the latest stable version to your local cache, use the following docker pull command: -``` +```sh docker pull ghcr.io/thenetworkguy/netbox-zabbix-sync:main ``` @@ -15,7 +15,7 @@ Make sure to specify the needed environment variables for the script to work (see [here](#set-environment-variables)) on the command line or use an [env file](https://docs.docker.com/reference/cli/docker/container/run/#env). -``` +```sh docker run -d -t -i -e ZABBIX_HOST='https://zabbix.local' \ -e ZABBIX_TOKEN='othersecrettoken' \ -e NETBOX_HOST='https://netbox.local' \ @@ -30,7 +30,7 @@ The image uses the default `config.py` for it's configuration, you can use a volume mount in the docker run command to override with your own config file if needed (see [config file](#config-file)): -``` +```sh docker run -d -t -i -v $(pwd)/config.py:/opt/netbox-zabbix/config.py ... ``` @@ -38,7 +38,7 @@ docker run -d -t -i -v $(pwd)/config.py:/opt/netbox-zabbix/config.py ... ### Cloning the repository -``` +```sh git clone https://github.com/TheNetworkGuy/netbox-zabbix-sync.git ``` @@ -48,7 +48,7 @@ Make sure that you have a python environment with the following packages installed. You can also use the `requirements.txt` file for installation with pip. -``` +```sh # Packages: pynetbox pyzabbix @@ -64,7 +64,7 @@ source .venv/bin/activate First time user? Copy the `config.py.example` file to `config.py`. This file is used for modifying filters and setting variables such as custom field names. -``` +```sh cp config.py.example config.py ``` @@ -72,7 +72,7 @@ cp config.py.example config.py Set the following environment variables: -``` +```sh export ZABBIX_HOST="https://zabbix.local" export ZABBIX_USER="username" export ZABBIX_PASS="Password" @@ -83,14 +83,14 @@ export NETBOX_TOKEN="secrettoken" Or, you can use a Zabbix API token to login instead of using a username and password. In that case `ZABBIX_USER` and `ZABBIX_PASS` will be ignored. -``` +```sh export ZABBIX_TOKEN=othersecrettoken ``` If you are using custom SSL certificates for NetBox and/or Zabbix, you can set the following environment variable to the path of your CA bundle file: -```bash +```sh export REQUESTS_CA_BUNDLE=/path/to/your/ca-bundle.crt ``` From 908e7eeda953e7d79b6db42b10c9e7e71b07e327 Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 14 Apr 2025 16:35:09 +0200 Subject: [PATCH 03/28] Added documentation line for unsupported Zabbix versions. --- README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/README.md b/README.md index 2b174a7..b482e7d 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,7 @@ # NetBox to Zabbix synchronization A script to create, update and delete Zabbix hosts using NetBox device objects. +Currently compatible with Zabbix 7.0. Zabbix 7.2 is unfortunately not supported and will result in the script failing. ## Installation via Docker From ea5b7d31967afdd841615cafd0696672dfdd21e2 Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 14 Apr 2025 20:13:15 +0200 Subject: [PATCH 04/28] Added initial unittesting PoC to see if Docker and Python are working correctly --- .github/workflows/unittesting.yml | 37 +++++++++++++++++++++++++++++++ 1 file changed, 37 insertions(+) create mode 100644 .github/workflows/unittesting.yml diff --git a/.github/workflows/unittesting.yml b/.github/workflows/unittesting.yml new file mode 100644 index 0000000..d4a2ad1 --- /dev/null +++ b/.github/workflows/unittesting.yml @@ -0,0 +1,37 @@ +--- +name: Unit testing and functional code control +on: + push: + branches: + - 'main' + - 'develop' + - 'unittesting' + +jobs: + integration: + runs-on: ubuntu-latest + services: + docker: + image: docker:dind + options: --privileged --shm-size=2g + volumes: + - /var/run/docker.sock:/var/run/docker.sock:ro + container: + image: ubuntu:latest + steps: + - uses: actions/checkout@v4 + + - name: Install Docker + run: | + apt-get update + apt-get install -y docker.io + + - name: Test Docker + run: | + docker version + - name: Setup Python + uses: actions/setup-python@v5 + with: + python-version: 3.12 + - name: Show Python version + run: python --version \ No newline at end of file From feb719542d8ca6cfe45f35f4c07739acef14f721 Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 14 Apr 2025 20:22:43 +0200 Subject: [PATCH 05/28] Added Netbox deployment config --- .github/workflows/unittesting.yml | 19 ++++++++++++++----- 1 file changed, 14 insertions(+), 5 deletions(-) diff --git a/.github/workflows/unittesting.yml b/.github/workflows/unittesting.yml index d4a2ad1..2283710 100644 --- a/.github/workflows/unittesting.yml +++ b/.github/workflows/unittesting.yml @@ -21,17 +21,26 @@ jobs: steps: - uses: actions/checkout@v4 - - name: Install Docker + - name: Install Docker and Docker-compose run: | apt-get update apt-get install -y docker.io + sudo apt-get install docker-compose -y - - name: Test Docker - run: | - docker version - name: Setup Python uses: actions/setup-python@v5 with: python-version: 3.12 - name: Show Python version - run: python --version \ No newline at end of file + run: python --version + - name: Test Docker + run: | + docker version + - name: configure and start Netbox + run: | + git clone -b release https://github.com/netbox-community/netbox-docker.git + mv netbox-docker/docker-compose.override.yml.example netbox-docker/docker-compose.override.yml + docker-compose -f netbox-docker/docker-compose.yml pull + docker-compose -f netbox-docker/docker-compose.yml up -d + - name: Wait 2 minutes for compose stack to build + run: sleep 120s \ No newline at end of file From 38d61dcde74fbf69dbbc787f2383ad6bcf5fb8ec Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 14 Apr 2025 20:25:02 +0200 Subject: [PATCH 06/28] Removed sudo statement --- .github/workflows/unittesting.yml | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/.github/workflows/unittesting.yml b/.github/workflows/unittesting.yml index 2283710..ef15ca3 100644 --- a/.github/workflows/unittesting.yml +++ b/.github/workflows/unittesting.yml @@ -25,8 +25,7 @@ jobs: run: | apt-get update apt-get install -y docker.io - sudo apt-get install docker-compose -y - + apt-get install docker-compose -y - name: Setup Python uses: actions/setup-python@v5 with: From f303e7e01d7e7cf76f437a80654888c9a304545c Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 14 Apr 2025 20:27:44 +0200 Subject: [PATCH 07/28] Moved to compose v2 --- .github/workflows/unittesting.yml | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/.github/workflows/unittesting.yml b/.github/workflows/unittesting.yml index ef15ca3..b970c5c 100644 --- a/.github/workflows/unittesting.yml +++ b/.github/workflows/unittesting.yml @@ -25,7 +25,7 @@ jobs: run: | apt-get update apt-get install -y docker.io - apt-get install docker-compose -y + apt install docker-compose-v2 -y - name: Setup Python uses: actions/setup-python@v5 with: @@ -39,7 +39,7 @@ jobs: run: | git clone -b release https://github.com/netbox-community/netbox-docker.git mv netbox-docker/docker-compose.override.yml.example netbox-docker/docker-compose.override.yml - docker-compose -f netbox-docker/docker-compose.yml pull - docker-compose -f netbox-docker/docker-compose.yml up -d + docker compose -f netbox-docker/docker-compose.yml pull + docker compose -f netbox-docker/docker-compose.yml up -d - name: Wait 2 minutes for compose stack to build run: sleep 120s \ No newline at end of file From 989f6fa96e2f7a03e18b243da956b73580ff9940 Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 14 Apr 2025 20:36:52 +0200 Subject: [PATCH 08/28] Moved compose override logic to infra folder --- .github/workflows/unittesting.yml | 2 +- infra/netbox-compose.yml | 22 ++++++++++++++++++++++ 2 files changed, 23 insertions(+), 1 deletion(-) create mode 100644 infra/netbox-compose.yml diff --git a/.github/workflows/unittesting.yml b/.github/workflows/unittesting.yml index b970c5c..8b0e033 100644 --- a/.github/workflows/unittesting.yml +++ b/.github/workflows/unittesting.yml @@ -38,7 +38,7 @@ jobs: - name: configure and start Netbox run: | git clone -b release https://github.com/netbox-community/netbox-docker.git - mv netbox-docker/docker-compose.override.yml.example netbox-docker/docker-compose.override.yml + mv infra/netbox-compose.yml netbox-docker/docker-compose.override.yml docker compose -f netbox-docker/docker-compose.yml pull docker compose -f netbox-docker/docker-compose.yml up -d - name: Wait 2 minutes for compose stack to build diff --git a/infra/netbox-compose.yml b/infra/netbox-compose.yml new file mode 100644 index 0000000..89e04cf --- /dev/null +++ b/infra/netbox-compose.yml @@ -0,0 +1,22 @@ +services: + netbox: + ports: + - "8000:8080" + # If you want the Nginx unit status page visible from the + # outside of the container add the following port mapping: + # - "8001:8081" + healthcheck: + # Time for which the health check can fail after the container is started. + # This depends mostly on the performance of your database. On the first start, + # when all tables need to be created the start_period should be higher than on + # subsequent starts. For the first start after major version upgrades of NetBox + # the start_period might also need to be set higher. + # Default value in our docker-compose.yml is 60s + start_period: 90s + # environment: + # SKIP_SUPERUSER: "false" + # SUPERUSER_API_TOKEN: "" + # SUPERUSER_EMAIL: "" + # SUPERUSER_NAME: "" + # SUPERUSER_PASSWORD: "" + From ad2ace942a713427a385203a5d76bffe9d5b87ec Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 14 Apr 2025 20:37:17 +0200 Subject: [PATCH 09/28] Increased start_period time of Netbox --- infra/netbox-compose.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/infra/netbox-compose.yml b/infra/netbox-compose.yml index 89e04cf..c1e8c96 100644 --- a/infra/netbox-compose.yml +++ b/infra/netbox-compose.yml @@ -12,7 +12,7 @@ services: # subsequent starts. For the first start after major version upgrades of NetBox # the start_period might also need to be set higher. # Default value in our docker-compose.yml is 60s - start_period: 90s + start_period: 150s # environment: # SKIP_SUPERUSER: "false" # SUPERUSER_API_TOKEN: "" From 4fd582970db3051d5781269f37e67dd0a2e5898c Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 14 Apr 2025 20:43:32 +0200 Subject: [PATCH 10/28] Container statement removed, added logs output --- .github/workflows/unittesting.yml | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/.github/workflows/unittesting.yml b/.github/workflows/unittesting.yml index 8b0e033..52db4bf 100644 --- a/.github/workflows/unittesting.yml +++ b/.github/workflows/unittesting.yml @@ -16,11 +16,8 @@ jobs: options: --privileged --shm-size=2g volumes: - /var/run/docker.sock:/var/run/docker.sock:ro - container: - image: ubuntu:latest steps: - uses: actions/checkout@v4 - - name: Install Docker and Docker-compose run: | apt-get update @@ -42,4 +39,6 @@ jobs: docker compose -f netbox-docker/docker-compose.yml pull docker compose -f netbox-docker/docker-compose.yml up -d - name: Wait 2 minutes for compose stack to build - run: sleep 120s \ No newline at end of file + run: | + sleep 120s + docker compose -f netbox-docker/docker-compose.yml logs netbox \ No newline at end of file From dad7d2911f5846ab52de4e7f91a1a459fcfaf9fc Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Wed, 23 Apr 2025 11:11:05 +0200 Subject: [PATCH 11/28] Reverted previous work --- .github/workflows/unittesting.yml | 44 ------------------------------- infra/netbox-compose.yml | 22 ---------------- 2 files changed, 66 deletions(-) delete mode 100644 .github/workflows/unittesting.yml delete mode 100644 infra/netbox-compose.yml diff --git a/.github/workflows/unittesting.yml b/.github/workflows/unittesting.yml deleted file mode 100644 index 52db4bf..0000000 --- a/.github/workflows/unittesting.yml +++ /dev/null @@ -1,44 +0,0 @@ ---- -name: Unit testing and functional code control -on: - push: - branches: - - 'main' - - 'develop' - - 'unittesting' - -jobs: - integration: - runs-on: ubuntu-latest - services: - docker: - image: docker:dind - options: --privileged --shm-size=2g - volumes: - - /var/run/docker.sock:/var/run/docker.sock:ro - steps: - - uses: actions/checkout@v4 - - name: Install Docker and Docker-compose - run: | - apt-get update - apt-get install -y docker.io - apt install docker-compose-v2 -y - - name: Setup Python - uses: actions/setup-python@v5 - with: - python-version: 3.12 - - name: Show Python version - run: python --version - - name: Test Docker - run: | - docker version - - name: configure and start Netbox - run: | - git clone -b release https://github.com/netbox-community/netbox-docker.git - mv infra/netbox-compose.yml netbox-docker/docker-compose.override.yml - docker compose -f netbox-docker/docker-compose.yml pull - docker compose -f netbox-docker/docker-compose.yml up -d - - name: Wait 2 minutes for compose stack to build - run: | - sleep 120s - docker compose -f netbox-docker/docker-compose.yml logs netbox \ No newline at end of file diff --git a/infra/netbox-compose.yml b/infra/netbox-compose.yml deleted file mode 100644 index c1e8c96..0000000 --- a/infra/netbox-compose.yml +++ /dev/null @@ -1,22 +0,0 @@ -services: - netbox: - ports: - - "8000:8080" - # If you want the Nginx unit status page visible from the - # outside of the container add the following port mapping: - # - "8001:8081" - healthcheck: - # Time for which the health check can fail after the container is started. - # This depends mostly on the performance of your database. On the first start, - # when all tables need to be created the start_period should be higher than on - # subsequent starts. For the first start after major version upgrades of NetBox - # the start_period might also need to be set higher. - # Default value in our docker-compose.yml is 60s - start_period: 150s - # environment: - # SKIP_SUPERUSER: "false" - # SUPERUSER_API_TOKEN: "" - # SUPERUSER_EMAIL: "" - # SUPERUSER_NAME: "" - # SUPERUSER_PASSWORD: "" - From 7383583c43343b8d1434f396e5197780a21657b6 Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Fri, 25 Apr 2025 14:43:35 +0200 Subject: [PATCH 12/28] Adjusted Gitignore, added config module, adjusted requirements for YAML support, added first unittests --- .gitignore | 2 +- config.yaml | 27 +++++++ modules/config.py | 37 +++++++++ requirements.txt | 1 + tests/__init__.py | 0 tests/test_device_deletion.py | 144 ++++++++++++++++++++++++++++++++++ 6 files changed, 210 insertions(+), 1 deletion(-) create mode 100644 config.yaml create mode 100644 modules/config.py create mode 100644 tests/__init__.py create mode 100644 tests/test_device_deletion.py diff --git a/.gitignore b/.gitignore index c3069c9..bc472c2 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,6 @@ *.log .venv -config.py +/config.py # Byte-compiled / optimized / DLL files __pycache__/ *.py[cod] diff --git a/config.yaml b/config.yaml new file mode 100644 index 0000000..db2f422 --- /dev/null +++ b/config.yaml @@ -0,0 +1,27 @@ +# Required: Custom Field name for Zabbix templates +template_cf: "zabbix_templates" + +# Required: Custom Field name for Zabbix device +device_cf: "zabbix_hostid" + +# Optional: Traverse site groups and assign Zabbix hostgroups based on site groups +traverse_site_groups: false + +# Optional: Traverse regions and assign Zabbix hostgroups based on region hierarchy +traverse_regions: false + +# Optional: Enable inventory syncing for host metadata +inventory_sync: true + +# Optional: Choose which inventory fields to sync ("enabled", "manual", "disabled") +inventory_mode: "manual" + +# Optional: Mapping of NetBox device fields to Zabbix inventory fields +# See: https://www.zabbix.com/documentation/current/en/manual/api/reference/host/object#host_inventory +inventory_map: + serial: "serial" + asset_tag: "asset_tag" + description: "comment" + location: "location" + contact: "contact" + site: "site" \ No newline at end of file diff --git a/modules/config.py b/modules/config.py new file mode 100644 index 0000000..5ee6b5d --- /dev/null +++ b/modules/config.py @@ -0,0 +1,37 @@ +""" +Module for parsing configuration from the top level config.yaml file +""" +from pathlib import Path +import yaml + +DEFAULT_CONFIG = { + "templates_config_context": False, + "templates_config_context_overrule": False, + "template_cf": "zabbix_template", + "device_cf": "zabbix_hostid", + "clustering": False, + "create_hostgroups": True, + "create_journal": False, + "sync_vms": False, + "zabbix_device_removal": ["Decommissioning", "Inventory"], + "zabbix_device_disable": ["Offline", "Planned", "Staged", "Failed"] +} + + +def load_config(config_path="config.yaml"): + """Loads config from YAML file and combines it with default config""" + # Get data from default config. + config = DEFAULT_CONFIG.copy() + # Set config path + config_file = Path(config_path) + # Check if file exists + if config_file.exists(): + try: + with open(config_file, "r", encoding="utf-8") as f: + user_config = yaml.safe_load(f) or {} + config.update(user_config) + except OSError: + # Probably some I/O error with user permissions etc. + # Ignore for now and return default config + pass + return config diff --git a/requirements.txt b/requirements.txt index 33f4b90..832b4b1 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,3 @@ pynetbox zabbix-utils==2.0.1 +pyyaml \ No newline at end of file diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_device_deletion.py b/tests/test_device_deletion.py new file mode 100644 index 0000000..f2c9438 --- /dev/null +++ b/tests/test_device_deletion.py @@ -0,0 +1,144 @@ +"""Testing device creation""" +from unittest.mock import MagicMock, patch, call +from modules.device import PhysicalDevice +from modules.config import load_config + +config = load_config() + + +def mock_nb_device(): + mock = MagicMock() + mock.id = 1 + mock.url = "http://netbox:8000/api/dcim/devices/1/" + mock.display_url = "http://netbox:8000/dcim/devices/1/" + mock.display = "SW01" + mock.name = "SW01" + + mock.device_type = MagicMock() + mock.device_type.id = 1 + mock.device_type.url = "http://netbox:8000/api/dcim/device-types/1/" + mock.device_type.display = "Catalyst 3750G-48TS-S" + mock.device_type.manufacturer = MagicMock() + mock.device_type.manufacturer.id = 1 + mock.device_type.manufacturer.url = "http://netbox:8000/api/dcim/manufacturers/1/" + mock.device_type.manufacturer.display = "Cisco" + mock.device_type.manufacturer.name = "Cisco" + mock.device_type.manufacturer.slug = "cisco" + mock.device_type.manufacturer.description = "" + mock.device_type.model = "Catalyst 3750G-48TS-S" + mock.device_type.slug = "cisco-ws-c3750g-48ts-s" + mock.device_type.description = "" + + mock.role = MagicMock() + mock.role.id = 1 + mock.role.url = "http://netbox:8000/api/dcim/device-roles/1/" + mock.role.display = "Switch" + mock.role.name = "Switch" + mock.role.slug = "switch" + mock.role.description = "" + + mock.tenant = None + mock.platform = None + mock.serial = "0031876" + mock.asset_tag = None + + mock.site = MagicMock() + mock.site.id = 2 + mock.site.url = "http://netbox:8000/api/dcim/sites/2/" + mock.site.display = "AMS01" + mock.site.name = "AMS01" + mock.site.slug = "ams01" + mock.site.description = "" + + mock.location = None + mock.rack = None + mock.position = None + mock.face = None + mock.latitude = None + mock.longitude = None + mock.parent_device = None + + mock.status = MagicMock() + mock.status.value = "decommissioning" + mock.status.label = "Decommissioning" + + mock.cluster = None + mock.virtual_chassis = None + mock.vc_position = None + mock.vc_priority = None + mock.description = "" + mock.comments = "" + mock.config_template = None + mock.config_context = {} + mock.local_context_data = None + mock.tags = [] + + mock.custom_fields = {"zabbix_hostid": 1956} + + def save(self): + pass + + return mock + +def mock_zabbix(): + mock = MagicMock() + mock.host.get.return_value = [{}] + mock.host.delete.return_value = True + + return mock + +netbox_journals = MagicMock() +nb_version = '4.2' +create_journal = MagicMock() +logger = MagicMock() + +def test_check_cluster_status(): + """Checks if the isCluster function is functioning properly""" + nb_device = mock_nb_device() + zabbix = mock_zabbix() + device = PhysicalDevice(nb_device, zabbix, None, None, + None, logger) + assert device.isCluster() == False + + +def test_device_deletion_host_exists(): + """Checks device deletion process""" + nb_device = mock_nb_device() + zabbix = mock_zabbix() + with patch.object(PhysicalDevice, 'create_journal_entry') as mock_journal: + # Create device + device = PhysicalDevice(nb_device, zabbix, netbox_journals, nb_version, + create_journal, logger) + device.cleanup() + # Check if Zabbix HostID is empty + assert device.nb.custom_fields[config["device_cf"]] is None + # Check if API calls are executed + device.zabbix.host.get.assert_called_once_with(filter={'hostid': 1956}, output=[]) + device.zabbix.host.delete.assert_called_once_with(1956) + # check logger + mock_journal.assert_called_once_with("warning", "Deleted host from Zabbix") + device.logger.info.assert_called_once_with("Host SW01: Deleted host from Zabbix.") + + +def test_device_deletion_host_notExists(): + nb_device = mock_nb_device() + zabbix = mock_zabbix() + zabbix.host.get.return_value = None + + with patch.object(PhysicalDevice, 'create_journal_entry') as mock_journal: + # Create new device + device = PhysicalDevice(nb_device, zabbix, netbox_journals, nb_version, + create_journal, logger) + # Try to clean the device up in Zabbix + device.cleanup() + # Confirm that a call was issued to Zabbix to check if the host exists + device.zabbix.host.get.assert_called_once_with(filter={'hostid': 1956}, output=[]) + # Confirm that no device was deleted in Zabbix + device.zabbix.host.delete.assert_not_called() + # Test logging + log_calls = [ + call('Host SW01: Deleted host from Zabbix.'), + call('Host SW01: was already deleted from Zabbix. Removed link in NetBox.') + ] + logger.info.assert_has_calls(log_calls) + assert logger.info.call_count == 2 From cb0500d0c0e398ee2ce8c969dbba8fb29479ffa2 Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 28 Apr 2025 10:47:52 +0200 Subject: [PATCH 13/28] Fixed test layout and added pipeline step to actually run tests --- .github/workflows/run_tests.yml | 22 ++++++++++++++ tests/test_device_deletion.py | 54 ++++++++++++++++----------------- 2 files changed, 48 insertions(+), 28 deletions(-) create mode 100644 .github/workflows/run_tests.yml diff --git a/.github/workflows/run_tests.yml b/.github/workflows/run_tests.yml new file mode 100644 index 0000000..2a35e78 --- /dev/null +++ b/.github/workflows/run_tests.yml @@ -0,0 +1,22 @@ +--- +name: Pytest code testing + +on: + workflow_call + +jobs: + test_code: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - name: Set up Python + uses: actions/setup-python@v5 + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install pytest pytest-mock + pip install -r requirements.txt + - name: Analysing the code with pylint + run: | + cp config.py.example config.py + pytest tests diff --git a/tests/test_device_deletion.py b/tests/test_device_deletion.py index f2c9438..41c4420 100644 --- a/tests/test_device_deletion.py +++ b/tests/test_device_deletion.py @@ -7,6 +7,7 @@ config = load_config() def mock_nb_device(): + """Function to mock Netbox device""" mock = MagicMock() mock.id = 1 mock.url = "http://netbox:8000/api/dcim/devices/1/" @@ -15,27 +16,20 @@ def mock_nb_device(): mock.name = "SW01" mock.device_type = MagicMock() - mock.device_type.id = 1 - mock.device_type.url = "http://netbox:8000/api/dcim/device-types/1/" mock.device_type.display = "Catalyst 3750G-48TS-S" mock.device_type.manufacturer = MagicMock() - mock.device_type.manufacturer.id = 1 - mock.device_type.manufacturer.url = "http://netbox:8000/api/dcim/manufacturers/1/" mock.device_type.manufacturer.display = "Cisco" mock.device_type.manufacturer.name = "Cisco" mock.device_type.manufacturer.slug = "cisco" mock.device_type.manufacturer.description = "" mock.device_type.model = "Catalyst 3750G-48TS-S" mock.device_type.slug = "cisco-ws-c3750g-48ts-s" - mock.device_type.description = "" mock.role = MagicMock() mock.role.id = 1 - mock.role.url = "http://netbox:8000/api/dcim/device-roles/1/" mock.role.display = "Switch" mock.role.name = "Switch" mock.role.slug = "switch" - mock.role.description = "" mock.tenant = None mock.platform = None @@ -43,19 +37,14 @@ def mock_nb_device(): mock.asset_tag = None mock.site = MagicMock() - mock.site.id = 2 - mock.site.url = "http://netbox:8000/api/dcim/sites/2/" mock.site.display = "AMS01" mock.site.name = "AMS01" mock.site.slug = "ams01" - mock.site.description = "" mock.location = None mock.rack = None mock.position = None mock.face = None - mock.latitude = None - mock.longitude = None mock.parent_device = None mock.status = MagicMock() @@ -71,34 +60,32 @@ def mock_nb_device(): mock.config_template = None mock.config_context = {} mock.local_context_data = None - mock.tags = [] mock.custom_fields = {"zabbix_hostid": 1956} - - def save(self): - pass - return mock + def mock_zabbix(): + """Function to mock Zabbix""" mock = MagicMock() mock.host.get.return_value = [{}] mock.host.delete.return_value = True - return mock + netbox_journals = MagicMock() -nb_version = '4.2' +NB_VERSION = '4.2' create_journal = MagicMock() logger = MagicMock() + def test_check_cluster_status(): """Checks if the isCluster function is functioning properly""" nb_device = mock_nb_device() zabbix = mock_zabbix() device = PhysicalDevice(nb_device, zabbix, None, None, None, logger) - assert device.isCluster() == False + assert device.isCluster() is False def test_device_deletion_host_exists(): @@ -107,38 +94,49 @@ def test_device_deletion_host_exists(): zabbix = mock_zabbix() with patch.object(PhysicalDevice, 'create_journal_entry') as mock_journal: # Create device - device = PhysicalDevice(nb_device, zabbix, netbox_journals, nb_version, + device = PhysicalDevice(nb_device, zabbix, netbox_journals, NB_VERSION, create_journal, logger) device.cleanup() # Check if Zabbix HostID is empty assert device.nb.custom_fields[config["device_cf"]] is None # Check if API calls are executed - device.zabbix.host.get.assert_called_once_with(filter={'hostid': 1956}, output=[]) + device.zabbix.host.get.assert_called_once_with(filter={'hostid': 1956}, + output=[]) device.zabbix.host.delete.assert_called_once_with(1956) # check logger - mock_journal.assert_called_once_with("warning", "Deleted host from Zabbix") - device.logger.info.assert_called_once_with("Host SW01: Deleted host from Zabbix.") + mock_journal.assert_called_once_with("warning", + "Deleted host from Zabbix") + device.logger.info.assert_called_once_with("Host SW01: Deleted " + "host from Zabbix.") -def test_device_deletion_host_notExists(): +def test_device_deletion_host_not_exists(): + """ + Test if device in Netbox gets unlinked + when host is not present in Zabbix + """ nb_device = mock_nb_device() zabbix = mock_zabbix() zabbix.host.get.return_value = None with patch.object(PhysicalDevice, 'create_journal_entry') as mock_journal: # Create new device - device = PhysicalDevice(nb_device, zabbix, netbox_journals, nb_version, + device = PhysicalDevice(nb_device, zabbix, netbox_journals, NB_VERSION, create_journal, logger) # Try to clean the device up in Zabbix device.cleanup() # Confirm that a call was issued to Zabbix to check if the host exists - device.zabbix.host.get.assert_called_once_with(filter={'hostid': 1956}, output=[]) + device.zabbix.host.get.assert_called_once_with(filter={'hostid': 1956}, + output=[]) # Confirm that no device was deleted in Zabbix device.zabbix.host.delete.assert_not_called() # Test logging log_calls = [ call('Host SW01: Deleted host from Zabbix.'), - call('Host SW01: was already deleted from Zabbix. Removed link in NetBox.') + call('Host SW01: was already deleted from Zabbix. ' + 'Removed link in NetBox.') ] logger.info.assert_has_calls(log_calls) assert logger.info.call_count == 2 + mock_journal.assert_called_once_with("warning", + "Deleted host from Zabbix") From 5fd89a1f8a3e75edb68938b8e5519db3731c17af Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 28 Apr 2025 13:32:28 +0200 Subject: [PATCH 14/28] Added .vscode as exception to gitignore --- .gitignore | 1 + 1 file changed, 1 insertion(+) diff --git a/.gitignore b/.gitignore index bc472c2..5fdbd95 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,4 @@ # Byte-compiled / optimized / DLL files __pycache__/ *.py[cod] +.vscode \ No newline at end of file From eb307337f68150a155cc52a94216aa15a637aee9 Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 28 Apr 2025 14:50:52 +0200 Subject: [PATCH 15/28] Removed YAML config logic, added python config logic with default fallback. Added ENV variable support for config parameters. --- config.yaml | 27 ------------ modules/config.py | 61 +++++++++++++++++++-------- modules/device.py | 84 +++++++++++++++++--------------------- modules/virtual_machine.py | 24 ++++------- netbox_zabbix_sync.py | 63 +++++++++++----------------- requirements.txt | 3 +- 6 files changed, 114 insertions(+), 148 deletions(-) delete mode 100644 config.yaml diff --git a/config.yaml b/config.yaml deleted file mode 100644 index db2f422..0000000 --- a/config.yaml +++ /dev/null @@ -1,27 +0,0 @@ -# Required: Custom Field name for Zabbix templates -template_cf: "zabbix_templates" - -# Required: Custom Field name for Zabbix device -device_cf: "zabbix_hostid" - -# Optional: Traverse site groups and assign Zabbix hostgroups based on site groups -traverse_site_groups: false - -# Optional: Traverse regions and assign Zabbix hostgroups based on region hierarchy -traverse_regions: false - -# Optional: Enable inventory syncing for host metadata -inventory_sync: true - -# Optional: Choose which inventory fields to sync ("enabled", "manual", "disabled") -inventory_mode: "manual" - -# Optional: Mapping of NetBox device fields to Zabbix inventory fields -# See: https://www.zabbix.com/documentation/current/en/manual/api/reference/host/object#host_inventory -inventory_map: - serial: "serial" - asset_tag: "asset_tag" - description: "comment" - location: "location" - contact: "contact" - site: "site" \ No newline at end of file diff --git a/modules/config.py b/modules/config.py index 5ee6b5d..3adda8b 100644 --- a/modules/config.py +++ b/modules/config.py @@ -2,7 +2,9 @@ Module for parsing configuration from the top level config.yaml file """ from pathlib import Path -import yaml +from importlib import util +from os import environ +from logging import getLogger DEFAULT_CONFIG = { "templates_config_context": False, @@ -18,20 +20,43 @@ DEFAULT_CONFIG = { } -def load_config(config_path="config.yaml"): - """Loads config from YAML file and combines it with default config""" - # Get data from default config. - config = DEFAULT_CONFIG.copy() - # Set config path - config_file = Path(config_path) - # Check if file exists - if config_file.exists(): - try: - with open(config_file, "r", encoding="utf-8") as f: - user_config = yaml.safe_load(f) or {} - config.update(user_config) - except OSError: - # Probably some I/O error with user permissions etc. - # Ignore for now and return default config - pass - return config +def load_config(): + """Returns combined config from all sources""" + # Overwrite default config with config.py + conf = load_config_file(config_default=DEFAULT_CONFIG) + # Overwrite default config and config.py with environment variables + for key in conf: + value_setting = load_env_variable(key) + if value_setting is not None: + conf[key] = value_setting + return conf + + +def load_env_variable(config_environvar): + """Returns config from environment variable""" + if config_environvar in environ: + return environ[config_environvar] + return None + + +def load_config_file(config_default, config_file="config.py"): + """Returns config from config.py file""" + # Check if config.py exists and load it + # If it does not exist, return the default config + config_path = Path(config_file) + if config_path.exists(): + dconf = config_default.copy() + # Dynamically import the config module + spec = util.spec_from_file_location("config", config_path) + config_module = util.module_from_spec(spec) + spec.loader.exec_module(config_module) + # Update DEFAULT_CONFIG with variables from the config module + for key in dconf: + if hasattr(config_module, key): + dconf[key] = getattr(config_module, key) + return dconf + else: + getLogger(__name__).warning( + "Config file %s not found. Using default config " + "and environment variables.", config_file) + return None diff --git a/modules/device.py b/modules/device.py index 2ed37e8..5d11c82 100644 --- a/modules/device.py +++ b/modules/device.py @@ -3,7 +3,6 @@ """ Device specific handeling for NetBox to Zabbix """ -from os import sys from re import search from logging import getLogger from zabbix_utils import APIRequestError @@ -11,19 +10,10 @@ from modules.exceptions import (SyncInventoryError, TemplateError, SyncExternalE InterfaceConfigError, JournalError) from modules.interface import ZabbixInterface from modules.hostgroups import Hostgroup -try: - from config import ( - template_cf, device_cf, - traverse_site_groups, - traverse_regions, - inventory_sync, - inventory_mode, - inventory_map - ) -except ModuleNotFoundError: - print("Configuration file config.py not found in main directory." - "Please create the file or rename the config.py.example file to config.py.") - sys.exit(0) +from modules.config import load_config + +config = load_config() + class PhysicalDevice(): # pylint: disable=too-many-instance-attributes, too-many-arguments, too-many-positional-arguments @@ -76,10 +66,10 @@ class PhysicalDevice(): raise SyncInventoryError(e) # Check if device has custom field for ZBX ID - if device_cf in self.nb.custom_fields: - self.zabbix_id = self.nb.custom_fields[device_cf] + if config["device_cf"] in self.nb.custom_fields: + self.zabbix_id = self.nb.custom_fields[config["device_cf"]] else: - e = f"Host {self.name}: Custom field {device_cf} not present" + e = f"Host {self.name}: Custom field {config["device_cf"]} not present" self.logger.warning(e) raise SyncInventoryError(e) @@ -87,7 +77,7 @@ class PhysicalDevice(): odd_character_list = ["ä", "ö", "ü", "Ä", "Ö", "Ü", "ß"] self.use_visible_name = False if (any(letter in self.name for letter in odd_character_list) or - bool(search('[\u0400-\u04FF]', self.name))): + bool(search('[\u0400-\u04FF]', self.name))): self.name = f"NETBOX_ID{self.id}" self.visible_name = self.nb.name self.use_visible_name = True @@ -101,8 +91,8 @@ class PhysicalDevice(): """Set the hostgroup for this device""" # Create new Hostgroup instance hg = Hostgroup("dev", self.nb, self.nb_api_version, logger=self.logger, - nested_sitegroup_flag=traverse_site_groups, - nested_region_flag=traverse_regions, + nested_sitegroup_flag=config["traverse_site_groups"], + nested_region_flag=config["traverse_regions"], nb_groups=nb_site_groups, nb_regions=nb_regions) # Generate hostgroup based on hostgroup format @@ -137,13 +127,13 @@ class PhysicalDevice(): # Get Zabbix templates from the device type device_type_cfs = self.nb.device_type.custom_fields # Check if the ZBX Template CF is present - if template_cf in device_type_cfs: + if config["template_cf"] in device_type_cfs: # Set value to template - return [device_type_cfs[template_cf]] + return [device_type_cfs[config["template_cf"]]] # Custom field not found, return error - e = (f"Custom field {template_cf} not " - f"found for {self.nb.device_type.manufacturer.name}" - f" - {self.nb.device_type.display}.") + e = (f"Custom field {config["template_cf"]} not " + f"found for {self.nb.device_type.manufacturer.name}" + f" - {self.nb.device_type.display}.") raise TemplateError(e) def get_templates_context(self): @@ -164,25 +154,25 @@ class PhysicalDevice(): def set_inventory(self, nbdevice): """ Set host inventory """ # Set inventory mode. Default is disabled (see class init function). - if inventory_mode == "disabled": - if inventory_sync: + if config["inventory_mode"] == "disabled": + if config["inventory_sync"]: self.logger.error(f"Host {self.name}: Unable to map NetBox inventory to Zabbix. " - "Inventory sync is enabled in config but inventory mode is disabled.") + "Inventory sync is enabled in config but inventory mode is disabled.") return True - if inventory_mode == "manual": + if config["inventory_mode"] == "manual": self.inventory_mode = 0 - elif inventory_mode == "automatic": + elif config["inventory_mode"] == "automatic": self.inventory_mode = 1 else: self.logger.error(f"Host {self.name}: Specified value for inventory mode in" - f" config is not valid. Got value {inventory_mode}") + f" config is not valid. Got value {config["inventory_mode"]}") return False self.inventory = {} - if inventory_sync and self.inventory_mode in [0,1]: + if config["inventory_sync"] and self.inventory_mode in [0, 1]: self.logger.debug(f"Host {self.name}: Starting inventory mapper") # Let's build an inventory dict for each property in the inventory_map - for nb_inv_field, zbx_inv_field in inventory_map.items(): - field_list = nb_inv_field.split("/") # convert str to list based on delimiter + for nb_inv_field, zbx_inv_field in config["inventory_map"].items(): + field_list = nb_inv_field.split("/") # convert str to list based on delimiter # start at the base of the dict... value = nbdevice # ... and step through the dict till we find the needed value @@ -191,8 +181,8 @@ class PhysicalDevice(): # Check if the result is usable and expected # We want to apply any int or float 0 values, # even if python thinks those are empty. - if ((value and isinstance(value, int | float | str )) or - (isinstance(value, int | float) and int(value) ==0)): + if ((value and isinstance(value, int | float | str)) or + (isinstance(value, int | float) and int(value) == 0)): self.inventory[zbx_inv_field] = str(value) elif not value: # empty value should just be an empty string for API compatibility @@ -204,7 +194,7 @@ class PhysicalDevice(): self.logger.error(f"Host {self.name}: Inventory lookup for '{nb_inv_field}'" " returned an unexpected type: it will be skipped.") self.logger.debug(f"Host {self.name}: Inventory mapping complete. " - f"Mapped {len(list(filter(None, self.inventory.values())))} field(s)") + f"Mapped {len(list(filter(None, self.inventory.values())))} field(s)") return True def isCluster(self): @@ -275,7 +265,7 @@ class PhysicalDevice(): # Return error should the template not be found in Zabbix if not template_match: e = (f"Unable to find template {nb_template} " - f"for host {self.name} in Zabbix. Skipping host...") + f"for host {self.name} in Zabbix. Skipping host...") self.logger.warning(e) raise SyncInventoryError(e) @@ -305,7 +295,7 @@ class PhysicalDevice(): zbx_host = bool(self.zabbix.host.get(filter={'hostid': self.zabbix_id}, output=[])) e = (f"Host {self.name}: was already deleted from Zabbix." - " Removed link in NetBox.") + " Removed link in NetBox.") if zbx_host: # Delete host should it exists self.zabbix.host.delete(self.zabbix_id) @@ -321,7 +311,7 @@ class PhysicalDevice(): def _zeroize_cf(self): """Sets the hostID custom field in NetBox to zero, effectively destroying the link""" - self.nb.custom_fields[device_cf] = None + self.nb.custom_fields[config["device_cf"]] = None self.nb.save() def _zabbixHostnameExists(self): @@ -366,7 +356,7 @@ class PhysicalDevice(): input: List of all proxies and proxy groups in standardized format """ # check if the key Zabbix is defined in the config context - if not "zabbix" in self.nb.config_context: + if "zabbix" not in self.nb.config_context: return False if ("proxy" in self.nb.config_context["zabbix"] and not self.nb.config_context["zabbix"]["proxy"]): @@ -448,7 +438,7 @@ class PhysicalDevice(): self.logger.error(e) raise SyncExternalError(e) from None # Set NetBox custom field to hostID value. - self.nb.custom_fields[device_cf] = int(self.zabbix_id) + self.nb.custom_fields[config["device_cf"]] = int(self.zabbix_id) self.nb.save() msg = f"Host {self.name}: Created host in Zabbix." self.logger.info(msg) @@ -542,7 +532,7 @@ class PhysicalDevice(): selectGroups=["groupid"], selectHostGroups=["groupid"], selectParentTemplates=["templateid"], - selectInventory=list(inventory_map.values())) + selectInventory=list(config["inventory_map"].values())) if len(host) > 1: e = (f"Got {len(host)} results for Zabbix hosts " f"with ID {self.zabbix_id} - hostname {self.name}.") @@ -645,9 +635,9 @@ class PhysicalDevice(): if proxy_set and not proxy_power: # Display error message self.logger.error(f"Host {self.name} is configured " - f"with proxy in Zabbix but not in NetBox. The" - " -p flag was ommited: no " - "changes have been made.") + f"with proxy in Zabbix but not in NetBox. The" + " -p flag was ommited: no " + "changes have been made.") if not proxy_set: self.logger.debug(f"Host {self.name}: proxy in-sync.") # Check host inventory mode @@ -656,7 +646,7 @@ class PhysicalDevice(): else: self.logger.warning(f"Host {self.name}: inventory_mode OUT of sync.") self.updateZabbixHost(inventory_mode=str(self.inventory_mode)) - if inventory_sync and self.inventory_mode in [0,1]: + if config["inventory_sync"] and self.inventory_mode in [0,1]: # Check host inventory mapping if host['inventory'] == self.inventory: self.logger.debug(f"Host {self.name}: inventory in-sync.") diff --git a/modules/virtual_machine.py b/modules/virtual_machine.py index 331a463..80dadc0 100644 --- a/modules/virtual_machine.py +++ b/modules/virtual_machine.py @@ -1,21 +1,15 @@ #!/usr/bin/env python3 # pylint: disable=duplicate-code """Module that hosts all functions for virtual machine processing""" - -from os import sys from modules.device import PhysicalDevice from modules.hostgroups import Hostgroup from modules.interface import ZabbixInterface -from modules.exceptions import TemplateError, InterfaceConfigError, SyncInventoryError -try: - from config import ( - traverse_site_groups, - traverse_regions - ) -except ModuleNotFoundError: - print("Configuration file config.py not found in main directory." - "Please create the file or rename the config.py.example file to config.py.") - sys.exit(0) +from modules.exceptions import (TemplateError, InterfaceConfigError, + SyncInventoryError) +from modules.config import load_config +# Load config +config = load_config() + class VirtualMachine(PhysicalDevice): """Model for virtual machines""" @@ -28,8 +22,8 @@ class VirtualMachine(PhysicalDevice): """Set the hostgroup for this device""" # Create new Hostgroup instance hg = Hostgroup("vm", self.nb, self.nb_api_version, logger=self.logger, - nested_sitegroup_flag=traverse_site_groups, - nested_region_flag=traverse_regions, + nested_sitegroup_flag=config["traverse_site_groups"], + nested_region_flag=config["traverse_regions"], nb_groups=nb_site_groups, nb_regions=nb_regions) # Generate hostgroup based on hostgroup format @@ -45,7 +39,7 @@ class VirtualMachine(PhysicalDevice): self.logger.warning(e) return True - def setInterfaceDetails(self): # pylint: disable=invalid-name + def setInterfaceDetails(self): # pylint: disable=invalid-name """ Overwrites device function to select an agent interface type by default Agent type interfaces are more likely to be used with VMs then SNMP diff --git a/netbox_zabbix_sync.py b/netbox_zabbix_sync.py index 935b55e..6129f92 100755 --- a/netbox_zabbix_sync.py +++ b/netbox_zabbix_sync.py @@ -10,28 +10,13 @@ from pynetbox import api from pynetbox.core.query import RequestError as NBRequestError from requests.exceptions import ConnectionError as RequestsConnectionError from zabbix_utils import ZabbixAPI, APIRequestError, ProcessingError +from modules.config import load_config from modules.device import PhysicalDevice from modules.virtual_machine import VirtualMachine from modules.tools import convert_recordset, proxy_prepper from modules.exceptions import EnvironmentVarError, HostgroupError, SyncError -try: - from config import ( - templates_config_context, - templates_config_context_overrule, - clustering, create_hostgroups, - create_journal, full_proxy_sync, - zabbix_device_removal, - zabbix_device_disable, - hostgroup_format, - vm_hostgroup_format, - nb_device_filter, - sync_vms, - nb_vm_filter - ) -except ModuleNotFoundError: - print("Configuration file config.py not found in main directory." - "Please create the file or rename the config.py.example file to config.py.") - sys.exit(1) + +config = load_config() # Set logging log_format = logging.Formatter('%(asctime)s - %(name)s - ' @@ -83,7 +68,7 @@ def main(arguments): # Set NetBox API netbox = api(netbox_host, token=netbox_token, threading=True) # Check if the provided Hostgroup layout is valid - hg_objects = hostgroup_format.split("/") + hg_objects = config["hostgroup_format"].split("/") allowed_objects = ["location", "role", "manufacturer", "region", "site", "site_group", "tenant", "tenant_group"] # Create API call to get all custom fields which are on the device objects @@ -130,11 +115,11 @@ def main(arguments): else: proxy_name = "name" # Get all Zabbix and NetBox data - netbox_devices = list(netbox.dcim.devices.filter(**nb_device_filter)) + netbox_devices = list(netbox.dcim.devices.filter(**config["nb_device_filter"])) netbox_vms = [] - if sync_vms: + if config["sync_vms"]: netbox_vms = list( - netbox.virtualization.virtual_machines.filter(**nb_vm_filter)) + netbox.virtualization.virtual_machines.filter(**config["nb_vm_filter"])) netbox_site_groups = convert_recordset((netbox.dcim.site_groups.all())) netbox_regions = convert_recordset(netbox.dcim.regions.all()) netbox_journals = netbox.extras.journal_entries @@ -160,19 +145,19 @@ def main(arguments): for nb_vm in netbox_vms: try: vm = VirtualMachine(nb_vm, zabbix, netbox_journals, nb_version, - create_journal, logger) + config["create_journal"], logger) logger.debug(f"Host {vm.name}: started operations on VM.") vm.set_vm_template() # Check if a valid template has been found for this VM. if not vm.zbx_template_names: continue - vm.set_hostgroup(vm_hostgroup_format, + vm.set_hostgroup(config["vm_hostgroup_format"], netbox_site_groups, netbox_regions) # Check if a valid hostgroup has been found for this VM. if not vm.hostgroup: continue # Checks if device is in cleanup state - if vm.status in zabbix_device_removal: + if vm.status in config["zabbix_device_removal"]: if vm.zabbix_id: # Delete device from Zabbix # and remove hostID from NetBox. @@ -185,16 +170,16 @@ def main(arguments): f"not in the active state.") continue # Check if the VM is in the disabled state - if vm.status in zabbix_device_disable: + if vm.status in config["zabbix_device_disable"]: vm.zabbix_state = 1 # Check if VM is already in Zabbix if vm.zabbix_id: vm.ConsistencyCheck(zabbix_groups, zabbix_templates, - zabbix_proxy_list, full_proxy_sync, - create_hostgroups) + zabbix_proxy_list, config["full_proxy_sync"], + config["create_hostgroups"]) continue # Add hostgroup is config is set - if create_hostgroups: + if config["create_hostgroups"]: # Create new hostgroup. Potentially multiple groups if nested hostgroups = vm.createZabbixHostgroup(zabbix_groups) # go through all newly created hostgroups @@ -211,22 +196,22 @@ def main(arguments): try: # Set device instance set data such as hostgroup and template information. device = PhysicalDevice(nb_device, zabbix, netbox_journals, nb_version, - create_journal, logger) + config["create_journal"], logger) logger.debug(f"Host {device.name}: started operations on device.") - device.set_template(templates_config_context, - templates_config_context_overrule) + device.set_template(config["templates_config_context"], + config["templates_config_context_overrule"]) # Check if a valid template has been found for this VM. if not device.zbx_template_names: continue device.set_hostgroup( - hostgroup_format, netbox_site_groups, netbox_regions) + config["hostgroup_format"], netbox_site_groups, netbox_regions) # Check if a valid hostgroup has been found for this VM. if not device.hostgroup: continue device.set_inventory(nb_device) # Checks if device is part of cluster. # Requires clustering variable - if device.isCluster() and clustering: + if device.isCluster() and config["clustering"]: # Check if device is primary or secondary if device.promoteMasterDevice(): e = (f"Device {device.name}: is " @@ -240,7 +225,7 @@ def main(arguments): logger.info(e) continue # Checks if device is in cleanup state - if device.status in zabbix_device_removal: + if device.status in config["zabbix_device_removal"]: if device.zabbix_id: # Delete device from Zabbix # and remove hostID from NetBox. @@ -253,16 +238,16 @@ def main(arguments): f"not in the active state.") continue # Check if the device is in the disabled state - if device.status in zabbix_device_disable: + if device.status in config["zabbix_device_disable"]: device.zabbix_state = 1 # Check if device is already in Zabbix if device.zabbix_id: device.ConsistencyCheck(zabbix_groups, zabbix_templates, - zabbix_proxy_list, full_proxy_sync, - create_hostgroups) + zabbix_proxy_list, config["full_proxy_sync"], + config["create_hostgroups"]) continue # Add hostgroup is config is set - if create_hostgroups: + if config["create_hostgroups"]: # Create new hostgroup. Potentially multiple groups if nested hostgroups = device.createZabbixHostgroup(zabbix_groups) # go through all newly created hostgroups diff --git a/requirements.txt b/requirements.txt index 832b4b1..8da5ce5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,2 @@ pynetbox -zabbix-utils==2.0.1 -pyyaml \ No newline at end of file +zabbix-utils==2.0.1 \ No newline at end of file From e91eecffaabf970234fad3b471f7efa66232e7f6 Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 28 Apr 2025 14:58:38 +0200 Subject: [PATCH 16/28] Fixed on statement on new testcode. --- .github/workflows/run_tests.yml | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/.github/workflows/run_tests.yml b/.github/workflows/run_tests.yml index 2a35e78..cae8b02 100644 --- a/.github/workflows/run_tests.yml +++ b/.github/workflows/run_tests.yml @@ -1,8 +1,9 @@ --- name: Pytest code testing -on: - workflow_call +on: + push: + pull_request: jobs: test_code: From 04a610cf8411aafc477e55043a7aae39c05a9204 Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 28 Apr 2025 15:10:48 +0200 Subject: [PATCH 17/28] Fixed some minor Flake8 errors --- modules/device.py | 36 +++++++++++++++++++----------------- 1 file changed, 19 insertions(+), 17 deletions(-) diff --git a/modules/device.py b/modules/device.py index 5d11c82..d410103 100644 --- a/modules/device.py +++ b/modules/device.py @@ -157,7 +157,8 @@ class PhysicalDevice(): if config["inventory_mode"] == "disabled": if config["inventory_sync"]: self.logger.error(f"Host {self.name}: Unable to map NetBox inventory to Zabbix. " - "Inventory sync is enabled in config but inventory mode is disabled.") + "Inventory sync is enabled in " + "config but inventory mode is disabled.") return True if config["inventory_mode"] == "manual": self.inventory_mode = 0 @@ -359,7 +360,7 @@ class PhysicalDevice(): if "zabbix" not in self.nb.config_context: return False if ("proxy" in self.nb.config_context["zabbix"] and - not self.nb.config_context["zabbix"]["proxy"]): + not self.nb.config_context["zabbix"]["proxy"]): return False # Proxy group takes priority over a proxy due # to it being HA and therefore being more reliable @@ -409,16 +410,17 @@ class PhysicalDevice(): # Set Zabbix proxy if defined self.setProxy(proxies) # Set basic data for host creation - create_data = {"host": self.name, - "name": self.visible_name, - "status": self.zabbix_state, - "interfaces": interfaces, - "groups": groups, - "templates": templateids, - "description": description, - "inventory_mode": self.inventory_mode, - "inventory": self.inventory - } + create_data = { + "host": self.name, + "name": self.visible_name, + "status": self.zabbix_state, + "interfaces": interfaces, + "groups": groups, + "templates": templateids, + "description": description, + "inventory_mode": self.inventory_mode, + "inventory": self.inventory + } # If a Zabbix proxy or Zabbix Proxy group has been defined if self.zbxproxy: # If a lower version than 7 is used, we can assume that @@ -518,7 +520,7 @@ class PhysicalDevice(): # Function returns true / false but also sets GroupID if not self.setZabbixGroupID(groups) and not create_hostgroups: e = (f"Host {self.name}: different hostgroup is required but " - "unable to create hostgroup without generation permission.") + "unable to create hostgroup without generation permission.") self.logger.warning(e) raise SyncInventoryError(e) # Prepare templates and proxy config @@ -569,7 +571,7 @@ class PhysicalDevice(): templateids.append({'templateid': template['templateid']}) # Update Zabbix with NB templates and clear any old / lost templates self.updateZabbixHost(templates_clear=host["parentTemplates"], - templates=templateids) + templates=templateids) else: self.logger.debug(f"Host {self.name}: template(s) in-sync.") @@ -594,7 +596,7 @@ class PhysicalDevice(): if self.zbxproxy: # Check if proxy or proxy group is defined if (self.zbxproxy["idtype"] in host and - host[self.zbxproxy["idtype"]] == self.zbxproxy["id"]): + host[self.zbxproxy["idtype"]] == self.zbxproxy["id"]): self.logger.debug(f"Host {self.name}: proxy in-sync.") # Backwards compatibility for Zabbix <= 6 elif "proxy_hostid" in host and host["proxy_hostid"] == self.zbxproxy["id"]: @@ -646,7 +648,7 @@ class PhysicalDevice(): else: self.logger.warning(f"Host {self.name}: inventory_mode OUT of sync.") self.updateZabbixHost(inventory_mode=str(self.inventory_mode)) - if config["inventory_sync"] and self.inventory_mode in [0,1]: + if config["inventory_sync"] and self.inventory_mode in [0, 1]: # Check host inventory mapping if host['inventory'] == self.inventory: self.logger.debug(f"Host {self.name}: inventory in-sync.") @@ -664,7 +666,7 @@ class PhysicalDevice(): if key in host["interfaces"][0]: # If SNMP is used, go through nested dict # to compare SNMP parameters - if isinstance(item,dict) and key == "details": + if isinstance(item, dict) and key == "details": for k, i in item.items(): if k in host["interfaces"][0][key]: # Set update if values don't match From 819126ce36b99fd119677bfe1e8c3c22942bfe30 Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 28 Apr 2025 15:35:51 +0200 Subject: [PATCH 18/28] Added tests for config file, added logger for config file --- modules/config.py | 4 +- tests/test_configuration_parsing.py | 130 ++++++++++++++++++++++++++++ 2 files changed, 133 insertions(+), 1 deletion(-) create mode 100644 tests/test_configuration_parsing.py diff --git a/modules/config.py b/modules/config.py index 3adda8b..0919b01 100644 --- a/modules/config.py +++ b/modules/config.py @@ -6,6 +6,8 @@ from importlib import util from os import environ from logging import getLogger +logger = getLogger(__name__) + DEFAULT_CONFIG = { "templates_config_context": False, "templates_config_context_overrule": False, @@ -56,7 +58,7 @@ def load_config_file(config_default, config_file="config.py"): dconf[key] = getattr(config_module, key) return dconf else: - getLogger(__name__).warning( + logger.warning( "Config file %s not found. Using default config " "and environment variables.", config_file) return None diff --git a/tests/test_configuration_parsing.py b/tests/test_configuration_parsing.py new file mode 100644 index 0000000..070d3dd --- /dev/null +++ b/tests/test_configuration_parsing.py @@ -0,0 +1,130 @@ +"""Tests for configuration parsing in the modules.config module.""" +from unittest.mock import patch, MagicMock +from pathlib import Path +import os +from modules.config import load_config, DEFAULT_CONFIG, load_config_file, load_env_variable + + +def test_load_config_defaults(): + """Test that load_config returns default values when no config file or env vars are present""" + with patch('modules.config.load_config_file', return_value=DEFAULT_CONFIG.copy()), \ + patch('modules.config.load_env_variable', return_value=None): + config = load_config() + assert config == DEFAULT_CONFIG + assert config["templates_config_context"] is False + assert config["create_hostgroups"] is True + + +def test_load_config_file(): + """Test that load_config properly loads values from config file""" + mock_config = DEFAULT_CONFIG.copy() + mock_config["templates_config_context"] = True + mock_config["sync_vms"] = True + + with patch('modules.config.load_config_file', return_value=mock_config), \ + patch('modules.config.load_env_variable', return_value=None): + config = load_config() + assert config["templates_config_context"] is True + assert config["sync_vms"] is True + # Unchanged values should remain as defaults + assert config["create_journal"] is False + + +def test_load_env_variables(): + """Test that load_config properly loads values from environment variables""" + # Mock env variable loading to return values for specific keys + def mock_load_env(key): + if key == "sync_vms": + return True + if key == "create_journal": + return True + return None + + with patch('modules.config.load_config_file', return_value=DEFAULT_CONFIG.copy()), \ + patch('modules.config.load_env_variable', side_effect=mock_load_env): + config = load_config() + assert config["sync_vms"] is True + assert config["create_journal"] is True + # Unchanged values should remain as defaults + assert config["templates_config_context"] is False + + +def test_env_vars_override_config_file(): + """Test that environment variables override values from config file""" + mock_config = DEFAULT_CONFIG.copy() + mock_config["templates_config_context"] = True + mock_config["sync_vms"] = False + + # Mock env variable that will override the config file value + def mock_load_env(key): + if key == "sync_vms": + return True + return None + + with patch('modules.config.load_config_file', return_value=mock_config), \ + patch('modules.config.load_env_variable', side_effect=mock_load_env): + config = load_config() + # This should be overridden by the env var + assert config["sync_vms"] is True + # This should remain from the config file + assert config["templates_config_context"] is True + + +def test_load_config_file_function(): + """Test the load_config_file function directly""" + # Test when the file exists + with patch('pathlib.Path.exists', return_value=True), \ + patch('importlib.util.spec_from_file_location') as mock_spec: + # Setup the mock module with attributes + mock_module = MagicMock() + mock_module.templates_config_context = True + mock_module.sync_vms = True + + # Setup the mock spec + mock_spec_instance = MagicMock() + mock_spec.return_value = mock_spec_instance + mock_spec_instance.loader.exec_module = lambda x: None + + # Patch module_from_spec to return our mock module + with patch('importlib.util.module_from_spec', return_value=mock_module): + config = load_config_file(DEFAULT_CONFIG.copy()) + assert config["templates_config_context"] is True + assert config["sync_vms"] is True + + +def test_load_config_file_not_found(): + """Test load_config_file when the config file doesn't exist""" + # Instead of trying to assert on the logger call, we'll just check the return value + # and verify the function works as expected in this case + with patch('pathlib.Path.exists', return_value=False): + result = load_config_file(DEFAULT_CONFIG.copy()) + assert result is None + + +def test_load_env_variable_function(): + """Test the load_env_variable function directly""" + # Test when the environment variable exists + with patch.dict(os.environ, {"templates_config_context": "True"}): + value = load_env_variable("templates_config_context") + assert value == "True" + + # Test when the environment variable doesn't exist + with patch.dict(os.environ, {}, clear=True): + value = load_env_variable("nonexistent_variable") + assert value is None + + +def test_load_config_file_exception_handling(): + """Test that load_config_file handles exceptions gracefully""" + # This test requires modifying the load_config_file function to handle exceptions + # For now, we're just checking that an exception is raised + with patch('pathlib.Path.exists', return_value=True), \ + patch('importlib.util.spec_from_file_location', side_effect=Exception("Import error")): + # Since the current implementation doesn't handle exceptions, we should + # expect an exception to be raised + try: + result = load_config_file(DEFAULT_CONFIG.copy()) + assert False, "An exception should have been raised" + except Exception: + # This is expected + pass From 0c715d4f9647292fa257959a1af64113210204e5 Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 28 Apr 2025 15:44:45 +0200 Subject: [PATCH 19/28] Fixed some basic Flake8 errors, added Pylinter exception, Fixed some minor logging bugs. --- modules/config.py | 7 +++---- modules/device.py | 10 ++++++---- tests/test_configuration_parsing.py | 21 ++++++++++----------- 3 files changed, 19 insertions(+), 19 deletions(-) diff --git a/modules/config.py b/modules/config.py index 0919b01..660bfec 100644 --- a/modules/config.py +++ b/modules/config.py @@ -57,8 +57,7 @@ def load_config_file(config_default, config_file="config.py"): if hasattr(config_module, key): dconf[key] = getattr(config_module, key) return dconf - else: - logger.warning( - "Config file %s not found. Using default config " - "and environment variables.", config_file) + logger.warning( + "Config file %s not found. Using default config " + "and environment variables.", config_file) return None diff --git a/modules/device.py b/modules/device.py index d410103..aa15a06 100644 --- a/modules/device.py +++ b/modules/device.py @@ -69,7 +69,7 @@ class PhysicalDevice(): if config["device_cf"] in self.nb.custom_fields: self.zabbix_id = self.nb.custom_fields[config["device_cf"]] else: - e = f"Host {self.name}: Custom field {config["device_cf"]} not present" + e = f'Host {self.name}: Custom field {config["device_cf"]} not present' self.logger.warning(e) raise SyncInventoryError(e) @@ -131,11 +131,13 @@ class PhysicalDevice(): # Set value to template return [device_type_cfs[config["template_cf"]]] # Custom field not found, return error - e = (f"Custom field {config["template_cf"]} not " + e = (f'Custom field {config["template_cf"]} not ' f"found for {self.nb.device_type.manufacturer.name}" f" - {self.nb.device_type.display}.") raise TemplateError(e) + + def get_templates_context(self): """ Get Zabbix templates from the device context """ if "zabbix" not in self.config_context: @@ -165,8 +167,8 @@ class PhysicalDevice(): elif config["inventory_mode"] == "automatic": self.inventory_mode = 1 else: - self.logger.error(f"Host {self.name}: Specified value for inventory mode in" - f" config is not valid. Got value {config["inventory_mode"]}") + self.logger.error(f"Host {self.name}: Specified value for inventory mode in " + f'config is not valid. Got value {config["inventory_mode"]}') return False self.inventory = {} if config["inventory_sync"] and self.inventory_mode in [0, 1]: diff --git a/tests/test_configuration_parsing.py b/tests/test_configuration_parsing.py index 070d3dd..4f97abf 100644 --- a/tests/test_configuration_parsing.py +++ b/tests/test_configuration_parsing.py @@ -1,6 +1,5 @@ """Tests for configuration parsing in the modules.config module.""" from unittest.mock import patch, MagicMock -from pathlib import Path import os from modules.config import load_config, DEFAULT_CONFIG, load_config_file, load_env_variable @@ -20,7 +19,7 @@ def test_load_config_file(): mock_config = DEFAULT_CONFIG.copy() mock_config["templates_config_context"] = True mock_config["sync_vms"] = True - + with patch('modules.config.load_config_file', return_value=mock_config), \ patch('modules.config.load_env_variable', return_value=None): config = load_config() @@ -39,7 +38,7 @@ def test_load_env_variables(): if key == "create_journal": return True return None - + with patch('modules.config.load_config_file', return_value=DEFAULT_CONFIG.copy()), \ patch('modules.config.load_env_variable', side_effect=mock_load_env): config = load_config() @@ -54,13 +53,13 @@ def test_env_vars_override_config_file(): mock_config = DEFAULT_CONFIG.copy() mock_config["templates_config_context"] = True mock_config["sync_vms"] = False - + # Mock env variable that will override the config file value def mock_load_env(key): if key == "sync_vms": return True return None - + with patch('modules.config.load_config_file', return_value=mock_config), \ patch('modules.config.load_env_variable', side_effect=mock_load_env): config = load_config() @@ -79,12 +78,12 @@ def test_load_config_file_function(): mock_module = MagicMock() mock_module.templates_config_context = True mock_module.sync_vms = True - + # Setup the mock spec mock_spec_instance = MagicMock() mock_spec.return_value = mock_spec_instance mock_spec_instance.loader.exec_module = lambda x: None - + # Patch module_from_spec to return our mock module with patch('importlib.util.module_from_spec', return_value=mock_module): config = load_config_file(DEFAULT_CONFIG.copy()) @@ -94,7 +93,7 @@ def test_load_config_file_function(): def test_load_config_file_not_found(): """Test load_config_file when the config file doesn't exist""" - # Instead of trying to assert on the logger call, we'll just check the return value + # Instead of trying to assert on the logger call, we'll just check the return value # and verify the function works as expected in this case with patch('pathlib.Path.exists', return_value=False): result = load_config_file(DEFAULT_CONFIG.copy()) @@ -107,7 +106,7 @@ def test_load_env_variable_function(): with patch.dict(os.environ, {"templates_config_context": "True"}): value = load_env_variable("templates_config_context") assert value == "True" - + # Test when the environment variable doesn't exist with patch.dict(os.environ, {}, clear=True): value = load_env_variable("nonexistent_variable") @@ -123,8 +122,8 @@ def test_load_config_file_exception_handling(): # Since the current implementation doesn't handle exceptions, we should # expect an exception to be raised try: - result = load_config_file(DEFAULT_CONFIG.copy()) + load_config_file(DEFAULT_CONFIG.copy()) assert False, "An exception should have been raised" - except Exception: + except Exception: # pylint: disable=broad-except # This is expected pass From 68cf28565d4c56f684f44421f8c78cc57b57e6bc Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 28 Apr 2025 15:47:37 +0200 Subject: [PATCH 20/28] Fixed some pipeline stuff --- .github/workflows/publish-image.yml | 3 --- .github/workflows/quality.yml | 7 ++++--- .github/workflows/run_tests.yml | 2 +- 3 files changed, 5 insertions(+), 7 deletions(-) diff --git a/.github/workflows/publish-image.yml b/.github/workflows/publish-image.yml index e9e6421..e6d07fe 100644 --- a/.github/workflows/publish-image.yml +++ b/.github/workflows/publish-image.yml @@ -4,9 +4,6 @@ on: push: branches: - main - - dockertest -# tags: -# - [0-9]+.* env: REGISTRY: ghcr.io diff --git a/.github/workflows/quality.yml b/.github/workflows/quality.yml index 7b01f6f..745f948 100644 --- a/.github/workflows/quality.yml +++ b/.github/workflows/quality.yml @@ -1,11 +1,12 @@ --- name: Pylint Quality control -on: - workflow_call +on: + push: + pull_request: jobs: - build: + python_quality_testing: runs-on: ubuntu-latest strategy: matrix: diff --git a/.github/workflows/run_tests.yml b/.github/workflows/run_tests.yml index cae8b02..6413d9c 100644 --- a/.github/workflows/run_tests.yml +++ b/.github/workflows/run_tests.yml @@ -17,7 +17,7 @@ jobs: python -m pip install --upgrade pip pip install pytest pytest-mock pip install -r requirements.txt - - name: Analysing the code with pylint + - name: Testing the code with PyTest run: | cp config.py.example config.py pytest tests From 772fef093089322e748472ac80d2aca37657123d Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 28 Apr 2025 15:57:11 +0200 Subject: [PATCH 21/28] Added prefix for environment variables --- modules/config.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/modules/config.py b/modules/config.py index 660bfec..3c74216 100644 --- a/modules/config.py +++ b/modules/config.py @@ -36,6 +36,8 @@ def load_config(): def load_env_variable(config_environvar): """Returns config from environment variable""" + prefix = "NZS_" + config_environvar = prefix + config_environvar.upper() if config_environvar in environ: return environ[config_environvar] return None From 98edf0ad9987b9ad8da76d2382431bbac8f80d81 Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 28 Apr 2025 17:23:51 +0200 Subject: [PATCH 22/28] Adjusted ENV prefix, fixed several linter errors in new tests --- .gitignore | 3 +- modules/config.py | 5 +- tests/test_configuration_parsing.py | 19 +- tests/test_device_deletion.py | 264 +++++++++++--------- tests/test_interface.py | 247 ++++++++++++++++++ tests/test_physical_device.py | 373 ++++++++++++++++++++++++++++ 6 files changed, 784 insertions(+), 127 deletions(-) create mode 100644 tests/test_interface.py create mode 100644 tests/test_physical_device.py diff --git a/.gitignore b/.gitignore index 5fdbd95..27761cd 100644 --- a/.gitignore +++ b/.gitignore @@ -4,4 +4,5 @@ # Byte-compiled / optimized / DLL files __pycache__/ *.py[cod] -.vscode \ No newline at end of file +.vscode +.flake \ No newline at end of file diff --git a/modules/config.py b/modules/config.py index 3c74216..3eca1d8 100644 --- a/modules/config.py +++ b/modules/config.py @@ -18,7 +18,8 @@ DEFAULT_CONFIG = { "create_journal": False, "sync_vms": False, "zabbix_device_removal": ["Decommissioning", "Inventory"], - "zabbix_device_disable": ["Offline", "Planned", "Staged", "Failed"] + "zabbix_device_disable": ["Offline", "Planned", "Staged", "Failed"], + "inventory_mode": "disabled" } @@ -36,7 +37,7 @@ def load_config(): def load_env_variable(config_environvar): """Returns config from environment variable""" - prefix = "NZS_" + prefix = "NBZX_" config_environvar = prefix + config_environvar.upper() if config_environvar in environ: return environ[config_environvar] diff --git a/tests/test_configuration_parsing.py b/tests/test_configuration_parsing.py index 4f97abf..23438b4 100644 --- a/tests/test_configuration_parsing.py +++ b/tests/test_configuration_parsing.py @@ -102,15 +102,26 @@ def test_load_config_file_not_found(): def test_load_env_variable_function(): """Test the load_env_variable function directly""" - # Test when the environment variable exists - with patch.dict(os.environ, {"templates_config_context": "True"}): + # Create a real environment variable for testing with correct prefix and uppercase + test_var = "NBZX_TEMPLATES_CONFIG_CONTEXT" + original_env = os.environ.get(test_var, None) + try: + # Set the environment variable with the proper prefix and case + os.environ[test_var] = "True" + + # Test that it's properly read (using lowercase in the function call) value = load_env_variable("templates_config_context") assert value == "True" - # Test when the environment variable doesn't exist - with patch.dict(os.environ, {}, clear=True): + # Test when the environment variable doesn't exist value = load_env_variable("nonexistent_variable") assert value is None + finally: + # Clean up - restore original environment + if original_env is not None: + os.environ[test_var] = original_env + else: + os.environ.pop(test_var, None) def test_load_config_file_exception_handling(): diff --git a/tests/test_device_deletion.py b/tests/test_device_deletion.py index 41c4420..392ba1a 100644 --- a/tests/test_device_deletion.py +++ b/tests/test_device_deletion.py @@ -1,142 +1,166 @@ -"""Testing device creation""" -from unittest.mock import MagicMock, patch, call +"""Tests for device deletion functionality in the PhysicalDevice class.""" +import unittest +from unittest.mock import MagicMock, patch +from zabbix_utils import APIRequestError from modules.device import PhysicalDevice -from modules.config import load_config - -config = load_config() +from modules.exceptions import SyncExternalError -def mock_nb_device(): - """Function to mock Netbox device""" - mock = MagicMock() - mock.id = 1 - mock.url = "http://netbox:8000/api/dcim/devices/1/" - mock.display_url = "http://netbox:8000/dcim/devices/1/" - mock.display = "SW01" - mock.name = "SW01" +class TestDeviceDeletion(unittest.TestCase): + """Test class for device deletion functionality.""" - mock.device_type = MagicMock() - mock.device_type.display = "Catalyst 3750G-48TS-S" - mock.device_type.manufacturer = MagicMock() - mock.device_type.manufacturer.display = "Cisco" - mock.device_type.manufacturer.name = "Cisco" - mock.device_type.manufacturer.slug = "cisco" - mock.device_type.manufacturer.description = "" - mock.device_type.model = "Catalyst 3750G-48TS-S" - mock.device_type.slug = "cisco-ws-c3750g-48ts-s" + def setUp(self): + """Set up test fixtures.""" + # Create mock NetBox device + self.mock_nb_device = MagicMock() + self.mock_nb_device.id = 123 + self.mock_nb_device.name = "test-device" + self.mock_nb_device.status.label = "Decommissioning" + self.mock_nb_device.custom_fields = {"zabbix_hostid": "456"} + self.mock_nb_device.config_context = {} - mock.role = MagicMock() - mock.role.id = 1 - mock.role.display = "Switch" - mock.role.name = "Switch" - mock.role.slug = "switch" + # Set up a primary IP + primary_ip = MagicMock() + primary_ip.address = "192.168.1.1/24" + self.mock_nb_device.primary_ip = primary_ip - mock.tenant = None - mock.platform = None - mock.serial = "0031876" - mock.asset_tag = None + # Create mock Zabbix API + self.mock_zabbix = MagicMock() + self.mock_zabbix.version = "6.0" - mock.site = MagicMock() - mock.site.display = "AMS01" - mock.site.name = "AMS01" - mock.site.slug = "ams01" + # Set up mock host.get response + self.mock_zabbix.host.get.return_value = [{"hostid": "456"}] - mock.location = None - mock.rack = None - mock.position = None - mock.face = None - mock.parent_device = None + # Mock NetBox journal class + self.mock_nb_journal = MagicMock() - mock.status = MagicMock() - mock.status.value = "decommissioning" - mock.status.label = "Decommissioning" + # Create logger mock + self.mock_logger = MagicMock() - mock.cluster = None - mock.virtual_chassis = None - mock.vc_position = None - mock.vc_priority = None - mock.description = "" - mock.comments = "" - mock.config_template = None - mock.config_context = {} - mock.local_context_data = None + # Create PhysicalDevice instance with mocks + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + self.device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + journal=True, + logger=self.mock_logger + ) - mock.custom_fields = {"zabbix_hostid": 1956} - return mock + def test_cleanup_successful_deletion(self): + """Test successful device deletion from Zabbix.""" + # Setup + self.mock_zabbix.host.get.return_value = [{"hostid": "456"}] + self.mock_zabbix.host.delete.return_value = {"hostids": ["456"]} + # Execute + self.device.cleanup() -def mock_zabbix(): - """Function to mock Zabbix""" - mock = MagicMock() - mock.host.get.return_value = [{}] - mock.host.delete.return_value = True - return mock + # Verify + self.mock_zabbix.host.get.assert_called_once_with(filter={'hostid': '456'}, output=[]) + self.mock_zabbix.host.delete.assert_called_once_with('456') + self.mock_nb_device.save.assert_called_once() + self.assertIsNone(self.mock_nb_device.custom_fields["zabbix_hostid"]) + self.mock_logger.info.assert_called_with(f"Host {self.device.name}: " + "Deleted host from Zabbix.") + def test_cleanup_device_already_deleted(self): + """Test cleanup when device is already deleted from Zabbix.""" + # Setup + self.mock_zabbix.host.get.return_value = [] # Empty list means host not found -netbox_journals = MagicMock() -NB_VERSION = '4.2' -create_journal = MagicMock() -logger = MagicMock() + # Execute + self.device.cleanup() + # Verify + self.mock_zabbix.host.get.assert_called_once_with(filter={'hostid': '456'}, output=[]) + self.mock_zabbix.host.delete.assert_not_called() + self.mock_nb_device.save.assert_called_once() + self.assertIsNone(self.mock_nb_device.custom_fields["zabbix_hostid"]) + self.mock_logger.info.assert_called_with( + f"Host {self.device.name}: was already deleted from Zabbix. Removed link in NetBox.") -def test_check_cluster_status(): - """Checks if the isCluster function is functioning properly""" - nb_device = mock_nb_device() - zabbix = mock_zabbix() - device = PhysicalDevice(nb_device, zabbix, None, None, - None, logger) - assert device.isCluster() is False + def test_cleanup_api_error(self): + """Test cleanup when Zabbix API returns an error.""" + # Setup + self.mock_zabbix.host.get.return_value = [{"hostid": "456"}] + self.mock_zabbix.host.delete.side_effect = APIRequestError("API Error") + # Execute and verify + with self.assertRaises(SyncExternalError): + self.device.cleanup() -def test_device_deletion_host_exists(): - """Checks device deletion process""" - nb_device = mock_nb_device() - zabbix = mock_zabbix() - with patch.object(PhysicalDevice, 'create_journal_entry') as mock_journal: - # Create device - device = PhysicalDevice(nb_device, zabbix, netbox_journals, NB_VERSION, - create_journal, logger) - device.cleanup() - # Check if Zabbix HostID is empty - assert device.nb.custom_fields[config["device_cf"]] is None - # Check if API calls are executed - device.zabbix.host.get.assert_called_once_with(filter={'hostid': 1956}, - output=[]) - device.zabbix.host.delete.assert_called_once_with(1956) - # check logger - mock_journal.assert_called_once_with("warning", - "Deleted host from Zabbix") - device.logger.info.assert_called_once_with("Host SW01: Deleted " - "host from Zabbix.") + # Verify correct calls were made + self.mock_zabbix.host.get.assert_called_once_with(filter={'hostid': '456'}, output=[]) + self.mock_zabbix.host.delete.assert_called_once_with('456') + self.mock_nb_device.save.assert_not_called() + self.mock_logger.error.assert_called() + def test_zeroize_cf(self): + """Test _zeroize_cf method that clears the custom field.""" + # Execute + self.device._zeroize_cf() # pylint: disable=protected-access -def test_device_deletion_host_not_exists(): - """ - Test if device in Netbox gets unlinked - when host is not present in Zabbix - """ - nb_device = mock_nb_device() - zabbix = mock_zabbix() - zabbix.host.get.return_value = None + # Verify + self.assertIsNone(self.mock_nb_device.custom_fields["zabbix_hostid"]) + self.mock_nb_device.save.assert_called_once() - with patch.object(PhysicalDevice, 'create_journal_entry') as mock_journal: - # Create new device - device = PhysicalDevice(nb_device, zabbix, netbox_journals, NB_VERSION, - create_journal, logger) - # Try to clean the device up in Zabbix - device.cleanup() - # Confirm that a call was issued to Zabbix to check if the host exists - device.zabbix.host.get.assert_called_once_with(filter={'hostid': 1956}, - output=[]) - # Confirm that no device was deleted in Zabbix - device.zabbix.host.delete.assert_not_called() - # Test logging - log_calls = [ - call('Host SW01: Deleted host from Zabbix.'), - call('Host SW01: was already deleted from Zabbix. ' - 'Removed link in NetBox.') - ] - logger.info.assert_has_calls(log_calls) - assert logger.info.call_count == 2 - mock_journal.assert_called_once_with("warning", - "Deleted host from Zabbix") + def test_create_journal_entry(self): + """Test create_journal_entry method.""" + # Setup + test_message = "Test journal entry" + + # Execute + result = self.device.create_journal_entry("info", test_message) + + # Verify + self.assertTrue(result) + self.mock_nb_journal.create.assert_called_once() + journal_entry = self.mock_nb_journal.create.call_args[0][0] + self.assertEqual(journal_entry["assigned_object_type"], "dcim.device") + self.assertEqual(journal_entry["assigned_object_id"], 123) + self.assertEqual(journal_entry["kind"], "info") + self.assertEqual(journal_entry["comments"], test_message) + + def test_create_journal_entry_invalid_severity(self): + """Test create_journal_entry with invalid severity.""" + # Execute + result = self.device.create_journal_entry("invalid", "Test message") + + # Verify + self.assertFalse(result) + self.mock_nb_journal.create.assert_not_called() + self.mock_logger.warning.assert_called() + + def test_create_journal_entry_when_disabled(self): + """Test create_journal_entry when journaling is disabled.""" + # Setup - create device with journal=False + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + journal=False, # Disable journaling + logger=self.mock_logger + ) + + # Execute + result = device.create_journal_entry("info", "Test message") + + # Verify + self.assertFalse(result) + self.mock_nb_journal.create.assert_not_called() + + def test_cleanup_updates_journal(self): + """Test that cleanup method creates a journal entry.""" + # Setup + self.mock_zabbix.host.get.return_value = [{"hostid": "456"}] + + # Execute + with patch.object(self.device, 'create_journal_entry') as mock_journal_entry: + self.device.cleanup() + + # Verify + mock_journal_entry.assert_called_once_with("warning", "Deleted host from Zabbix") diff --git a/tests/test_interface.py b/tests/test_interface.py new file mode 100644 index 0000000..ff55218 --- /dev/null +++ b/tests/test_interface.py @@ -0,0 +1,247 @@ +"""Tests for the ZabbixInterface class in the interface module.""" +import unittest +from modules.interface import ZabbixInterface +from modules.exceptions import InterfaceConfigError + + +class TestZabbixInterface(unittest.TestCase): + """Test class for ZabbixInterface functionality.""" + + def setUp(self): + """Set up test fixtures.""" + self.test_ip = "192.168.1.1" + self.empty_context = {} + self.default_interface = ZabbixInterface(self.empty_context, self.test_ip) + + # Create some test contexts for different scenarios + self.snmpv2_context = { + "zabbix": { + "interface_type": 2, + "interface_port": "161", + "snmp": { + "version": 2, + "community": "public", + "bulk": 1 + } + } + } + + self.snmpv3_context = { + "zabbix": { + "interface_type": 2, + "snmp": { + "version": 3, + "securityname": "snmpuser", + "securitylevel": "authPriv", + "authprotocol": "SHA", + "authpassphrase": "authpass123", + "privprotocol": "AES", + "privpassphrase": "privpass123", + "contextname": "context1" + } + } + } + + self.agent_context = { + "zabbix": { + "interface_type": 1, + "interface_port": "10050" + } + } + + def test_init(self): + """Test initialization of ZabbixInterface.""" + interface = ZabbixInterface(self.empty_context, self.test_ip) + + # Check basic properties + self.assertEqual(interface.ip, self.test_ip) + self.assertEqual(interface.context, self.empty_context) + self.assertEqual(interface.interface["ip"], self.test_ip) + self.assertEqual(interface.interface["main"], "1") + self.assertEqual(interface.interface["useip"], "1") + self.assertEqual(interface.interface["dns"], "") + + def test_get_context_empty(self): + """Test get_context with empty context.""" + interface = ZabbixInterface(self.empty_context, self.test_ip) + result = interface.get_context() + self.assertFalse(result) + + def test_get_context_with_interface_type(self): + """Test get_context with interface_type but no port.""" + context = {"zabbix": {"interface_type": 2}} + interface = ZabbixInterface(context, self.test_ip) + + # Should set type and default port + result = interface.get_context() + self.assertTrue(result) + self.assertEqual(interface.interface["type"], 2) + self.assertEqual(interface.interface["port"], "161") # Default port for SNMP + + def test_get_context_with_interface_type_and_port(self): + """Test get_context with both interface_type and port.""" + context = {"zabbix": {"interface_type": 1, "interface_port": "12345"}} + interface = ZabbixInterface(context, self.test_ip) + + # Should set type and specified port + result = interface.get_context() + self.assertTrue(result) + self.assertEqual(interface.interface["type"], 1) + self.assertEqual(interface.interface["port"], "12345") + + def test_set_default_port(self): + """Test _set_default_port for different interface types.""" + interface = ZabbixInterface(self.empty_context, self.test_ip) + + # Test for agent type (1) + interface.interface["type"] = 1 + interface._set_default_port() # pylint: disable=protected-access + self.assertEqual(interface.interface["port"], "10050") + + # Test for SNMP type (2) + interface.interface["type"] = 2 + interface._set_default_port() # pylint: disable=protected-access + self.assertEqual(interface.interface["port"], "161") + + # Test for IPMI type (3) + interface.interface["type"] = 3 + interface._set_default_port() # pylint: disable=protected-access + self.assertEqual(interface.interface["port"], "623") + + # Test for JMX type (4) + interface.interface["type"] = 4 + interface._set_default_port() # pylint: disable=protected-access + self.assertEqual(interface.interface["port"], "12345") + + # Test for unsupported type + interface.interface["type"] = 99 + result = interface._set_default_port() # pylint: disable=protected-access + self.assertFalse(result) + + def test_set_snmp_v2(self): + """Test set_snmp with SNMPv2 configuration.""" + interface = ZabbixInterface(self.snmpv2_context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp + interface.set_snmp() + + # Check SNMP details + self.assertEqual(interface.interface["details"]["version"], "2") + self.assertEqual(interface.interface["details"]["community"], "public") + self.assertEqual(interface.interface["details"]["bulk"], "1") + + def test_set_snmp_v3(self): + """Test set_snmp with SNMPv3 configuration.""" + interface = ZabbixInterface(self.snmpv3_context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp + interface.set_snmp() + + # Check SNMP details + self.assertEqual(interface.interface["details"]["version"], "3") + self.assertEqual(interface.interface["details"]["securityname"], "snmpuser") + self.assertEqual(interface.interface["details"]["securitylevel"], "authPriv") + self.assertEqual(interface.interface["details"]["authprotocol"], "SHA") + self.assertEqual(interface.interface["details"]["authpassphrase"], "authpass123") + self.assertEqual(interface.interface["details"]["privprotocol"], "AES") + self.assertEqual(interface.interface["details"]["privpassphrase"], "privpass123") + self.assertEqual(interface.interface["details"]["contextname"], "context1") + + def test_set_snmp_no_snmp_config(self): + """Test set_snmp with missing SNMP configuration.""" + # Create context with interface type but no SNMP config + context = {"zabbix": {"interface_type": 2}} + interface = ZabbixInterface(context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp - should raise exception + with self.assertRaises(InterfaceConfigError): + interface.set_snmp() + + def test_set_snmp_unsupported_version(self): + """Test set_snmp with unsupported SNMP version.""" + # Create context with invalid SNMP version + context = { + "zabbix": { + "interface_type": 2, + "snmp": { + "version": 4 # Invalid version + } + } + } + interface = ZabbixInterface(context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp - should raise exception + with self.assertRaises(InterfaceConfigError): + interface.set_snmp() + + def test_set_snmp_no_version(self): + """Test set_snmp with missing SNMP version.""" + # Create context without SNMP version + context = { + "zabbix": { + "interface_type": 2, + "snmp": { + "community": "public" # No version specified + } + } + } + interface = ZabbixInterface(context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp - should raise exception + with self.assertRaises(InterfaceConfigError): + interface.set_snmp() + + def test_set_snmp_non_snmp_interface(self): + """Test set_snmp with non-SNMP interface type.""" + interface = ZabbixInterface(self.agent_context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp - should raise exception + with self.assertRaises(InterfaceConfigError): + interface.set_snmp() + + def test_set_default_snmp(self): + """Test set_default_snmp method.""" + interface = ZabbixInterface(self.empty_context, self.test_ip) + interface.set_default_snmp() + + # Check interface properties + self.assertEqual(interface.interface["type"], "2") + self.assertEqual(interface.interface["port"], "161") + self.assertEqual(interface.interface["details"]["version"], "2") + self.assertEqual(interface.interface["details"]["community"], "{$SNMP_COMMUNITY}") + self.assertEqual(interface.interface["details"]["bulk"], "1") + + def test_set_default_agent(self): + """Test set_default_agent method.""" + interface = ZabbixInterface(self.empty_context, self.test_ip) + interface.set_default_agent() + + # Check interface properties + self.assertEqual(interface.interface["type"], "1") + self.assertEqual(interface.interface["port"], "10050") + + def test_snmpv2_no_community(self): + """Test SNMPv2 with no community string specified.""" + # Create context with SNMPv2 but no community + context = { + "zabbix": { + "interface_type": 2, + "snmp": { + "version": 2 + } + } + } + interface = ZabbixInterface(context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp + interface.set_snmp() + + # Should use default community string + self.assertEqual(interface.interface["details"]["community"], "{$SNMP_COMMUNITY}") diff --git a/tests/test_physical_device.py b/tests/test_physical_device.py new file mode 100644 index 0000000..4fe8ce3 --- /dev/null +++ b/tests/test_physical_device.py @@ -0,0 +1,373 @@ +"""Tests for the PhysicalDevice class in the device module.""" +import unittest +from unittest.mock import MagicMock, patch +from modules.device import PhysicalDevice +from modules.exceptions import TemplateError, SyncInventoryError + + +class TestPhysicalDevice(unittest.TestCase): + """Test class for PhysicalDevice functionality.""" + + def setUp(self): + """Set up test fixtures.""" + # Create mock NetBox device + self.mock_nb_device = MagicMock() + self.mock_nb_device.id = 123 + self.mock_nb_device.name = "test-device" + self.mock_nb_device.status.label = "Active" + self.mock_nb_device.custom_fields = {"zabbix_hostid": None} + self.mock_nb_device.config_context = {} + + # Set up a primary IP + primary_ip = MagicMock() + primary_ip.address = "192.168.1.1/24" + self.mock_nb_device.primary_ip = primary_ip + + # Create mock Zabbix API + self.mock_zabbix = MagicMock() + self.mock_zabbix.version = "6.0" + + # Mock NetBox journal class + self.mock_nb_journal = MagicMock() + + # Create logger mock + self.mock_logger = MagicMock() + + # Create PhysicalDevice instance with mocks + with patch('modules.device.config', + {"device_cf": "zabbix_hostid", + "template_cf": "zabbix_template", + "templates_config_context": False, + "templates_config_context_overrule": False, + "traverse_regions": False, + "traverse_site_groups": False, + "inventory_mode": "disabled", + "inventory_sync": False, + "inventory_map": {} + }): + self.device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + journal=True, + logger=self.mock_logger + ) + + def test_init(self): + """Test the initialization of the PhysicalDevice class.""" + # Check that basic properties are set correctly + self.assertEqual(self.device.name, "test-device") + self.assertEqual(self.device.id, 123) + self.assertEqual(self.device.status, "Active") + self.assertEqual(self.device.ip, "192.168.1.1") + self.assertEqual(self.device.cidr, "192.168.1.1/24") + + def test_init_no_primary_ip(self): + """Test initialization when device has no primary IP.""" + # Set primary_ip to None + self.mock_nb_device.primary_ip = None + + # Creating device should raise SyncInventoryError + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + with self.assertRaises(SyncInventoryError): + PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + def test_set_basics_with_special_characters(self): + """Test _setBasics when device name contains special characters.""" + # Set name with special characters that + # will actually trigger the special character detection + self.mock_nb_device.name = "test-devïce" + + # We need to patch the search function to simulate finding special characters + with patch('modules.device.search') as mock_search, \ + patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + # Make the search function return True to simulate special characters + mock_search.return_value = True + + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # With the mocked search function, the name should be changed to NETBOX_ID format + self.assertEqual(device.name, f"NETBOX_ID{self.mock_nb_device.id}") + # And visible_name should be set to the original name + self.assertEqual(device.visible_name, "test-devïce") + # use_visible_name flag should be set + self.assertTrue(device.use_visible_name) + + def test_get_templates_context(self): + """Test get_templates_context with valid config.""" + # Set up config_context with valid template data + self.mock_nb_device.config_context = { + "zabbix": { + "templates": ["Template1", "Template2"] + } + } + + # Create device with the updated mock + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Test that templates are returned correctly + templates = device.get_templates_context() + self.assertEqual(templates, ["Template1", "Template2"]) + + def test_get_templates_context_with_string(self): + """Test get_templates_context with a string instead of list.""" + # Set up config_context with a string template + self.mock_nb_device.config_context = { + "zabbix": { + "templates": "Template1" + } + } + + # Create device with the updated mock + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Test that template is wrapped in a list + templates = device.get_templates_context() + self.assertEqual(templates, ["Template1"]) + + def test_get_templates_context_no_zabbix_key(self): + """Test get_templates_context when zabbix key is missing.""" + # Set up config_context without zabbix key + self.mock_nb_device.config_context = {} + + # Create device with the updated mock + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Test that TemplateError is raised + with self.assertRaises(TemplateError): + device.get_templates_context() + + def test_get_templates_context_no_templates_key(self): + """Test get_templates_context when templates key is missing.""" + # Set up config_context without templates key + self.mock_nb_device.config_context = {"zabbix": {}} + + # Create device with the updated mock + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Test that TemplateError is raised + with self.assertRaises(TemplateError): + device.get_templates_context() + + def test_set_template_with_config_context(self): + """Test set_template with templates_config_context=True.""" + # Set up config_context with templates + self.mock_nb_device.config_context = { + "zabbix": { + "templates": ["Template1"] + } + } + + # Mock get_templates_context to return expected templates + with patch.object(PhysicalDevice, 'get_templates_context', return_value=["Template1"]): + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Call set_template with prefer_config_context=True + result = device.set_template(prefer_config_context=True, overrule_custom=False) + + # Check result and template names + self.assertTrue(result) + self.assertEqual(device.zbx_template_names, ["Template1"]) + + def test_set_inventory_disabled_mode(self): + """Test set_inventory with inventory_mode=disabled.""" + # Configure with disabled inventory mode + config_patch = { + "device_cf": "zabbix_hostid", + "inventory_mode": "disabled", + "inventory_sync": False + } + + with patch('modules.device.config', config_patch): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Call set_inventory with the config patch still active + with patch('modules.device.config', config_patch): + result = device.set_inventory({}) + + # Check result + self.assertTrue(result) + # Default value for disabled inventory + self.assertEqual(device.inventory_mode, -1) + + def test_set_inventory_manual_mode(self): + """Test set_inventory with inventory_mode=manual.""" + # Configure with manual inventory mode + config_patch = { + "device_cf": "zabbix_hostid", + "inventory_mode": "manual", + "inventory_sync": False + } + + with patch('modules.device.config', config_patch): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Call set_inventory with the config patch still active + with patch('modules.device.config', config_patch): + result = device.set_inventory({}) + + # Check result + self.assertTrue(result) + self.assertEqual(device.inventory_mode, 0) # Manual mode + + def test_set_inventory_automatic_mode(self): + """Test set_inventory with inventory_mode=automatic.""" + # Configure with automatic inventory mode + config_patch = { + "device_cf": "zabbix_hostid", + "inventory_mode": "automatic", + "inventory_sync": False + } + + with patch('modules.device.config', config_patch): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Call set_inventory with the config patch still active + with patch('modules.device.config', config_patch): + result = device.set_inventory({}) + + # Check result + self.assertTrue(result) + self.assertEqual(device.inventory_mode, 1) # Automatic mode + + def test_set_inventory_with_inventory_sync(self): + """Test set_inventory with inventory_sync=True.""" + # Configure with inventory sync enabled + config_patch = { + "device_cf": "zabbix_hostid", + "inventory_mode": "manual", + "inventory_sync": True, + "inventory_map": { + "name": "name", + "serial": "serialno_a" + } + } + + with patch('modules.device.config', config_patch): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Create a mock device with the required attributes + mock_device_data = { + "name": "test-device", + "serial": "ABC123" + } + + # Call set_inventory with the config patch still active + with patch('modules.device.config', config_patch): + result = device.set_inventory(mock_device_data) + + # Check result + self.assertTrue(result) + self.assertEqual(device.inventory_mode, 0) # Manual mode + self.assertEqual(device.inventory, { + "name": "test-device", + "serialno_a": "ABC123" + }) + + def test_iscluster_true(self): + """Test isCluster when device is part of a cluster.""" + # Set up virtual_chassis + self.mock_nb_device.virtual_chassis = MagicMock() + + # Create device with the updated mock + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Check isCluster result + self.assertTrue(device.isCluster()) + + def test_is_cluster_false(self): + """Test isCluster when device is not part of a cluster.""" + # Set virtual_chassis to None + self.mock_nb_device.virtual_chassis = None + + # Create device with the updated mock + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Check isCluster result + self.assertFalse(device.isCluster()) From d60eb1cb2dfb6fdf3971290da9eed9c2f033e768 Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 28 Apr 2025 22:18:59 +0200 Subject: [PATCH 23/28] Removed python test files for linter. --- .github/workflows/quality.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/quality.yml b/.github/workflows/quality.yml index 745f948..4421765 100644 --- a/.github/workflows/quality.yml +++ b/.github/workflows/quality.yml @@ -24,4 +24,4 @@ jobs: pip install -r requirements.txt - name: Analysing the code with pylint run: | - pylint --module-naming-style=any $(git ls-files '*.py') + pylint --module-naming-style=any modules/* netbox_zabbix_sync.py From 2998dfde549a8f9b6ec11e286d94df458287c19d Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 28 Apr 2025 22:21:30 +0200 Subject: [PATCH 24/28] Specifiek Python version in pipeline test step --- .github/workflows/run_tests.yml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.github/workflows/run_tests.yml b/.github/workflows/run_tests.yml index 6413d9c..9093c96 100644 --- a/.github/workflows/run_tests.yml +++ b/.github/workflows/run_tests.yml @@ -12,6 +12,8 @@ jobs: - uses: actions/checkout@v4 - name: Set up Python uses: actions/setup-python@v5 + with: + python-version: 3.12 - name: Install dependencies run: | python -m pip install --upgrade pip From bbe28d97053aff1940e3b2e76524dec92a9c9ad2 Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 28 Apr 2025 22:31:36 +0200 Subject: [PATCH 25/28] Added all default config statements and added warning to any curious users. --- modules/config.py | 27 ++++++++++++++++++++++++++- 1 file changed, 26 insertions(+), 1 deletion(-) diff --git a/modules/config.py b/modules/config.py index 3eca1d8..6528d67 100644 --- a/modules/config.py +++ b/modules/config.py @@ -8,6 +8,8 @@ from logging import getLogger logger = getLogger(__name__) +# PLEASE NOTE: This is a sample config file. You should create your own config.py + DEFAULT_CONFIG = { "templates_config_context": False, "templates_config_context_overrule": False, @@ -17,9 +19,32 @@ DEFAULT_CONFIG = { "create_hostgroups": True, "create_journal": False, "sync_vms": False, + "vm_hostgroup_format": "cluster_type/cluster/role", + "full_proxy_sync": False, "zabbix_device_removal": ["Decommissioning", "Inventory"], "zabbix_device_disable": ["Offline", "Planned", "Staged", "Failed"], - "inventory_mode": "disabled" + "hostgroup_format": "site/manufacturer/role", + "traverse_regions": False, + "traverse_site_groups": False, + "nb_device_filter": {"name__n": "null"}, + "nb_vm_filter": {"name__n": "null"}, + "inventory_mode": "disabled", + "inventory_sync": False, + "inventory_map": { + "asset_tag": "asset_tag", + "virtual_chassis/name": "chassis", + "status/label": "deployment_status", + "location/name": "location", + "latitude": "location_lat", + "longitude": "location_lon", + "comments": "notes", + "name": "name", + "rack/name": "site_rack", + "serial": "serialno_a", + "device_type/model": "type", + "device_type/manufacturer/name": "vendor", + "oob_ip/address": "oob_ip" + } } From 539ad64c8d517a9981a1fd294d0419692ddb744a Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Mon, 28 Apr 2025 22:49:04 +0200 Subject: [PATCH 26/28] Adds 2 new tests for virtual chassis --- tests/test_physical_device.py | 56 +++++++++++++++++++++++++++++++++++ 1 file changed, 56 insertions(+) diff --git a/tests/test_physical_device.py b/tests/test_physical_device.py index 4fe8ce3..d0ba43b 100644 --- a/tests/test_physical_device.py +++ b/tests/test_physical_device.py @@ -371,3 +371,59 @@ class TestPhysicalDevice(unittest.TestCase): # Check isCluster result self.assertFalse(device.isCluster()) + + + def test_promote_master_device_primary(self): + """Test promoteMasterDevice when device is primary in cluster.""" + # Set up virtual chassis with master device + mock_vc = MagicMock() + mock_vc.name = "virtual-chassis-1" + mock_master = MagicMock() + mock_master.id = self.mock_nb_device.id # Set master ID to match the current device + mock_vc.master = mock_master + self.mock_nb_device.virtual_chassis = mock_vc + + # Create device with the updated mock + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Call promoteMasterDevice and check the result + result = device.promoteMasterDevice() + + # Should return True for primary device + self.assertTrue(result) + # Device name should be updated to virtual chassis name + self.assertEqual(device.name, "virtual-chassis-1") + + + def test_promote_master_device_secondary(self): + """Test promoteMasterDevice when device is secondary in cluster.""" + # Set up virtual chassis with a different master device + mock_vc = MagicMock() + mock_vc.name = "virtual-chassis-1" + mock_master = MagicMock() + mock_master.id = self.mock_nb_device.id + 1 # Different ID than the current device + mock_vc.master = mock_master + self.mock_nb_device.virtual_chassis = mock_vc + + # Create device with the updated mock + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Call promoteMasterDevice and check the result + result = device.promoteMasterDevice() + + # Should return False for secondary device + self.assertFalse(result) + # Device name should not be modified + self.assertEqual(device.name, "test-device") From 9e1a90833d984ff19aa39d5a9e862f1e2da5d46e Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Sun, 8 Jun 2025 21:45:45 +0200 Subject: [PATCH 27/28] Added new config parameters to base template --- modules/config.py | 39 ++++++++++++++++++++++++++++++++++++--- 1 file changed, 36 insertions(+), 3 deletions(-) diff --git a/modules/config.py b/modules/config.py index 6528d67..586227d 100644 --- a/modules/config.py +++ b/modules/config.py @@ -1,5 +1,5 @@ """ -Module for parsing configuration from the top level config.yaml file +Module for parsing configuration from the top level config.py file """ from pathlib import Path from importlib import util @@ -8,7 +8,8 @@ from logging import getLogger logger = getLogger(__name__) -# PLEASE NOTE: This is a sample config file. You should create your own config.py +# PLEASE NOTE: This is a sample config file. Please do NOT make any edits in this file! +# You should create your own config.py and it will overwrite the default config. DEFAULT_CONFIG = { "templates_config_context": False, @@ -30,7 +31,7 @@ DEFAULT_CONFIG = { "nb_vm_filter": {"name__n": "null"}, "inventory_mode": "disabled", "inventory_sync": False, - "inventory_map": { + "device_inventory_map": { "asset_tag": "asset_tag", "virtual_chassis/name": "chassis", "status/label": "deployment_status", @@ -44,6 +45,38 @@ DEFAULT_CONFIG = { "device_type/model": "type", "device_type/manufacturer/name": "vendor", "oob_ip/address": "oob_ip" + }, + "vm_inventory_map": { + "status/label": "deployment_status", + "comments": "notes", + "name": "name" + }, + "usermacro_sync": False, + "device_usermacro_map": { + "serial": "{$HW_SERIAL}", + "role/name": "{$DEV_ROLE}", + "url": "{$NB_URL}", + "id": "{$NB_ID}" + }, + "vm_usermacro_map": { + "memory": "{$TOTAL_MEMORY}", + "role/name": "{$DEV_ROLE}", + "url": "{$NB_URL}", + "id": "{$NB_ID}" + }, + "tag_sync": False, + "tag_lower": True, + "tag_name": 'NetBox', + "tag_value": "name", + "device_tag_map": { + "site/name": "site", + "rack/name": "rack", + "platform/name": "target" + }, + "vm_tag_map": { + "site/name": "site", + "cluster/name": "cluster", + "platform/name": "target" } } From a325863aecb99067e16137cc4913976d8604733d Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Sun, 8 Jun 2025 22:13:36 +0200 Subject: [PATCH 28/28] Fixed several config errors, double exception imports, Linter stuff and edited test for new device_inventory_map key --- modules/device.py | 28 ++++++++++++---------------- modules/virtual_machine.py | 13 +++++-------- tests/test_physical_device.py | 4 ++-- 3 files changed, 19 insertions(+), 26 deletions(-) diff --git a/modules/device.py b/modules/device.py index 8bea73d..971a5b3 100644 --- a/modules/device.py +++ b/modules/device.py @@ -124,8 +124,8 @@ class PhysicalDevice: self.nb, self.nb_api_version, logger=self.logger, - nested_sitegroup_flag=traverse_site_groups, - nested_region_flag=traverse_regions, + nested_sitegroup_flag=config['traverse_site_groups'], + nested_region_flag=config['traverse_regions'], nb_groups=nb_site_groups, nb_regions=nb_regions, ) @@ -166,7 +166,7 @@ class PhysicalDevice: return [device_type_cfs[config["template_cf"]]] # Custom field not found, return error e = ( - f"Custom field {template_cf} not " + f"Custom field {config['template_cf']} not " f"found for {self.nb.device_type.manufacturer.name}" f" - {self.nb.device_type.display}." ) @@ -394,7 +394,7 @@ class PhysicalDevice: macros = ZabbixUsermacros( self.nb, self._usermacro_map(), - usermacro_sync, + config['usermacro_sync'], logger=self.logger, host=self.name, ) @@ -411,10 +411,10 @@ class PhysicalDevice: tags = ZabbixTags( self.nb, self._tag_map(), - tag_sync, - tag_lower, - tag_name=tag_name, - tag_value=tag_value, + config['tag_sync'], + config['tag_lower'], + tag_name=config['tag_name'], + tag_value=config['tag_value'], logger=self.logger, host=self.name, ) @@ -771,15 +771,11 @@ class PhysicalDevice: self.updateZabbixHost(inventory=self.inventory) # Check host usermacros - if usermacro_sync: + if config['usermacro_sync']: macros_filtered = [] # Do not re-sync secret usermacros unless sync is set to 'full' - if str(usermacro_sync).lower() != "full": - for m in - - - - (self.usermacros): + if str(config['usermacro_sync']).lower() != "full": + for m in deepcopy(self.usermacros): if m["type"] == str(1): # Remove the value as the api doesn't return it # this will allow us to only update usermacros that don't exist @@ -792,7 +788,7 @@ class PhysicalDevice: self.updateZabbixHost(macros=self.usermacros) # Check host usermacros - if tag_sync: + if config['tag_sync']: if remove_duplicates(host["tags"], sortkey="tag") == self.tags: self.logger.debug(f"Host {self.name}: tags in-sync.") else: diff --git a/modules/virtual_machine.py b/modules/virtual_machine.py index acabd1d..7ee6659 100644 --- a/modules/virtual_machine.py +++ b/modules/virtual_machine.py @@ -4,9 +4,6 @@ from modules.device import PhysicalDevice from modules.exceptions import InterfaceConfigError, SyncInventoryError, TemplateError from modules.hostgroups import Hostgroup from modules.interface import ZabbixInterface - -from modules.exceptions import (TemplateError, InterfaceConfigError, - SyncInventoryError) from modules.config import load_config # Load config config = load_config() @@ -22,15 +19,15 @@ class VirtualMachine(PhysicalDevice): def _inventory_map(self): """use VM inventory maps""" - return vm_inventory_map + return config["vm_inventory_map"] def _usermacro_map(self): """use VM usermacro maps""" - return vm_usermacro_map + return config["vm_usermacro_map"] def _tag_map(self): """use VM tag maps""" - return vm_tag_map + return config["vm_tag_map"] def set_hostgroup(self, hg_format, nb_site_groups, nb_regions): """Set the hostgroup for this device""" @@ -40,8 +37,8 @@ class VirtualMachine(PhysicalDevice): self.nb, self.nb_api_version, logger=self.logger, - nested_sitegroup_flag=traverse_site_groups, - nested_region_flag=traverse_regions, + nested_sitegroup_flag=config["traverse_site_groups"], + nested_region_flag=config["traverse_regions"], nb_groups=nb_site_groups, nb_regions=nb_regions, ) diff --git a/tests/test_physical_device.py b/tests/test_physical_device.py index d0ba43b..1b79ad8 100644 --- a/tests/test_physical_device.py +++ b/tests/test_physical_device.py @@ -43,7 +43,7 @@ class TestPhysicalDevice(unittest.TestCase): "traverse_site_groups": False, "inventory_mode": "disabled", "inventory_sync": False, - "inventory_map": {} + "device_inventory_map": {} }): self.device = PhysicalDevice( self.mock_nb_device, @@ -303,7 +303,7 @@ class TestPhysicalDevice(unittest.TestCase): "device_cf": "zabbix_hostid", "inventory_mode": "manual", "inventory_sync": True, - "inventory_map": { + "device_inventory_map": { "name": "name", "serial": "serialno_a" }