Merge pull request #3 from sown/infra

WIP: Netbox integration

.github/workflows/test.yml

@@ -36,4 +36,6 @@ jobs:
       - name: Lint
         run: make lint
       - name: Django Static Checks
-        run: make check
+        run: make check
+      - name: Unit tests
+        run: make test

Makefile

@@ -1,10 +1,10 @@
-.PHONY: all check clean dev format format-check lint lint-fix
+.PHONY: all check clean dev format format-check lint lint-fix test test-cov
 
 PYMODULE:=kmicms
 MANAGEPY:=$(CMD) ./$(PYMODULE)/manage.py
 APPS:=kmicms accounts core integrations pages
 
-all: format lint check
+all: format lint check test
 
 format:
 	ruff format $(PYMODULE)
@@ -27,4 +27,10 @@ dev:
 	$(MANAGEPY) runserver
 
 clean:
-	git clean -Xdf # Delete all files in .gitignore
+	git clean -Xdf # Delete all files in .gitignore
+
+test: | $(PYMODULE)
+	cd kmicms && DJANGO_SETTINGS_MODULE=kmicms.settings pytest --cov=. $(APPS) $(PYMODULE)
+
+test-cov:
+	cd kmicms && DJANGO_SETTINGS_MODULE=kmicms.settings pytest --cov=. $(APPS) $(PYMODULE) --cov-report html

kmicms/integrations/netbox/__init__.py

@@ -0,0 +1,7 @@
+from .client import NetboxClient
+from .exceptions import NetboxRequestError
+
+__all__ = [
+    "NetboxClient",
+    "NetboxRequestError",
+]

kmicms/integrations/netbox/client.py

@@ -0,0 +1,129 @@
+from __future__ import annotations
+
+from typing import Any
+
+import requests
+from django.core.cache import cache
+
+from .exceptions import NetboxRequestError
+from .queries import GET_DEVICE_QUERY, GET_VM_QUERY, LIST_DEVICE_QUERY, LIST_VM_QUERY
+
+
+class NetboxClient:
+    def __init__(
+        self, *, graphql_endpoint: str, api_token: str, cache_ttl_seconds: int, request_timeout: float
+    ) -> None:
+        self.graphql_endpoint = graphql_endpoint
+        self.api_token = api_token
+        self.cache_ttl_seconds = cache_ttl_seconds
+        self.request_timeout = request_timeout
+
+    def _get_headers(self) -> dict[str, str]:
+        return {
+            "Authorization": f"Token {self.api_token}",
+            "Accept": "application/json",
+            "Content-Type": "application/json",
+        }
+
+    def _get_payload(self, query: str, variables: dict[str, str] | None = None) -> dict[str, str]:
+        payload = {
+            "query": query,
+        }
+
+        if variables is not None:
+            payload["variables"] = variables
+
+        return payload
+
+    def _query(self, query: str, variables: dict[str, str] | None = None) -> dict[str, Any]:  # noqa: FA102
+        payload = self._get_payload(query, variables)
+
+        try:
+            resp = requests.post(
+                self.graphql_endpoint,
+                headers=self._get_headers(),
+                json=payload,
+                timeout=self.request_timeout,
+            )
+        except ConnectionError as e:
+            raise NetboxRequestError("Unable to connect to netbox") from e
+        except requests.RequestException as e:
+            raise NetboxRequestError("Error when requesting data from netbox") from e
+
+        try:
+            resp.raise_for_status()
+        except requests.RequestException as e:
+            # Include GraphQL errors in the exception message.
+            try:
+                data = resp.json()
+            except requests.exceptions.JSONDecodeError:
+                message = str(e)
+            else:
+                errors = data.get("errors")
+                message = f"{e}: GraphQL errors: {errors}"
+            raise NetboxRequestError(message) from e
+
+        try:
+            gql_response = resp.json()
+        except requests.exceptions.JSONDecodeError as e:
+            raise NetboxRequestError("Netbox returned invalid JSON") from e
+
+        # Check for and raise any GraphQL errors from successful responses.
+        if "errors" in gql_response:
+            errors = gql_response["errors"]
+            raise NetboxRequestError(f"Invalid GraphQL response: {errors}")
+
+        try:
+            return gql_response["data"]
+        except KeyError as e:
+            raise NetboxRequestError(
+                "Netbox API response did not contain data.",
+            ) from e
+
+    def get_device(self, device_id: int) -> dict[str, Any]:
+        cache_key = f"netbox__device__{device_id}"
+
+        if cache_value := cache.get(cache_key):
+            return cache_value
+
+        device = self._query(GET_DEVICE_QUERY, {"deviceId": device_id})["device"]
+        cache.set(cache_key, device, timeout=self.cache_ttl_seconds)
+
+        return device
+
+    def get_vm(self, vm_id: int) -> dict[str, Any]:
+        cache_key = f"netbox__vm__{vm_id}"
+
+        if cache_value := cache.get(cache_key):
+            return cache_value
+
+        vm = self._query(GET_VM_QUERY, {"VMId": vm_id})["virtual_machine"]
+        cache.set(cache_key, vm, timeout=self.cache_ttl_seconds)
+
+        return vm
+
+    def list_devices(
+        self,
+    ) -> dict[str, Any]:
+        cache_key = "netbox__device_list"
+
+        if cache_value := cache.get(cache_key):
+            return cache_value
+
+        device = self._query(LIST_DEVICE_QUERY)["device_list"]
+        cache.set(cache_key, device, timeout=self.cache_ttl_seconds)
+
+        return device
+
+    def list_vms(
+        self,
+    ) -> dict[str, Any]:
+        cache_key = "netbox__vm_list"
+
+        if cache_value := cache.get(cache_key):
+            return cache_value
+
+        vms = self._query(LIST_VM_QUERY)["virtual_machine_list"]
+        cache.set(cache_key, vms, timeout=self.cache_ttl_seconds)
+
+        return vms

