diff --git a/.github/workflows/py2pr.yml b/.github/workflows/py2pr.yml new file mode 100644 index 00000000..93ecc98f --- /dev/null +++ b/.github/workflows/py2pr.yml @@ -0,0 +1,26 @@ +name: PR Test + +on: [pull_request] + +jobs: + build: + + runs-on: ubuntu-latest + strategy: + matrix: + python: [2.7] + + steps: + - uses: actions/checkout@v2 + + - name: Setup Python + uses: actions/setup-python@v2 + with: + python-version: ${{ matrix.python }} + + - name: Install pynetbox and testing packages. + run: pip install . mock + + - name: Run Tests + run: python -m unittest discover + diff --git a/.github/workflows/pr.yml b/.github/workflows/py3pr.yml similarity index 66% rename from .github/workflows/pr.yml rename to .github/workflows/py3pr.yml index c42d8d29..2e5ecdfd 100644 --- a/.github/workflows/pr.yml +++ b/.github/workflows/py3pr.yml @@ -8,7 +8,7 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - python: [2.7, 3.6] + python: [3.6, 3.8] steps: - uses: actions/checkout@v2 @@ -19,13 +19,11 @@ jobs: python-version: ${{ matrix.python }} - name: Install pynetbox and testing packages. - run: pip install . pycodestyle mock + run: pip install . black pytest - name: Run Linter - run: | - pycodestyle pynetbox - pycodestyle --ignore=E501 tests + run: black --check . - name: Run Tests - run: python -m unittest discover + run: pytest diff --git a/docs/advanced.rst b/docs/advanced.rst new file mode 100644 index 00000000..991eb17d --- /dev/null +++ b/docs/advanced.rst @@ -0,0 +1,75 @@ +Custom Sessions +=============== + +Custom sessions can be used to modify the default HTTP behavior. Below are a few examples, most of them from `here `_. + +Headers +******* + +To set a custom header on all requests. These headers are automatically merged with headers pynetbox sets itself. + +:Example: + +>>> import pynetbox +>>> import requests +>>> session = requests.Session() +>>> session.headers = {"mycustomheader": "test"} +>>> nb = pynetbox.api( +... 'http://localhost:8000', +... private_key_file='/path/to/private-key.pem', +... token='d6f4e314a5b5fefd164995169f28ae32d987704f' +... ) +>>> nb.http_session = session + + +SSL Verification +**************** + +To disable SSL verification. See `the docs `_. + +:Example: + +>>> import pynetbox +>>> import requests +>>> session = requests.Session() +>>> session.verify = False +>>> nb = pynetbox.api( +... 'http://localhost:8000', +... private_key_file='/path/to/private-key.pem', +... token='d6f4e314a5b5fefd164995169f28ae32d987704f' +... ) +>>> nb.http_session = session + + +Timeouts +******** + +Setting timeouts requires the use of Adapters. + +:Example: + +.. code-block:: python + + from requests.adapters import HTTPAdapter + + class TimeoutHTTPAdapter(HTTPAdapter): + def __init__(self, *args, **kwargs): + self.timeout = kwargs.get("timeout", 5) + super().__init__(*args, **kwargs) + + def send(self, request, **kwargs): + kwargs['timeout'] = self.timeout + return super().send(request, **kwargs) + + adapter = TimeoutHTTPAdapter() + session = requests.Session() + session.mount("http://", adapter) + session.mount("https://", adapter) + + nb = pynetbox.api( + 'http://localhost:8000', + private_key_file='/path/to/private-key.pem', + token='d6f4e314a5b5fefd164995169f28ae32d987704f' + ) + nb.http_session = session + diff --git a/docs/conf.py b/docs/conf.py index c1d27184..05b09fce 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -19,7 +19,8 @@ import os import sys from pkg_resources import get_distribution -sys.path.insert(0, os.path.abspath('../')) + +sys.path.insert(0, os.path.abspath("../")) # -- General configuration ------------------------------------------------ @@ -31,33 +32,33 @@ # Add any Sphinx extension module names here, as strings. They can be # extensions coming with Sphinx (named 'sphinx.ext.*') or your custom # ones. -extensions = ['sphinx.ext.autodoc'] +extensions = ["sphinx.ext.autodoc"] # Add any paths that contain templates here, relative to this directory. -templates_path = ['_templates'] +templates_path = ["_templates"] # The suffix(es) of source filenames. # You can specify multiple suffix as a list of string: # # source_suffix = ['.rst', '.md'] -source_suffix = '.rst' +source_suffix = ".rst" # The master toctree document. -master_doc = 'index' +master_doc = "index" # General information about the project. -project = u'pynetbox' -copyright = u'2017, DigitalOcean' -author = u'Zach Moody' +project = u"pynetbox" +copyright = u"2017, DigitalOcean" +author = u"Zach Moody" # The version info for the project you're documenting, acts as replacement for # |version| and |release|, also used in various other places throughout the # built documents. # The full version, including alpha/beta/rc tags. -release = get_distribution('pynetbox').version +release = get_distribution("pynetbox").version # # The short X.Y version. -version = '.'.join(release.split('.')[:2]) +version = ".".join(release.split(".")[:2]) # The language for content autogenerated by Sphinx. Refer to documentation # for a list of supported languages. @@ -69,10 +70,10 @@ # List of patterns, relative to source directory, that match files and # directories to ignore when looking for source files. # This patterns also effect to html_static_path and html_extra_path -exclude_patterns = ['_build', 'Thumbs.db', '.DS_Store'] +exclude_patterns = ["_build", "Thumbs.db", ".DS_Store"] # The name of the Pygments (syntax highlighting) style to use. -pygments_style = 'sphinx' +pygments_style = "sphinx" # If true, `todo` and `todoList` produce output, else they produce nothing. todo_include_todos = False @@ -96,18 +97,14 @@ # so a file named "default.css" will overwrite the builtin "default.css". # html_static_path = ['_static'] -html_sidebars = {'**': [ - 'globaltoc.html', - 'relations.html', - 'sourcelink.html', - 'searchbox.html' - ] +html_sidebars = { + "**": ["globaltoc.html", "relations.html", "sourcelink.html", "searchbox.html"] } # -- Options for HTMLHelp output ------------------------------------------ # Output file base name for HTML help builder. -htmlhelp_basename = 'pynetboxdoc' +htmlhelp_basename = "pynetboxdoc" # -- Options for LaTeX output --------------------------------------------- @@ -116,15 +113,12 @@ # The paper size ('letterpaper' or 'a4paper'). # # 'papersize': 'letterpaper', - # The font size ('10pt', '11pt' or '12pt'). # # 'pointsize': '10pt', - # Additional stuff for the LaTeX preamble. # # 'preamble': '', - # Latex figure (float) alignment # # 'figure_align': 'htbp', @@ -134,8 +128,7 @@ # (source start file, target name, title, # author, documentclass [howto, manual, or own class]). latex_documents = [ - (master_doc, 'pynetbox.tex', u'pynetbox Documentation', - u'Zach Moody', 'manual'), + (master_doc, "pynetbox.tex", u"pynetbox Documentation", u"Zach Moody", "manual"), ] @@ -143,10 +136,7 @@ # One entry per manual page. List of tuples # (source start file, name, description, authors, manual section). -man_pages = [ - (master_doc, 'pynetbox', u'pynetbox Documentation', - [author], 1) -] +man_pages = [(master_doc, "pynetbox", u"pynetbox Documentation", [author], 1)] # -- Options for Texinfo output ------------------------------------------- @@ -155,7 +145,13 @@ # (source start file, target name, title, author, # dir menu entry, description, category) texinfo_documents = [ - (master_doc, 'pynetbox', u'pynetbox Documentation', - author, 'pynetbox', 'A python library for NetBox.', - 'Miscellaneous'), + ( + master_doc, + "pynetbox", + u"pynetbox Documentation", + author, + "pynetbox", + "A python library for NetBox.", + "Miscellaneous", + ), ] diff --git a/docs/index.rst b/docs/index.rst index 3b3f6bfc..36886b72 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -14,13 +14,13 @@ Instantiate the :py:class:`.Api`. Use the methods available on :py:class:`.Endpo API === -.. autoclass:: pynetbox.api.Api +.. autoclass:: pynetbox.core.api.Api :members: App === -.. autoclass:: pynetbox.api.App +.. autoclass:: pynetbox.core.app.App :members: Indices and tables diff --git a/pynetbox/__init__.py b/pynetbox/__init__.py index 610400bd..0e1f5100 100644 --- a/pynetbox/__init__.py +++ b/pynetbox/__init__.py @@ -1,7 +1,7 @@ from pkg_resources import get_distribution, DistributionNotFound from pynetbox.core.query import RequestError, AllocationError, ContentError -from pynetbox.api import Api as api +from pynetbox.core.api import Api as api try: __version__ = get_distribution(__name__).version diff --git a/pynetbox/api.py b/pynetbox/api.py deleted file mode 100644 index 32449d9c..00000000 --- a/pynetbox/api.py +++ /dev/null @@ -1,232 +0,0 @@ -""" -(c) 2017 DigitalOcean - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -""" -import sys - -import requests - -from pynetbox.core.endpoint import Endpoint -from pynetbox.core.query import Request -from pynetbox.models import dcim, ipam, virtualization, circuits, extras - - -class App(object): - """ Represents apps in NetBox. - - Calls to attributes are returned as Endpoint objects. - - :returns: :py:class:`.Endpoint` matching requested attribute. - :raises: :py:class:`.RequestError` - if requested endpoint doesn't exist. - """ - def __init__(self, api, name): - self.api = api - self.name = name - self._choices = None - self._setmodel() - - models = { - "dcim": dcim, - "ipam": ipam, - "circuits": circuits, - "virtualization": virtualization, - "extras": extras - } - - def _setmodel(self): - self.model = App.models[self.name] if self.name in App.models else None - - def __getstate__(self): - return { - 'api': self.api, - 'name': self.name, - '_choices': self._choices - } - - def __setstate__(self, d): - self.__dict__.update(d) - self._setmodel() - - def __getattr__(self, name): - return Endpoint(self.api, self, name, model=self.model) - - def choices(self): - """ Returns _choices response from App - - .. note:: - - This method is deprecated and only works with NetBox version 2.7.x - or older. The ``choices()`` method in :py:class:`.Endpoint` is - compatible with all NetBox versions. - - :Returns: Raw response from NetBox's _choices endpoint. - """ - if self._choices: - return self._choices - - self._choices = Request( - base="{}/{}/_choices/".format(self.api.base_url, self.name), - token=self.api.token, - private_key=self.api.private_key, - ssl_verify=self.api.ssl_verify, - http_session=self.api.http_session, - ).get() - - return self._choices - - def custom_choices(self): - """ Returns _custom_field_choices response from app - - :Returns: Raw response from NetBox's _custom_field_choices endpoint. - :Raises: :py:class:`.RequestError` if called for an invalid endpoint. - :Example: - - >>> nb.extras.custom_choices() - {'Testfield1': {'Testvalue2': 2, 'Testvalue1': 1}, - 'Testfield2': {'Othervalue2': 4, 'Othervalue1': 3}} - """ - custom_field_choices = Request( - base="{}/{}/_custom_field_choices/".format( - self.api.base_url, - self.name, - ), - token=self.api.token, - private_key=self.api.private_key, - ssl_verify=self.api.ssl_verify, - http_session=self.api.http_session, - ).get() - return custom_field_choices - - -class Api(object): - """ The API object is the point of entry to pynetbox. - - After instantiating the Api() with the appropriate named arguments - you can specify which app and endpoint you wish to interact with. - - Valid attributes currently are: - * dcim - * ipam - * circuits - * secrets - * tenancy - * extras - * virtualization - - Calling any of these attributes will return - :py:class:`.App` which exposes endpoints as attributes. - - :type ssl_verify: bool or str - :param str url: The base URL to the instance of NetBox you - wish to connect to. - :param str token: Your NetBox token. - :param str,optional private_key_file: The path to your private - key file. - :param str,optional private_key: Your private key. - :param bool/str,optional ssl_verify: Specify SSL verification behavior - see: requests_. - :param bool,optional threading: Set to True to use threading in ``.all()`` - and ``.filter()`` requests. - :raises ValueError: If *private_key* and *private_key_file* are both - specified. - :raises AttributeError: If app doesn't exist. - :Examples: - - >>> import pynetbox - >>> nb = pynetbox.api( - ... 'http://localhost:8000', - ... private_key_file='/path/to/private-key.pem', - ... token='d6f4e314a5b5fefd164995169f28ae32d987704f' - ... ) - >>> nb.dcim.devices.all() - - .. _requests: http://docs.python-requests.org/en/master/user/advanced/#ssl-cert-verification - """ # noqa - - def __init__( - self, - url, - token=None, - private_key=None, - private_key_file=None, - ssl_verify=True, - threading=False, - ): - if private_key and private_key_file: - raise ValueError( - '"private_key" and "private_key_file" cannot be used together.' - ) - base_url = "{}/api".format(url if url[-1] != "/" else url[:-1]) - self.token = token - self.private_key = private_key - self.private_key_file = private_key_file - self.base_url = base_url - self.ssl_verify = ssl_verify - self.session_key = None - self.http_session = requests.Session() - if threading and sys.version_info.major == 2: - raise NotImplementedError("Threaded pynetbox calls not supported \ - in Python 2") - self.threading = threading - - if self.private_key_file: - with open(self.private_key_file, "r") as kf: - private_key = kf.read() - self.private_key = private_key - - req = Request( - base=base_url, - token=token, - private_key=private_key, - ssl_verify=ssl_verify, - http_session=self.http_session - ) - if self.token and self.private_key: - self.session_key = req.get_session_key() - - self.dcim = App(self, "dcim") - self.ipam = App(self, "ipam") - self.circuits = App(self, "circuits") - self.secrets = App(self, "secrets") - self.tenancy = App(self, "tenancy") - self.extras = App(self, "extras") - self.virtualization = App(self, "virtualization") - - @property - def version(self): - """ Gets the API version of NetBox. - - Can be used to check the NetBox API version if there are - version-dependent features or syntaxes in the API. - - :Returns: Version number as a string. - :Example: - - >>> import pynetbox - >>> nb = pynetbox.api( - ... 'http://localhost:8000', - ... private_key_file='/path/to/private-key.pem', - ... token='d6f4e314a5b5fefd164995169f28ae32d987704f' - ... ) - >>> nb.version - '2.6' - >>> - """ - version = Request( - base=self.base_url, - ssl_verify=self.ssl_verify, - http_session=self.http_session, - ).get_version() - return version diff --git a/pynetbox/core/api.py b/pynetbox/core/api.py new file mode 100644 index 00000000..c557a773 --- /dev/null +++ b/pynetbox/core/api.py @@ -0,0 +1,149 @@ +""" +(c) 2017 DigitalOcean + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" +import sys + +import requests + +from pynetbox.core.query import Request +from pynetbox.core.app import App + + +class Api(object): + """The API object is the point of entry to pynetbox. + + After instantiating the Api() with the appropriate named arguments + you can specify which app and endpoint you wish to interact with. + + Valid attributes currently are: + * dcim + * ipam + * circuits + * secrets + * tenancy + * extras + * virtualization + + Calling any of these attributes will return + :py:class:`.App` which exposes endpoints as attributes. + + **Additional Attributes**: + * **http_session(requests.Session)**: + Override the default session with your own. This is used to control + a number of HTTP behaviors such as SSL verification, custom headers, + retires, and timeouts. + See `custom sessions `__ for more info. + + :param str url: The base URL to the instance of NetBox you + wish to connect to. + :param str token: Your NetBox token. + :param str,optional private_key_file: The path to your private + key file. + :param str,optional private_key: Your private key. + :param bool,optional threading: Set to True to use threading in ``.all()`` + and ``.filter()`` requests. + :raises ValueError: If *private_key* and *private_key_file* are both + specified. + :raises AttributeError: If app doesn't exist. + :Examples: + + >>> import pynetbox + >>> nb = pynetbox.api( + ... 'http://localhost:8000', + ... private_key_file='/path/to/private-key.pem', + ... token='d6f4e314a5b5fefd164995169f28ae32d987704f' + ... ) + >>> nb.dcim.devices.all() + """ + + def __init__( + self, url, token=None, private_key=None, private_key_file=None, threading=False, + ): + if private_key and private_key_file: + raise ValueError( + '"private_key" and "private_key_file" cannot be used together.' + ) + base_url = "{}/api".format(url if url[-1] != "/" else url[:-1]) + self.token = token + self.private_key = private_key + self.private_key_file = private_key_file + self.base_url = base_url + self.session_key = None + self.http_session = requests.Session() + if threading and sys.version_info.major == 2: + raise NotImplementedError( + "Threaded pynetbox calls not supported in Python 2" + ) + self.threading = threading + + if self.private_key_file: + with open(self.private_key_file, "r") as kf: + private_key = kf.read() + self.private_key = private_key + + self.dcim = App(self, "dcim") + self.ipam = App(self, "ipam") + self.circuits = App(self, "circuits") + self.secrets = App(self, "secrets") + self.tenancy = App(self, "tenancy") + self.extras = App(self, "extras") + self.virtualization = App(self, "virtualization") + + @property + def version(self): + """Gets the API version of NetBox. + + Can be used to check the NetBox API version if there are + version-dependent features or syntaxes in the API. + + :Returns: Version number as a string. + :Example: + + >>> import pynetbox + >>> nb = pynetbox.api( + ... 'http://localhost:8000', + ... private_key_file='/path/to/private-key.pem', + ... token='d6f4e314a5b5fefd164995169f28ae32d987704f' + ... ) + >>> nb.version + '2.6' + >>> + """ + version = Request( + base=self.base_url, http_session=self.http_session, + ).get_version() + return version + + def openapi(self): + """Returns the OpenAPI spec. + + Quick helper function to pull down the entire OpenAPI spec. + + :Returns: dict + :Example: + + >>> import pynetbox + >>> nb = pynetbox.api( + ... 'http://localhost:8000', + ... private_key_file='/path/to/private-key.pem', + ... token='d6f4e314a5b5fefd164995169f28ae32d987704f' + ... ) + >>> nb.openapi() + {...} + >>> + """ + return Request( + base=self.base_url, http_session=self.http_session, + ).get_openapi() diff --git a/pynetbox/core/app.py b/pynetbox/core/app.py new file mode 100644 index 00000000..c51e7ed8 --- /dev/null +++ b/pynetbox/core/app.py @@ -0,0 +1,111 @@ +""" +(c) 2017 DigitalOcean + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" +from pynetbox.core.endpoint import Endpoint +from pynetbox.core.query import Request +from pynetbox.models import dcim, ipam, virtualization, circuits, extras + + +class App(object): + """ Represents apps in NetBox. + + Calls to attributes are returned as Endpoint objects. + + :returns: :py:class:`.Endpoint` matching requested attribute. + :raises: :py:class:`.RequestError` + if requested endpoint doesn't exist. + """ + + def __init__(self, api, name): + self.api = api + self.name = name + self._choices = None + self._setmodel() + + models = { + "dcim": dcim, + "ipam": ipam, + "circuits": circuits, + "virtualization": virtualization, + "extras": extras, + } + + def _setmodel(self): + self.model = App.models[self.name] if self.name in App.models else None + + def __getstate__(self): + return {"api": self.api, "name": self.name, "_choices": self._choices} + + def __setstate__(self, d): + self.__dict__.update(d) + self._setmodel() + + def __getattr__(self, name): + if name == "secrets": + self._set_session_key() + return Endpoint(self.api, self, name, model=self.model) + + def _set_session_key(self): + if getattr(self.api, "session_key"): + return + if self.api.token and self.api.private_key: + self.api.session_key = Request( + base=self.api.base_url, + token=self.api.token, + private_key=self.api.private_key, + http_session=self.api.http_session, + ).get_session_key() + + def choices(self): + """ Returns _choices response from App + + .. note:: + + This method is deprecated and only works with NetBox version 2.7.x + or older. The ``choices()`` method in :py:class:`.Endpoint` is + compatible with all NetBox versions. + + :Returns: Raw response from NetBox's _choices endpoint. + """ + if self._choices: + return self._choices + + self._choices = Request( + base="{}/{}/_choices/".format(self.api.base_url, self.name), + token=self.api.token, + private_key=self.api.private_key, + http_session=self.api.http_session, + ).get() + + return self._choices + + def custom_choices(self): + """ Returns _custom_field_choices response from app + + :Returns: Raw response from NetBox's _custom_field_choices endpoint. + :Raises: :py:class:`.RequestError` if called for an invalid endpoint. + :Example: + + >>> nb.extras.custom_choices() + {'Testfield1': {'Testvalue2': 2, 'Testvalue1': 1}, + 'Testfield2': {'Othervalue2': 4, 'Othervalue1': 3}} + """ + custom_field_choices = Request( + base="{}/{}/_custom_field_choices/".format(self.api.base_url, self.name,), + token=self.api.token, + private_key=self.api.private_key, + http_session=self.api.http_session, + ).get() + return custom_field_choices diff --git a/pynetbox/core/endpoint.py b/pynetbox/core/endpoint.py index ea742c2d..a88b722c 100644 --- a/pynetbox/core/endpoint.py +++ b/pynetbox/core/endpoint.py @@ -13,12 +13,18 @@ See the License for the specific language governing permissions and limitations under the License. """ -from pynetbox.core.query import Request, url_param_builder +from pynetbox.core.query import Request, RequestError, url_param_builder from pynetbox.core.response import Record RESERVED_KWARGS = ("id", "pk", "limit", "offset") +def response_loader(req, return_obj, endpoint): + if isinstance(req, list): + return [return_obj(i, endpoint.api, endpoint) for i in req] + return return_obj(req, endpoint.api, endpoint) + + class Endpoint(object): """Represent actions available on endpoints in the Netbox API. @@ -47,11 +53,8 @@ def __init__(self, api, app, name, model=None): self.base_url = api.base_url self.token = api.token self.session_key = api.session_key - self.ssl_verify = api.ssl_verify self.url = "{base_url}/{app}/{endpoint}".format( - base_url=self.base_url, - app=app.name, - endpoint=self.name, + base_url=self.base_url, app=app.name, endpoint=self.name, ) self._choices = None @@ -74,9 +77,6 @@ def _lookup_ret_obj(self, name, model): ret = Record return ret - def _response_loader(self, values): - return self.return_obj(values, self.api, self) - def all(self): """Queries the 'ListView' of a given endpoint. @@ -94,12 +94,11 @@ def all(self): base="{}/".format(self.url), token=self.token, session_key=self.session_key, - ssl_verify=self.ssl_verify, http_session=self.api.http_session, threading=self.api.threading, ) - return [self._response_loader(i) for i in req.get()] + return response_loader(req.get(), self.return_obj, self) def get(self, *args, **kwargs): r"""Queries the DetailsView of a given endpoint. @@ -148,16 +147,18 @@ def get(self, *args, **kwargs): return filter_lookup[0] return None - req = Request( - key=key, - base=self.url, - token=self.token, - session_key=self.session_key, - ssl_verify=self.ssl_verify, - http_session=self.api.http_session, - ) + try: + req = Request( + key=key, + base=self.url, + token=self.token, + session_key=self.session_key, + http_session=self.api.http_session, + ) + except RequestError: + return None - return self._response_loader(req.get()) + return response_loader(req.get(), self.return_obj, self) def filter(self, *args, **kwargs): r"""Queries the 'ListView' of a given endpoint. @@ -205,9 +206,7 @@ def filter(self, *args, **kwargs): kwargs.update({"q": args[0]}) if not kwargs: - raise ValueError( - "filter must be passed kwargs. Perhaps use all() instead." - ) + raise ValueError("filter must be passed kwargs. Perhaps use all() instead.") if any(i in RESERVED_KWARGS for i in kwargs): raise ValueError( "A reserved {} kwarg was passed. Please remove it " @@ -219,13 +218,11 @@ def filter(self, *args, **kwargs): base=self.url, token=self.token, session_key=self.session_key, - ssl_verify=self.ssl_verify, http_session=self.api.http_session, threading=self.api.threading, ) - ret = [self._response_loader(i) for i in req.get()] - return ret + return response_loader(req.get(), self.return_obj, self) def create(self, *args, **kwargs): r"""Creates an object on an endpoint. @@ -282,14 +279,10 @@ def create(self, *args, **kwargs): base=self.url, token=self.token, session_key=self.session_key, - ssl_verify=self.ssl_verify, http_session=self.api.http_session, ).post(args[0] if args else kwargs) - if isinstance(req, list): - return [self._response_loader(i) for i in req] - - return self._response_loader(req) + return response_loader(req, self.return_obj, self) def choices(self): """ Returns all choices from the endpoint. @@ -327,21 +320,18 @@ def choices(self): base=self.url, token=self.api.token, private_key=self.api.private_key, - ssl_verify=self.api.ssl_verify, http_session=self.api.http_session, ).options() try: - post_data = req['actions']['POST'] + post_data = req["actions"]["POST"] except KeyError: raise ValueError( - "Unexpected format in the OPTIONS response at {}".format( - self.url - ) + "Unexpected format in the OPTIONS response at {}".format(self.url) ) self._choices = {} for prop in post_data: - if 'choices' in post_data[prop]: - self._choices[prop] = post_data[prop]['choices'] + if "choices" in post_data[prop]: + self._choices[prop] = post_data[prop]["choices"] return self._choices @@ -390,7 +380,6 @@ def count(self, *args, **kwargs): base=self.url, token=self.token, session_key=self.session_key, - ssl_verify=self.ssl_verify, http_session=self.api.http_session, ) @@ -407,14 +396,11 @@ class DetailEndpoint(object): def __init__(self, parent_obj, name, custom_return=None): self.parent_obj = parent_obj self.custom_return = custom_return - self.url = "{}/{}/{}/".format( - parent_obj.endpoint.url, parent_obj.id, name - ) + self.url = "{}/{}/{}/".format(parent_obj.endpoint.url, parent_obj.id, name) self.request_kwargs = dict( base=self.url, token=parent_obj.api.token, session_key=parent_obj.api.session_key, - ssl_verify=parent_obj.api.ssl_verify, http_session=parent_obj.api.http_session, ) @@ -434,16 +420,7 @@ def list(self, **kwargs): req = Request(**self.request_kwargs).get(add_params=kwargs) if self.custom_return: - if isinstance(req, list): - return [ - self.custom_return( - i, self.parent_obj.api, self.parent_obj.endpoint - ) - for i in req - ] - return self.custom_return( - req, self.parent_obj.api, self.parent_obj.endpoint - ) + return response_loader(req, self.custom_return, self.parent_obj.endpoint) return req def create(self, data=None): @@ -459,13 +436,13 @@ def create(self, data=None): :returns: A dictionary or list of dictionaries its created in NetBox. """ - if not data: - return Request(**self.request_kwargs).post({}) - return Request(**self.request_kwargs).post(data) + data = data or {} + req = Request(**self.request_kwargs).post(data) + if self.custom_return: + return response_loader(req, self.custom_return, self.parent_obj.endpoint) + return req class RODetailEndpoint(DetailEndpoint): def create(self, data): - raise NotImplementedError( - "Writes are not supported for this endpoint." - ) + raise NotImplementedError("Writes are not supported for this endpoint.") diff --git a/pynetbox/core/query.py b/pynetbox/core/query.py index 28f2771d..f4222a2d 100644 --- a/pynetbox/core/query.py +++ b/pynetbox/core/query.py @@ -58,9 +58,7 @@ def __init__(self, message): req = message if req.status_code == 404: - message = "The requested url: {} could not be found.".format( - req.url - ) + message = "The requested url: {} could not be found.".format(req.url) else: try: message = "The request failed with code {} {}: {}".format( @@ -113,8 +111,7 @@ def __init__(self, message): req = message message = ( - "The server returned invalid (non-json) data. Maybe not " - "a NetBox server?" + "The server returned invalid (non-json) data. Maybe not " "a NetBox server?" ) super(ContentError, self).__init__(message) @@ -148,8 +145,6 @@ def __init__( token=None, private_key=None, session_key=None, - ssl_verify=True, - url=None, threading=False, ): """ @@ -171,11 +166,24 @@ def __init__( self.token = token self.private_key = private_key self.session_key = session_key - self.ssl_verify = ssl_verify self.http_session = http_session self.url = self.base if not key else "{}{}/".format(self.base, key) self.threading = threading + def get_openapi(self): + """ Gets the OpenAPI Spec """ + headers = { + "Content-Type": "application/json;", + } + req = self.http_session.get( + "{}docs/?format=openapi".format(self.normalize_url(self.base)), + headers=headers, + ) + if req.ok: + return req.json() + else: + raise RequestError(req) + def get_version(self): """ Gets the API version of NetBox. @@ -189,11 +197,7 @@ def get_version(self): headers = { "Content-Type": "application/json;", } - req = requests.get( - self.normalize_url(self.base), - headers=headers, - verify=self.ssl_verify, - ) + req = requests.get(self.normalize_url(self.base), headers=headers,) if req.ok: return req.headers.get("API-Version", "") else: @@ -215,7 +219,6 @@ def get_session_key(self): "Content-Type": "application/x-www-form-urlencoded", }, data=urlencode({"private_key": self.private_key.strip("\n")}), - verify=self.ssl_verify, ) if req.ok: try: @@ -233,9 +236,7 @@ def normalize_url(self, url): return url - def _make_call( - self, verb="get", url_override=None, add_params=None, data=None - ): + def _make_call(self, verb="get", url_override=None, add_params=None, data=None): if verb in ("post", "put"): headers = {"Content-Type": "application/json;"} else: @@ -254,8 +255,7 @@ def _make_call( params.update(add_params) req = getattr(self.http_session, verb)( - url_override or self.url, headers=headers, verify=self.ssl_verify, - params=params, json=data + url_override or self.url, headers=headers, params=params, json=data ) if req.status_code == 204 and verb == "post": @@ -309,10 +309,12 @@ def req_all(): # passed in here because results from detail routes aren't # paginated, thus far. if first_run: - req = self._make_call(add_params={ - "limit": req["count"], - "offset": len(req["results"]) - }) + req = self._make_call( + add_params={ + "limit": req["count"], + "offset": len(req["results"]), + } + ) else: req = self._make_call(url_override=req["next"]) first_run = False diff --git a/pynetbox/core/response.py b/pynetbox/core/response.py index c89eee6b..52856853 100644 --- a/pynetbox/core/response.py +++ b/pynetbox/core/response.py @@ -13,6 +13,8 @@ See the License for the specific language governing permissions and limitations under the License. """ +import pynetbox.core.app +from six.moves.urllib.parse import urlsplit from pynetbox.core.query import Request from pynetbox.core.util import Hashabledict @@ -54,14 +56,14 @@ def get_return(lookup, return_fields=None): def flatten_custom(custom_dict): return { - k: v if not isinstance(v, dict) else v["value"] - for k, v in custom_dict.items() + k: v if not isinstance(v, dict) else v["value"] for k, v in custom_dict.items() } class JsonField(object): """Explicit field type for values that are not to be converted to a Record object""" + _json_field = True @@ -163,9 +165,12 @@ def __init__(self, values, api, endpoint): self._full_cache = [] self._init_cache = [] self.api = api - self.endpoint = endpoint self.default_ret = Record - + self.endpoint = ( + self._endpoint_from_url(values["url"]) + if values and "url" in values + else endpoint + ) if values: self._parse_values(values) @@ -203,9 +208,7 @@ def __getitem__(self, item): return item def __str__(self): - return ( - getattr(self, "name", None) or getattr(self, "label", None) or "" - ) + return getattr(self, "name", None) or getattr(self, "label", None) or "" def __repr__(self): return str(self) @@ -220,7 +223,7 @@ def __key__(self): if hasattr(self, "id"): return (self.endpoint.name, self.id) else: - return (self.endpoint.name) + return self.endpoint.name def __hash__(self): return hash(self.__key__()) @@ -268,6 +271,10 @@ def list_parser(list_item): self._add_cache((k, v)) setattr(self, k, v) + def _endpoint_from_url(self, url): + app, name = urlsplit(url).path.split("/")[2:4] + return getattr(pynetbox.core.app.App(self.api, app), name) + def _compare(self): """Compares current attributes to values at instantiation. @@ -297,7 +304,6 @@ def full_details(self): base=self.url, token=self.api.token, session_key=self.api.session_key, - ssl_verify=self.api.ssl_verify, http_session=self.api.http_session, ) self._parse_values(req.get()) @@ -335,14 +341,11 @@ def serialize(self, nested=False, init=False): ret[i] = flatten_custom(current_val) else: if isinstance(current_val, Record): - current_val = getattr(current_val, "serialize")( - nested=True - ) + current_val = getattr(current_val, "serialize")(nested=True) if isinstance(current_val, list): current_val = [ - v.id if isinstance(v, Record) else v - for v in current_val + v.id if isinstance(v, Record) else v for v in current_val ] if i in LIST_AS_SET: current_val = list(set(current_val)) @@ -357,9 +360,7 @@ def fmt_dict(k, v): return k, ",".join(map(str, v)) return k, v - current = Hashabledict( - {fmt_dict(k, v) for k, v in self.serialize().items()} - ) + current = Hashabledict({fmt_dict(k, v) for k, v in self.serialize().items()}) init = Hashabledict( {fmt_dict(k, v) for k, v in self.serialize(init=True).items()} ) @@ -388,10 +389,9 @@ def save(self): serialized = self.serialize() req = Request( key=self.id if not self.url else None, - base=self.url or self.endpoint.url, + base=self.endpoint.url, token=self.api.token, session_key=self.api.session_key, - ssl_verify=self.api.ssl_verify, http_session=self.api.http_session, ) if req.patch({i: serialized[i] for i in diff}): @@ -437,10 +437,9 @@ def delete(self): """ req = Request( key=self.id if not self.url else None, - base=self.url or self.endpoint.url, + base=self.endpoint.url, token=self.api.token, session_key=self.api.session_key, - ssl_verify=self.api.ssl_verify, http_session=self.api.http_session, ) return True if req.delete() else False diff --git a/pynetbox/models/dcim.py b/pynetbox/models/dcim.py index 9d9d2089..cae761cb 100644 --- a/pynetbox/models/dcim.py +++ b/pynetbox/models/dcim.py @@ -96,7 +96,6 @@ def __str__(self): class RUs(Record): - device = Devices diff --git a/pynetbox/models/ipam.py b/pynetbox/models/ipam.py index 0322c9a3..d2511aba 100644 --- a/pynetbox/models/ipam.py +++ b/pynetbox/models/ipam.py @@ -55,7 +55,7 @@ def available_ips(self): >>> len(create) 2 """ - return DetailEndpoint(self, "available-ips") + return DetailEndpoint(self, "available-ips", custom_return=IpAddresses) @property def available_prefixes(self): @@ -86,7 +86,7 @@ def available_prefixes(self): u'10.1.1.56/29' """ - return DetailEndpoint(self, "available-prefixes") + return DetailEndpoint(self, "available-prefixes", custom_return=Prefixes) class Aggregates(Record): diff --git a/setup.py b/setup.py index f01b5456..92daf485 100644 --- a/setup.py +++ b/setup.py @@ -1,29 +1,25 @@ from setuptools import setup, find_packages setup( - name='pynetbox', - description='NetBox API client library', - url='https://github.com/digitalocean/pynetbox', - author='Zach Moody', - author_email='zmoody@do.co', - license='Apache2', + name="pynetbox", + description="NetBox API client library", + url="https://github.com/digitalocean/pynetbox", + author="Zach Moody", + author_email="zmoody@do.co", + license="Apache2", include_package_data=True, use_scm_version=True, - setup_requires=['setuptools_scm'], + setup_requires=["setuptools_scm"], packages=find_packages(), - install_requires=[ - 'requests>=2.20.0,<3.0', - 'six==1.*', - ], + install_requires=["requests>=2.20.0,<3.0", "six==1.*",], zip_safe=False, - keywords=['netbox'], + keywords=["netbox"], classifiers=[ - 'Intended Audience :: Developers', - 'Development Status :: 5 - Production/Stable', - 'Programming Language :: Python :: 2', - 'Programming Language :: Python :: 2.7', - 'Programming Language :: Python :: 3', - 'Programming Language :: Python :: 3.6', + "Intended Audience :: Developers", + "Development Status :: 5 - Production/Stable", + "Programming Language :: Python :: 2", + "Programming Language :: Python :: 2.7", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.6", ], - ) diff --git a/tests/fixtures/dcim/interface.json b/tests/fixtures/dcim/interface.json index 711d1caf..7ccfc66e 100644 --- a/tests/fixtures/dcim/interface.json +++ b/tests/fixtures/dcim/interface.json @@ -42,5 +42,26 @@ "id": 1, "url": "http://localhost:8000/api/dcim/cables/1/", "label": "" - } + }, + "mode": { + "value": "tagged", + "label": "Tagged", + "id": 200 + }, + "untagged_vlan": { + "id": 2, + "url": "http://localhost:8000/api/ipam/vlans/792/", + "vid": 2069, + "name": "v2069", + "display_name": "2069 (v2069)" + }, + "tagged_vlans": [ + { + "id": 3, + "url": "http://localhost:8000/api/ipam/vlans/248/", + "vid": 1210, + "name": "v1210", + "display_name": "1210 (v1210)" + } + ] } \ No newline at end of file diff --git a/tests/fixtures/ipam/vlan.json b/tests/fixtures/ipam/vlan.json index 9be0a0c9..7227a0c0 100644 --- a/tests/fixtures/ipam/vlan.json +++ b/tests/fixtures/ipam/vlan.json @@ -1,5 +1,5 @@ { - "id": 1, + "id": 3, "site": { "id": 1, "url": "http://localhost:8000/api/dcim/sites/1/", @@ -7,8 +7,8 @@ "slug": "test1" }, "group": null, - "vid": 999, - "name": "TEST", + "vid": 1210, + "name": "v1210", "tenant": null, "status": { "value": 1, @@ -21,6 +21,6 @@ "slug": "lab-network" }, "description": "", - "display_name": "999 (TEST)", + "display_name": "1210 (v1210)", "custom_fields": {} } \ No newline at end of file diff --git a/tests/fixtures/ipam/vlans.json b/tests/fixtures/ipam/vlans.json index fa09503a..053aacfe 100644 --- a/tests/fixtures/ipam/vlans.json +++ b/tests/fixtures/ipam/vlans.json @@ -1,5 +1,5 @@ { - "count": 1, + "count": 3, "next": null, "previous": null, "results": [ @@ -28,6 +28,58 @@ "description": "", "display_name": "999 (TEST)", "custom_fields": {} + }, + { + "id": 2, + "site": { + "id": 1, + "url": "http://localhost:8000/api/dcim/sites/1/", + "name": "TEST1", + "slug": "test1" + }, + "group": null, + "vid": 2069, + "name": "v2069", + "tenant": null, + "status": { + "value": 1, + "label": "Active" + }, + "role": { + "id": 1, + "url": "http://localhost:8000/api/ipam/roles/1/", + "name": "Lab Network", + "slug": "lab-network" + }, + "description": "", + "display_name": "2069 (v2069)", + "custom_fields": {} + }, + { + "id": 3, + "site": { + "id": 1, + "url": "http://localhost:8000/api/dcim/sites/1/", + "name": "TEST1", + "slug": "test1" + }, + "group": null, + "vid": 1210, + "name": "v1210", + "tenant": null, + "status": { + "value": 1, + "label": "Active" + }, + "role": { + "id": 1, + "url": "http://localhost:8000/api/ipam/roles/1/", + "name": "Lab Network", + "slug": "lab-network" + }, + "description": "", + "display_name": "1210 (v1210)", + "custom_fields": {} } ] } \ No newline at end of file diff --git a/tests/test_api.py b/tests/test_api.py index 41a45526..59d6a65f 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -12,8 +12,8 @@ host = "http://localhost:8000" def_kwargs = { - 'token': 'abc123', - 'private_key_file': 'tests/fixtures/api/get_session_key.json', + "token": "abc123", + "private_key_file": "tests/fixtures/api/get_session_key.json", } # Keys are app names, values are arbitrarily selected endpoints @@ -21,100 +21,46 @@ # and circuits because it does not. We don't add other apps/endpoints # beyond 'circuits' as they all use the same code as each other endpoints = { - 'dcim': 'devices', - 'ipam': 'prefixes', - 'circuits': 'circuits', + "dcim": "devices", + "ipam": "prefixes", + "circuits": "circuits", } class ApiTestCase(unittest.TestCase): - @patch( - 'pynetbox.core.query.requests.sessions.Session.post', - return_value=Response(fixture='api/get_session_key.json') + "pynetbox.core.query.requests.sessions.Session.post", + return_value=Response(fixture="api/get_session_key.json"), ) def test_get(self, *_): - api = pynetbox.api( - host, - **def_kwargs - ) + api = pynetbox.api(host, **def_kwargs) self.assertTrue(api) @patch( - 'pynetbox.core.query.requests.sessions.Session.post', - return_value=Response(fixture='api/get_session_key.json') + "pynetbox.core.query.requests.sessions.Session.post", + return_value=Response(fixture="api/get_session_key.json"), ) def test_sanitize_url(self, *_): - api = pynetbox.api( - 'http://localhost:8000/', - **def_kwargs - ) + api = pynetbox.api("http://localhost:8000/", **def_kwargs) self.assertTrue(api) - self.assertEqual(api.base_url, 'http://localhost:8000/api') - - -class ApiArgumentsTestCase(unittest.TestCase): - - @patch( - 'pynetbox.core.query.requests.sessions.Session.post', - return_value=Response(fixture='api/get_session_key.json') - ) - def common_arguments(self, kwargs, arg, expect, *_): - ''' - Ensures the api and endpoint instances have ssl_verify set - as expected - ''' - api = pynetbox.api( - host, - **kwargs - ) - self.assertIs(getattr(api, arg, "fail"), expect) - for app, endpoint in endpoints.items(): - ep = getattr(getattr(api, app), endpoint) - self.assertIs(getattr(ep, arg), expect) - - def test_ssl_verify_default(self): - self.common_arguments(def_kwargs, 'ssl_verify', True) - - def test_ssl_verify_true(self): - kwargs = dict(def_kwargs, **{'ssl_verify': True}) - self.common_arguments(kwargs, 'ssl_verify', True) - - def test_ssl_verify_false(self): - kwargs = dict(def_kwargs, **{'ssl_verify': False}) - self.common_arguments(kwargs, 'ssl_verify', False) - - def test_ssl_verify_string(self): - kwargs = dict(def_kwargs, **{'ssl_verify': '/path/to/bundle'}) - self.common_arguments(kwargs, 'ssl_verify', '/path/to/bundle') + self.assertEqual(api.base_url, "http://localhost:8000/api") class ApiVersionTestCase(unittest.TestCase): - class ResponseHeadersWithVersion: headers = {"API-Version": "1.999"} ok = True - @patch( - 'requests.get', - return_value=ResponseHeadersWithVersion() - ) + @patch("requests.get", return_value=ResponseHeadersWithVersion()) def test_api_version(self, *_): - api = pynetbox.api( - host, - ) + api = pynetbox.api(host,) self.assertEqual(api.version, "1.999") class ResponseHeadersWithoutVersion: headers = {} ok = True - @patch( - 'requests.get', - return_value=ResponseHeadersWithoutVersion() - ) + @patch("requests.get", return_value=ResponseHeadersWithoutVersion()) def test_api_version_not_found(self, *_): - api = pynetbox.api( - host, - ) + api = pynetbox.api(host,) self.assertEqual(api.version, "") diff --git a/tests/test_app.py b/tests/test_app.py index 237ad38a..632ee82f 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -12,24 +12,20 @@ host = "http://localhost:8000" def_kwargs = { - 'token': 'abc123', + "token": "abc123", } class AppCustomChoicesTestCase(unittest.TestCase): - @patch( - 'pynetbox.core.query.Request.get', + "pynetbox.core.query.Request.get", return_value={ "Testfield1": {"TF1_1": 1, "TF1_2": 2}, "Testfield2": {"TF2_1": 3, "TF2_2": 4}, - } + }, ) def test_custom_choices(self, *_): - api = pynetbox.api( - host, - **def_kwargs - ) + api = pynetbox.api(host, **def_kwargs) choices = api.extras.custom_choices() self.assertEqual(len(choices), 2) self.assertEqual(sorted(choices.keys()), ["Testfield1", "Testfield2"]) diff --git a/tests/test_circuits.py b/tests/test_circuits.py index add97dd9..e0ceb924 100644 --- a/tests/test_circuits.py +++ b/tests/test_circuits.py @@ -9,119 +9,102 @@ else: from mock import patch -api = pynetbox.api( - "http://localhost:8000", -) +api = pynetbox.api("http://localhost:8000",) nb = api.circuits -HEADERS = { - 'accept': 'application/json;' -} +HEADERS = {"accept": "application/json;"} class Generic(object): class Tests(unittest.TestCase): - name = '' + name = "" ret = pynetbox.core.response.Record - app = 'circuits' + app = "circuits" def test_get_all(self): with patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='{}/{}.json'.format( - self.app, - self.name - )) + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="{}/{}.json".format(self.app, self.name)), ) as mock: ret = getattr(nb, self.name).all() self.assertTrue(ret) self.assertTrue(isinstance(ret, list)) self.assertTrue(isinstance(ret[0], self.ret)) mock.assert_called_with( - 'http://localhost:8000/api/{}/{}/'.format( - self.app, - self.name.replace('_', '-') + "http://localhost:8000/api/{}/{}/".format( + self.app, self.name.replace("_", "-") ), params={}, json=None, headers=HEADERS, - verify=True ) def test_filter(self): with patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='{}/{}.json'.format( - self.app, - self.name - )) + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="{}/{}.json".format(self.app, self.name)), ) as mock: - ret = getattr(nb, self.name).filter(name='test') + ret = getattr(nb, self.name).filter(name="test") self.assertTrue(ret) self.assertTrue(isinstance(ret, list)) self.assertTrue(isinstance(ret[0], self.ret)) mock.assert_called_with( - 'http://localhost:8000/api/{}/{}/'.format( - self.app, - self.name.replace('_', '-') + "http://localhost:8000/api/{}/{}/".format( + self.app, self.name.replace("_", "-") ), params={"name": "test"}, json=None, headers=HEADERS, - verify=True ) def test_get(self): with patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='{}/{}.json'.format( - self.app, - self.name[:-1] - )) + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response( + fixture="{}/{}.json".format(self.app, self.name[:-1]) + ), ) as mock: ret = getattr(nb, self.name).get(1) self.assertTrue(ret) self.assertTrue(isinstance(ret, self.ret)) mock.assert_called_with( - 'http://localhost:8000/api/{}/{}/1/'.format( - self.app, - self.name.replace('_', '-') + "http://localhost:8000/api/{}/{}/1/".format( + self.app, self.name.replace("_", "-") ), params={}, json=None, headers=HEADERS, - verify=True ) class CircuitsTestCase(Generic.Tests): - name = 'circuits' + name = "circuits" @patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='circuits/circuit.json') + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="circuits/circuit.json"), ) def test_repr(self, _): test = nb.circuits.get(1) - self.assertEqual(str(test), '123456') + self.assertEqual(str(test), "123456") class ProviderTestCase(Generic.Tests): - name = 'providers' + name = "providers" class CircuitTypeTestCase(Generic.Tests): - name = 'circuit_types' + name = "circuit_types" class CircuitTerminationsTestCase(Generic.Tests): - name = 'circuit_terminations' + name = "circuit_terminations" @patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='circuits/circuit_termination.json') + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="circuits/circuit_termination.json"), ) def test_repr(self, _): test = nb.circuit_terminations.get(1) - self.assertEqual(str(test), '123456') + self.assertEqual(str(test), "123456") diff --git a/tests/test_dcim.py b/tests/test_dcim.py index 94001c9c..47119ea5 100644 --- a/tests/test_dcim.py +++ b/tests/test_dcim.py @@ -10,78 +10,64 @@ from mock import patch -api = pynetbox.api( - "http://localhost:8000", - token="abc123", -) +api = pynetbox.api("http://localhost:8000", token="abc123",) nb = api.dcim HEADERS = { - 'accept': 'application/json;', - 'authorization': 'Token abc123', + "accept": "application/json;", + "authorization": "Token abc123", } class Generic(object): class Tests(unittest.TestCase): - name = '' + name = "" ret = pynetbox.core.response.Record - app = 'dcim' + app = "dcim" def test_get_all(self): with patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='{}/{}.json'.format( - self.app, - self.name - )) + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="{}/{}.json".format(self.app, self.name)), ) as mock: ret = getattr(nb, self.name).all() self.assertTrue(ret) self.assertTrue(isinstance(ret, list)) self.assertTrue(isinstance(ret[0], self.ret)) mock.assert_called_with( - 'http://localhost:8000/api/{}/{}/'.format( - self.app, - self.name.replace('_', '-') + "http://localhost:8000/api/{}/{}/".format( + self.app, self.name.replace("_", "-") ), params={}, json=None, headers=HEADERS, - verify=True ) def test_filter(self): with patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='{}/{}.json'.format( - self.app, - self.name - )) + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="{}/{}.json".format(self.app, self.name)), ) as mock: - ret = getattr(nb, self.name).filter(name='test') + ret = getattr(nb, self.name).filter(name="test") self.assertTrue(ret) self.assertTrue(isinstance(ret, list)) self.assertTrue(isinstance(ret[0], self.ret)) mock.assert_called_with( - 'http://localhost:8000/api/{}/{}/'.format( - self.app, - self.name.replace('_', '-') + "http://localhost:8000/api/{}/{}/".format( + self.app, self.name.replace("_", "-") ), params={"name": "test"}, json=None, headers=HEADERS, - verify=True ) def test_get(self): with patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='{}/{}.json'.format( - self.app, - self.name[:-1] - )) + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response( + fixture="{}/{}.json".format(self.app, self.name[:-1]) + ), ) as mock: ret = getattr(nb, self.name).get(1) self.assertTrue(ret) @@ -89,54 +75,48 @@ def test_get(self): self.assertTrue(isinstance(str(ret), str)) self.assertTrue(isinstance(dict(ret), dict)) mock.assert_called_with( - 'http://localhost:8000/api/{}/{}/1/'.format( - self.app, - self.name.replace('_', '-') + "http://localhost:8000/api/{}/{}/1/".format( + self.app, self.name.replace("_", "-") ), params={}, json=None, headers=HEADERS, - verify=True ) def test_delete(self): with patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='{}/{}.json'.format( - self.app, - self.name[:-1] - )) - ) as mock, patch('pynetbox.core.query.requests.sessions.Session.delete') as delete: + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response( + fixture="{}/{}.json".format(self.app, self.name[:-1]) + ), + ) as mock, patch( + "pynetbox.core.query.requests.sessions.Session.delete" + ) as delete: ret = getattr(nb, self.name).get(1) self.assertTrue(ret.delete()) mock.assert_called_with( - 'http://localhost:8000/api/{}/{}/1/'.format( - self.app, - self.name.replace('_', '-') + "http://localhost:8000/api/{}/{}/1/".format( + self.app, self.name.replace("_", "-") ), params={}, json=None, headers=HEADERS, - verify=True ) delete.assert_called_with( - 'http://localhost:8000/api/{}/{}/1/'.format( - self.app, - self.name.replace('_', '-') + "http://localhost:8000/api/{}/{}/1/".format( + self.app, self.name.replace("_", "-") ), params={}, json=None, headers=HEADERS, - verify=True ) def test_compare(self): with patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='{}/{}.json'.format( - self.app, - self.name[:-1] - )) + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response( + fixture="{}/{}.json".format(self.app, self.name[:-1]) + ), ): ret = getattr(nb, self.name).get(1) self.assertTrue(ret) @@ -144,11 +124,10 @@ def test_compare(self): def test_serialize(self): with patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='{}/{}.json'.format( - self.app, - self.name[:-1] - )) + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response( + fixture="{}/{}.json".format(self.app, self.name[:-1]) + ), ): ret = getattr(nb, self.name).get(1) self.assertTrue(ret) @@ -156,11 +135,11 @@ def test_serialize(self): class DeviceTestCase(Generic.Tests): - name = 'devices' + name = "devices" @patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='dcim/device.json') + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="dcim/device.json"), ) def test_get(self, mock): ret = getattr(nb, self.name).get(1) @@ -172,194 +151,166 @@ def test_get(self, mock): self.assertTrue(isinstance(ret.custom_fields, dict)) self.assertTrue(isinstance(ret.local_context_data, dict)) mock.assert_called_with( - 'http://localhost:8000/api/{}/{}/1/'.format( - self.app, - self.name.replace('_', '-') + "http://localhost:8000/api/{}/{}/1/".format( + self.app, self.name.replace("_", "-") ), params={}, json=None, headers=HEADERS, - verify=True ) @patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='dcim/devices.json') + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="dcim/devices.json"), ) def test_multi_filter(self, mock): - ret = getattr(nb, self.name).filter(role=['test', 'test1'], site='TEST#1') + ret = getattr(nb, self.name).filter(role=["test", "test1"], site="TEST#1") self.assertTrue(ret) self.assertTrue(isinstance(ret, list)) self.assertTrue(isinstance(ret[0], self.ret)) mock.assert_called_with( - 'http://localhost:8000/api/{}/{}/'.format( - self.app, - self.name.replace('_', '-') + "http://localhost:8000/api/{}/{}/".format( + self.app, self.name.replace("_", "-") ), - params={'role': ['test', 'test1'], 'site': 'TEST#1'}, + params={"role": ["test", "test1"], "site": "TEST#1"}, json=None, headers=HEADERS, - verify=True ) @patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='dcim/device.json') + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="dcim/device.json"), ) def test_modify(self, *_): ret = nb.devices.get(1) - ret.serial = '123123123123' + ret.serial = "123123123123" ret_serialized = ret.serialize() self.assertTrue(ret_serialized) self.assertFalse(ret._compare()) - self.assertEqual(ret_serialized['serial'], '123123123123') + self.assertEqual(ret_serialized["serial"], "123123123123") @patch( - 'pynetbox.core.query.requests.sessions.Session.post', - return_value=Response(fixture='dcim/device.json') + "pynetbox.core.query.requests.sessions.Session.post", + return_value=Response(fixture="dcim/device.json"), ) def test_create(self, *_): data = { - 'name': 'test-device', - 'site': 1, - 'device_type': 1, - 'device_role': 1, + "name": "test-device", + "site": 1, + "device_type": 1, + "device_role": 1, } ret = nb.devices.create(**data) self.assertTrue(ret) @patch( - 'pynetbox.core.query.requests.sessions.Session.post', - return_value=Response(fixture='dcim/device_bulk_create.json') + "pynetbox.core.query.requests.sessions.Session.post", + return_value=Response(fixture="dcim/device_bulk_create.json"), ) def test_create_device_bulk(self, *_): data = [ - { - 'name': 'test-device', - 'site': 1, - 'device_type': 1, - 'device_role': 1, - }, - { - 'name': 'test-device1', - 'site': 1, - 'device_type': 1, - 'device_role': 1, - }, + {"name": "test-device", "site": 1, "device_type": 1, "device_role": 1,}, + {"name": "test-device1", "site": 1, "device_type": 1, "device_role": 1,}, ] ret = nb.devices.create(data) self.assertTrue(ret) self.assertTrue(len(ret), 2) @patch( - 'pynetbox.core.query.requests.sessions.Session.get', + "pynetbox.core.query.requests.sessions.Session.get", side_effect=[ - Response(fixture='dcim/device.json'), - Response(fixture='dcim/rack.json'), - ] + Response(fixture="dcim/device.json"), + Response(fixture="dcim/rack.json"), + ], ) def test_get_recurse(self, *_): - '''Test that automatic recursion works, and that nested items + """Test that automatic recursion works, and that nested items are converted to Response() objects. - ''' + """ ret = getattr(nb, self.name).get(1) self.assertTrue(ret) self.assertTrue(isinstance(ret, self.ret)) - self.assertTrue(isinstance( - ret.rack.role, - self.ret - )) + self.assertTrue(isinstance(ret.rack.role, self.ret)) @patch( - 'pynetbox.core.query.requests.sessions.Session.get', + "pynetbox.core.query.requests.sessions.Session.get", side_effect=[ - Response(fixture='dcim/device.json'), - Response(fixture='dcim/napalm.json'), - ] + Response(fixture="dcim/device.json"), + Response(fixture="dcim/napalm.json"), + ], ) def test_get_napalm(self, mock): test = nb.devices.get(1) - ret = test.napalm.list(method='get_facts') + ret = test.napalm.list(method="get_facts") mock.assert_called_with( - 'http://localhost:8000/api/dcim/devices/1/napalm/', + "http://localhost:8000/api/dcim/devices/1/napalm/", params={"method": "get_facts"}, json=None, headers=HEADERS, - verify=True, ) self.assertTrue(ret) - self.assertTrue(ret['get_facts']) + self.assertTrue(ret["get_facts"]) class SiteTestCase(Generic.Tests): - name = 'sites' + name = "sites" @patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='dcim/site.json') + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="dcim/site.json"), ) def test_modify_custom(self, *_): - '''Test modifying a custom field. - ''' + """Test modifying a custom field. + """ ret = getattr(nb, self.name).get(1) - ret.custom_fields['test_custom'] = "Testing" + ret.custom_fields["test_custom"] = "Testing" self.assertFalse(ret._compare()) self.assertTrue(ret.serialize()) - self.assertEqual( - ret.custom_fields['test_custom'], 'Testing' - ) + self.assertEqual(ret.custom_fields["test_custom"], "Testing") @patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='dcim/site.json') + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="dcim/site.json"), ) def test_custom_selection_serializer(self, _): - '''Tests serializer with custom selection fields. - ''' + """Tests serializer with custom selection fields. + """ ret = getattr(nb, self.name).get(1) - ret.custom_fields['test_custom'] = "Testing" + ret.custom_fields["test_custom"] = "Testing" test = ret.serialize() - self.assertEqual( - test['custom_fields']['test_selection'], - 2 - ) + self.assertEqual(test["custom_fields"]["test_selection"], 2) @patch( - 'pynetbox.core.query.requests.sessions.Session.post', - return_value=Response(fixture='dcim/site.json') + "pynetbox.core.query.requests.sessions.Session.post", + return_value=Response(fixture="dcim/site.json"), ) def test_create(self, *_): - data = { - 'name': 'TEST1', - 'custom_fields': { - 'test_custom': 'Testing' - } - } + data = {"name": "TEST1", "custom_fields": {"test_custom": "Testing"}} ret = nb.sites.create(**data) self.assertTrue(ret) class InterfaceTestCase(Generic.Tests): - name = 'interfaces' + name = "interfaces" @patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='dcim/interface.json') + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="dcim/interface.json"), ) def test_modify(self, *_): ret = nb.interfaces.get(1) - ret.description = 'Testing' + ret.description = "Testing" ret_serialized = ret.serialize() self.assertTrue(ret) - self.assertEqual(ret_serialized['description'], 'Testing') - self.assertEqual(ret_serialized['form_factor'], 1400) + self.assertEqual(ret_serialized["description"], "Testing") + self.assertEqual(ret_serialized["form_factor"], 1400) @patch( - 'pynetbox.core.query.requests.sessions.Session.get', + "pynetbox.core.query.requests.sessions.Session.get", side_effect=[ - Response(fixture='dcim/{}.json'.format(name + '_1')), - Response(fixture='dcim/{}.json'.format(name + '_2')), - ] + Response(fixture="dcim/{}.json".format(name + "_1")), + Response(fixture="dcim/{}.json".format(name + "_2")), + ], ) def test_get_all(self, mock): ret = getattr(nb, self.name).all() @@ -368,136 +319,129 @@ def test_get_all(self, mock): self.assertTrue(isinstance(ret[0], self.ret)) self.assertEqual(len(ret), 71) mock.assert_called_with( - 'http://localhost:8000/api/dcim/interfaces/', + "http://localhost:8000/api/dcim/interfaces/", params={"limit": 221, "offset": 50}, json=None, headers=HEADERS, - verify=True ) class RackTestCase(Generic.Tests): - name = 'racks' + name = "racks" @patch( - 'pynetbox.core.query.requests.sessions.Session.get', + "pynetbox.core.query.requests.sessions.Session.get", side_effect=[ - Response(fixture='dcim/rack.json'), - Response(fixture='dcim/rack_u.json'), - ] + Response(fixture="dcim/rack.json"), + Response(fixture="dcim/rack_u.json"), + ], ) def test_get_units(self, mock): test = nb.racks.get(1) ret = test.units.list() mock.assert_called_with( - 'http://localhost:8000/api/dcim/racks/1/units/', + "http://localhost:8000/api/dcim/racks/1/units/", params={}, json=None, headers=HEADERS, - verify=True, ) self.assertTrue(ret) - self.assertTrue( - isinstance(ret[0].device, pynetbox.models.dcim.Devices) - ) + self.assertTrue(isinstance(ret[0].device, pynetbox.models.dcim.Devices)) @patch( - 'pynetbox.core.query.requests.sessions.Session.get', + "pynetbox.core.query.requests.sessions.Session.get", side_effect=[ - Response(fixture='dcim/rack.json'), - Response(fixture='dcim/rack_u.json'), - ] + Response(fixture="dcim/rack.json"), + Response(fixture="dcim/rack_u.json"), + ], ) def test_get_elevation(self, mock): test = nb.racks.get(1) ret = test.elevation.list() mock.assert_called_with( - 'http://localhost:8000/api/dcim/racks/1/elevation/', + "http://localhost:8000/api/dcim/racks/1/elevation/", params={}, json=None, headers=HEADERS, - verify=True, ) self.assertTrue(ret) - self.assertTrue( - isinstance(ret[0].device, pynetbox.models.dcim.Devices) - ) + self.assertTrue(isinstance(ret[0].device, pynetbox.models.dcim.Devices)) class RackRoleTestCase(Generic.Tests): - name = 'rack_roles' + name = "rack_roles" class RegionTestCase(Generic.Tests): - name = 'regions' + name = "regions" class RackGroupsTestCase(Generic.Tests): - name = 'rack_groups' + name = "rack_groups" class RackReservationsTestCase(Generic.Tests): - name = 'rack_reservations' + name = "rack_reservations" class ManufacturersTestCase(Generic.Tests): - name = 'manufacturers' + name = "manufacturers" class DeviceTypeTestCase(Generic.Tests): - name = 'device_types' + name = "device_types" class ConsolePortTemplateTestCase(Generic.Tests): - name = 'console_port_templates' + name = "console_port_templates" class ConsoleServerPortTemplateTestCase(Generic.Tests): - name = 'console_server_port_templates' + name = "console_server_port_templates" class PowerPortTemplateTestCase(Generic.Tests): - name = 'power_port_templates' + name = "power_port_templates" class PowerOutletTemplateTestCase(Generic.Tests): - name = 'power_outlet_templates' + name = "power_outlet_templates" class InterfaceTemplateTestCase(Generic.Tests): - name = 'interface_templates' + name = "interface_templates" class DeviceBayTemplateTestCase(Generic.Tests): - name = 'device_bay_templates' + name = "device_bay_templates" class DeviceRolesTestCase(Generic.Tests): - name = 'device_roles' + name = "device_roles" class PlatformsTestCase(Generic.Tests): - name = 'platforms' + name = "platforms" class ConsolePortsTestCase(Generic.Tests): - name = 'console_ports' + name = "console_ports" class ConsoleServerPortsTestCase(Generic.Tests): - name = 'console_server_ports' + name = "console_server_ports" class PowerPortsTestCase(Generic.Tests): - name = 'power_ports' + name = "power_ports" class PowerOutletsTestCase(Generic.Tests): - name = 'power_outlets' + name = "power_outlets" class DeviceBaysTestCase(Generic.Tests): - name = 'device_bays' + name = "device_bays" # class InventoryItemsTestCase(Generic.Tests): @@ -505,7 +449,7 @@ class DeviceBaysTestCase(Generic.Tests): class InterfaceConnectionsTestCase(Generic.Tests): - name = 'interface_connections' + name = "interface_connections" # class ConnectedDevicesTestCase(Generic.Tests): @@ -513,75 +457,69 @@ class InterfaceConnectionsTestCase(Generic.Tests): class VirtualChassisTestCase(Generic.Tests): - name = 'virtual_chassis_devices' + name = "virtual_chassis_devices" class Choices(unittest.TestCase): - def test_get(self): with patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='{}/{}.json'.format( - 'dcim', - 'choices' - )) + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="{}/{}.json".format("dcim", "choices")), ) as mock: ret = nb.choices() self.assertTrue(ret) mock.assert_called_with( - 'http://localhost:8000/api/dcim/_choices/', + "http://localhost:8000/api/dcim/_choices/", params={}, json=None, headers=HEADERS, - verify=True ) class CablesTestCase(Generic.Tests): - name = 'cables' + name = "cables" def test_get_circuit(self): - response_obj = Response(content={ - "id": 1, - "termination_a_type": "circuits.circuittermination", - "termination_a_id": 1, - "termination_a": { + response_obj = Response( + content={ "id": 1, - "url": "http://localhost:8000/api/circuits/circuit-terminations/1/", - "circuit": { - "id": 346, - "url": "http://localhost:8000/api/circuits/circuits/1/", - "cid": "TEST123321" + "termination_a_type": "circuits.circuittermination", + "termination_a_id": 1, + "termination_a": { + "id": 1, + "url": "http://localhost:8000/api/circuits/circuit-terminations/1/", + "circuit": { + "id": 346, + "url": "http://localhost:8000/api/circuits/circuits/1/", + "cid": "TEST123321", + }, + "term_side": "A", }, - "term_side": "A", - }, - "termination_b_type": "dcim.interface", - "termination_b_id": 2, - "termination_b": { - "id": 2, - "url": "http://localhost:8000/api/dcim/interfaces/2/", - "device": { + "termination_b_type": "dcim.interface", + "termination_b_id": 2, + "termination_b": { "id": 2, - "url": "http://localhost:8000/api/dcim/devices/2/", - "name": "tst1-test2", - "display_name": "tst1-test2" + "url": "http://localhost:8000/api/dcim/interfaces/2/", + "device": { + "id": 2, + "url": "http://localhost:8000/api/dcim/devices/2/", + "name": "tst1-test2", + "display_name": "tst1-test2", + }, + "name": "xe-0/0/0", + "cable": 1, }, - "name": "xe-0/0/0", - "cable": 1 - }, - "type": None, - "status": { - "value": True, - "label": "Connected" - }, - "label": "", - "color": "", - "length": None, - "length_unit": None - }) + "type": None, + "status": {"value": True, "label": "Connected"}, + "label": "", + "color": "", + "length": None, + "length_unit": None, + } + ) with patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=response_obj + "pynetbox.core.query.requests.sessions.Session.get", + return_value=response_obj, ) as mock: ret = getattr(nb, self.name).get(1) self.assertTrue(ret) @@ -589,12 +527,10 @@ def test_get_circuit(self): self.assertTrue(isinstance(str(ret), str)) self.assertTrue(isinstance(dict(ret), dict)) mock.assert_called_with( - 'http://localhost:8000/api/{}/{}/1/'.format( - self.app, - self.name.replace('_', '-') + "http://localhost:8000/api/{}/{}/1/".format( + self.app, self.name.replace("_", "-") ), headers=HEADERS, params={}, json=None, - verify=True ) diff --git a/tests/test_ipam.py b/tests/test_ipam.py index 287a7063..8cc12004 100644 --- a/tests/test_ipam.py +++ b/tests/test_ipam.py @@ -12,84 +12,72 @@ from mock import patch -api = pynetbox.api( - "http://localhost:8000", - token="abc123", -) +api = pynetbox.api("http://localhost:8000", token="abc123",) nb = api.ipam HEADERS = { - 'accept': 'application/json;', - 'authorization': 'Token abc123', + "accept": "application/json;", + "authorization": "Token abc123", } POST_HEADERS = { - 'Content-Type': 'application/json;', - 'authorization': 'Token abc123', + "Content-Type": "application/json;", + "authorization": "Token abc123", } class Generic(object): class Tests(unittest.TestCase): - name = '' + name = "" name_singular = None ret = pynetbox.core.response.Record - app = 'ipam' + app = "ipam" def test_get_all(self): with patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='{}/{}.json'.format( - self.app, - self.name - )) + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="{}/{}.json".format(self.app, self.name)), ) as mock: ret = getattr(nb, self.name).all() self.assertTrue(ret) self.assertTrue(isinstance(ret, list)) self.assertTrue(isinstance(ret[0], self.ret)) mock.assert_called_with( - 'http://localhost:8000/api/{}/{}/'.format( - self.app, - self.name.replace('_', '-') + "http://localhost:8000/api/{}/{}/".format( + self.app, self.name.replace("_", "-") ), params={}, json=None, headers=HEADERS, - verify=True ) def test_filter(self): with patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='{}/{}.json'.format( - self.app, - self.name - )) + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="{}/{}.json".format(self.app, self.name)), ) as mock: - ret = getattr(nb, self.name).filter(name='test') + ret = getattr(nb, self.name).filter(name="test") self.assertTrue(ret) self.assertTrue(isinstance(ret, list)) self.assertTrue(isinstance(ret[0], self.ret)) mock.assert_called_with( - 'http://localhost:8000/api/{}/{}/'.format( - self.app, - self.name.replace('_', '-') + "http://localhost:8000/api/{}/{}/".format( + self.app, self.name.replace("_", "-") ), params={"name": "test"}, json=None, headers=HEADERS, - verify=True ) def test_get(self): with patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='{}/{}.json'.format( - self.app, - self.name_singular or self.name[:-1] - )) + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response( + fixture="{}/{}.json".format( + self.app, self.name_singular or self.name[:-1] + ) + ), ) as mock: ret = getattr(nb, self.name).get(1) self.assertTrue(ret) @@ -97,40 +85,38 @@ def test_get(self): self.assertTrue(isinstance(dict(ret), dict)) self.assertTrue(isinstance(str(ret), str)) mock.assert_called_with( - 'http://localhost:8000/api/{}/{}/1/'.format( - self.app, - self.name.replace('_', '-') + "http://localhost:8000/api/{}/{}/1/".format( + self.app, self.name.replace("_", "-") ), params={}, json=None, headers=HEADERS, - verify=True ) class PrefixTestCase(Generic.Tests): - name = 'prefixes' - name_singular = 'prefix' + name = "prefixes" + name_singular = "prefix" @patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='ipam/prefix.json') + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="ipam/prefix.json"), ) def test_modify(self, *_): ret = nb.prefixes.get(1) - ret.prefix = '10.1.2.0/24' + ret.prefix = "10.1.2.0/24" ret_serialized = ret.serialize() self.assertTrue(ret_serialized) self.assertFalse(ret._compare()) - self.assertEqual(ret_serialized['prefix'], '10.1.2.0/24') + self.assertEqual(ret_serialized["prefix"], "10.1.2.0/24") @patch( - 'pynetbox.core.query.requests.sessions.Session.put', - return_value=Response(fixture='ipam/prefix.json') + "pynetbox.core.query.requests.sessions.Session.put", + return_value=Response(fixture="ipam/prefix.json"), ) @patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='ipam/prefix.json') + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="ipam/prefix.json"), ) def test_idempotence(self, *_): ret = nb.prefixes.get(1) @@ -138,144 +124,145 @@ def test_idempotence(self, *_): self.assertFalse(test) @patch( - 'pynetbox.core.query.requests.sessions.Session.get', + "pynetbox.core.query.requests.sessions.Session.get", side_effect=[ - Response(fixture='ipam/prefix.json'), - Response(fixture='ipam/available-ips.json'), - ] + Response(fixture="ipam/prefix.json"), + Response(fixture="ipam/available-ips.json"), + ], ) def test_get_available_ips(self, mock): pfx = nb.prefixes.get(1) ret = pfx.available_ips.list() mock.assert_called_with( - 'http://localhost:8000/api/ipam/prefixes/1/available-ips/', + "http://localhost:8000/api/ipam/prefixes/1/available-ips/", params={}, json=None, headers=HEADERS, - verify=True ) self.assertTrue(ret) self.assertEqual(len(ret), 3) @patch( - 'pynetbox.core.query.requests.sessions.Session.post', - return_value=Response(fixture='ipam/available-ips-post.json') + "pynetbox.core.query.requests.sessions.Session.post", + return_value=Response(fixture="ipam/available-ips-post.json"), ) @patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='ipam/prefix.json'), + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="ipam/prefix.json"), ) def test_create_available_ips(self, _, post): - expected_result = { - 'status': 1, - 'description': '', - 'nat_inside': None, - 'role': None, - 'vrf': None, - 'address': - '10.1.1.1/32', - 'interface': None, - 'id': 1, - 'tenant': None - } - create_parms = dict( - status=2, - ) + create_parms = dict(status=2,) pfx = nb.prefixes.get(1) ret = pfx.available_ips.create(create_parms) post.assert_called_with( - 'http://localhost:8000/api/ipam/prefixes/1/available-ips/', + "http://localhost:8000/api/ipam/prefixes/1/available-ips/", params={}, headers=POST_HEADERS, json=create_parms, - verify=True ) self.assertTrue(ret) - self.assertEqual(ret, expected_result) + self.assertTrue(isinstance(ret, pynetbox.models.ipam.IpAddresses)) @patch( - 'pynetbox.core.query.requests.sessions.Session.get', + "pynetbox.core.query.requests.sessions.Session.get", side_effect=[ - Response(fixture='ipam/prefix.json'), - Response(fixture='ipam/available-prefixes.json'), - ] + Response(fixture="ipam/prefix.json"), + Response(fixture="ipam/available-prefixes.json"), + ], ) def test_get_available_prefixes(self, mock): pfx = nb.prefixes.get(1) ret = pfx.available_prefixes.list() mock.assert_called_with( - 'http://localhost:8000/api/ipam/prefixes/1/available-prefixes/', + "http://localhost:8000/api/ipam/prefixes/1/available-prefixes/", params={}, json=None, headers=HEADERS, - verify=True ) self.assertTrue(ret) @patch( - 'pynetbox.core.query.requests.sessions.Session.post', - return_value=Response(fixture='ipam/available-prefixes-post.json') + "pynetbox.core.query.requests.sessions.Session.post", + return_value=Response(fixture="ipam/available-prefixes-post.json"), ) @patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='ipam/prefix.json'), + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="ipam/prefix.json"), ) def test_create_available_prefixes(self, _, post): - create_parms = dict( - prefix_length=30, - ) + create_parms = dict(prefix_length=30,) pfx = nb.prefixes.get(1) ret = pfx.available_prefixes.create(create_parms) post.assert_called_with( - 'http://localhost:8000/api/ipam/prefixes/1/available-prefixes/', + "http://localhost:8000/api/ipam/prefixes/1/available-prefixes/", params={}, headers=POST_HEADERS, json=create_parms, - verify=True ) self.assertTrue(ret) + self.assertTrue(isinstance(ret[0], pynetbox.models.ipam.Prefixes)) class IPAddressTestCase(Generic.Tests): - name = 'ip_addresses' - name_singular = 'ip_address' + name = "ip_addresses" + name_singular = "ip_address" @patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='ipam/ip_address.json') + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="ipam/ip_address.json"), ) def test_modify(self, *_): ret = nb.prefixes.get(1) - ret.description = 'testing' + ret.description = "testing" ret_serialized = ret.serialize() self.assertTrue(ret_serialized) self.assertFalse(ret._compare()) - self.assertEqual(ret_serialized['address'], '10.0.255.1/32') - self.assertEqual(ret_serialized['description'], 'testing') + self.assertEqual(ret_serialized["address"], "10.0.255.1/32") + self.assertEqual(ret_serialized["description"], "testing") class RoleTestCase(Generic.Tests): - name = 'roles' + name = "roles" class RIRTestCase(Generic.Tests): - name = 'rirs' + name = "rirs" class AggregatesTestCase(Generic.Tests): - name = 'aggregates' + name = "aggregates" class VlanTestCase(Generic.Tests): - name = 'vlans' + name = "vlans" + + @patch( + "pynetbox.core.query.requests.sessions.Session.get", + side_effect=[ + Response(fixture="ipam/vlan.json"), + Response(fixture="dcim/interface.json"), + ], + ) + def test_vlan_in_interface(self, mock): + vlan = nb.vlans.get(3) + interface = api.dcim.interfaces.get(1) + mock.assert_called_with( + "http://localhost:8000/api/dcim/interfaces/1/", + params={}, + json=None, + headers=HEADERS, + ) + self.assertEqual(vlan.vid, interface.tagged_vlans[0].vid) + self.assertEqual(vlan.id, interface.tagged_vlans[0].id) + self.assertTrue(vlan in interface.tagged_vlans) class VlanGroupsTestCase(Generic.Tests): - name = 'vlan_groups' + name = "vlan_groups" class VRFTestCase(Generic.Tests): - name = 'vrfs' + name = "vrfs" # class ServicesTestCase(Generic.Tests): diff --git a/tests/test_query.py b/tests/test_query.py index 6d30388a..60165180 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -5,11 +5,6 @@ class TestQuery(unittest.TestCase): - def test_url_param_builder(self): - test_params = OrderedDict([ - ('abc', '123'), - ('xyz', '321'), - ('enc', '#12'), - ]) - self.assertEqual(url_param_builder(test_params), '?abc=123&xyz=321&enc=%2312') + test_params = OrderedDict([("abc", "123"), ("xyz", "321"), ("enc", "#12"),]) + self.assertEqual(url_param_builder(test_params), "?abc=123&xyz=321&enc=%2312") diff --git a/tests/test_tenancy.py b/tests/test_tenancy.py index 8b51d1d6..7a006894 100644 --- a/tests/test_tenancy.py +++ b/tests/test_tenancy.py @@ -10,95 +10,78 @@ from mock import patch -api = pynetbox.api( - "http://localhost:8000", -) +api = pynetbox.api("http://localhost:8000",) nb = api.tenancy -HEADERS = { - 'accept': 'application/json;' -} +HEADERS = {"accept": "application/json;"} class Generic(object): class Tests(unittest.TestCase): - name = '' + name = "" ret = pynetbox.core.response.Record - app = 'tenancy' + app = "tenancy" def test_get_all(self): with patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='{}/{}.json'.format( - self.app, - self.name - )) + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="{}/{}.json".format(self.app, self.name)), ) as mock: ret = getattr(nb, self.name).all() self.assertTrue(ret) self.assertTrue(isinstance(ret, list)) self.assertTrue(isinstance(ret[0], self.ret)) mock.assert_called_with( - 'http://localhost:8000/api/{}/{}/'.format( - self.app, - self.name.replace('_', '-') + "http://localhost:8000/api/{}/{}/".format( + self.app, self.name.replace("_", "-") ), params={}, json=None, headers=HEADERS, - verify=True ) def test_filter(self): with patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='{}/{}.json'.format( - self.app, - self.name - )) + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="{}/{}.json".format(self.app, self.name)), ) as mock: - ret = getattr(nb, self.name).filter(name='test') + ret = getattr(nb, self.name).filter(name="test") self.assertTrue(ret) self.assertTrue(isinstance(ret, list)) self.assertTrue(isinstance(ret[0], self.ret)) mock.assert_called_with( - 'http://localhost:8000/api/{}/{}/'.format( - self.app, - self.name.replace('_', '-') + "http://localhost:8000/api/{}/{}/".format( + self.app, self.name.replace("_", "-") ), params={"name": "test"}, json=None, headers=HEADERS, - verify=True ) def test_get(self): with patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='{}/{}.json'.format( - self.app, - self.name[:-1] - )) + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response( + fixture="{}/{}.json".format(self.app, self.name[:-1]) + ), ) as mock: ret = getattr(nb, self.name).get(1) self.assertTrue(ret) self.assertTrue(isinstance(ret, self.ret)) mock.assert_called_with( - 'http://localhost:8000/api/{}/{}/1/'.format( - self.app, - self.name.replace('_', '-') + "http://localhost:8000/api/{}/{}/1/".format( + self.app, self.name.replace("_", "-") ), params={}, json=None, headers=HEADERS, - verify=True ) class TenantsTestCase(Generic.Tests): - name = 'tenants' + name = "tenants" class TenantGroupsTestCase(Generic.Tests): - name = 'tenant_groups' + name = "tenant_groups" diff --git a/tests/test_virtualization.py b/tests/test_virtualization.py index 63e19f42..8749b9ec 100644 --- a/tests/test_virtualization.py +++ b/tests/test_virtualization.py @@ -10,107 +10,90 @@ from mock import patch -api = pynetbox.api( - "http://localhost:8000", -) +api = pynetbox.api("http://localhost:8000",) nb = api.virtualization -HEADERS = { - 'accept': 'application/json;' -} +HEADERS = {"accept": "application/json;"} class Generic(object): class Tests(unittest.TestCase): - name = '' + name = "" ret = pynetbox.core.response.Record - app = 'virtualization' + app = "virtualization" def test_get_all(self): with patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='{}/{}.json'.format( - self.app, - self.name - )) + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="{}/{}.json".format(self.app, self.name)), ) as mock: ret = getattr(nb, self.name).all() self.assertTrue(ret) self.assertTrue(isinstance(ret, list)) self.assertTrue(isinstance(ret[0], self.ret)) mock.assert_called_with( - 'http://localhost:8000/api/{}/{}/'.format( - self.app, - self.name.replace('_', '-') + "http://localhost:8000/api/{}/{}/".format( + self.app, self.name.replace("_", "-") ), params={}, json=None, headers=HEADERS, - verify=True ) def test_filter(self): with patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='{}/{}.json'.format( - self.app, - self.name - )) + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response(fixture="{}/{}.json".format(self.app, self.name)), ) as mock: - ret = getattr(nb, self.name).filter(name='test') + ret = getattr(nb, self.name).filter(name="test") self.assertTrue(ret) self.assertTrue(isinstance(ret, list)) self.assertTrue(isinstance(ret[0], self.ret)) mock.assert_called_with( - 'http://localhost:8000/api/{}/{}/'.format( - self.app, - self.name.replace('_', '-') + "http://localhost:8000/api/{}/{}/".format( + self.app, self.name.replace("_", "-") ), params={"name": "test"}, json=None, headers=HEADERS, - verify=True ) def test_get(self): with patch( - 'pynetbox.core.query.requests.sessions.Session.get', - return_value=Response(fixture='{}/{}.json'.format( - self.app, - self.name[:-1] - )) + "pynetbox.core.query.requests.sessions.Session.get", + return_value=Response( + fixture="{}/{}.json".format(self.app, self.name[:-1]) + ), ) as mock: ret = getattr(nb, self.name).get(1) self.assertTrue(ret) self.assertTrue(isinstance(ret, self.ret)) mock.assert_called_with( - 'http://localhost:8000/api/{}/{}/1/'.format( - self.app, - self.name.replace('_', '-') + "http://localhost:8000/api/{}/{}/1/".format( + self.app, self.name.replace("_", "-") ), params={}, json=None, headers=HEADERS, - verify=True ) class ClusterTypesTestCase(Generic.Tests): - name = 'cluster_types' + name = "cluster_types" class ClusterGroupsTestCase(Generic.Tests): - name = 'cluster_groups' + name = "cluster_groups" class ClustersTestCase(Generic.Tests): - name = 'clusters' + name = "clusters" class VirtualMachinesTestCase(Generic.Tests): - name = 'virtual_machines' + name = "virtual_machines" class InterfacesTestCase(Generic.Tests): - name = 'interfaces' + name = "interfaces" diff --git a/tests/unit/test_endpoint.py b/tests/unit/test_endpoint.py index 4ce10de9..abdc9605 100644 --- a/tests/unit/test_endpoint.py +++ b/tests/unit/test_endpoint.py @@ -11,14 +11,11 @@ class EndPointTestCase(unittest.TestCase): - def test_filter(self): - with patch( - "pynetbox.core.query.Request.get", return_value=Mock() - ) as mock: + with patch("pynetbox.core.query.Request.get", return_value=Mock()) as mock: api = Mock(base_url="http://localhost:8000/api") app = Mock(name="test") - mock.return_value = [{'id': 123}, {'id': 321}] + mock.return_value = [{"id": 123}, {"id": 321}] test_obj = Endpoint(api, app, "test") test = test_obj.filter(test="test") self.assertEqual(len(test), 2) @@ -40,9 +37,7 @@ def test_filter_reserved_kwargs(self): test_obj.filter(id=1) def test_choices(self): - with patch( - "pynetbox.core.query.Request.options", return_value=Mock() - ) as mock: + with patch("pynetbox.core.query.Request.options", return_value=Mock()) as mock: api = Mock(base_url="http://localhost:8000/api") app = Mock(name="test") mock.return_value = { diff --git a/tests/unit/test_query.py b/tests/unit/test_query.py index e2605486..4f7fc173 100644 --- a/tests/unit/test_query.py +++ b/tests/unit/test_query.py @@ -27,7 +27,6 @@ def test_get_count(self): "http://localhost:8001/api/dcim/devices/", params={"q": "abcd", "limit": 1}, headers={"accept": "application/json;"}, - verify=True, ) test_obj.http_session.get.ok = True test = test_obj.get_count() @@ -37,13 +36,11 @@ def test_get_count(self): params={"q": "abcd", "limit": 1}, headers={"accept": "application/json;"}, json=None, - verify=True, ) def test_get_count_no_filters(self): test_obj = Request( - http_session=Mock(), - base="http://localhost:8001/api/dcim/devices", + http_session=Mock(), base="http://localhost:8001/api/dcim/devices", ) test_obj.http_session.get.return_value.json.return_value = { "count": 42, @@ -59,5 +56,4 @@ def test_get_count_no_filters(self): params={"limit": 1}, headers={"accept": "application/json;"}, json=None, - verify=True, ) diff --git a/tests/unit/test_request.py b/tests/unit/test_request.py new file mode 100644 index 00000000..89bcabdc --- /dev/null +++ b/tests/unit/test_request.py @@ -0,0 +1,20 @@ +import unittest + +import six + +from pynetbox.core.query import Request + +if six.PY3: + from unittest.mock import Mock +else: + from mock import Mock + + +class RequestTestCase(unittest.TestCase): + def test_get_openapi(self): + test = Request("http://localhost:8080/api", Mock()) + test.get_openapi() + test.http_session.get.assert_called_with( + "http://localhost:8080/api/docs/?format=openapi", + headers={"Content-Type": "application/json;"}, + ) diff --git a/tests/unit/test_response.py b/tests/unit/test_response.py index dce2c329..c93452c5 100644 --- a/tests/unit/test_response.py +++ b/tests/unit/test_response.py @@ -31,7 +31,7 @@ def test_serialize_list_of_records(self): }, ], } - test_obj = Record(test_values, None, None) + test_obj = Record(test_values, Mock(base_url="test"), None) test = test_obj.serialize() self.assertEqual(test["tagged_vlans"], [1, 2]) @@ -75,7 +75,7 @@ def test_diff_append_records_list(self): } ], } - test_obj = Record(test_values, None, None) + test_obj = Record(test_values, Mock(base_url="test"), None) test_obj.tagged_vlans.append(1) test = test_obj._diff() self.assertFalse(test) @@ -115,10 +115,7 @@ def test_dict(self): def test_choices_idempotence_prev27(self): test_values = { "id": 123, - "choices_test": { - "value": 1, - "label": "test", - }, + "choices_test": {"value": 1, "label": "test",}, } test = Record(test_values, None, None) test.choices_test = 1 @@ -127,11 +124,7 @@ def test_choices_idempotence_prev27(self): def test_choices_idempotence_v27(self): test_values = { "id": 123, - "choices_test": { - "value": "test", - "label": "test", - "id": 1, - }, + "choices_test": {"value": "test", "label": "test", "id": 1,}, } test = Record(test_values, None, None) test.choices_test = "test" @@ -140,10 +133,7 @@ def test_choices_idempotence_v27(self): def test_choices_idempotence_v28(self): test_values = { "id": 123, - "choices_test": { - "value": "test", - "label": "test", - }, + "choices_test": {"value": "test", "label": "test",}, } test = Record(test_values, None, None) test.choices_test = "test" @@ -179,18 +169,39 @@ def test_compare(self): def test_nested_write(self): app = Mock() - app.token = 'abc123' + app.token = "abc123" + app.base_url = "http://localhost:8080/api" endpoint = Mock() endpoint.name = "test-endpoint" - test = Record({ - 'id': 123, - 'name': 'test', - 'child': { - 'id': 321, - 'name': 'test123', - 'url': 'http://localhost:8080/test', + test = Record( + { + "id": 123, + "name": "test", + "child": { + "id": 321, + "name": "test123", + "url": "http://localhost:8080/api/test-app/test-endpoint/", + }, }, - }, app, endpoint) - test.child.name = 'test321' + app, + endpoint, + ) + test.child.name = "test321" test.child.save() - self.assertEqual(app.http_session.patch.call_args[0][0], "http://localhost:8080/test/") + self.assertEqual( + app.http_session.patch.call_args[0][0], + "http://localhost:8080/api/test-app/test-endpoint/", + ) + + def test_endpoint_from_url(self): + test = Record( + { + "id": 123, + "name": "test", + "url": "http://localhost:8080/api/test-app/test-endpoint/1/", + }, + Mock(), + None, + ) + ret = test._endpoint_from_url(test.url) + self.assertEqual(ret.name, "test-endpoint") diff --git a/tests/util.py b/tests/util.py index 3e481594..6c08f576 100644 --- a/tests/util.py +++ b/tests/util.py @@ -8,7 +8,7 @@ def __init__(self, fixture=None, status_code=200, ok=True, content=None): self.ok = ok def load_fixture(self, path): - with open("tests/fixtures/{}".format(path), 'r') as f: + with open("tests/fixtures/{}".format(path), "r") as f: return f.read() def json(self):