diff --git a/.coveragerc b/.coveragerc index c3b3ba5..4dc55b3 100644 --- a/.coveragerc +++ b/.coveragerc @@ -1,4 +1,5 @@ [run] +parallel = true omit = /*/test* /tests diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..d949732 --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,14 @@ +version: 2 +updates: + - package-ecosystem: pip + directory: / + schedule: + interval: monthly + commit-message: + prefix: "[deps] " + - package-ecosystem: github-actions + directory: / + schedule: + interval: monthly + commit-message: + prefix: "[ci] " diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md new file mode 100644 index 0000000..44e9369 --- /dev/null +++ b/.github/pull_request_template.md @@ -0,0 +1,20 @@ +## Checklist + +- [ ] I have read the [OpenWISP Contributing Guidelines](http://openwisp.io/docs/developer/contributing.html). +- [ ] I have manually tested the changes proposed in this pull request. +- [ ] I have written new test cases for new code and/or updated existing tests for changes to existing code. +- [ ] I have updated the documentation. + +## Reference to Existing Issue + +Closes #. + +Please [open a new issue](https://github.com/openwisp/netengine/issues/new/choose) if there isn't an existing issue yet. + +## Description of Changes + +Please describe these changes. + +## Screenshot + +Please include any relevant screenshots. diff --git a/.github/workflows/backport.yml b/.github/workflows/backport.yml new file mode 100644 index 0000000..556ec7d --- /dev/null +++ b/.github/workflows/backport.yml @@ -0,0 +1,42 @@ +name: Backport fixes to stable branch + +on: + push: + branches: + - master + issue_comment: + types: [created] + +concurrency: + group: backport-${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: false + +permissions: + contents: write + pull-requests: write + +jobs: + backport-on-push: + if: github.event_name == 'push' + uses: openwisp/openwisp-utils/.github/workflows/reusable-backport.yml@master + with: + commit_sha: ${{ github.sha }} + secrets: + app_id: ${{ secrets.OPENWISP_BOT_APP_ID }} + private_key: ${{ secrets.OPENWISP_BOT_PRIVATE_KEY }} + + backport-on-comment: + if: > + github.event_name == 'issue_comment' && + github.event.issue.pull_request && + github.event.issue.pull_request.merged_at != null && + github.event.issue.state == 'closed' && + contains(fromJSON('["MEMBER", "OWNER"]'), github.event.comment.author_association) && + startsWith(github.event.comment.body, '/backport') + uses: openwisp/openwisp-utils/.github/workflows/reusable-backport.yml@master + with: + pr_number: ${{ github.event.issue.number }} + comment_body: ${{ github.event.comment.body }} + secrets: + app_id: ${{ secrets.OPENWISP_BOT_APP_ID }} + private_key: ${{ secrets.OPENWISP_BOT_PRIVATE_KEY }} diff --git a/.github/workflows/bot-changelog-trigger.yml b/.github/workflows/bot-changelog-trigger.yml index 9efa56f..873b6be 100644 --- a/.github/workflows/bot-changelog-trigger.yml +++ b/.github/workflows/bot-changelog-trigger.yml @@ -20,7 +20,7 @@ jobs: env: PR_TITLE: ${{ github.event.pull_request.title }} run: | - if echo "$PR_TITLE" | grep -qiE '^\[(feature|fix|change)\]'; then + if echo "$PR_TITLE" | grep -qiE '^\[(feature|fix|change!?)\]'; then echo "has_noteworthy=true" >> $GITHUB_OUTPUT fi diff --git a/.github/workflows/bot-ci-failure.yml b/.github/workflows/bot-ci-failure.yml new file mode 100644 index 0000000..c3e9189 --- /dev/null +++ b/.github/workflows/bot-ci-failure.yml @@ -0,0 +1,87 @@ +name: CI Failure Bot + +on: + workflow_run: + workflows: ["Netengine CI Build"] + types: + - completed + +permissions: + pull-requests: read + actions: read + contents: read + +concurrency: + group: ci-failure-${{ github.repository }}-${{ github.event.workflow_run.pull_requests[0].number || github.event.workflow_run.head_branch }} + cancel-in-progress: true + +jobs: + find-pr: + runs-on: ubuntu-latest + if: ${{ github.event.workflow_run.conclusion == 'failure' && github.event.workflow_run.event == 'pull_request' }} + outputs: + pr_number: ${{ steps.pr.outputs.number }} + pr_author: ${{ steps.pr.outputs.author }} + steps: + - name: Find PR Number + id: pr + env: + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + REPO: ${{ github.repository }} + PR_NUMBER_PAYLOAD: ${{ github.event.workflow_run.pull_requests[0].number }} + EVENT_HEAD_SHA: ${{ github.event.workflow_run.head_sha }} + run: | + emit_pr() { + local pr_number="$1" + local pr_author + pr_author=$(gh pr view "$pr_number" --repo "$REPO" --json author --jq '.author.login // empty' 2>/dev/null || echo "") + if [ -z "$pr_author" ] || [ "$pr_author" = "null" ]; then + echo "::warning::Could not fetch PR author for PR #$pr_number" + fi + echo "number=$pr_number" >> "$GITHUB_OUTPUT" + echo "author=$pr_author" >> "$GITHUB_OUTPUT" + } + PR_NUMBER="$PR_NUMBER_PAYLOAD" + if [ -n "$PR_NUMBER" ]; then + echo "Found PR #$PR_NUMBER from workflow payload." + emit_pr "$PR_NUMBER" + exit 0 + fi + HEAD_SHA="$EVENT_HEAD_SHA" + echo "Payload empty. Searching for PR via Commits API..." + PR_NUMBER=$(gh api repos/$REPO/commits/$HEAD_SHA/pulls -q '.[0].number' 2>/dev/null || true) + if [ -n "$PR_NUMBER" ] && [ "$PR_NUMBER" != "null" ]; then + echo "Found PR #$PR_NUMBER using Commits API." + emit_pr "$PR_NUMBER" + exit 0 + fi + echo "API lookup failed/empty. Scanning open PRs for matching head SHA..." + PR_NUMBER=$(gh pr list --repo "$REPO" --state open --limit 100 --json number,headRefOid --jq ".[] | select(.headRefOid == \"$HEAD_SHA\") | .number" | head -n 1) + if [ -n "$PR_NUMBER" ]; then + echo "Found PR #$PR_NUMBER by scanning open PRs." + emit_pr "$PR_NUMBER" + exit 0 + fi + echo "::warning::No open PR found. This workflow run might not be attached to an open PR." + exit 0 + + call-ci-failure-bot: + needs: find-pr + if: ${{ needs.find-pr.outputs.pr_number != '' }} + permissions: + pull-requests: write + actions: write + contents: read + uses: openwisp/openwisp-utils/.github/workflows/reusable-bot-ci-failure.yml@master + with: + pr_number: ${{ needs.find-pr.outputs.pr_number }} + head_sha: ${{ github.event.workflow_run.head_sha }} + head_repo: ${{ github.event.workflow_run.head_repository.full_name }} + base_repo: ${{ github.repository }} + run_id: ${{ github.event.workflow_run.id }} + pr_author: ${{ needs.find-pr.outputs.pr_author }} + actor: ${{ github.event.workflow_run.actor.login }} + secrets: + GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }} + APP_ID: ${{ secrets.OPENWISP_BOT_APP_ID }} + PRIVATE_KEY: ${{ secrets.OPENWISP_BOT_PRIVATE_KEY }} diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7b41574..4d4d799 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1,5 +1,4 @@ --- - name: Netengine CI Build on: @@ -11,61 +10,63 @@ on: - master jobs: - build: name: Python==${{ matrix.python-version }} - runs-on: ubuntu-20.04 + runs-on: ubuntu-24.04 strategy: fail-fast: false matrix: python-version: - - 3.6 - - 3.7 - - 3.8 - - 3.9 + - "3.10" + - "3.11" + - "3.12" + - "3.13" steps: - - uses: actions/checkout@v2 - with: - ref: ${{ github.event.pull_request.head.sha }} + - uses: actions/checkout@v7 + with: + ref: ${{ github.event.pull_request.head.sha }} - - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v2 - with: - python-version: ${{ matrix.python-version }} + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v6 + with: + python-version: ${{ matrix.python-version }} + cache: "pip" + cache-dependency-path: | + **/requirements*.txt - - name: Install Dependencies - id: deps - run: | - pip install -U wheel setuptools - pip install -e . - pip install -U -r requirements-test.txt + - name: Install Dependencies + id: deps + run: | + pip install -U pip wheel setuptools + pip install -U -r requirements-test.txt + pip install -U -e . - - name: Run QA Checks - run: ./run-qa-checks + - name: Run QA Checks + run: ./run-qa-checks - - name: Run tests - if: ${{ !cancelled() && steps.deps.conclusion == 'success' }} - run: coverage run --source=netengine ./runtests.py + - name: Run tests + if: ${{ !cancelled() && steps.deps.conclusion == 'success' }} + run: | + coverage run runtests.py + coverage combine + coverage xml - - name: Upload Coverage - if: ${{ success() }} - run: coveralls --service=github - env: - GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - COVERALLS_FLAG_NAME: python-${{ matrix.python-version }} - COVERALLS_PARALLEL: true + - name: Upload Coverage + if: ${{ success() }} + uses: coverallsapp/github-action@v2 + with: + parallel: true + format: cobertura + flag-name: python-${{ matrix.python-version }} + github-token: ${{ secrets.GITHUB_TOKEN }} coveralls: - name: Finish Coveralls needs: build runs-on: ubuntu-latest - container: python:3-slim steps: - - name: Finished - run: | - python3 -m pip install --upgrade coveralls - coveralls --finish - env: - GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + - name: Coveralls Finished + uses: coverallsapp/github-action@v2 + with: + parallel-finished: true diff --git a/.github/workflows/version-branch.yml b/.github/workflows/version-branch.yml new file mode 100644 index 0000000..ddef2b7 --- /dev/null +++ b/.github/workflows/version-branch.yml @@ -0,0 +1,13 @@ +name: Replicate Commits to Version Branch + +on: + push: + branches: + - master + +jobs: + version-branch: + uses: openwisp/openwisp-utils/.github/workflows/reusable-version-branch.yml@master + with: + module_name: netengine + install_package: true diff --git a/README.rst b/README.rst index 735765a..da2fcf6 100644 --- a/README.rst +++ b/README.rst @@ -1,4 +1,3 @@ -========= netengine ========= @@ -13,7 +12,7 @@ netengine .. image:: https://badge.fury.io/py/netengine.svg :target: http://badge.fury.io/py/netengine ------------------------------- +---- .. image:: https://raw.githubusercontent.com/openwisp/netengine/master/docs/source/images/netengine-logo.png @@ -21,8 +20,13 @@ Abstraction layer for extracting information from network devices. Documentation: http://netengine.rtfd.org +Supported Python Versions +------------------------- + +NetEngine supports Python 3.10, 3.11, 3.12, and 3.13. + Contribute -========== +---------- 1. Join the `OpenWISP mailing list`_ 2. Fork this repo @@ -34,5 +38,6 @@ Contribute 8. Document your changes 9. Send pull request -.. _PEP8, Style Guide for Python Code: http://www.python.org/dev/peps/pep-0008/ -.. _OpenWISP mailing list: https://groups.google.com/g/openwisp +.. _openwisp mailing list: https://groups.google.com/g/openwisp + +.. _pep8, style guide for python code: http://www.python.org/dev/peps/pep-0008/ diff --git a/docs/source/conf.py b/docs/source/conf.py index 757c429..67d08d7 100755 --- a/docs/source/conf.py +++ b/docs/source/conf.py @@ -31,29 +31,29 @@ extensions = [] # Add any paths that contain templates here, relative to this directory. -templates_path = ['_templates'] +templates_path = ["_templates"] # The suffix of source filenames. -source_suffix = '.rst' +source_suffix = ".rst" # The encoding of source files. # source_encoding = 'utf-8-sig' # The master toctree document. -master_doc = 'index' +master_doc = "index" # General information about the project. -project = u'netengine' -copyright = u'OpenWISP.org' +project = "netengine" +copyright = "OpenWISP.org" # 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 short X.Y version. -version = '0.1' +version = "0.1" # The full version, including alpha/beta/rc tags. -release = '0.1' +release = "0.1" # The language for content autogenerated by Sphinx. Refer to documentation # for a list of supported languages. @@ -85,7 +85,7 @@ # show_authors = False # The name of the Pygments (syntax highlighting) style to use. -pygments_style = 'sphinx' +pygments_style = "sphinx" # A list of ignored prefixes for module index sorting. # modindex_common_prefix = [] @@ -98,7 +98,7 @@ # The theme to use for HTML and HTML Help pages. See the documentation for # a list of builtin themes. -html_theme = 'default' +html_theme = "default" # Theme options are theme-specific and customize the look and feel of a theme # further. For a list of options available for each theme, see the @@ -127,7 +127,7 @@ # Add any paths that contain custom static files (such as style sheets) here, # relative to this directory. They are copied after the builtin static files, # so a file named "default.css" will overwrite the builtin "default.css". -html_static_path = ['_static'] +html_static_path = ["_static"] # Add any extra paths that contain custom files (such as robots.txt or # .htaccess) here, relative to this directory. These files are copied @@ -176,7 +176,7 @@ # html_file_suffix = None # Output file base name for HTML help builder. -htmlhelp_basename = 'netenginedoc' +htmlhelp_basename = "netenginedoc" # -- Options for LaTeX output --------------------------------------------- @@ -195,11 +195,11 @@ # author, documentclass [howto, manual, or own class]). latex_documents = [ ( - 'index', - 'netengine.tex', - u'netengine Documentation', - u'Alessandro Bucciarelli, Federico Capoano', - 'manual', + "index", + "netengine.tex", + "netengine Documentation", + "Alessandro Bucciarelli, Federico Capoano", + "manual", ), ] @@ -230,10 +230,10 @@ # (source start file, name, description, authors, manual section). man_pages = [ ( - 'index', - 'netengine', - u'netengine Documentation', - [u'Alessandro Bucciarelli, Federico Capoano'], + "index", + "netengine", + "netengine Documentation", + ["Alessandro Bucciarelli, Federico Capoano"], 1, ) ] @@ -249,13 +249,13 @@ # dir menu entry, description, category) texinfo_documents = [ ( - 'index', - 'netengine', - u'netengine Documentation', - u'Alessandro Bucciarelli, Federico Capoano', - 'netengine', - 'One line description of project.', - 'Miscellaneous', + "index", + "netengine", + "netengine Documentation", + "Alessandro Bucciarelli, Federico Capoano", + "netengine", + "One line description of project.", + "Miscellaneous", ), ] diff --git a/docs/source/index.rst b/docs/source/index.rst index 0c70a47..b3f1b04 100755 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -1,34 +1,36 @@ -========= Netengine ========= -**Netengine** is a python library that aims to provide a single API to extract common -information from network devices with different firwmares (eg: OpenWRT, AirOS) using different protocols -such as the Simple Network Management Protocol (SNMP), and the ability to easily add other backends +**Netengine** is a python library that aims to provide a single API to +extract common information from network devices with different firwmares +(eg: OpenWRT, AirOS) using different protocols such as the Simple Network +Management Protocol (SNMP), and the ability to easily add other backends like SSH and HTTP (`read more <#status-of-this-project>`_). -You can immagine **Netengine** as a read-only ORM (Object Relational Mapper) equivalent for networks. +You can immagine **Netengine** as a read-only ORM (Object Relational +Mapper) equivalent for networks. -=========== Motivations =========== -While dealing with networks in the real world, it's highly probable that you will -deal with a network which is made with very different routers, switches and servers. -Some may support standard SNMP mibs, some may not, some may implement other HTTP APIs, -some may even implement obscure/custom SNMP mibs. +While dealing with networks in the real world, it's highly probable that +you will deal with a network which is made with very different routers, +switches and servers. Some may support standard SNMP mibs, some may not, +some may implement other HTTP APIs, some may even implement obscure/custom +SNMP mibs. -If you need to develop a web application that automates some networking tasks, you -don't want to deal with all those differences in the application code, because it -would become hard to mantain very soon. You also might not want to tie your web -app code to a specific vendor or firmware because that would make your software unflexible. +If you need to develop a web application that automates some networking +tasks, you don't want to deal with all those differences in the +application code, because it would become hard to mantain very soon. You +also might not want to tie your web app code to a specific vendor or +firmware because that would make your software unflexible. -If we had a single API we could let web developers focus on the task they need to accomplish -rather than dealing with different firmwares, different linux distributions and so on. +If we had a single API we could let web developers focus on the task they +need to accomplish rather than dealing with different firmwares, different +linux distributions and so on. The goal of this project is to build that single API. -====================== Status of this project ====================== @@ -38,26 +40,29 @@ The 0.1 final version will be out by August 2021. .. note:: - The legacy versions of this project had support for SSH and HTTP for extracting information from - devices. To see how it worked, visit the - `0.1.0 alpha release `_ page on + The legacy versions of this project had support for SSH and HTTP for + extracting information from devices. To see how it worked, visit the + `0.1.0 alpha release + `_ page on github. -======= Install ======= -Install the development version (tarball):: +Install the development version (tarball): + +:: pip install https://github.com/openwisp/netengine/tarball/master -Alternatively, you can install via pip using git:: +Alternatively, you can install via pip using git: + +:: pip install -e git+git://github.com/openwisp/netengine#egg=netengine -========== Contents: -========== +========= .. toctree:: :maxdepth: 2 @@ -66,7 +71,7 @@ Contents: /topics/snmp Indices and tables -================== +------------------ -* :ref:`genindex` -* :ref:`search` +- :ref:`genindex` +- :ref:`search` diff --git a/docs/source/topics/snmp.rst b/docs/source/topics/snmp.rst index e584ac9..2727590 100644 --- a/docs/source/topics/snmp.rst +++ b/docs/source/topics/snmp.rst @@ -1,50 +1,44 @@ - -************** SNMP backend -************** +============ SNMP -======= - -SNMP (Simple Network Management Protocol) is a network protocol very useful for retrieving info from a device. -All the information is retrieved by using codes called MIBs. All MIBs have a tree-like structure, every main information is the root and by adding more detail to the info -the tree gains more depth. -Obviously, by getting the smallest MIB which is "1" or simply " . " one can get all the tree. - - +---- +SNMP (Simple Network Management Protocol) is a network protocol very +useful for retrieving info from a device. All the information is retrieved +by using codes called MIBs. All MIBs have a tree-like structure, every +main information is the root and by adding more detail to the info the +tree gains more depth. Obviously, by getting the smallest MIB which is "1" +or simply " . " one can get all the tree. The SNMP backend provides support for 2 firmwares: - * AirOS - * OpenWRT - - - + - AirOS + - OpenWRT AirOS example -============= +------------- :: - from netengine.backends.snmp import AirOS - device = AirOS("10.40.0.130") - device.name - 'RM5PomeziaSNode' - device.uptime_tuple - (121, 0, 5) # a tuple containing device uptime hours, mins and seconds + from netengine.backends.snmp import AirOS + device = AirOS("10.40.0.130") + device.name + 'RM5PomeziaSNode' + device.uptime_tuple + (121, 0, 5) # a tuple containing device uptime hours, mins and seconds We have just called two simple properties on **device**, but we can ask **device** for more specific values or portions of the SNMP tree not included in the API, just type:: - device.next("1.3.6") + device.next("1.3.6") Otherwise, if you want simply a value of the tree just type:: - device.get_value("oid_you_want_to_ask_for") - - - + device.get_value("oid_you_want_to_ask_for") OpenWRT example -================ +--------------- + +The same instructions typed above can be applied to OpenWRT itself, just +remember to import the correct firmware by typing: -The same instructions typed above can be applied to OpenWRT itself, just remember to import the correct firmware by typing:: +:: - from netengine.backends.snmp import OpenWRT + from netengine.backends.snmp import OpenWRT diff --git a/docs/source/topics/usage.rst b/docs/source/topics/usage.rst index b3a5baa..c6e2d9d 100755 --- a/docs/source/topics/usage.rst +++ b/docs/source/topics/usage.rst @@ -1,61 +1,79 @@ -***** Usage -***** +===== -The usage of Netengine module requires it to be installed properly as explained in :doc:`index<../index>`. -If you have an installation under a virtualenv, enter the folder /bin and type:: +The usage of Netengine module requires it to be installed properly as +explained in :doc:`index <../index>`. If you have an installation under a +virtualenv, enter the folder /bin and type: + +:: source activate -otherwise (if you have installed globally) just open an editor as bpython and you we are ready to go. +otherwise (if you have installed globally) just open an editor as bpython +and you we are ready to go. These are the main steps to follow to use the module: - * import the correct backend and supported framework - * declare a device using the proper constructor - * invoke methods over the device just declared + - import the correct backend and supported framework + - declare a device using the proper constructor + - invoke methods over the device just declared + +So we have: -So we have:: +:: - from netengine.backends. import + from netengine.backends. import - = supported_firmware_constructor + = supported_firmware_constructor -To invoke methods over the just declared device it's necessary to use the dot notation as:: +To invoke methods over the just declared device it's necessary to use the +dot notation as: - . +:: + . Further example will be found inside dedicated docs for every backend -************* Running tests -************* +============= -Install test reqirements:: +Install test reqirements: + +:: pip install -r reqirements.txt pip install -r requirements-test.txt -Clone repo:: +Clone repo: + +:: git clone git://github.com/openwisp/netengine ./runtests.py -To run tests on real devices, first copy the settings file:: +To run tests on real devices, first copy the settings file: + +:: cp test-settings.example.json test-settings.json -Then change the credentials accordingly, now run tests with:: +Then change the credentials accordingly, now run tests with: + +:: DISABLE_MOCKS=1 TEST_SETTINGS_FILE='test-settings.json' ./runtests.py -See test coverage with:: +See test coverage with: + +:: nose2 --with-coverage -Run specific tests by specifying the relative path:: +Run specific tests by specifying the relative path: + +:: # base tests nose2 tests.base diff --git a/netengine/__init__.py b/netengine/__init__.py index dd80d7d..d001086 100644 --- a/netengine/__init__.py +++ b/netengine/__init__.py @@ -1,18 +1,18 @@ -VERSION = (0, 1, 0, 'beta') +VERSION = (0, 1, 0, "beta") __version__ = VERSION def get_version(): - version = '%s.%s' % (VERSION[0], VERSION[1]) + version = "%s.%s" % (VERSION[0], VERSION[1]) if VERSION[2]: - version = '%s.%s' % (version, VERSION[2]) - if VERSION[3:] == ('alpha', 0): - version = '%s pre-alpha' % version + version = "%s.%s" % (version, VERSION[2]) + if VERSION[3:] == ("alpha", 0): + version = "%s pre-alpha" % version else: - if VERSION[3] != 'final': + if VERSION[3] != "final": try: rev = VERSION[4] except IndexError: rev = 0 - version = '%s%s%s' % (version, VERSION[3][0:1], rev) + version = "%s%s%s" % (version, VERSION[3][0:1], rev) return version diff --git a/netengine/backends/__init__.py b/netengine/backends/__init__.py index 41896b3..1a2ed34 100644 --- a/netengine/backends/__init__.py +++ b/netengine/backends/__init__.py @@ -1,4 +1,4 @@ from .base import BaseBackend from .dummy import Dummy -__all__ = ['BaseBackend', 'Dummy'] +__all__ = ["BaseBackend", "Dummy"] diff --git a/netengine/backends/base.py b/netengine/backends/base.py index 86509b0..fafbca2 100644 --- a/netengine/backends/base.py +++ b/netengine/backends/base.py @@ -3,29 +3,27 @@ from netaddr import EUI, NotRegisteredError -__all__ = ['BaseBackend'] +__all__ = ["BaseBackend"] class BaseBackend(object): - """ - Base NetEngine Backend - """ + """Base NetEngine Backend""" __netengine__ = True _dict = OrderedDict def __str__(self): - raise NotImplementedError('Not implemented, must be extended') + raise NotImplementedError("Not implemented, must be extended") def __repr__(self): """returns unicode string represantation""" return self.__str__() def validate(self): - raise NotImplementedError('Not implemented') + raise NotImplementedError("Not implemented") def to_dict(self): - raise NotImplementedError('Not implemented') + raise NotImplementedError("Not implemented") def to_json(self, **kwargs): dictionary = self.to_dict() @@ -33,105 +31,99 @@ def to_json(self, **kwargs): @property def os(self): - """ - Not Implemented + """Not Implemented - should return a tuple in which - the first element is the OS name and - the second element is the OS version + should return a tuple in which the first element is the OS name + and the second element is the OS version """ - raise NotImplementedError('Not implemented') + raise NotImplementedError("Not implemented") @property def name(self): - """ - Not Implemented + """Not Implemented should return a string containing the device name """ - raise NotImplementedError('Not implemented') + raise NotImplementedError("Not implemented") @property def model(self): - """ - Not Implemented + """Not Implemented should return a string containing the device model """ - raise NotImplementedError('Not implemented') + raise NotImplementedError("Not implemented") @property def RAM_total(self): - """ - Not Implemented + """Not Implemented should return a string containing the device RAM in bytes """ - raise NotImplementedError('Not implemented') + raise NotImplementedError("Not implemented") @property def uptime(self): - """ - Not Implemented + """Not Implemented - should return an integer representing the number of seconds of uptime + should return an integer representing the number of seconds of + uptime """ - raise NotImplementedError('Not implemented') + raise NotImplementedError("Not implemented") @property def uptime_tuple(self): - """ - Not Implemented + """Not Implemented should return tuple (days, hours, minutes) """ - raise NotImplementedError('Not implemented') + raise NotImplementedError("Not implemented") @property def ethernet_standard(self): - raise NotImplementedError('Not implemented') + raise NotImplementedError("Not implemented") @property def ethernet_duplex(self): - raise NotImplementedError('Not implemented') + raise NotImplementedError("Not implemented") @property def wireless_channel_width(self): - raise NotImplementedError('Not implemented') + raise NotImplementedError("Not implemented") @property def wireless_mode(self): - raise NotImplementedError('Not implemented') + raise NotImplementedError("Not implemented") @property def wireless_channel(self): - raise NotImplementedError('Not implemented') + raise NotImplementedError("Not implemented") @property def wireless_output_power(self): - raise NotImplementedError('Not implemented') + raise NotImplementedError("Not implemented") @property def wireless_dbm(self): - raise NotImplementedError('Not implemented') + raise NotImplementedError("Not implemented") @property def wireless_noise(self): - raise NotImplementedError('Not implemented') + raise NotImplementedError("Not implemented") # TODO: this sucks @property def olsr(self): - raise NotImplementedError('Not implemented') + raise NotImplementedError("Not implemented") def get_interfaces(self): - raise NotImplementedError('Not implemented') + raise NotImplementedError("Not implemented") def get_manufacturer(self, mac_address): """returns the manufacturer of the network interface""" if not mac_address: - return '' + return "" try: return EUI(mac_address).oui.registration().org except NotRegisteredError: - return '' + return "" diff --git a/netengine/backends/dummy.py b/netengine/backends/dummy.py index 2173234..938c4c2 100644 --- a/netengine/backends/dummy.py +++ b/netengine/backends/dummy.py @@ -2,9 +2,7 @@ class Dummy(BaseBackend): - """ - Dummy backend - """ + """Dummy backend""" def __init__(self, host, port=0): """dummy netengine backend for development or testing""" @@ -12,106 +10,103 @@ def __init__(self, host, port=0): self.port = port def validate(self): - """ - raises NetEngineError exception if anything is wrong with the connection - for example: wrong host, invalid credentials - """ + """Raise NetEngineError when the connection is invalid.""" pass def __str__(self): """print a human readable object description""" - return f'' + return f"" def get_interfaces(self): return [ {}, { - 'ipv6_address_link': '', - 'hardware_address': '00:16:3E:26:9D:13', - 'rx_packets': '147684', - 'broadcast_address': '', - 'rx_bytes': '12956143', - 'link_encap': 'Ethernet', - 'metric': '1', - 'txqueuelen': '1000', - 'net_mask': '', - 'ip_address': '', - 'collisions': '0', - 'interface': 'eth0', - 'tx_bytes': '12523266', - 'mtu': '1500', - 'tx_packets': '132602', - 'ipv6_address_global': '', + "ipv6_address_link": "", + "hardware_address": "00:16:3E:26:9D:13", + "rx_packets": "147684", + "broadcast_address": "", + "rx_bytes": "12956143", + "link_encap": "Ethernet", + "metric": "1", + "txqueuelen": "1000", + "net_mask": "", + "ip_address": "", + "collisions": "0", + "interface": "eth0", + "tx_bytes": "12523266", + "mtu": "1500", + "tx_packets": "132602", + "ipv6_address_global": "", }, { - 'ipv6_address_link': '', - 'hardware_address': '', - 'rx_packets': '', - 'broadcast_address': '', - 'rx_bytes': '', - 'link_encap': 'Local', - 'metric': '', - 'txqueuelen': '', - 'net_mask': '', - 'ip_address': '', - 'collisions': '', - 'interface': 'lo', - 'tx_bytes': '', - 'mtu': '', - 'tx_packets': '', - 'ipv6_address_global': '', + "ipv6_address_link": "", + "hardware_address": "", + "rx_packets": "", + "broadcast_address": "", + "rx_bytes": "", + "link_encap": "Local", + "metric": "", + "txqueuelen": "", + "net_mask": "", + "ip_address": "", + "collisions": "", + "interface": "lo", + "tx_bytes": "", + "mtu": "", + "tx_packets": "", + "ipv6_address_global": "", }, ] def to_dict(self): return self._dict( { - 'name': 'dummy', - 'type': 'radio', # maybe remove - 'os': 'dummyOS', - 'os_version': '0.1', - 'manufacturer': 'dummy inc.', - 'model': 'dummy model', - 'RAM_total': 65536, - 'uptime': 0, - 'uptime_tuple': (0, 0, 0), - 'interfaces': [ + "name": "dummy", + "type": "radio", # maybe remove + "os": "dummyOS", + "os_version": "0.1", + "manufacturer": "dummy inc.", + "model": "dummy model", + "RAM_total": 65536, + "uptime": 0, + "uptime_tuple": (0, 0, 0), + "interfaces": [ { - 'type': 'wireless', - 'name': 'wifi0', - 'mac_address': 'de:9f:db:30:c9:c5', - 'mtu': 1500, - 'standard': '802.11n', - 'channel': 5745, - 'channel_width': 20, - 'mode': 'ap', - 'output_power': 18, - 'tx_rate': None, - 'rx_rate': None, - 'dbm': -27, - 'noise': -97, - 'ip': [ - {'version': 4, 'address': '192.168.1.1'}, - {'version': 6, 'address': '2001:4c00:893b:fede::1'}, + "type": "wireless", + "name": "wifi0", + "mac_address": "de:9f:db:30:c9:c5", + "mtu": 1500, + "standard": "802.11n", + "channel": 5745, + "channel_width": 20, + "mode": "ap", + "output_power": 18, + "tx_rate": None, + "rx_rate": None, + "dbm": -27, + "noise": -97, + "ip": [ + {"version": 4, "address": "192.168.1.1"}, + {"version": 6, "address": "2001:4c00:893b:fede::1"}, ], - 'vap': [{'essid': 'dummyssid', 'bssid': '', 'encryption': ''}], + "vap": [{"essid": "dummyssid", "bssid": "", "encryption": ""}], }, { - 'type': 'ethernet', - 'name': 'eth0', - 'mac_address': 'de:9f:db:30:c9:c4', - 'mtu': 1500, - 'standard': 'fast', - 'duplex': 'full', - 'tx_rate': None, - 'rx_rate': None, - 'ip': [ - {'version': 4, 'address': '192.168.1.2'}, - {'version': 6, 'address': '2001:4c00:893b:fede::2'}, + "type": "ethernet", + "name": "eth0", + "mac_address": "de:9f:db:30:c9:c4", + "mtu": 1500, + "standard": "fast", + "duplex": "full", + "tx_rate": None, + "rx_rate": None, + "ip": [ + {"version": 4, "address": "192.168.1.2"}, + {"version": 6, "address": "2001:4c00:893b:fede::2"}, ], }, ], - 'antennas': [], - 'routing_protocols': [{'name': 'olsr', 'version': 'dummy version'}], + "antennas": [], + "routing_protocols": [{"name": "olsr", "version": "dummy version"}], } ) diff --git a/netengine/backends/snmp/__init__.py b/netengine/backends/snmp/__init__.py index f78c332..d6182f3 100644 --- a/netengine/backends/snmp/__init__.py +++ b/netengine/backends/snmp/__init__.py @@ -2,4 +2,4 @@ from .base import SNMP from .openwrt import OpenWRT -__all__ = ['SNMP', 'OpenWRT', 'AirOS'] +__all__ = ["SNMP", "OpenWRT", "AirOS"] diff --git a/netengine/backends/snmp/airos.py b/netengine/backends/snmp/airos.py index 24c7fd1..9c85cfa 100644 --- a/netengine/backends/snmp/airos.py +++ b/netengine/backends/snmp/airos.py @@ -1,8 +1,6 @@ -""" -NetEngine SNMP Ubiquiti Air OS backend -""" +"""NetEngine SNMP Ubiquiti Air OS backend""" -__all__ = ['AirOS'] +__all__ = ["AirOS"] import binascii @@ -15,114 +13,91 @@ class AirOS(SNMP): - """ - Ubiquiti AirOS SNMP backend - """ + """Ubiquiti AirOS SNMP backend""" - _oid_to_retrieve = '1.3.6.1.2.1.1.9.1.1' + _oid_to_retrieve = "1.3.6.1.2.1.1.9.1.1" def __str__(self): """print a human readable object description""" - return f'' + return f"" def validate(self): - """ - raises NetEngineError exception if anything is wrong with the connection - for example: wrong host, invalid community - """ + """Raise NetEngineError when the connection is invalid.""" # this triggers a connection which # will raise an exception if anything is wrong return self.name @property def os(self): - """ - returns (os_name, os_version) - """ - os_name = 'AirOS' - os_version = self.get_value('1.3.6.1.2.1.1.1.0').split('#')[0].strip() + """returns (os_name, os_version)""" + os_name = "AirOS" + os_version = self.get_value("1.3.6.1.2.1.1.1.0").split("#")[0].strip() return os_name, os_version @property def name(self): - """ - returns a string containing the device name - """ - return self.get_value('1.3.6.1.2.1.1.5.0') + """returns a string containing the device name""" + return self.get_value("1.3.6.1.2.1.1.5.0") @property def model(self): - """ - returns a string containing the device model - """ - oids = ['1.2.840.10036.3.1.2.1.3.5', '1.2.840.10036.3.1.2.1.3.8'] + """returns a string containing the device model""" + oids = ["1.2.840.10036.3.1.2.1.3.5", "1.2.840.10036.3.1.2.1.3.8"] for oid in oids: model = self.get_value(oid) - if model != '': + if model != "": return model @property def firmware(self): - """ - returns a string containing the device firmware - """ - oids = ['1.2.840.10036.3.1.2.1.4.5', '1.2.840.10036.3.1.2.1.4.8'] + """returns a string containing the device firmware""" + oids = ["1.2.840.10036.3.1.2.1.4.5", "1.2.840.10036.3.1.2.1.4.8"] for oid in oids: - tmp = self.get_value(oid).split('.') + tmp = self.get_value(oid).split(".") if tmp is not None: length = len(tmp) i = 0 for piece in tmp: - if 'v' in piece: - return 'AirOS ' + '.'.join(tmp[i:length]) + if "v" in piece: + return "AirOS " + ".".join(tmp[i:length]) i = i + 1 @property def manufacturer(self): - return self.get_manufacturer(self.interfaces_MAC[1]['mac_address']) + return self.get_manufacturer(self.interfaces_MAC[1]["mac_address"]) @property def ssid(self): - """ - returns a string containing the wireless ssid - """ - oids = ['1.2.840.10036.1.1.1.9.5', '1.2.840.10036.1.1.1.9.8'] + """returns a string containing the wireless ssid""" + oids = ["1.2.840.10036.1.1.1.9.5", "1.2.840.10036.1.1.1.9.8"] for oid in oids: - if self.get_value(oid) != '': + if self.get_value(oid) != "": return self.get_value(oid) @property def uptime(self): - """ - returns an integer representing the number of seconds of uptime - """ - return int(self.get_value('1.3.6.1.2.1.1.3.0')) // 100 + """returns an integer representing the number of seconds of uptime""" + return int(self.get_value("1.3.6.1.2.1.1.3.0")) // 100 @property def uptime_tuple(self): - """ - returns (days, hours, minutes) - """ + """returns (days, hours, minutes)""" td = timedelta(seconds=self.uptime) return td.days, td.seconds // 3600, (td.seconds // 60) % 60 @property def interfaces_number(self): - """ - Returns the number of the network interfaces - """ - return int(self.get_value('1.3.6.1.2.1.2.1.0')) + """Returns the number of the network interfaces""" + return int(self.get_value("1.3.6.1.2.1.2.1.0")) _interfaces = None def get_interfaces(self): - """ - returns the list of all the interfaces of the device - """ + """returns the list of all the interfaces of the device""" if self._interfaces is None: interfaces = [] - value_to_get = '1.3.6.1.2.1.2.2.1.2.' + value_to_get = "1.3.6.1.2.1.2.2.1.2." for i in self._value_to_retrieve(): value_to_get1 = value_to_get + str(i) @@ -137,21 +112,19 @@ def get_interfaces(self): @property def interfaces_mtu(self): - """ - Returns an ordereed dict with the interface and its MTU - """ + """Returns an ordereed dict with the interface and its MTU""" if self._interfaces_mtu is None: results = [] - starting = '1.3.6.1.2.1.2.2.1.2.' + starting = "1.3.6.1.2.1.2.2.1.2." tmp = list(starting) tmp[18] = str(4) - to = ''.join(tmp) + to = "".join(tmp) for i in self._value_to_retrieve(): result = self._dict( { - 'name': self.get_value(starting + str(i)), - 'mtu': int(self.get_value(to + str(i))), + "name": self.get_value(starting + str(i)), + "mtu": int(self.get_value(to + str(i))), } ) results.append(result) @@ -164,30 +137,28 @@ def interfaces_mtu(self): @property def interfaces_state(self): - """ - Returns an ordereed dict with the interfaces and their state (up, down) - """ + """Returns an ordereed dict with the interfaces and their state (up, down)""" if self._interfaces_state is None: results = [] - starting = '1.3.6.1.2.1.2.2.1.2.' - operative = '1.3.6.1.2.1.2.2.1.8.' + starting = "1.3.6.1.2.1.2.2.1.2." + operative = "1.3.6.1.2.1.2.2.1.8." tmp = list(starting) tmp[18] = str(4) for i in self._value_to_retrieve(): - if self.get_value(starting + str(i)) != '': + if self.get_value(starting + str(i)) != "": if int(self.get_value(operative + str(i))) == 1: result = self._dict( - {'name': self.get_value(starting + str(i)), 'state': 'up'} + {"name": self.get_value(starting + str(i)), "state": "up"} ) else: result = self._dict( { - 'name': self.get_value(starting + str(i)), - 'state': 'down', + "name": self.get_value(starting + str(i)), + "state": "down", } ) - elif self.get_value(starting + str(i)) == '': - result = self._dict({'name': '', 'state': ''}) + elif self.get_value(starting + str(i)) == "": + result = self._dict({"name": "", "state": ""}) # append result to list results.append(result) @@ -199,19 +170,17 @@ def interfaces_state(self): @property def interfaces_speed(self): - """ - Returns an ordered dict with the interface and ist speed in bps - """ + """Returns an ordered dict with the interface and ist speed in bps""" if self._interfaces_speed is None: results = [] - starting = '1.3.6.1.2.1.2.2.1.2.' - starting_speed = '1.3.6.1.2.1.2.2.1.5.' + starting = "1.3.6.1.2.1.2.2.1.2." + starting_speed = "1.3.6.1.2.1.2.2.1.5." for i in self._value_to_retrieve(): result = self._dict( { - 'name': self.get_value(starting + str(i)), - 'speed': int(self.get_value(starting_speed + str(i))), + "name": self.get_value(starting + str(i)), + "speed": int(self.get_value(starting_speed + str(i))), } ) results.append(result) @@ -224,21 +193,19 @@ def interfaces_speed(self): @property def interfaces_bytes(self): - """ - Returns an ordereed dict with the interface and its tx and rx octets (1 octet = 1 byte = 8 bits) - """ + """Returns an ordereed dict with the interface and its tx and rx octets (1 octet = 1 byte = 8 bits)""" if self._interfaces_bytes is None: results = [] - starting = '1.3.6.1.2.1.2.2.1.2.' - starting_rx = '1.3.6.1.2.1.2.2.1.10.' - starting_tx = '1.3.6.1.2.1.2.2.1.16.' + starting = "1.3.6.1.2.1.2.2.1.2." + starting_rx = "1.3.6.1.2.1.2.2.1.10." + starting_tx = "1.3.6.1.2.1.2.2.1.16." for i in self._value_to_retrieve(): result = self._dict( { - 'name': self.get_value(starting + str(i)), - 'tx': int(self.get_value(starting_tx + str(i))), - 'rx': int(self.get_value(starting_rx + str(i))), + "name": self.get_value(starting + str(i)), + "tx": int(self.get_value(starting_tx + str(i))), + "rx": int(self.get_value(starting_rx + str(i))), } ) results.append(result) @@ -250,13 +217,11 @@ def interfaces_bytes(self): @property def interfaces_MAC(self): - """ - Returns an ordered dict with the hardware address of every interface - """ + """Returns an ordered dict with the hardware address of every interface""" if self._interfaces_MAC is None: results = [] - starting = '1.3.6.1.2.1.2.2.1.2.' - starting_mac = '1.3.6.1.2.1.2.2.1.6.' + starting = "1.3.6.1.2.1.2.2.1.2." + starting_mac = "1.3.6.1.2.1.2.2.1.6." for i in self._value_to_retrieve(): mac = binascii.b2a_hex( @@ -265,13 +230,13 @@ def interfaces_MAC(self): # now we are going to format mac as the canonical way as a MAC # address is intended by inserting ':' every two chars of mac # to obtain something as 00:11:22:22:33:44:55 - mac_transformed = ':'.join( - mac[slice(j, j + 2)] for j in range(0, 12, 2) if mac != '' + mac_transformed = ":".join( + mac[slice(j, j + 2)] for j in range(0, 12, 2) if mac != "" ) result = self._dict( { - 'name': self.get_value(starting + str(i)), - 'mac_address': mac_transformed, + "name": self.get_value(starting + str(i)), + "mac_address": mac_transformed, } ) results.append(result) @@ -284,20 +249,18 @@ def interfaces_MAC(self): @property def interfaces_type(self): - """ - Returns an ordered dict with the interface type (e.g Ethernet, loopback) - """ + """Returns an ordered dict with the interface type (e.g Ethernet, loopback)""" if self._interfaces_type is None: - types = {'6': 'ethernetCsmacd', '24': 'softwareLoopback'} + types = {"6": "ethernetCsmacd", "24": "softwareLoopback"} results = [] - starting = '1.3.6.1.2.1.2.2.1.2.' - types_oid = '1.3.6.1.2.1.2.2.1.3.' + starting = "1.3.6.1.2.1.2.2.1.2." + types_oid = "1.3.6.1.2.1.2.2.1.3." for i in self._value_to_retrieve(): result = self._dict( { - 'name': self.get_value(starting + str(i)), - 'type': types[self.get_value(types_oid + str(i))], + "name": self.get_value(starting + str(i)), + "type": types[self.get_value(types_oid + str(i))], } ) results.append(result) @@ -308,22 +271,20 @@ def interfaces_type(self): @property def interfaces_to_dict(self): - """ - Returns an ordered dict with all the information available about the interface - """ + """Returns an ordered dict with all the information available about the interface""" results = [] for i in range(0, len(self.get_interfaces())): - logger.info(f'===== {i} =====') + logger.info(f"===== {i} =====") result = self._dict( { - 'name': self.interfaces_MAC[i]['name'], - 'type': self.interfaces_type[i]['type'], - 'mac_address': self.interfaces_MAC[i]['mac_address'], - 'rx_bytes': int(self.interfaces_bytes[i]['rx']), - 'tx_bytes': int(self.interfaces_bytes[i]['tx']), - 'state': self.interfaces_state[i]['state'], - 'mtu': int(self.interfaces_mtu[i]['mtu']), - 'speed': int(self.interfaces_speed[i]['speed']), + "name": self.interfaces_MAC[i]["name"], + "type": self.interfaces_type[i]["type"], + "mac_address": self.interfaces_MAC[i]["mac_address"], + "rx_bytes": int(self.interfaces_bytes[i]["rx"]), + "tx_bytes": int(self.interfaces_bytes[i]["tx"]), + "state": self.interfaces_state[i]["state"], + "mtu": int(self.interfaces_mtu[i]["mtu"]), + "speed": int(self.interfaces_speed[i]["speed"]), } ) results.append(result) @@ -331,10 +292,8 @@ def interfaces_to_dict(self): @property def wireless_dbm(self): - """ - returns a list with the wireless signal (dbm) of the link/s - """ - res = self.next('1.3.6.1.4.1.14988.1.1.1.2.1.3.0') + """returns a list with the wireless signal (dbm) of the link/s""" + res = self.next("1.3.6.1.4.1.14988.1.1.1.2.1.3.0") dbm = [] for i in range(0, len(res[3])): dbm.append(int(res[3][i][0][1])) @@ -342,12 +301,10 @@ def wireless_dbm(self): @property def wireless_links(self): - ''' - Returns an ordered dict with all the infos about the wireless link/s - ''' + """Returns an ordered dict with all the infos about the wireless link/s""" final = [] - results = self.next('1.3.6.1.4.1.14988.1.1.1.2.1') - link_number = len(self.next('1.3.6.1.4.1.14988.1.1.1.2.1.3')[3]) + results = self.next("1.3.6.1.4.1.14988.1.1.1.2.1") + link_number = len(self.next("1.3.6.1.4.1.14988.1.1.1.2.1.3")[3]) separated_by_meaning = [] dbm = [] tx_bytes = [] @@ -372,13 +329,13 @@ def wireless_links(self): for i in range(0, link_number): result = self._dict( { - 'dbm': dbm[i], - 'tx_bytes': tx_bytes[i], - 'rx_bytes': rx_bytes[i], - 'tx_packets': tx_packets[i], - 'rx_packets': rx_packets[i], - 'tx_rate': tx_rate[i], - 'rx_rate': rx_rate[i], + "dbm": dbm[i], + "tx_bytes": tx_bytes[i], + "rx_bytes": rx_bytes[i], + "tx_packets": tx_packets[i], + "rx_packets": rx_packets[i], + "tx_rate": tx_rate[i], + "rx_rate": rx_rate[i], } ) final.append(result) @@ -386,37 +343,33 @@ def wireless_links(self): @property def RAM_total(self): - """ - Returns the total RAM of the device - """ - total = self.get_value('1.3.6.1.4.1.10002.1.1.1.1.1.0') + """Returns the total RAM of the device""" + total = self.get_value("1.3.6.1.4.1.10002.1.1.1.1.1.0") return int(total) @property def RAM_free(self): - """ - Returns the free RAM of the device - """ - free = self.get_value('1.3.6.1.4.1.10002.1.1.1.1.2.0') + """Returns the free RAM of the device""" + free = self.get_value("1.3.6.1.4.1.10002.1.1.1.1.2.0") return int(free) def to_dict(self): return self._dict( { - 'name': self.name, - 'type': 'radio', - 'os': self.os[0], - 'os_version': self.os[1], - 'manufacturer': self.manufacturer, - 'model': self.model, - 'RAM_total': self.RAM_total, - 'RAM_free': self.RAM_free, - 'uptime': self.uptime, - 'uptime_tuple': self.uptime_tuple, - 'interfaces': self.interfaces_to_dict, - 'antennas': [], - 'wireless_dbm': self.wireless_dbm, - 'wireless_links': self.wireless_links, - 'routing_protocols': None, + "name": self.name, + "type": "radio", + "os": self.os[0], + "os_version": self.os[1], + "manufacturer": self.manufacturer, + "model": self.model, + "RAM_total": self.RAM_total, + "RAM_free": self.RAM_free, + "uptime": self.uptime, + "uptime_tuple": self.uptime_tuple, + "interfaces": self.interfaces_to_dict, + "antennas": [], + "wireless_dbm": self.wireless_dbm, + "wireless_links": self.wireless_links, + "routing_protocols": None, } ) diff --git a/netengine/backends/snmp/base.py b/netengine/backends/snmp/base.py index 64f21d7..16e8589 100644 --- a/netengine/backends/snmp/base.py +++ b/netengine/backends/snmp/base.py @@ -1,103 +1,129 @@ try: - from pysnmp.entity.rfc3413.oneliner import cmdgen + from pysnmp.hlapi.v3arch.asyncio import ( + CommunityData, + ContextData, + ObjectIdentity, + ObjectType, + SnmpEngine, + UdpTransportTarget, + get_cmd, + walk_cmd, + ) except ImportError: raise ImportError( 'pysnmp library is not installed, install it with "pip install pysnmp"' ) +import asyncio import logging from netengine.backends import BaseBackend from netengine.exceptions import NetEngineError -__all__ = ['SNMP'] +__all__ = ["SNMP"] logger = logging.getLogger(__name__) class SNMP(BaseBackend): - """ - SNMP base backend - """ + """SNMP base backend""" _oid_to_retrieve = None - def __init__(self, host, community='public', agent='my-agent', port=161): - """ - :host string: required + def __init__(self, host, community="public", agent="my-agent", port=161): + """:host string: required :community string: defaults to public :agent string: defaults to my-agent :port integer: defaults to 161 """ self.host = host - self.community = cmdgen.CommunityData(agent, community, 0) - self.transport = cmdgen.UdpTransportTarget((host, port)) + self.community = CommunityData(agent, community, mpModel=0) self.port = port def __str__(self): """prints a human readable object description""" - return f'' - - @property - def _command(self): - """ - alias to cmdgen.CommandGenerator() - """ - return cmdgen.CommandGenerator() + return f"" + + async def _command(self, command, oid): + transport = await UdpTransportTarget.create((self.host, self.port)) + return await command( + SnmpEngine(), + self.community, + transport, + ContextData(), + ObjectType(ObjectIdentity(oid)), + ) + + async def _walk(self, oid): + transport = await UdpTransportTarget.create((self.host, self.port)) + result = (None, 0, 0, []) + async for error_indication, error_status, error_index, var_binds in walk_cmd( + SnmpEngine(), + self.community, + transport, + ContextData(), + ObjectType(ObjectIdentity(oid)), + ): + result = (error_indication, error_status, error_index, result[3]) + if error_indication or error_status: + return result + result[3].append(var_binds) + return result def _oid(self, oid): - """ - returns valid oid value to be passed to getCmd() or nextCmd() - """ + """returns valid oid value to be passed to getCmd() or nextCmd()""" if type(oid) not in (str, tuple, list): - raise AttributeError('get accepts only strings, tuples or lists') + raise AttributeError("get accepts only strings, tuples or lists") # allow string representations of oids with commas , elif isinstance(oid, str): # ignore spaces - oid = oid.replace(' ', '').replace(',', '.') + oid = oid.replace(" ", "").replace(",", ".") # convert lists and tuples into strings else: # convert each list item to string oid = [str(element) for element in oid] - oid = '.'.join(oid) + oid = ".".join(oid) # ensure is string (could be unicode) return str(oid) def get(self, oid): - """ - alias to cmdgen.CommandGenerator().getCmd - :oid string|tuple|list: string, tuple or list representing the OID to get + """Execute an SNMP GET request. + + :oid string|tuple|list: string, tuple or list representing the OID + to get example of valid oid parameters: - * '1,3,6,1,2,1,1,5,0' - * '1, 3, 6, 1, 2, 1, 1, 5, 0' - * '1.3.6.1.2.1.1.5.0' - * [1, 3, 6, 1, 2, 1, 1, 5, 0] - * (1, 3, 6, 1, 2, 1, 1, 5, 0) + - '1,3,6,1,2,1,1,5,0' + - '1, 3, 6, 1, 2, 1, 1, 5, 0' + - '1.3.6.1.2.1.1.5.0' + - [1, 3, 6, 1, 2, 1, 1, 5, 0] + - (1, 3, 6, 1, 2, 1, 1, 5, 0) """ - logger.info(f'DEBUG: SNMP GET {self._oid(oid)}') - return self._command.getCmd(self.community, self.transport, self._oid(oid)) + logger.info(f"DEBUG: SNMP GET {self._oid(oid)}") + return asyncio.run(self._command(get_cmd, self._oid(oid))) def next(self, oid): - """ - alias to cmdgen.CommandGenerator().nextCmd - :oid string|tuple|list: string, tuple or list representing the OID to get + """Execute an SNMP walk request. + + :oid string|tuple|list: string, tuple or list representing the OID + to get example of valid oid parameters: - * '1,3,6,1,2,1,1,5,0' - * '1, 3, 6, 1, 2, 1, 1, 5, 0' - * '1.3.6.1.2.1.1.5.0' - * [1, 3, 6, 1, 2, 1, 1, 5, 0] - * (1, 3, 6, 1, 2, 1, 1, 5, 0) + - '1,3,6,1,2,1,1,5,0' + - '1, 3, 6, 1, 2, 1, 1, 5, 0' + - '1.3.6.1.2.1.1.5.0' + - [1, 3, 6, 1, 2, 1, 1, 5, 0] + - (1, 3, 6, 1, 2, 1, 1, 5, 0) """ - logger.info(f'DEBUG: SNMP NEXT {self._oid(oid)}') - return self._command.nextCmd(self.community, self.transport, self._oid(oid)) + logger.info(f"DEBUG: SNMP NEXT {self._oid(oid)}") + return asyncio.run(self._walk(self._oid(oid))) def get_value(self, oid): - """ - returns value of oid, or raises NetEngineError Exception is anything wrong - :oid string|tuple|list: string, tuple or list representing the OID to get + """Return the OID value or raise NetEngineError. + + :oid string|tuple|list: string, tuple or list representing the OID + to get """ result = self.get(oid) try: @@ -106,14 +132,13 @@ def get_value(self, oid): raise NetEngineError(str(result[0])) def _value_to_retrieve(self): - """ - return the final SNMP indexes for the interfaces to be used in the other methods and properties - """ + """return the final SNMP indexes for the interfaces to be used in the other methods and properties""" value_to_retr = [] if self._oid_to_retrieve is None: raise NetEngineError( - 'Please fix properly the _oid_to_retrieve string in OpenWRT or AirOS SNMP backend' + "Please fix properly the _oid_to_retrieve string in OpenWRT " + "or AirOS SNMP backend" ) indexes = self.next(self._oid_to_retrieve)[3] diff --git a/netengine/backends/snmp/openwrt.py b/netengine/backends/snmp/openwrt.py index 2e21adb..afc04ae 100644 --- a/netengine/backends/snmp/openwrt.py +++ b/netengine/backends/snmp/openwrt.py @@ -1,8 +1,6 @@ -""" -NetEngine SNMP OpenWRT backend -""" +"""NetEngine SNMP OpenWRT backend""" -__all__ = ['OpenWRT'] +__all__ = ["OpenWRT"] import binascii @@ -15,59 +13,46 @@ class OpenWRT(SNMP): - """ - OpenWRT SNMP backend - """ + """OpenWRT SNMP backend""" - _oid_to_retrieve = '1.3.6.1.2.1.2.2.1.1' + _oid_to_retrieve = "1.3.6.1.2.1.2.2.1.1" _interface_dict = {} def __str__(self): """print a human readable object description""" - return f'' + return f"" def validate(self): - """ - raises NetEngineError exception if anything is wrong with the connection - for example: wrong host, invalid community - """ + """Raise NetEngineError when the connection is invalid.""" # this triggers a connection which # will raise an exception if anything is wrong return self.name @property def os(self): - """ - returns (os_name, os_version) - """ - os_name = 'OpenWRT' - os_version = self.get_value('1.3.6.1.2.1.1.1.0').split('#')[0].strip() + """returns (os_name, os_version)""" + os_name = "OpenWRT" + os_version = self.get_value("1.3.6.1.2.1.1.1.0").split("#")[0].strip() return os_name, os_version @property def manufacturer(self): # TODO: this is dangerous, it might not work in all cases - return self.get_manufacturer(self.interfaces_MAC[1]['mac_address']) + return self.get_manufacturer(self.interfaces_MAC[1]["mac_address"]) @property def name(self): - """ - returns a string containing the device name - """ - return self.get_value('1.3.6.1.2.1.1.5.0') + """returns a string containing the device name""" + return self.get_value("1.3.6.1.2.1.1.5.0") @property def uptime(self): - """ - returns an integer representing the number of seconds of uptime - """ - return int(self.get_value('1.3.6.1.2.1.1.3.0')) // 100 + """returns an integer representing the number of seconds of uptime""" + return int(self.get_value("1.3.6.1.2.1.1.3.0")) // 100 @property def uptime_tuple(self): - """ - returns (days, hours, minutes) - """ + """returns (days, hours, minutes)""" td = timedelta(seconds=self.uptime) return td.days, td.seconds // 3600, (td.seconds // 60) % 60 @@ -75,12 +60,10 @@ def uptime_tuple(self): _interfaces = None def get_interfaces(self): - """ - returns the list of all the interfaces of the device - """ + """returns the list of all the interfaces of the device""" if self._interfaces is None: interfaces = [] - value_to_get = '1.3.6.1.2.1.2.2.1.2.' + value_to_get = "1.3.6.1.2.1.2.2.1.2." for i in self._value_to_retrieve(): value_to_get1 = value_to_get + str(i) @@ -96,30 +79,28 @@ def get_interfaces(self): @property def interfaces_MAC(self): - """ - Returns an ordered dict with the hardware address of every interface - """ + """Returns an ordered dict with the hardware address of every interface""" if self._interfaces_MAC is None: results = [] mac1 = [] - mac = self.next('1.3.6.1.2.1.2.2.1.6.')[3] + mac = self.next("1.3.6.1.2.1.2.2.1.6.")[3] for i in range(1, len(mac) + 1): - mac1.append(self.get_value('1.3.6.1.2.1.2.2.1.6.' + str(i))) + mac1.append(self.get_value("1.3.6.1.2.1.2.2.1.6." + str(i))) mac_trans = [] for i in mac1: mac_string = binascii.b2a_hex(i.encode()).decode() mac_trans.append( - ':'.join( + ":".join( [ mac_string[slice(i, i + 2)] for i in range(0, 12, 2) - if i != '' + if i != "" ] ) ) for i in range(0, len(self.get_interfaces())): result = self._dict( - {'name': self.get_interfaces()[i], 'mac_address': mac_trans[i]} + {"name": self.get_interfaces()[i], "mac_address": mac_trans[i]} ) results.append(result) @@ -131,21 +112,19 @@ def interfaces_MAC(self): @property def interfaces_mtu(self): - """ - Returns an ordereed dict with the interface and its MTU - """ + """Returns an ordereed dict with the interface and its MTU""" if self._interfaces_mtu is None: results = [] - starting = '1.3.6.1.2.1.2.2.1.2.' + starting = "1.3.6.1.2.1.2.2.1.2." tmp = list(starting) tmp[18] = str(4) - to = ''.join(tmp) + to = "".join(tmp) for i in self._value_to_retrieve(): result = self._dict( { - 'name': self.get_value(starting + str(i)), - 'mtu': int(self.get_value(to + str(i))), + "name": self.get_value(starting + str(i)), + "mtu": int(self.get_value(to + str(i))), } ) results.append(result) @@ -158,13 +137,11 @@ def interfaces_mtu(self): @property def interfaces_speed(self): - """ - Returns an ordered dict with the interface and ist speed in bps - """ + """Returns an ordered dict with the interface and ist speed in bps""" if self._interfaces_speed is None: results = [] - starting = '1.3.6.1.2.1.2.2.1.2.' - starting_speed = '1.3.6.1.2.1.2.2.1.5.' + starting = "1.3.6.1.2.1.2.2.1.2." + starting_speed = "1.3.6.1.2.1.2.2.1.5." STOP_AFTER_FAILS = 3 @@ -181,7 +158,7 @@ def interfaces_speed(self): name = self.get_value(starting + str(i)) # if nothing found - if name == '': + if name == "": # increment fail counter consecutive_fails += 1 # increment i @@ -195,7 +172,7 @@ def interfaces_speed(self): # get speed and convert to int speed = int(self.get_value(starting_speed + str(i))) - result = self._dict({'name': name, 'speed': speed}) + result = self._dict({"name": name, "speed": speed}) results.append(result) # increment i @@ -209,32 +186,30 @@ def interfaces_speed(self): @property def interfaces_state(self): - """ - Returns an ordereed dict with the interfaces and their state (up, down) - """ + """Returns an ordereed dict with the interfaces and their state (up, down)""" if self._interfaces_state is None: results = [] - starting = '1.3.6.1.2.1.2.2.1.2.' - operative = '1.3.6.1.2.1.2.2.1.8.' + starting = "1.3.6.1.2.1.2.2.1.2." + operative = "1.3.6.1.2.1.2.2.1.8." tmp = list(starting) tmp[18] = str(4) for i in self._value_to_retrieve(): - if self.get_value(starting + str(i)) != '': + if self.get_value(starting + str(i)) != "": if int(self.get_value(operative + str(i))) == 1: result = self._dict( - {'name': self.get_value(starting + str(i)), 'state': 'up'} + {"name": self.get_value(starting + str(i)), "state": "up"} ) results.append(result) else: result = self._dict( { - 'name': self.get_value(starting + str(i)), - 'state': 'down', + "name": self.get_value(starting + str(i)), + "state": "down", } ) results.append(result) - elif self.get_value(starting + str(i)) == '': - result = self._dict({'name': '', 'state': ''}) + elif self.get_value(starting + str(i)) == "": + result = self._dict({"name": "", "state": ""}) results.append(result) self._interfaces_state = results @@ -245,21 +220,19 @@ def interfaces_state(self): @property def interfaces_bytes(self): - """ - Returns an ordereed dict with the interface and its tx and rx octets (1 octet = 1 byte = 8 bits) - """ + """Returns an ordereed dict with the interface and its tx and rx octets (1 octet = 1 byte = 8 bits)""" if self._interfaces_bytes is None: results = [] - starting = '1.3.6.1.2.1.2.2.1.2.' - starting_rx = '1.3.6.1.2.1.2.2.1.10.' - starting_tx = '1.3.6.1.2.1.2.2.1.16.' + starting = "1.3.6.1.2.1.2.2.1.2." + starting_rx = "1.3.6.1.2.1.2.2.1.10." + starting_tx = "1.3.6.1.2.1.2.2.1.16." for i in self._value_to_retrieve(): result = self._dict( { - 'name': self.get_value(starting + str(i)), - 'tx': int(self.get_value(starting_tx + str(i))), - 'rx': int(self.get_value(starting_rx + str(i))), + "name": self.get_value(starting + str(i)), + "tx": int(self.get_value(starting_tx + str(i))), + "rx": int(self.get_value(starting_rx + str(i))), } ) results.append(result) @@ -272,23 +245,21 @@ def interfaces_bytes(self): @property def interfaces_type(self): - """ - Returns an ordered dict with the interface type (e.g Ethernet, loopback) - """ + """Returns an ordered dict with the interface type (e.g Ethernet, loopback)""" if self._interfaces_type is None: types = { - '6': 'ethernetCsmacd', - '24': 'softwareLoopback', - '131': 'tunnel', + "6": "ethernetCsmacd", + "24": "softwareLoopback", + "131": "tunnel", } results = [] - starting = '1.3.6.1.2.1.2.2.1.2.' - types_oid = '1.3.6.1.2.1.2.2.1.3.' + starting = "1.3.6.1.2.1.2.2.1.2." + types_oid = "1.3.6.1.2.1.2.2.1.3." for i in self._value_to_retrieve(): result = self._dict( { - 'name': self.get_value(starting + str(i)), - 'type': types[self.get_value(types_oid + str(i))], + "name": self.get_value(starting + str(i)), + "type": types[self.get_value(types_oid + str(i))], } ) results.append(result) @@ -300,30 +271,28 @@ def interfaces_type(self): @property def interface_addr_and_mask(self): - """ - TODO: this method needs to be simplified and explained - """ + """TODO: this method needs to be simplified and explained""" if self._interface_addr_and_mask is None: interface_name = self.get_interfaces() for i in range(0, len(interface_name)): self._interface_dict[self._value_to_retrieve()[i]] = interface_name[i] - interface_ip_address = self.next('1.3.6.1.2.1.4.20.1.1')[3] - interface_index = self.next('1.3.6.1.2.1.4.20.1.2')[3] - interface_netmask = self.next('1.3.6.1.2.1.4.20.1.3')[3] + interface_ip_address = self.next("1.3.6.1.2.1.4.20.1.1")[3] + interface_index = self.next("1.3.6.1.2.1.4.20.1.2")[3] + interface_netmask = self.next("1.3.6.1.2.1.4.20.1.3")[3] results = {} for i in range(0, len(interface_ip_address)): a = interface_ip_address[i][0][1].asNumbers() - ip_address = '.'.join(str(a[i]) for i in range(0, len(a))) + ip_address = ".".join(str(a[i]) for i in range(0, len(a))) b = interface_netmask[i][0][1].asNumbers() - netmask = '.'.join(str(b[i]) for i in range(0, len(b))) + netmask = ".".join(str(b[i]) for i in range(0, len(b))) name = self._interface_dict[int(interface_index[i][0][1])] - results[name] = {'address': ip_address, 'netmask': netmask} + results[name] = {"address": ip_address, "netmask": netmask} self._interface_addr_and_mask = results @@ -331,52 +300,50 @@ def interface_addr_and_mask(self): @property def interfaces_to_dict(self): - """ - Returns an ordered dict with all the information available about the interface - """ + """Returns an ordered dict with all the information available about the interface""" results = [] for i in range(0, len(self.get_interfaces())): - logger.info(f'====== {i} ======') - - logger.info('... name ...') - name = self.interfaces_MAC[i]['name'] - logger.info('... if_type ...') - if_type = self.interfaces_type[i]['type'] - logger.info('... mac_address ...') - mac_address = self.interfaces_MAC[i]['mac_address'] - logger.info('... rx_bytes ...') - rx_bytes = int(self.interfaces_bytes[i]['rx']) - logger.info('... tx_bytes ...') - tx_bytes = int(self.interfaces_bytes[i]['tx']) - logger.info('... state ...') - state = self.interfaces_state[i]['state'] - logger.info('... mtu ...') - mtu = int(self.interfaces_mtu[i]['mtu']) - logger.info('... speed ...') - speed = int(self.interfaces_speed[i]['speed']) - logger.info('... ip address & subnet ...') + logger.info(f"====== {i} ======") + + logger.info("... name ...") + name = self.interfaces_MAC[i]["name"] + logger.info("... if_type ...") + if_type = self.interfaces_type[i]["type"] + logger.info("... mac_address ...") + mac_address = self.interfaces_MAC[i]["mac_address"] + logger.info("... rx_bytes ...") + rx_bytes = int(self.interfaces_bytes[i]["rx"]) + logger.info("... tx_bytes ...") + tx_bytes = int(self.interfaces_bytes[i]["tx"]) + logger.info("... state ...") + state = self.interfaces_state[i]["state"] + logger.info("... mtu ...") + mtu = int(self.interfaces_mtu[i]["mtu"]) + logger.info("... speed ...") + speed = int(self.interfaces_speed[i]["speed"]) + logger.info("... ip address & subnet ...") ip_and_netmask = self.interface_addr_and_mask if name in list(ip_and_netmask.keys()): - ip_address = ip_and_netmask[name]['address'] - netmask = ip_and_netmask[name]['netmask'] + ip_address = ip_and_netmask[name]["address"] + netmask = ip_and_netmask[name]["netmask"] else: ip_address = None netmask = None result = self._dict( { - 'name': name, - 'type': if_type, - 'mac_address': mac_address, - 'ip_address': ip_address, - 'netmask': netmask, - 'rx_bytes': rx_bytes, - 'tx_bytes': tx_bytes, - 'state': state, - 'mtu': mtu, - 'speed': speed, + "name": name, + "type": if_type, + "mac_address": mac_address, + "ip_address": ip_address, + "netmask": netmask, + "rx_bytes": rx_bytes, + "tx_bytes": tx_bytes, + "state": state, + "mtu": mtu, + "speed": speed, } ) results.append(result) @@ -384,25 +351,23 @@ def interfaces_to_dict(self): @property def RAM_total(self): - """ - returns the total RAM of the device - """ - return int(self.get_value('1.3.6.1.2.1.25.2.3.1.5.1')) + """returns the total RAM of the device""" + return int(self.get_value("1.3.6.1.2.1.25.2.3.1.5.1")) def to_dict(self): return self._dict( { - 'name': self.name, - 'type': 'radio', - 'os': self.os[0], - 'os_version': self.os[1], - 'manufacturer': self.manufacturer, - 'model': None, - 'RAM_total': self.RAM_total, - 'uptime': self.uptime, - 'uptime_tuple': self.uptime_tuple, - 'interfaces': self.get_interfaces(), - 'antennas': [], - 'routing_protocols': None, + "name": self.name, + "type": "radio", + "os": self.os[0], + "os_version": self.os[1], + "manufacturer": self.manufacturer, + "model": None, + "RAM_total": self.RAM_total, + "uptime": self.uptime, + "uptime_tuple": self.uptime_tuple, + "interfaces": self.get_interfaces(), + "antennas": [], + "routing_protocols": None, } ) diff --git a/netengine/exceptions.py b/netengine/exceptions.py index 0e1d3a5..399a275 100644 --- a/netengine/exceptions.py +++ b/netengine/exceptions.py @@ -1,6 +1,4 @@ -""" -netengine exception classes. -""" +"""netengine exception classes.""" class NetEngineError(Exception): diff --git a/requirements-test.txt b/requirements-test.txt index fed2167..6bf8241 100644 --- a/requirements-test.txt +++ b/requirements-test.txt @@ -1,5 +1,4 @@ -nose2~=0.10.0 -coverage~=5.5 -sphinx~=4.0.2 -openwisp-utils[qa]~=0.7.4 -pylinkvalidator~=0.3.0 +nose2[coverage_plugin]>=0.16.0 +coveralls +sphinx +openwisp-utils[qa] @ https://github.com/openwisp/openwisp-utils/archive/refs/heads/1.3.tar.gz diff --git a/requirements.txt b/requirements.txt index 81d4504..07c80ee 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,2 @@ -netaddr~=0.8.0 -pysnmp~=4.4.12 +netaddr~=1.3.0 +pysnmp~=7.1.28 diff --git a/run-qa-checks b/run-qa-checks index 329fdbb..5e3b20e 100755 --- a/run-qa-checks +++ b/run-qa-checks @@ -2,19 +2,7 @@ set -e +rm -rf docs/_build openwisp-qa-check --skip-checkmigrations - -# test sphinx docs mkdir -p docs/source/_static -make -C docs html - -PYTHON_VERSION=$(python -c 'import sys; print(".".join(map(str, sys.version_info[:2])))') -if [[ $PYTHON_VERSION != 3.6 ]]; then - # check for broken links - # remove condition when 3.6 is dropped - python -m http.server 8001 -d docs/build/html/ &>/dev/null & - pid=$! - sleep 4 - pylinkvalidate.py http://localhost:8001/ - kill "${pid}" 2> /dev/null || true -fi +sphinx-build -W -b html docs/source docs/_build diff --git a/runtests.py b/runtests.py index 6493dbb..dd88df2 100755 --- a/runtests.py +++ b/runtests.py @@ -11,8 +11,11 @@ """ raise ImportError(message) -if __name__ == '__main__': +if __name__ == "__main__": file_path = os.path.abspath(__file__) - tests_path = os.path.join(os.path.abspath(os.path.dirname(file_path)), 'tests',) + tests_path = os.path.join( + os.path.abspath(os.path.dirname(file_path)), + "tests", + ) nose2.discover() result = nose2.main() diff --git a/setup.py b/setup.py index 8d69b17..cd67284 100755 --- a/setup.py +++ b/setup.py @@ -6,17 +6,15 @@ def get_install_requires(): - """ - parse requirements.txt, ignore links, exclude comments - """ + """parse requirements.txt, ignore links, exclude comments""" requirements = [] - for line in open('requirements.txt').readlines(): + for line in open("requirements.txt").readlines(): # skip to next iteration if comment or empty line if ( - line.startswith('#') - or line == '' - or line.startswith('http') - or line.startswith('git') + line.startswith("#") + or line == "" + or line.startswith("http") + or line.startswith("git") ): continue # add line to requirements @@ -25,25 +23,30 @@ def get_install_requires(): setup( - name='netengine', + name="netengine", version=get_version(), - description='Abstraction layer for extracting information from network devices.', - long_description=open('README.rst').read(), - author='OpenWISP and Ninux.org Contributors', - author_email='support@openwisp.io', - license='MIT', - url='https://github.com/openwisp/netengine', - packages=find_packages(exclude=['tests', 'tests.*', 'docs', 'docs.*']), + description="Abstraction layer for extracting information from network devices.", + long_description=open("README.rst").read(), + author="OpenWISP and Ninux.org Contributors", + author_email="support@openwisp.io", + license="MIT", + url="https://github.com/openwisp/netengine", + packages=find_packages(exclude=["tests", "tests.*", "docs", "docs.*"]), install_requires=get_install_requires(), + python_requires=">=3.10", zip_safe=False, classifiers=[ - 'Development Status :: 1 - Planning', - 'Environment :: Console', - 'Intended Audience :: Developers', - 'License :: OSI Approved :: MIT License', - 'Operating System :: OS Independent', - 'Programming Language :: Python', - 'Topic :: System :: Networking', + "Development Status :: 1 - Planning", + "Environment :: Console", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Operating System :: OS Independent", + "Programming Language :: Python", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Topic :: System :: Networking", ], - test_suite='nose2.collector.collector', + test_suite="nose2.collector.collector", ) diff --git a/test-settings.example.json b/test-settings.example.json index 9c7d366..3cfeb6e 100644 --- a/test-settings.example.json +++ b/test-settings.example.json @@ -1,17 +1,17 @@ { - "base-snmp": { - "host": "0.0.0.0", - "community": "public", - "port": 161 - }, - "airos-snmp": { - "host": "0.0.0.0", - "community": "public", - "port": 161 - }, - "openwrt-snmp": { - "host": "0.0.0.0", - "community": "public", - "port": 161 - } + "base-snmp": { + "host": "0.0.0.0", + "community": "public", + "port": 161 + }, + "airos-snmp": { + "host": "0.0.0.0", + "community": "public", + "port": 161 + }, + "openwrt-snmp": { + "host": "0.0.0.0", + "community": "public", + "port": 161 + } } diff --git a/tests/settings.py b/tests/settings.py index da8f734..b055e1d 100644 --- a/tests/settings.py +++ b/tests/settings.py @@ -1,6 +1,6 @@ import json import os -settings_file = os.getenv('TEST_SETTINGS_FILE', './test-settings.example.json') +settings_file = os.getenv("TEST_SETTINGS_FILE", "./test-settings.example.json") settings = json.loads(open(settings_file).read()) -settings['disable_mocks'] = os.getenv('DISABLE_MOCKS', '0') == '1' +settings["disable_mocks"] = os.getenv("DISABLE_MOCKS", "0") == "1" diff --git a/tests/static/test-openwrt-snmp-oid.json b/tests/static/test-openwrt-snmp-oid.json index 2db1140..3067cfd 100644 --- a/tests/static/test-openwrt-snmp-oid.json +++ b/tests/static/test-openwrt-snmp-oid.json @@ -1,49 +1,49 @@ { - "1.3.6.1.2.1.25.2.3.1.5.1": "115080", - "1.3.6.1.2.1.2.2.1.2.1": "lo", - "1.3.6.1.2.1.2.2.1.2.2": "Device 8086:100e", - "1.3.6.1.2.1.2.2.1.2.3": "Device 8086:100e", - "1.3.6.1.2.1.2.2.1.2.4": "Device 8086:100e", - "1.3.6.1.2.1.2.2.1.2.5": "br-lan", - "1.3.6.1.2.1.2.2.1.16.1": "719914", - "1.3.6.1.2.1.2.2.1.10.1": "719914", - "1.3.6.1.2.1.2.2.1.16.2": "806244", - "1.3.6.1.2.1.2.2.1.10.2": "758983", - "1.3.6.1.2.1.2.2.1.16.3": "3326302", - "1.3.6.1.2.1.2.2.1.10.3": "9723560", - "1.3.6.1.2.1.2.2.1.10.5": "647519", - "1.3.6.1.2.1.2.2.1.16.4": "0", - "1.3.6.1.2.1.2.2.1.10.4": "0", - "1.3.6.1.2.1.2.2.1.16.5": "805932", - "1.3.6.1.2.1.2.2.1.8.1": "1", - "1.3.6.1.2.1.2.2.1.8.2": "1", - "1.3.6.1.2.1.2.2.1.8.3": "1", - "1.3.6.1.2.1.2.2.1.8.4": "2", - "1.3.6.1.2.1.2.2.1.8.5": "1", - "1.3.6.1.2.1.2.2.1.4.1": "65536", - "1.3.6.1.2.1.2.2.1.4.2": "1500", - "1.3.6.1.2.1.2.2.1.4.3": "1500", - "1.3.6.1.2.1.2.2.1.4.4": "1500", - "1.3.6.1.2.1.2.2.1.4.5": "1500", - "1.3.6.1.2.1.2.2.1.5.1": "10000000", - "1.3.6.1.2.1.2.2.1.5.2": "1000000000", - "1.3.6.1.2.1.2.2.1.5.3": "1000000000", - "1.3.6.1.2.1.2.2.1.5.4": "1000000000", - "1.3.6.1.2.1.2.2.1.5.5": "0", - "1.3.6.1.2.1.2.2.1.2.6": "", - "1.3.6.1.2.1.2.2.1.2.7": "", - "1.3.6.1.2.1.2.2.1.2.8": "", - "1.3.6.1.2.1.2.2.1.3.1": "24", - "1.3.6.1.2.1.2.2.1.3.2": "6", - "1.3.6.1.2.1.2.2.1.3.3": "6", - "1.3.6.1.2.1.2.2.1.3.4": "6", - "1.3.6.1.2.1.2.2.1.3.5": "6", - "1.3.6.1.2.1.1.3.0": "1033939", - "1.3.6.1.2.1.1.5.0": "HeartOfGold", - "1.3.6.1.2.1.1.1.0": "Linux 08-00-27-0A-F7-6A 4.14.221 #0 SMP Mon Feb 15 15:22:37 2021 x86_64", - "1.3.6.1.2.1.2.2.1.6.1": "\b\u0000''�\u0010\u0014", - "1.3.6.1.2.1.2.2.1.6.2": "\b\u0000''�\u0010\u0000", - "1.3.6.1.2.1.2.2.1.6.3": "\b\u0000''�\u0010\u0014", - "1.3.6.1.2.1.2.2.1.6.4": "\b\u0000''�\u0010\u0000", - "1.3.6.1.2.1.2.2.1.6.5": "\b\u0000''�\u0010\u0015" + "1.3.6.1.2.1.25.2.3.1.5.1": "115080", + "1.3.6.1.2.1.2.2.1.2.1": "lo", + "1.3.6.1.2.1.2.2.1.2.2": "Device 8086:100e", + "1.3.6.1.2.1.2.2.1.2.3": "Device 8086:100e", + "1.3.6.1.2.1.2.2.1.2.4": "Device 8086:100e", + "1.3.6.1.2.1.2.2.1.2.5": "br-lan", + "1.3.6.1.2.1.2.2.1.16.1": "719914", + "1.3.6.1.2.1.2.2.1.10.1": "719914", + "1.3.6.1.2.1.2.2.1.16.2": "806244", + "1.3.6.1.2.1.2.2.1.10.2": "758983", + "1.3.6.1.2.1.2.2.1.16.3": "3326302", + "1.3.6.1.2.1.2.2.1.10.3": "9723560", + "1.3.6.1.2.1.2.2.1.10.5": "647519", + "1.3.6.1.2.1.2.2.1.16.4": "0", + "1.3.6.1.2.1.2.2.1.10.4": "0", + "1.3.6.1.2.1.2.2.1.16.5": "805932", + "1.3.6.1.2.1.2.2.1.8.1": "1", + "1.3.6.1.2.1.2.2.1.8.2": "1", + "1.3.6.1.2.1.2.2.1.8.3": "1", + "1.3.6.1.2.1.2.2.1.8.4": "2", + "1.3.6.1.2.1.2.2.1.8.5": "1", + "1.3.6.1.2.1.2.2.1.4.1": "65536", + "1.3.6.1.2.1.2.2.1.4.2": "1500", + "1.3.6.1.2.1.2.2.1.4.3": "1500", + "1.3.6.1.2.1.2.2.1.4.4": "1500", + "1.3.6.1.2.1.2.2.1.4.5": "1500", + "1.3.6.1.2.1.2.2.1.5.1": "10000000", + "1.3.6.1.2.1.2.2.1.5.2": "1000000000", + "1.3.6.1.2.1.2.2.1.5.3": "1000000000", + "1.3.6.1.2.1.2.2.1.5.4": "1000000000", + "1.3.6.1.2.1.2.2.1.5.5": "0", + "1.3.6.1.2.1.2.2.1.2.6": "", + "1.3.6.1.2.1.2.2.1.2.7": "", + "1.3.6.1.2.1.2.2.1.2.8": "", + "1.3.6.1.2.1.2.2.1.3.1": "24", + "1.3.6.1.2.1.2.2.1.3.2": "6", + "1.3.6.1.2.1.2.2.1.3.3": "6", + "1.3.6.1.2.1.2.2.1.3.4": "6", + "1.3.6.1.2.1.2.2.1.3.5": "6", + "1.3.6.1.2.1.1.3.0": "1033939", + "1.3.6.1.2.1.1.5.0": "HeartOfGold", + "1.3.6.1.2.1.1.1.0": "Linux 08-00-27-0A-F7-6A 4.14.221 #0 SMP Mon Feb 15 15:22:37 2021 x86_64", + "1.3.6.1.2.1.2.2.1.6.1": "\b\u0000''�\u0010\u0014", + "1.3.6.1.2.1.2.2.1.6.2": "\b\u0000''�\u0010\u0000", + "1.3.6.1.2.1.2.2.1.6.3": "\b\u0000''�\u0010\u0014", + "1.3.6.1.2.1.2.2.1.6.4": "\b\u0000''�\u0010\u0000", + "1.3.6.1.2.1.2.2.1.6.5": "\b\u0000''�\u0010\u0015" } diff --git a/tests/test_base.py b/tests/test_base.py index a3883db..f680131 100644 --- a/tests/test_base.py +++ b/tests/test_base.py @@ -5,7 +5,7 @@ from netengine import __version__, get_version from netengine.backends import BaseBackend -__all__ = ['TestBaseBackend'] +__all__ = ["TestBaseBackend"] class TestBaseBackend(unittest.TestCase): @@ -86,4 +86,4 @@ def test_base_backend(self): def test_get_manufacturer_unicode(self): device = BaseBackend() with self.assertRaises(AddrFormatError): - device.get_manufacturer('wrong MAC') + device.get_manufacturer("wrong MAC") diff --git a/tests/test_dummy.py b/tests/test_dummy.py index 7c5120c..1318561 100644 --- a/tests/test_dummy.py +++ b/tests/test_dummy.py @@ -3,15 +3,15 @@ from netengine.backends import Dummy -__all__ = ['TestDummyBackend'] +__all__ = ["TestDummyBackend"] class TestDummyBackend(unittest.TestCase): def setUp(self): - self.dummy = Dummy('10.40.0.1') + self.dummy = Dummy("10.40.0.1") def test_str(self): - self.assertIn('Dummy NetEngine', str(self.dummy)) + self.assertIn("Dummy NetEngine", str(self.dummy)) def test_validate(self): self.dummy.validate() @@ -20,8 +20,8 @@ def test_to_dict(self): self.assertTrue(isinstance(self.dummy.to_dict(), dict)) def test_get_manufacturer(self): - dummy_addr = self.dummy.get_interfaces()[1]['hardware_address'] - self.assertIn('Xensource, Inc.', str(self.dummy.get_manufacturer(dummy_addr))) + dummy_addr = self.dummy.get_interfaces()[1]["hardware_address"] + self.assertIn("Xensource, Inc.", str(self.dummy.get_manufacturer(dummy_addr))) def test_to_json(self): json_string = self.dummy.to_json() diff --git a/tests/test_snmp/test_airos.py b/tests/test_snmp/test_airos.py index 4ea3227..1059bd1 100644 --- a/tests/test_snmp/test_airos.py +++ b/tests/test_snmp/test_airos.py @@ -1,7 +1,6 @@ import unittest from unittest.mock import patch -from pysnmp.entity.rfc3413.oneliner import cmdgen from pysnmp.smi.error import NoSuchObjectError from netengine.backends.snmp import AirOS @@ -10,28 +9,26 @@ from ..settings import settings from ..utils import MockOutputMixin, SpyMock -__all__ = ['TestSNMPAirOS'] +__all__ = ["TestSNMPAirOS"] class TestSNMPAirOS(unittest.TestCase, MockOutputMixin): def setUp(self): - self.host = settings['airos-snmp']['host'] - self.community = settings['airos-snmp']['community'] - self.port = settings['airos-snmp'].get('port', 161) + self.host = settings["airos-snmp"]["host"] + self.community = settings["airos-snmp"]["community"] + self.port = settings["airos-snmp"].get("port", 161) self.device = AirOS(self.host, self.community, port=self.port) # mock calls being made to devices - self.oid_mock_data = self._load_mock_json('/static/test-airos-snmp.json') - self.nextcmd_patcher = SpyMock._patch( - target=cmdgen.CommandGenerator, - attribute='nextCmd', - wrap_obj=self.device._command, - return_value=[0, 0, 0, [[[0, 1]]] * 5], + self.oid_mock_data = self._load_mock_json("/static/test-airos-snmp.json") + self.nextcmd_patcher = patch( + "netengine.backends.snmp.base.walk_cmd", + side_effect=lambda *args: self._get_mocked_walkcmd( + [0, 0, 0, [[[0, 1]]] * 5] + ), ) - self.getcmd_patcher = SpyMock._patch( - target=cmdgen.CommandGenerator, - attribute='getCmd', - wrap_obj=self.device._command, + self.getcmd_patcher = patch( + "netengine.backends.snmp.base.get_cmd", side_effect=lambda *args: self._get_mocked_getcmd( data=self.oid_mock_data, input=args ), @@ -40,12 +37,15 @@ def setUp(self): def test_get_value_error(self): self.getcmd_patcher.stop() - with self.assertRaises(NoSuchObjectError): - self.device.get_value('.') + with patch( + "netengine.backends.snmp.base.get_cmd", side_effect=NoSuchObjectError + ): + with self.assertRaises(NoSuchObjectError): + self.device.get_value(".") def test_validate_negative_result(self): self.getcmd_patcher.stop() - wrong = AirOS('10.40.0.254', 'wrong', 'wrong') + wrong = AirOS("10.40.0.254", "wrong", "wrong") self.assertRaises(NetEngineError, wrong.validate) def test_validate_positive_result(self): @@ -56,8 +56,8 @@ def test_get(self): self.device.get({}) with self.assertRaises(AttributeError): self.device.get(object) - self.device.get('1,3,6,1,2,1,1,5,0') - self.device.get('1,3,6,1,2,1,1,5,0') + self.device.get("1,3,6,1,2,1,1,5,0") + self.device.get("1,3,6,1,2,1,1,5,0") self.device.get((1, 3, 6, 1, 2, 1, 1, 5, 0)) self.device.get([1, 3, 6, 1, 2, 1, 1, 5, 0]) @@ -120,8 +120,8 @@ def test_wireless_to_dict(self): with self.nextcmd_patcher as np: SpyMock._update_patch( np, - _mock_side_effect=lambda *args: self._get_mocked_wireless_links( - data=args + _mock_side_effect=lambda *args: self._get_mocked_walkcmd( + self._get_mocked_wireless_links(data=args) ), ) self.assertIsInstance(self.device.wireless_links, list) @@ -136,8 +136,8 @@ def test_to_dict(self): with self.nextcmd_patcher as np: SpyMock._update_patch( np, - _mock_side_effect=lambda *args: self._get_mocked_wireless_links( - data=args + _mock_side_effect=lambda *args: self._get_mocked_walkcmd( + self._get_mocked_wireless_links(data=args) ), ) self.assertTrue(isinstance(self.device.to_dict(), dict)) @@ -146,11 +146,11 @@ def test_manufacturer_to_dict(self): with self.nextcmd_patcher as np: SpyMock._update_patch( np, - _mock_side_effect=lambda *args: self._get_mocked_wireless_links( - data=args + _mock_side_effect=lambda *args: self._get_mocked_walkcmd( + self._get_mocked_wireless_links(data=args) ), ) - self.assertIsNotNone(self.device.to_dict()['manufacturer']) + self.assertIsNotNone(self.device.to_dict()["manufacturer"]) def test_manufacturer(self): with self.nextcmd_patcher: diff --git a/tests/test_snmp/test_base.py b/tests/test_snmp/test_base.py index 7c7a82c..c42e22a 100644 --- a/tests/test_snmp/test_base.py +++ b/tests/test_snmp/test_base.py @@ -1,67 +1,38 @@ import unittest +from unittest.mock import patch -from netengine.backends.snmp import SNMP +from netengine.backends.snmp import AirOS from netengine.exceptions import NetEngineError -from ..settings import settings - -__all__ = ['TestSNMP'] - class TestSNMP(unittest.TestCase): def setUp(self): - self.host = settings['base-snmp']['host'] - self.community = settings['base-snmp']['community'] - self.port = settings['base-snmp'].get('port', 161) - - def test_instantiation(self): - device = SNMP(self.host, self.community, self.port) - self.assertTrue(device.__netengine__) - self.assertIn('SNMP', str(device)) - - def test_not_implemented_methods(self): - device = SNMP(self.host, self.community) - - with self.assertRaises(NotImplementedError): - device.os - with self.assertRaises(NotImplementedError): - device.name - with self.assertRaises(NotImplementedError): - device.model - with self.assertRaises(NotImplementedError): - device.RAM_total - with self.assertRaises(NotImplementedError): - device.ethernet_standard - with self.assertRaises(NotImplementedError): - device.ethernet_duplex - with self.assertRaises(NotImplementedError): - device.wireless_channel_width - with self.assertRaises(NotImplementedError): - device.wireless_mode - with self.assertRaises(NotImplementedError): - device.wireless_channel - with self.assertRaises(NotImplementedError): - device.wireless_output_power - with self.assertRaises(NotImplementedError): - device.wireless_dbm - with self.assertRaises(NotImplementedError): - device.wireless_noise - - def test_raised_exception(self): - class WrongSNMPBackend(SNMP): - pass - - device = WrongSNMPBackend(self.host, self.community) - - with self.assertRaises(NetEngineError): - device._value_to_retrieve() - - # this time define the _oid_to_retrieve attribute - class RightSNMPBackend(SNMP): - _oid_to_retrieve = '' - - device = RightSNMPBackend(self.host, self.community) - - # now we expect a different kind of error - with self.assertRaises(IndexError): - device._value_to_retrieve() + self.device = AirOS("192.0.2.1") + + def test_get_value_error_response(self): + with patch.object( + self.device, + "get", + return_value=(Exception("request timed out"), 0, 0, ()), + ): + with self.assertRaisesRegex(NetEngineError, "request timed out"): + self.device.get_value("1.3.6.1.2.1.1.5.0") + + def test_walk_error_response(self): + with patch.object( + self.device, + "next", + return_value=(Exception("request timed out"), 0, 0, ()), + ): + self.assertEqual(self.device._value_to_retrieve(), []) + + def test_next_collects_walk_responses(self): + async def walk_response(*args): + yield None, 0, 0, ((0, 1),) + yield None, 0, 0, ((0, 2),) + + with patch("netengine.backends.snmp.base.walk_cmd", side_effect=walk_response): + self.assertEqual( + self.device.next("1.3.6.1.2.1.1.5.0"), + (None, 0, 0, [((0, 1),), ((0, 2),)]), + ) diff --git a/tests/test_snmp/test_openwrt.py b/tests/test_snmp/test_openwrt.py index f12588e..dde183d 100644 --- a/tests/test_snmp/test_openwrt.py +++ b/tests/test_snmp/test_openwrt.py @@ -1,35 +1,35 @@ import unittest from unittest.mock import patch -from pysnmp.entity.rfc3413.oneliner import cmdgen - from netengine.backends.snmp import OpenWRT from ..settings import settings from ..utils import MockOutputMixin, SpyMock -__all__ = ['TestSNMPOpenWRT'] +__all__ = ["TestSNMPOpenWRT"] class TestSNMPOpenWRT(unittest.TestCase, MockOutputMixin): def setUp(self): - self.host = settings['openwrt-snmp']['host'] - self.community = settings['openwrt-snmp']['community'] - self.port = settings['openwrt-snmp'].get('port', 161) - self.device = OpenWRT(host=self.host, community=self.community, port=self.port,) + self.host = settings["openwrt-snmp"]["host"] + self.community = settings["openwrt-snmp"]["community"] + self.port = settings["openwrt-snmp"].get("port", 161) + self.device = OpenWRT( + host=self.host, + community=self.community, + port=self.port, + ) # mock calls being made to devices - self.oid_mock_data = self._load_mock_json('/static/test-openwrt-snmp-oid.json') - self.nextcmd_patcher = SpyMock._patch( - target=cmdgen.CommandGenerator, - attribute='nextCmd', - wrap_obj=self.device._command, - return_value=[0, 0, 0, [[[0, 1]]] * 5], + self.oid_mock_data = self._load_mock_json("/static/test-openwrt-snmp-oid.json") + self.nextcmd_patcher = patch( + "netengine.backends.snmp.base.walk_cmd", + side_effect=lambda *args: self._get_mocked_walkcmd( + [0, 0, 0, [[[0, 1]]] * 5] + ), ) - self.getcmd_patcher = SpyMock._patch( - target=cmdgen.CommandGenerator, - attribute='getCmd', - wrap_obj=self.device._command, + self.getcmd_patcher = patch( + "netengine.backends.snmp.base.get_cmd", side_effect=lambda *args: self._get_mocked_getcmd( data=self.oid_mock_data, input=args ), @@ -81,12 +81,12 @@ def test_interfaces_state(self): def test_interfaces_to_dict(self): with self.nextcmd_patcher as p: - p.return_value = (0, 0, 0, []) + p._mock_side_effect = lambda *args: self._get_mocked_walkcmd((0, 0, 0, [])) self.assertIsInstance(self.device.interfaces_to_dict, list) def test_interface_addr_and_mask(self): with self.nextcmd_patcher as p: - p.return_value = (0, 0, 0, []) + p._mock_side_effect = lambda *args: self._get_mocked_walkcmd((0, 0, 0, [])) self.assertIsInstance(self.device.interface_addr_and_mask, dict) def test_RAM_total(self): @@ -94,17 +94,28 @@ def test_RAM_total(self): def test_to_dict(self): with self.nextcmd_patcher as p: - SpyMock._update_patch(p, _mock_return_value=[0, 0, 0, [[[0, 1]]] * 5]) + SpyMock._update_patch( + p, + _mock_side_effect=lambda *args: self._get_mocked_walkcmd( + [0, 0, 0, [[[0, 1]]] * 5] + ), + ) device_dict = self.device.to_dict() self.assertTrue(isinstance(device_dict, dict)) self.assertEqual( - len(device_dict['interfaces']), len(self.device.get_interfaces()), + len(device_dict["interfaces"]), + len(self.device.get_interfaces()), ) def test_manufacturer_to_dict(self): with self.nextcmd_patcher as p: - SpyMock._update_patch(p, _mock_return_value=[0, 0, 0, [[[0, 1]]] * 5]) - self.assertIsNotNone(self.device.to_dict()['manufacturer']) + SpyMock._update_patch( + p, + _mock_side_effect=lambda *args: self._get_mocked_walkcmd( + [0, 0, 0, [[[0, 1]]] * 5] + ), + ) + self.assertIsNotNone(self.device.to_dict()["manufacturer"]) def tearDown(self): patch.stopall() diff --git a/tests/utils.py b/tests/utils.py index 6c3b548..423a726 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -8,19 +8,26 @@ class SpyMock: @staticmethod def _patch(*args, **kwargs): - if not settings['disable_mocks']: + if not settings["disable_mocks"]: return mock.patch.object(*args, **kwargs) - wraps = getattr(kwargs['wrap_obj'], kwargs['attribute']) - return mock.patch.object(kwargs['target'], kwargs['attribute'], wraps=wraps) + wraps = getattr(kwargs["wrap_obj"], kwargs["attribute"]) + return mock.patch.object(kwargs["target"], kwargs["attribute"], wraps=wraps) @staticmethod def _update_patch(mock_obj, *args, **kwargs): - if settings['disable_mocks']: + if settings["disable_mocks"]: return mock_obj.__dict__.update(*args, **kwargs) class MockOutputMixin(object): + @staticmethod + def _get_oid(input): + var_bind = input[-1] + return var_bind.__dict__["_ObjectType__args"][0].__dict__[ + "_ObjectIdentity__args" + ][0] + @staticmethod def _load_mock_json(file): base_dir = os.path.dirname(os.path.abspath(__file__)) @@ -30,19 +37,24 @@ def _load_mock_json(file): @staticmethod def _get_mocked_getcmd(data, input): - oid = input[2] + oid = MockOutputMixin._get_oid(input) result = data[oid] - if type(result) == list: - result = '\n'.join(result[0:]) + if isinstance(result, list): + result = "\n".join(result[0:]) return [0, 0, 0, [[0, result]]] + @staticmethod + async def _get_mocked_walkcmd(result): + for row in result[3]: + yield result[0], result[1], result[2], row + @staticmethod def _get_mocked_wireless_links(data): - oid = data[2] + oid = MockOutputMixin._get_oid(data) return_data = { - '1.3.6.1.4.1.14988.1.1.1.2.1': [0, 0, 0, [[[0, 0], 0]] * 28], - '1.3.6.1.4.1.14988.1.1.1.2.1.3': [0, 0, 0, [0, 0]], - '1.3.6.1.4.1.14988.1.1.1.2.1.3.0': [None, 0, 0, []], - '1.3.6.1.2.1.1.9.1.1': [0, 0, 0, [[[0, 1]]] * 5], + "1.3.6.1.4.1.14988.1.1.1.2.1": [0, 0, 0, [[[0, 0], 0]] * 28], + "1.3.6.1.4.1.14988.1.1.1.2.1.3": [0, 0, 0, [0, 0]], + "1.3.6.1.4.1.14988.1.1.1.2.1.3.0": [None, 0, 0, []], + "1.3.6.1.2.1.1.9.1.1": [0, 0, 0, [[[0, 1]]] * 5], } return return_data[oid]