Adds support for NetBox v3.5+ (#564)

* Fix unused `PowerPorts` model (#535)

Updated models/mapper.py: set `PowerPorts` for "dcim.powerport" in the
map-dict.

* migrate from pkg_resources to importlib

* adds core app

* adds endpoints added in 3.5

* adds support for 3.5

* lint fixes

* updates openapi tests

* adds testing for 3.4 and 3.5

* updates pytest.skip

* updates docker tags for testing

* fixes superuser account creation for testing

* fixed requirements

* updates the docstring for render-config endpoint

* removes extra semicolon from content type value

* migrate from pkg_resources to importlib

* adds core app

* adds endpoints added in 3.5

* adds support for 3.5

* lint fixes

* updates openapi tests

* adds testing for 3.4 and 3.5

* updates pytest.skip

* updates docker tags for testing

* fixes superuser account creation for testing

* fixed requirements

* updates the docstring for render-config endpoint

* removes extra semicolon from content type value

---------

Co-authored-by: nautics889 <cyberukr@gmail.com>

Author: abhi1693

.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")

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": "admin@example.com",
+                        "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: