Adds support for NetBox v3.5+

.github/workflows/py3.yml

@@ -13,7 +13,7 @@ jobs:
     strategy:
       matrix:
         python: ["3.8", "3.9", "3.10"]
-        netbox: ["3.3"]
+        netbox: ["3.3", "3.4", "3.5"]
 
     steps:
       - uses: actions/checkout@v2

pynetbox/__init__.py

@@ -1,9 +1,6 @@
-from pkg_resources import get_distribution, DistributionNotFound
+from importlib.metadata import metadata
 
 from pynetbox.core.query import RequestError, AllocationError, ContentError
 from pynetbox.core.api import Api as api
 
-try:
-    __version__ = get_distribution(__name__).version
-except DistributionNotFound:
-    pass
+__version__ = metadata(__name__).get("Version")

pynetbox/core/api.py

@@ -27,6 +27,7 @@ class Api:
     you can specify which app and endpoint you wish to interact with.
 
     Valid attributes currently are:
+        * core (NetBox 3.5+)
         * dcim
         * ipam
         * circuits
@@ -74,6 +75,7 @@ def __init__(
         self.base_url = base_url
         self.http_session = requests.Session()
         self.threading = threading
+        self.core = App(self, "core")
         self.dcim = App(self, "dcim")
         self.ipam = App(self, "ipam")
         self.circuits = App(self, "circuits")

pynetbox/core/query.py

@@ -15,6 +15,7 @@
 """
 import concurrent.futures as cf
 import json
+from packaging import version
 
 
 def calc_pages(limit, count):
@@ -153,12 +154,22 @@ def __init__(
     def get_openapi(self):
         """Gets the OpenAPI Spec"""
         headers = {
-            "Content-Type": "application/json;",
+            "Accept": "application/json",
+            "Content-Type": "application/json",
         }
-        req = self.http_session.get(
-            "{}docs/?format=openapi".format(self.normalize_url(self.base)),
-            headers=headers,
-        )
+
+        current_version = version.parse(self.get_version())
+        if current_version >= version.parse("3.5"):
+            req = self.http_session.get(
+                "{}schema/".format(self.normalize_url(self.base)),
+                headers=headers,
+            )
+        else:
+            req = self.http_session.get(
+                "{}docs/?format=openapi".format(self.normalize_url(self.base)),
+                headers=headers,
+            )
+
         if req.ok:
             return req.json()
         else:
@@ -175,7 +186,7 @@ def get_version(self):
         present in the headers.
         """
         headers = {
-            "Content-Type": "application/json;",
+            "Content-Type": "application/json",
         }
         req = self.http_session.get(
             self.normalize_url(self.base),
@@ -192,7 +203,7 @@ def get_status(self):
         :Returns: Dictionary as returned by NetBox.
         :Raises: RequestError if request is not successful.
         """
-        headers = {"Content-Type": "application/json;"}
+        headers = {"Content-Type": "application/json"}
         if self.token:
             headers["authorization"] = "Token {}".format(self.token)
         req = self.http_session.get(
@@ -213,9 +224,9 @@ def normalize_url(self, url):
 
     def _make_call(self, verb="get", url_override=None, add_params=None, data=None):
         if verb in ("post", "put") or verb == "delete" and data:
-            headers = {"Content-Type": "application/json;"}
+            headers = {"Content-Type": "application/json"}
         else:
-            headers = {"accept": "application/json;"}
+            headers = {"accept": "application/json"}
 
         if self.token:
             headers["authorization"] = "Token {}".format(self.token)

pynetbox/models/dcim.py

@@ -17,7 +17,7 @@
 
 from pynetbox.core.query import Request
 from pynetbox.core.response import Record, JsonField
-from pynetbox.core.endpoint import RODetailEndpoint
+from pynetbox.core.endpoint import RODetailEndpoint, DetailEndpoint
 from pynetbox.models.ipam import IpAddresses
 from pynetbox.models.circuits import Circuits
 
@@ -121,6 +121,23 @@ def napalm(self):
         """
         return RODetailEndpoint(self, "napalm")
 
+    @property
+    def render_config(self):
+        """
+        Represents the ``render-config`` detail endpoint.
+
+        Returns a DetailEndpoint object that is the interface for
+        viewing response from the render-config endpoint.
+
+        :returns: :py:class:`.DetailEndpoint`
+
+        :Examples:
+
+        >>> device = nb.ipam.devices.get(123)
+        >>> device.render_config.create()
+        """
+        return DetailEndpoint(self, "render-config")
+
 
 class InterfaceConnections(Record):
     def __str__(self):

pynetbox/models/ipam.py

@@ -162,3 +162,33 @@ def available_vlans(self):
         NewVLAN (10)
         """
         return DetailEndpoint(self, "available-vlans", custom_return=Vlans)
+
+
+class AsnRanges(Record):
+    @property
+    def available_asns(self):
+        """
+        Represents the ``available-asns`` detail endpoint.
+
+        Returns a DetailEndpoint object that is the interface for
+        viewing and creating ASNs inside an ASN range.
+
+        :returns: :py:class:`.DetailEndpoint`
+
+        :Examples:
+
+        >>> asn_range = nb.ipam.asn_ranges.get(1)
+        >>> asn_range.available_asns.list()
+        [64512, 64513, 64514]
+
+        To create a new ASN:
+
+        >>> asn_range.available_asns.create()
+        64512
+
+        To create multiple ASNs:
+
+        >>> asn_range.available_asns.create([{} for i in range(2)])
+        [64513, 64514]
+        """
+        return DetailEndpoint(self, "available-asns")

pynetbox/models/mapper.py

@@ -59,7 +59,7 @@
     "dcim.poweroutlet": PowerOutlets,
     "dcim.poweroutlettemplate": None,
     "dcim.powerpanel": None,
-    "dcim.powerport": ConsolePorts,
+    "dcim.powerport": PowerPorts,
     "dcim.powerporttemplate": None,
     "dcim.rack": Racks,
     "dcim.rackreservation": RackReservations,

requirements-dev.txt

@@ -1,4 +1,4 @@
 black~=22.10
 pytest==7.1.*
 pytest-docker==1.0.*
-PyYAML==6.0
+PyYAML==6.0.1

requirements.txt

@@ -1 +1,2 @@
 requests>=2.20.0,<3.0
+packaging<24.0

setup.py

@@ -13,6 +13,7 @@
     packages=find_packages(exclude=["tests", "tests.*"]),
     install_requires=[
         "requests>=2.20.0,<3.0",
+        "packaging<24.0"
     ],
     zip_safe=False,
     keywords=["netbox"],

tests/integration/conftest.py

@@ -28,7 +28,11 @@ def get_netbox_docker_version_tag(netbox_version):
     major, minor = netbox_version.major, netbox_version.minor
 
     if (major, minor) == (3, 3):
-        tag = "2.2.0"
+        tag = "2.3.0"
+    elif (major, minor) == (3, 4):
+        tag = "2.5.3"
+    elif (major, minor) == (3, 5):
+        tag = "2.6.1"
     else:
         raise NotImplementedError(
             "Version %s is not currently supported" % netbox_version
@@ -48,7 +52,7 @@ def git_toplevel():
     try:
         subp.check_call(["which", "git"])
     except subp.CalledProcessError:
-        pytest.skip(msg="git executable was not found on the host")
+        pytest.skip(reason="git executable was not found on the host")
     return (
         subp.check_output(["git", "rev-parse", "--show-toplevel"])
         .decode("utf-8")
@@ -73,7 +77,7 @@ def netbox_docker_repo_dirpaths(pytestconfig, git_toplevel):
     try:
         subp.check_call(["which", "docker"])
     except subp.CalledProcessError:
-        pytest.skip(msg="docker executable was not found on the host")
+        pytest.skip(reason="docker executable was not found on the host")
     netbox_versions_by_repo_dirpaths = {}
     for netbox_version in pytestconfig.option.netbox_versions:
         repo_version_tag = get_netbox_docker_version_tag(netbox_version=netbox_version)
@@ -248,6 +252,14 @@ def docker_compose_file(pytestconfig, netbox_docker_repo_dirpaths):
                         "netboxcommunity/netbox:v%s" % netbox_version
                     )
 
+                    new_services[new_service_name]["environment"] = {
+                        "SKIP_SUPERUSER": "false",
+                        "SUPERUSER_API_TOKEN": "0123456789abcdef0123456789abcdef01234567",
+                        "SUPERUSER_EMAIL": "[email protected]",
+                        "SUPERUSER_NAME": "admin",
+                        "SUPERUSER_PASSWORD": "admin",
+                    }
+
                 if service_name == "netbox":
                     # ensure the netbox container listens on a random port
                     new_services[new_service_name]["ports"] = ["8080"]
@@ -341,7 +353,7 @@ def docker_compose_file(pytestconfig, netbox_docker_repo_dirpaths):
 
 
 def netbox_is_responsive(url):
-    """Chack if the HTTP service is up and responsive."""
+    """Check if the HTTP service is up and responsive."""
     try:
         response = requests.get(url)
         if response.status_code == 200:

tests/test_circuits.py

@@ -11,7 +11,7 @@
 
 nb = api.circuits
 
-HEADERS = {"accept": "application/json;"}
+HEADERS = {"accept": "application/json"}
 
 
 class Generic:

tests/test_tenancy.py

@@ -11,7 +11,7 @@
 
 nb = api.tenancy
 
-HEADERS = {"accept": "application/json;"}
+HEADERS = {"accept": "application/json"}
 
 
 class Generic:

tests/test_users.py

@@ -11,7 +11,7 @@
 
 nb = api.users
 
-HEADERS = {"accept": "application/json;"}
+HEADERS = {"accept": "application/json"}
 
 
 class Generic:

tests/test_virtualization.py

@@ -11,7 +11,7 @@
 
 nb = api.virtualization
 
-HEADERS = {"accept": "application/json;"}
+HEADERS = {"accept": "application/json"}
 
 
 class Generic:

tests/test_wireless.py

@@ -9,7 +9,7 @@
 
 nb_app = api.wireless
 
-HEADERS = {"accept": "application/json;"}
+HEADERS = {"accept": "application/json"}
 
 
 class Generic:

tests/unit/test_query.py

@@ -20,15 +20,15 @@ def test_get_count(self):
         expected = call(
             "http://localhost:8001/api/dcim/devices/",
             params={"q": "abcd", "limit": 1},
-            headers={"accept": "application/json;"},
+            headers={"accept": "application/json"},
         )
         test_obj.http_session.get.ok = True
         test = test_obj.get_count()
         self.assertEqual(test, 42)
         test_obj.http_session.get.assert_called_with(
             "http://localhost:8001/api/dcim/devices/",
             params={"q": "abcd", "limit": 1},
-            headers={"accept": "application/json;"},
+            headers={"accept": "application/json"},
             json=None,
         )
 
@@ -49,7 +49,7 @@ def test_get_count_no_filters(self):
         test_obj.http_session.get.assert_called_with(
             "http://localhost:8001/api/dcim/devices/",
             params={"limit": 1},
-            headers={"accept": "application/json;"},
+            headers={"accept": "application/json"},
             json=None,
         )
 
@@ -69,14 +69,14 @@ def test_get_manual_pagination(self):
         expected = call(
             "http://localhost:8001/api/dcim/devices/",
             params={"offset": 20, "limit": 10},
-            headers={"accept": "application/json;"},
+            headers={"accept": "application/json"},
         )
         test_obj.http_session.get.ok = True
         generator = test_obj.get()
         self.assertEqual(len(list(generator)), 4)
         test_obj.http_session.get.assert_called_with(
             "http://localhost:8001/api/dcim/devices/",
             params={"offset": 20, "limit": 10},
-            headers={"accept": "application/json;"},
+            headers={"accept": "application/json"},
             json=None,
         )

tests/unit/test_request.py

@@ -5,10 +5,32 @@
 
 
 class RequestTestCase(unittest.TestCase):
-    def test_get_openapi(self):
+    def test_get_openapi_version_less_than_3_5(self):
         test = Request("http://localhost:8080/api", Mock())
+        test.get_version = Mock(return_value="3.4")
+
+        # Mock the HTTP response
+        response_mock = Mock()
+        response_mock.ok = True
+        test.http_session.get.return_value = response_mock
+
         test.get_openapi()
         test.http_session.get.assert_called_with(
             "http://localhost:8080/api/docs/?format=openapi",
-            headers={"Content-Type": "application/json;"},
+            headers={"Accept": "application/json", "Content-Type": "application/json"},
+        )
+
+    def test_get_openapi_version_3_5_or_greater(self):
+        test = Request("http://localhost:8080/api", Mock())
+        test.get_version = Mock(return_value="3.5")
+
+        # Mock the HTTP response
+        response_mock = Mock()
+        response_mock.ok = True
+        test.http_session.get.return_value = response_mock
+
+        test.get_openapi()
+        test.http_session.get.assert_called_with(
+            "http://localhost:8080/api/schema/",
+            headers={"Accept": "application/json", "Content-Type": "application/json"},
         )