kmicms/integrations/netbox/exceptions.py

@@ -0,0 +1,2 @@
+class NetboxRequestError(Exception):
+    """An error occurred in a request to Netbox."""

kmicms/integrations/netbox/queries.py

@@ -0,0 +1,128 @@
+GET_DEVICE_QUERY = """
+query DeviceInfo($deviceId: Int!) {
+  device(id: $deviceId) {
+    name
+    role {
+      name
+      color
+    }
+    description
+    comments
+    platform {
+      name
+    }
+    interfaces {
+      description
+      name
+      enabled
+      ip_addresses {
+        display
+        dns_name
+      }
+    }
+    device_type {
+      model
+      manufacturer {
+        name
+      }
+      front_image
+      rear_image
+    }
+    tags {
+      name
+      color
+    }
+    rack {
+      name
+    }
+    location {
+      name
+    }
+    position
+    status
+  }
+}
+"""
+
+LIST_DEVICE_QUERY = """
+query listDevices {
+  device_list {
+    id
+    name
+    role {
+      name
+      color
+    }
+    platform {
+      name
+    }
+    device_type {
+      model
+      manufacturer {
+        name
+      }
+    }
+    rack {
+      name
+    }
+    status
+  }
+}
+"""
+
+GET_VM_QUERY = """
+query VMInfo($VMId: Int!) {
+  virtual_machine(id: $VMId) {
+    name
+    cluster {
+      name
+    }
+    role {
+      name
+      color
+    }
+    description
+    comments
+    platform {
+      name
+    }
+    interfaces {
+      description
+      name
+      enabled
+      ip_addresses {
+        display
+        dns_name
+      }
+    }
+    vcpus
+    memory
+    disk
+    tags {
+      name
+      color
+    }
+    status
+  }
+}
+"""
+
+LIST_VM_QUERY = """
+query listVMs {
+  virtual_machine_list {
+    id
+    name
+    role {
+      name
+      color
+    }
+    platform {
+      name
+    }
+    cluster {
+      name
+    }
+    status
+  }
+}
+"""

kmicms/integrations/tests/conftest.py

@@ -0,0 +1,10 @@
+from collections.abc import Iterator
+
+import pytest
+import responses
+
+
+@pytest.fixture
+def mocked_responses() -> Iterator[responses.RequestsMock]:
+    with responses.RequestsMock() as requests_mock:
+        yield requests_mock

kmicms/integrations/tests/test_netbox.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+from http import HTTPStatus
+from typing import Any
+
+import pytest
+from responses import RequestsMock, matchers
+
+from integrations.netbox import NetboxClient
+
+
+class TestNetboxClient:
+    @pytest.fixture
+    def client(self) -> NetboxClient:
+        return NetboxClient(
+            graphql_endpoint="https://netbox.invalid/graphql/",
+            api_token="abc",  # noqa: S106
+            cache_ttl_seconds=300,
+            request_timeout=0.5,
+        )
+
+    def test_get_headers(self, client: NetboxClient) -> None:
+        expected_headers = {
+            "Authorization": "Token abc",
+            "Accept": "application/json",
+            "Content-Type": "application/json",
+        }
+        assert expected_headers == client._get_headers()
+
+    @pytest.mark.parametrize(
+        ("query", "variables", "expected_payload"),
+        [
+            pytest.param("query", None, {"query": "query"}, id="no-variables"),
+            pytest.param("query", {}, {"query": "query", "variables": {}}, id="empty-variables"),
+            pytest.param("query", {"a": 1}, {"query": "query", "variables": {"a": 1}}, id="variables"),
+        ],
+    )
+    def test_get_payload(
+        self,
+        client: NetboxClient,
+        query: str,
+        variables: dict[str, str] | None,
+        expected_payload: dict[str, Any],
+    ) -> None:
+        assert client._get_payload(query, variables) == expected_payload
+
+    def test_query(self, client: NetboxClient, mocked_responses: RequestsMock) -> None:
+        mocked_responses.post(
+            "https://netbox.invalid/graphql/",
+            match=[
+                matchers.header_matcher(
+                    {"Content-Type": "application/json", "Accept": "application/json", "Authorization": "Token abc"},
+                ),
+            ],
+            json={"data": {"a": "b"}},
+            status=HTTPStatus.OK,
+            content_type="application/json",
+        )
+        data = client._query("query")
+        assert data == {"a": "b"}

kmicms/kmicms/configuration.dev.py

@@ -31,3 +31,8 @@
 
 RECAPTCHA_PUBLIC_KEY = ""
 RECAPTCHA_PRIVATE_KEY = ""
+
+NETBOX_GRAPHQL_ENDPOINT = "https://netbox.example.com/graphql/"
+NETBOX_API_TOKEN = "abc"  # noqa: S105
+NETBOX_REQUEST_TIMEOUT = 0.5
+NETBOX_CACHE_TTL = 300

kmicms/kmicms/configuration.example.py

@@ -75,3 +75,10 @@
 
 # Time zone (default: UTC)
 TIME_ZONE = "UTC"
+
+# Netbox
+
+NETBOX_GRAPHQL_ENDPOINT = "https://netbox.example.com/graphql/"
+NETBOX_API_TOKEN = "abc"  # noqa: S105
+NETBOX_REQUEST_TIMEOUT = 0.5
+NETBOX_CACHE_TTL = 300