NetBox 4.6 support

.github/workflows/ci.yml

@@ -23,7 +23,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        netbox_version: &nb_versions [v4.3.7, v4.4.10, v4.5.5]
+        netbox_version: &nb_versions [v4.4.10, v4.5.8, v4.6.0]
     steps:
       - name: Checkout
         uses: actions/checkout@v3

requirements/dev.txt

@@ -2,9 +2,9 @@ debugpy==1.8.19
 factory_boy==3.3.3
 ipython
 pre-commit==4.5.1
-pytest==9.0.2
-pytest-cov==7.0.0
-pytest-django==4.11.1
+pytest==9.0.3
+pytest-cov==7.1.0
+pytest-django==4.12.0
 pytest-subtests==0.15.0
 ruff==0.14.11
 tomli==2.4.0

validity/__init__.py

@@ -19,8 +19,8 @@ class NetBoxValidityConfig(PluginConfig):
     version = "3.4.3"
     base_url = "validity"
     django_apps = ["django_bootstrap5"]
-    min_version = "4.3.0"
-    max_version = "4.5.99"
+    min_version = "4.4.0"
+    max_version = "4.6.99"
     graphql_schema = "graphql.schema"
 
     # custom field

validity/api/helpers.py

@@ -1,5 +1,4 @@
 from django.core.exceptions import ValidationError
-from django.db.models import ManyToManyField
 from rest_framework.permissions import BasePermission
 from rest_framework.relations import PrimaryKeyRelatedField
 from rest_framework.serializers import HyperlinkedIdentityField, JSONField, ModelSerializer
@@ -82,7 +81,7 @@ class SubformValidationMixin:
     def _validate(self, attrs):
         instance = self.instance or self.Meta.model()
         for field, field_value in attrs.items():
-            if not isinstance(instance._meta.get_field(field), ManyToManyField):
+            if not instance._meta.get_field(field).many_to_many:
                 setattr(instance, field, field_value)
         if not instance.subform_type:
             return

validity/choices.py

@@ -1,11 +1,12 @@
 from typing import Optional, TypeVar
 
 from django.db.models import IntegerChoices, TextChoices
-from django.db.models.enums import ChoicesMeta
 from django.utils.translation import gettext_lazy as _
 
+from validity.netbox_changes import ChoicesType
 
-class ColoredChoiceMeta(ChoicesMeta):
+
+class ColoredChoiceMeta(ChoicesType):
     """
     Allows to write choice fields with a color like that:
         option1 = 'red'

validity/netbox_changes/__init__.py

@@ -16,9 +16,9 @@ def get_base_table_kwargs(self):
 
 StrFilterLookup = locate("strawberry_django.StrFilterLookup") if config.netbox_version >= "4.5.5" else FilterLookup[str]
 
-if config.netbox_version >= "4.5.0":
+if config.netbox_version >= "4.6.0":
     from .current import *
-elif config.netbox_version >= "4.4.0":
+elif config.netbox_version >= "4.5.0":
     from .old import *
 else:
     from .oldest import *

validity/netbox_changes/current.py

@@ -1,8 +1,8 @@
-# NetBox 4.5
+# NetBox 4.6
+
 from pydoc import locate
 
 from .old import *
 
 
-BaseModelFilter = locate("netbox.graphql.filters.BaseModelFilter")
-NetBoxModelFilter = locate("netbox.graphql.filters.NetBoxModelFilter")
+ChoicesType = locate("django.db.models.enums.ChoicesType")

validity/netbox_changes/old.py

@@ -1,23 +1,8 @@
-# NetBox 4.4
-from typing import TYPE_CHECKING
+# NetBox 4.5
+from pydoc import locate
 
 from .oldest import *
 
 
-if TYPE_CHECKING:
-    from validity.models import ComplianceReport
-    from validity.scripts.data_models import RequestInfo
-
-
-def get_logs(job):
-    return job.log_entries
-
-
-def set_logs(job, logs):
-    job.log_entries = logs
-
-
-def enqueue(report: "ComplianceReport", request: "RequestInfo"):
-    queue = {}
-    enqueue_event(queue, report, request, "object_created")
-    flush_events(queue.values())
+BaseModelFilter = locate("netbox.graphql.filters.BaseModelFilter")
+NetBoxModelFilter = locate("netbox.graphql.filters.NetBoxModelFilter")

validity/netbox_changes/oldest.py

@@ -1,30 +1,7 @@
-# NetBox 4.3
+# NetBox 4.4
 from pydoc import locate
-from typing import TYPE_CHECKING
-
-from extras.events import enqueue_event, flush_events
-
-
-if TYPE_CHECKING:
-    from validity.models import ComplianceReport
-    from validity.scripts.data_models import RequestInfo
 
 
 BaseModelFilter = locate("netbox.graphql.filter_mixins.BaseFilterMixin")
 NetBoxModelFilter = locate("netbox.graphql.filter_mixins.NetBoxModelFilterMixin")
-
-
-def get_logs(job):
-    return job.data["log"]
-
-
-def set_logs(job, logs):
-    if not isinstance(job.data, dict):
-        job.data = {}
-    job.data["log"] = logs
-
-
-def enqueue(report: "ComplianceReport", request: "RequestInfo"):
-    queue = {}
-    enqueue_event(queue, report, request.get_user(), request.id, "object_created")
-    flush_events(queue.values())
+ChoicesType = locate("django.db.models.enums.ChoicesMeta")

validity/scripts/keeper.py

@@ -5,7 +5,6 @@
 from core.models import Job
 
 from validity import di
-from validity.netbox_changes import set_logs
 from validity.utils.logger import Logger
 from .exceptions import AbortScript
 
@@ -48,7 +47,7 @@ def terminate_job(
         self, status: str = JobStatusChoices.STATUS_COMPLETED, error: str | None = None, output=None
     ) -> None:
         self.job.data = self.job.data or {}
-        set_logs(self.job, [log.serialized for log in self.logger.messages])
+        self.job.log_entries = [log.serialized for log in self.logger.messages]
         output = output or self.job.data.get("output")
         self.job.data["output"] = output
         self.job.terminate(status, error)

validity/scripts/runtests/combine.py

@@ -10,10 +10,10 @@
 from dimi import Singleton
 from django.db.models import QuerySet
 from django.urls import reverse
+from extras.events import enqueue_event, flush_events
 
 from validity import di
 from validity.models import ComplianceReport, ComplianceTestResult
-from validity.netbox_changes import enqueue
 from validity.utils.logger import Logger, Message
 from ..data_models import FullRunTestsParams, RequestInfo, TestResultRatio
 from ..exceptions import AbortScript
@@ -22,6 +22,12 @@
 from ..parent_jobs import JobExtractor
 
 
+def enqueue(report: "ComplianceReport", request: "RequestInfo"):
+    queue = {}
+    enqueue_event(queue, report, request, "object_created")
+    flush_events(queue.values())
+
+
 @di.dependency(scope=Singleton)
 @dataclass(repr=False, kw_only=True)
 class CombineWorker:

validity/tests/test_scripts/runtests/test_combine.py

@@ -5,7 +5,6 @@
 import pytest
 from django.utils import timezone
 
-from validity.netbox_changes import get_logs
 from validity.scripts.data_models import ExecutionResult, Message
 from validity.scripts.data_models import TestResultRatio as ResultRatio
 from validity.scripts.runtests.combine import CombineWorker
@@ -59,7 +58,7 @@ def test_call_abort(worker, full_runtests_params, job_extractor, monkeypatch):
     worker(full_runtests_params)
     job = full_runtests_params.get_job()
     assert job.status == "errored"
-    assert get_logs(job) == [
+    assert job.log_entries == [
         {"message": "m-3", "status": "info", "time": "2000-01-01T00:00:00"},
         {"message": "m-4", "status": "info", "time": "2000-01-01T00:00:00"},
         {"message": "ApplyWorkerError", "status": "failure", "time": "2020-01-01T00:00:00"},
@@ -78,7 +77,7 @@ def test_successful_call(worker, full_runtests_params, job_extractor, monkeypatc
     job.refresh_from_db()
     assert job.status == "completed"
     assert job.data["output"] == {"statistics": {"total": 7, "passed": 3}}
-    assert get_logs(job) == [
+    assert job.log_entries == [
         *[m.serialized for m in messages],
         {
             "time": "2020-01-01T00:00:00",

validity/tests/test_scripts/test_backup.py

@@ -7,7 +7,6 @@
 
 from validity.data_backup import BackupBackend
 from validity.integrations.errors import IntegrationError
-from validity.netbox_changes import get_logs
 from validity.scripts.backup import perform_backup
 from validity.scripts.data_models import FullBackUpParams
 
@@ -35,7 +34,7 @@ def test_backup_success(di, params):
     assert bp.last_status == "completed"
     assert bp.last_uploaded < timezone.now()
     assert job.status == "completed"
-    assert get_logs(job)
+    assert job.log_entries
 
 
 @pytest.mark.django_db

validity/tests/test_scripts/test_keeper.py

@@ -3,7 +3,6 @@
 import pytest
 from factories import DSBackupJobFactory
 
-from validity.netbox_changes import get_logs
 from validity.scripts.exceptions import AbortScript
 from validity.scripts.keeper import JobKeeper
 from validity.utils.logger import Logger, Message
@@ -17,7 +16,7 @@ def test_keeper_noerror(timezone_now):
         keeper.logger.info("msg")
     keeper.job.refresh_from_db()
     assert keeper.job.status == "completed"
-    assert get_logs(keeper.job) == [{"time": "2000-01-01T01:00:00+00:00", "status": "info", "message": "msg"}]
+    assert keeper.job.log_entries == [{"time": "2000-01-01T01:00:00+00:00", "status": "info", "message": "msg"}]
     assert keeper.logger.messages == []
 
 
@@ -29,7 +28,7 @@ def test_keeper_abort(timezone_now):
         raise AbortScript("abort_msg", logs=[Message("warning", "extra_msg")])
     keeper.job.refresh_from_db()
     assert keeper.job.status == "failed"
-    assert get_logs(keeper.job) == [
+    assert keeper.job.log_entries == [
         {"time": "2000-01-01T01:00:00+00:00", "status": "info", "message": "msg1"},
         {"time": "2000-01-01T01:00:00+00:00", "status": "warning", "message": "extra_msg"},
         {"time": "2000-01-01T01:00:00+00:00", "status": "failure", "message": "abort_msg"},
@@ -46,7 +45,7 @@ def test_keeper_exception(timezone_now):
             raise ValueError("unexpected")
     keeper.job.refresh_from_db()
     assert keeper.job.status == "errored"
-    logs = get_logs(keeper.job)
+    logs = keeper.job.log_entries
     assert len(logs) == 2
     assert logs[0] == {"time": "2000-01-01T01:00:00+00:00", "status": "info", "message": "msg1"}
     assert logs[1]["status"] == "failure"

validity/views/script.py

@@ -12,7 +12,7 @@
 
 from validity import di
 from validity.forms import RunTestsForm
-from validity.netbox_changes import get_base_table_kwargs, get_logs
+from validity.netbox_changes import get_base_table_kwargs
 from validity.scripts import Launcher, RunTestsParams, ScriptParams
 from validity.tables import ScriptResultTable
 from .base import LauncherMixin
@@ -51,7 +51,7 @@ class ScriptResultView(PermissionRequiredMixin, TableMixin, ObjectView):
     permission_required = "core.view_job"
 
     def get_table(self, job, request, bulk_actions=False):
-        logs = [entry | {"index": i} for i, entry in enumerate(get_logs(job), start=1)]
+        logs = [entry | {"index": i} for i, entry in enumerate(job.log_entries, start=1)]
         table = self.table_class(logs, **self.get_table_kwargs())
         table.configure(request)
         return table