hetzner: fix disk sizes before NetBox 4.1 and handle API errors

module/sources/hetzner/client.py

@@ -15,5 +15,11 @@ class HetznerClient:
     def __init__(self, token):
         self.client = Client(token=token)
 
+    def check_connection(self):
+        """
+        Fails if the API can't be reached or doesn't accept the token
+        """
+        self.client.servers.get_list(per_page=1)
+
     def get_servers(self):
         return self.client.servers.get_all()

module/sources/hetzner/connection.py

@@ -7,7 +7,11 @@
 #  For a copy, see file LICENSE.txt included in this
 #  repository or visit: <https://opensource.org/licenses/MIT>.
 
+from hcloud import HCloudException
+from requests.exceptions import RequestException
+
 from module.common.logging import get_logger
+from module.common.misc import do_error_exit
 from module.sources.common.source_base import SourceBase
 from module.sources.hetzner.client import HetznerClient
 from module.sources.hetzner.config import HetznerConfig
@@ -66,6 +70,15 @@ def __init__(self, name=None):
             log.info(f"Source '{name}' is currently disabled. Skipping")
             return
 
+        self.client = HetznerClient(token=self.settings.api_token)
+
+        try:
+            self.client.check_connection()
+        except (HCloudException, RequestException) as e:
+            log.error(f"Unable to connect to the Hetzner Cloud API for source '{name}': {e}")
+            log.info(f"Source '{name}' is currently unavailable. Skipping")
+            return
+
         self.init_successful = True
 
     @classmethod
@@ -74,13 +87,16 @@ def implements(cls, source_type):
 
     def apply(self):
 
-        token = self.settings.api_token
-
-        if not token:
-            self.log.error("Hetzner api_token not defined in settings.ini")
-            return
+        try:
+            self.sync_servers()
+        except (HCloudException, RequestException) as e:
+            # pruning is already decided and part of this source may be in the inventory,
+            # so stop the whole run instead of writing a partial sync to NetBox
+            message = f"Hetzner source '{self.name}': {e}. No NetBox writes or pruning will be performed."
+            log.error(message)
+            do_error_exit(message)
 
-        self.client = HetznerClient(token=token)
+    def sync_servers(self):
 
         servers = self.client.get_servers()
 

module/sources/hetzner/disk.py

@@ -7,6 +7,8 @@
 #  For a copy, see file LICENSE.txt included in this
 #  repository or visit: <https://opensource.org/licenses/MIT>.
 
+from packaging import version
+
 from module.netbox.inventory import NBVirtualDisk
 
 
@@ -23,12 +25,16 @@ def sync_vm_disks(handler, vm, server):
     for volume in server.volumes:
 
         disk_name = f"{server.name}-{volume.name}"[:60]
-        size_mb = int(volume.size) * 1024  # Hetzner size = GB
+
+        # Hetzner reports volume sizes in GB, NetBox stores virtual disk sizes in GB before 4.1.0 and in MB since
+        disk_size = int(volume.size)
+        if version.parse(inventory.netbox_api_version) >= version.parse("4.1.0"):
+            disk_size *= 1024
 
         disk_data = {
             "name": disk_name,
             "virtual_machine": vm,   # object, не id
-            "size": size_mb,
+            "size": disk_size,
         }
 
         existing_disk = None

tests/test_hetzner_source.py

@@ -0,0 +1,119 @@
+# -*- coding: utf-8 -*-
+#  Copyright (c) 2020 - 2026 netbox-sync team. All rights reserved.
+#
+#  netbox-sync.py
+#
+#  This work is licensed under the terms of the MIT license.
+#  For a copy, see file LICENSE.txt included in this
+#  repository or visit: <https://opensource.org/licenses/MIT>.
+
+import json
+from urllib.parse import urlsplit
+
+import pytest
+import requests
+
+from module.netbox import NBVirtualDisk
+from module.sources import instantiate_sources
+
+TOKEN = "test-only"
+
+
+def hetzner_project():
+    """One server with a 10 GB volume attached, answered the way the Hetzner Cloud API does."""
+    return {
+        "/servers": (200, {
+            "servers": [{
+                "id": 42, "name": "srv1", "status": "running",
+                "public_net": {
+                    "ipv4": {"id": 4201, "ip": "203.0.113.10", "blocked": False, "dns_ptr": "srv1.example.com"},
+                    "ipv6": None, "floating_ips": [], "firewalls": [],
+                },
+                "private_net": [],
+                "volumes": [7],
+            }],
+            "meta": {"pagination": {"page": 1, "per_page": 50, "previous_page": None, "next_page": None,
+                                    "last_page": 1, "total_entries": 1}},
+        }),
+        "/volumes/7": (200, {"volume": {"id": 7, "name": "data", "size": 10, "server": 42}}),
+    }
+
+
+@pytest.fixture
+def hetzner_api(monkeypatch):
+    """
+    Stands in for the Hetzner Cloud API behind hcloud. Maps an API token to the
+    responses of its project (API path -> (HTTP status, JSON body)), any other
+    token gets the 401 the real API answers with.
+    """
+    projects = dict()
+
+    def request(session, method, url, headers=None, **kwargs):
+        project = projects.get(headers["Authorization"].removeprefix("Bearer "))
+        if isinstance(project, Exception):
+            raise project
+        if project is None:
+            status, body = 401, {"error": {"code": "unauthorized", "message": "unable to authenticate"}}
+        else:
+            status, body = project[urlsplit(url).path.removeprefix("/v1")]
+        response = requests.Response()
+        response.status_code = status
+        response._content = json.dumps(body).encode()
+        return response
+
+    monkeypatch.setattr(requests.Session, "request", request)
+    return projects
+
+
+@pytest.fixture
+def hetzner_sources(inventory, load_config, hetzner_api):
+    def make(*sources):
+        load_config("".join(f"[source/{name}]\ntype = hetzner\napi_token = {token}\n" for name, token in sources))
+        return instantiate_sources()
+    return make
+
+
+def test_volume_size_follows_netbox_disk_unit(hetzner_api, hetzner_sources, inventory):
+    # NetBox stored virtual disk sizes in GB and switched to MB with 4.1.0
+    hetzner_api[TOKEN] = hetzner_project()
+    source, = hetzner_sources(("test", TOKEN))
+
+    sizes = dict()
+    for api_version in ["3.7.0", "4.0.11", "4.1.0", "4.3.0"]:
+        inventory.netbox_api_version = api_version
+        source.apply()
+        sizes[api_version] = [disk.data["size"] for disk in inventory.get_all_items(NBVirtualDisk)]
+
+    assert sizes == {"3.7.0": [10], "4.0.11": [10], "4.1.0": [10240], "4.3.0": [10240]}
+
+
+@pytest.mark.parametrize("failure", ["unauthorized", "unreachable"])
+def test_unusable_api_skips_source_and_keeps_the_others(hetzner_api, hetzner_sources, inventory, caplog, failure):
+    hetzner_api[TOKEN] = hetzner_project()
+    if failure == "unreachable":
+        hetzner_api["other-token"] = requests.ConnectionError("Failed to resolve 'api.hetzner.cloud'")
+
+    sources = hetzner_sources(("good", TOKEN), ("broken", "other-token"))
+
+    assert [source.name for source in sources] == ["good"]
+    # an enabled source which failed to initialize makes netbox-sync disable pruning
+    broken, = [source for source in inventory.source_list if source.name == "broken"]
+    assert broken.settings.enabled is True and broken.init_successful is False
+    reason = {"unauthorized": "unable to authenticate (unauthorized)",
+              "unreachable": "Failed to resolve 'api.hetzner.cloud'"}[failure]
+    assert any(record.levelname == "ERROR" and "'broken'" in record.getMessage() and reason in record.getMessage()
+               for record in caplog.records)
+
+
+def test_api_error_during_apply_exits_cleanly(hetzner_api, hetzner_sources, caplog):
+    project = hetzner_project()
+    project["/volumes/7"] = (503, {"error": {"code": "unavailable", "message": "service unavailable"}})
+    hetzner_api[TOKEN] = project
+    source, = hetzner_sources(("test", TOKEN))
+
+    with pytest.raises(SystemExit) as exit_info:
+        source.apply()
+
+    assert exit_info.value.code == 1
+    assert ("Hetzner source 'test': service unavailable (unavailable). "
+            "No NetBox writes or pruning will be performed.") in caplog.text