diff --git a/.coveragerc b/.coveragerc new file mode 100644 index 0000000..c3b3ba5 --- /dev/null +++ b/.coveragerc @@ -0,0 +1,6 @@ +[run] +omit = + /*/test* + /tests + /*/__init__.py + /setup.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..fae03d3 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,71 @@ +--- + +name: Netengine CI Build + +on: + push: + branches: + - master + pull_request: + branches: + - master + +jobs: + + build: + name: Python==${{ matrix.python-version }} + runs-on: ubuntu-20.04 + + strategy: + fail-fast: false + matrix: + python-version: + - 3.6 + - 3.7 + - 3.8 + - 3.9 + + steps: + - uses: actions/checkout@v2 + 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: Upgrade python system packages + run: pip install -U wheel setuptools + + - name: Install netengine + run: pip install -e . + + - name: Install test dependencies + run: pip install -U -r requirements-test.txt + + - name: Run QA Checks + run: ./run-qa-checks + + - name: Run tests + run: coverage run --source=netengine ./runtests.py + + - name: Upload Coverage + run: coveralls --service=github + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + COVERALLS_FLAG_NAME: python-${{ matrix.python-version }} + COVERALLS_PARALLEL: true + + 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 }} diff --git a/LICENSE b/LICENSE index bc9ff35..75ad7fa 100644 --- a/LICENSE +++ b/LICENSE @@ -20,4 +20,4 @@ FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN -THE SOFTWARE. \ No newline at end of file +THE SOFTWARE. diff --git a/docs/source/conf.py b/docs/source/conf.py index c7deaa0..757c429 100755 --- a/docs/source/conf.py +++ b/docs/source/conf.py @@ -12,18 +12,18 @@ # All configuration values have a default; values that are commented out # serve to show the default. -import sys import os +import sys # If extensions (or modules to document with autodoc) are in another directory, # add these directories to sys.path here. If the directory is relative to the # documentation root, use os.path.abspath to make it absolute, like shown here. -#sys.path.insert(0, os.path.abspath('.')) +# sys.path.insert(0, os.path.abspath('.')) # -- General configuration ------------------------------------------------ # If your documentation needs a minimal Sphinx version, state it here. -#needs_sphinx = '1.0' +# needs_sphinx = '1.0' # Add any Sphinx extension module names here, as strings. They can be # extensions coming with Sphinx (named 'sphinx.ext.*') or your custom @@ -37,7 +37,7 @@ source_suffix = '.rst' # The encoding of source files. -#source_encoding = 'utf-8-sig' +# source_encoding = 'utf-8-sig' # The master toctree document. master_doc = 'index' @@ -57,13 +57,13 @@ # The language for content autogenerated by Sphinx. Refer to documentation # for a list of supported languages. -#language = None +# language = None # There are two options for replacing |today|: either, you set today to some # non-false value, then it is used: -#today = '' +# today = '' # Else, today_fmt is used as the format for a strftime call. -#today_fmt = '%B %d, %Y' +# today_fmt = '%B %d, %Y' # List of patterns, relative to source directory, that match files and # directories to ignore when looking for source files. @@ -71,27 +71,27 @@ # The reST default role (used for this markup: `text`) to use for all # documents. -#default_role = None +# default_role = None # If true, '()' will be appended to :func: etc. cross-reference text. -#add_function_parentheses = True +# add_function_parentheses = True # If true, the current module name will be prepended to all description # unit titles (such as .. function::). -#add_module_names = True +# add_module_names = True # If true, sectionauthor and moduleauthor directives will be shown in the # output. They are ignored by default. -#show_authors = False +# show_authors = False # The name of the Pygments (syntax highlighting) style to use. pygments_style = 'sphinx' # A list of ignored prefixes for module index sorting. -#modindex_common_prefix = [] +# modindex_common_prefix = [] # If true, keep warnings as "system message" paragraphs in the built documents. -#keep_warnings = False +# keep_warnings = False # -- Options for HTML output ---------------------------------------------- @@ -103,26 +103,26 @@ # 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 # documentation. -#html_theme_options = {} +# html_theme_options = {} # Add any paths that contain custom themes here, relative to this directory. -#html_theme_path = [] +# html_theme_path = [] # The name for this set of Sphinx documents. If None, it defaults to # " v documentation". -#html_title = None +# html_title = None # A shorter title for the navigation bar. Default is the same as html_title. -#html_short_title = None +# html_short_title = None # The name of an image file (relative to this directory) to place at the top # of the sidebar. -#html_logo = None +# html_logo = None # The name of an image file (within the static path) to use as favicon of the # docs. This file should be a Windows icon file (.ico) being 16x16 or 32x32 # pixels large. -#html_favicon = None +# html_favicon = None # 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, @@ -132,48 +132,48 @@ # Add any extra paths that contain custom files (such as robots.txt or # .htaccess) here, relative to this directory. These files are copied # directly to the root of the documentation. -#html_extra_path = [] +# html_extra_path = [] # If not '', a 'Last updated on:' timestamp is inserted at every page bottom, # using the given strftime format. -#html_last_updated_fmt = '%b %d, %Y' +# html_last_updated_fmt = '%b %d, %Y' # If true, SmartyPants will be used to convert quotes and dashes to # typographically correct entities. -#html_use_smartypants = True +# html_use_smartypants = True # Custom sidebar templates, maps document names to template names. -#html_sidebars = {} +# html_sidebars = {} # Additional templates that should be rendered to pages, maps page names to # template names. -#html_additional_pages = {} +# html_additional_pages = {} # If false, no module index is generated. -#html_domain_indices = True +# html_domain_indices = True # If false, no index is generated. -#html_use_index = True +# html_use_index = True # If true, the index is split into individual pages for each letter. -#html_split_index = False +# html_split_index = False # If true, links to the reST sources are added to the pages. -#html_show_sourcelink = True +# html_show_sourcelink = True # If true, "Created using Sphinx" is shown in the HTML footer. Default is True. -#html_show_sphinx = True +# html_show_sphinx = True # If true, "(C) Copyright ..." is shown in the HTML footer. Default is True. -#html_show_copyright = True +# html_show_copyright = True # If true, an OpenSearch description file will be output, and all pages will # contain a tag referring to it. The value of this option must be the # base URL from which the finished HTML is served. -#html_use_opensearch = '' +# html_use_opensearch = '' # This is the file name suffix for HTML files (e.g. ".xhtml"). -#html_file_suffix = None +# html_file_suffix = None # Output file base name for HTML help builder. htmlhelp_basename = 'netenginedoc' @@ -182,43 +182,46 @@ # -- Options for LaTeX output --------------------------------------------- latex_elements = { -# The paper size ('letterpaper' or 'a4paper'). -#'papersize': 'letterpaper', - -# The font size ('10pt', '11pt' or '12pt'). -#'pointsize': '10pt', - -# Additional stuff for the LaTeX preamble. -#'preamble': '', + # The paper size ('letterpaper' or 'a4paper'). + #'papersize': 'letterpaper', + # The font size ('10pt', '11pt' or '12pt'). + #'pointsize': '10pt', + # Additional stuff for the LaTeX preamble. + #'preamble': '', } # Grouping the document tree into LaTeX files. List of tuples # (source start file, target name, title, # author, documentclass [howto, manual, or own class]). latex_documents = [ - ('index', 'netengine.tex', u'netengine Documentation', - u'Alessandro Bucciarelli, Federico Capoano', 'manual'), + ( + 'index', + 'netengine.tex', + u'netengine Documentation', + u'Alessandro Bucciarelli, Federico Capoano', + 'manual', + ), ] # The name of an image file (relative to this directory) to place at the top of # the title page. -#latex_logo = None +# latex_logo = None # For "manual" documents, if this is true, then toplevel headings are parts, # not chapters. -#latex_use_parts = False +# latex_use_parts = False # If true, show page references after internal links. -#latex_show_pagerefs = False +# latex_show_pagerefs = False # If true, show URL addresses after external links. -#latex_show_urls = False +# latex_show_urls = False # Documents to append as an appendix to all manuals. -#latex_appendices = [] +# latex_appendices = [] # If false, no module index is generated. -#latex_domain_indices = True +# latex_domain_indices = True # -- Options for manual page output --------------------------------------- @@ -226,12 +229,17 @@ # One entry per manual page. List of tuples # (source start file, name, description, authors, manual section). man_pages = [ - ('index', 'netengine', u'netengine Documentation', - [u'Alessandro Bucciarelli, Federico Capoano'], 1) + ( + 'index', + 'netengine', + u'netengine Documentation', + [u'Alessandro Bucciarelli, Federico Capoano'], + 1, + ) ] # If true, show URL addresses after external links. -#man_show_urls = False +# man_show_urls = False # -- Options for Texinfo output ------------------------------------------- @@ -240,19 +248,25 @@ # (source start file, target name, title, author, # 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', + u'netengine Documentation', + u'Alessandro Bucciarelli, Federico Capoano', + 'netengine', + 'One line description of project.', + 'Miscellaneous', + ), ] # Documents to append as an appendix to all manuals. -#texinfo_appendices = [] +# texinfo_appendices = [] # If false, no module index is generated. -#texinfo_domain_indices = True +# texinfo_domain_indices = True # How to display URL addresses: 'footnote', 'no', or 'inline'. -#texinfo_show_urls = 'footnote' +# texinfo_show_urls = 'footnote' # If true, do not generate a @detailmenu in the "Top" node's menu. -#texinfo_no_detailmenu = False +# texinfo_no_detailmenu = False diff --git a/docs/source/index.rst b/docs/source/index.rst index c1a8c0b..0c70a47 100755 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -69,5 +69,4 @@ Indices and tables ================== * :ref:`genindex` -* :ref:`modindex` * :ref:`search` diff --git a/docs/source/topics/usage.rst b/docs/source/topics/usage.rst index 6c97c54..b3a5baa 100755 --- a/docs/source/topics/usage.rst +++ b/docs/source/topics/usage.rst @@ -2,9 +2,11 @@ Usage ***** -The usage of Netengine module requires it to be installed properly as explained in :ref:`index_reference`. +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 + + source activate + 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: @@ -51,17 +53,17 @@ Then change the credentials accordingly, now run tests with:: See test coverage with:: - nosetests --with-coverage --cover-package=netengine + nose2 --with-coverage Run specific tests by specifying the relative path:: # base tests - nosetests tests.base + nose2 tests.base # snmp tests - nosetests tests.snmp + nose2 tests.snmp # snmp openwrt specific tests - nosetests tests.snmp.openwrt + nose2 tests.snmp.openwrt # run without mocks with a custom test file - DISABLE_MOCKS=1 TEST_SETTINGS_FILE='test-settings.json' nosetests tests.snmp + DISABLE_MOCKS=1 TEST_SETTINGS_FILE='test-settings.json' nose2 tests.snmp diff --git a/netengine/backends/__init__.py b/netengine/backends/__init__.py index 0992f50..41896b3 100644 --- a/netengine/backends/__init__.py +++ b/netengine/backends/__init__.py @@ -1,5 +1,4 @@ from .base import BaseBackend from .dummy import Dummy - -__all__ = ['BaseBackend', 'Dummy'] \ No newline at end of file +__all__ = ['BaseBackend', 'Dummy'] diff --git a/netengine/backends/base.py b/netengine/backends/base.py index da893c0..86509b0 100644 --- a/netengine/backends/base.py +++ b/netengine/backends/base.py @@ -1,32 +1,26 @@ import json +from collections import OrderedDict -from netengine.shortcuts import OrderedDict from netaddr import EUI, NotRegisteredError - -__all__ = [ - 'BaseBackend' -] +__all__ = ['BaseBackend'] class BaseBackend(object): """ 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 """ + """returns unicode string represantation""" return self.__str__() - def __unicode__(self): - """ unicode __str__() for python2.7 """ - return unicode(self.__str__()) - def validate(self): raise NotImplementedError('Not implemented') @@ -134,7 +128,7 @@ def get_interfaces(self): raise NotImplementedError('Not implemented') def get_manufacturer(self, mac_address): - """ returns the manufacturer of the network interface """ + """returns the manufacturer of the network interface""" if not mac_address: return '' try: diff --git a/netengine/backends/dummy.py b/netengine/backends/dummy.py index ea7aacd..2173234 100644 --- a/netengine/backends/dummy.py +++ b/netengine/backends/dummy.py @@ -5,8 +5,9 @@ class Dummy(BaseBackend): """ Dummy backend """ + def __init__(self, host, port=0): - """ dummy netengine backend for development or testing """ + """dummy netengine backend for development or testing""" self.host = host self.port = port @@ -18,120 +19,99 @@ def validate(self): pass def __str__(self): - """ print a human readable object description """ - return u"" % self.host + """print a human readable object description""" + return f'' def get_interfaces(self): return [ {}, { - "ipv6_address_link": "", - "hardware_address": u"00:16:3E:26:9D:13", - "rx_packets": u"147684", - "broadcast_address": "", - "rx_bytes": u"12956143", - "link_encap": u"Ethernet", - "metric": u"1", - "txqueuelen": u"1000", - "net_mask": "", - "ip_address": "", - "collisions": u"0", - "interface": u"eth0", - "tx_bytes": u"12523266", - "mtu": u"1500", - "tx_packets": u"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": u"Local", - "metric": "", - "txqueuelen": "", - "net_mask": "", - "ip_address": "", - "collisions": "", - "interface": u"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": [ - { - "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": "" - } - ] - }, - { - "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" - } - ] - }) + 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': [ + { + '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': ''}], + }, + { + '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'}], + } + ) diff --git a/netengine/backends/snmp/__init__.py b/netengine/backends/snmp/__init__.py index 23fd560..f78c332 100644 --- a/netengine/backends/snmp/__init__.py +++ b/netengine/backends/snmp/__init__.py @@ -1,11 +1,5 @@ +from .airos import AirOS from .base import SNMP from .openwrt import OpenWRT -from .airos import AirOS - - -__all__ = [ - 'SNMP', - 'OpenWRT', - 'AirOS' -] +__all__ = ['SNMP', 'OpenWRT', 'AirOS'] diff --git a/netengine/backends/snmp/airos.py b/netengine/backends/snmp/airos.py index 03445a3..24c7fd1 100644 --- a/netengine/backends/snmp/airos.py +++ b/netengine/backends/snmp/airos.py @@ -6,21 +6,25 @@ import binascii +import logging from datetime import timedelta -from netengine.backends.snmp import SNMP + +from .base import SNMP + +logger = logging.getLogger(__name__) class AirOS(SNMP): """ Ubiquiti AirOS SNMP backend """ - + _oid_to_retrieve = '1.3.6.1.2.1.1.9.1.1' - + def __str__(self): - """ print a human readable object description """ - return u"" % self.host - + """print a human readable object description""" + return f'' + def validate(self): """ raises NetEngineError exception if anything is wrong with the connection @@ -29,7 +33,7 @@ def validate(self): # this triggers a connection which # will raise an exception if anything is wrong return self.name - + @property def os(self): """ @@ -38,45 +42,45 @@ def os(self): 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') - + @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'] + 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 != '': 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'] + 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('.') if tmp is not None: length = len(tmp) i = 0 for piece in tmp: - if "v" in piece: + 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']) - + @property def ssid(self): """ @@ -86,32 +90,32 @@ def ssid(self): for oid in oids: 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 - + return int(self.get_value('1.3.6.1.2.1.1.3.0')) // 100 + @property def uptime_tuple(self): """ returns (days, hours, minutes) """ td = timedelta(seconds=self.uptime) - - return td.days, td.seconds//3600, (td.seconds//60)%60 - + + 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')) - + 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 @@ -119,18 +123,18 @@ def get_interfaces(self): if self._interfaces is None: interfaces = [] 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) + value_to_get1 = value_to_get + str(i) if value_to_get1: interfaces.append(self.get_value(value_to_get1)) - + self._interfaces = interfaces - + return self._interfaces - + _interfaces_mtu = None - + @property def interfaces_mtu(self): """ @@ -138,24 +142,26 @@ def interfaces_mtu(self): """ 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) - + 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))) - }) + result = self._dict( + { + 'name': self.get_value(starting + str(i)), + 'mtu': int(self.get_value(to + str(i))), + } + ) results.append(result) - + self._interfaces_mtu = results - + return self._interfaces_mtu - + _interfaces_state = None - + @property def interfaces_state(self): """ @@ -163,36 +169,34 @@ def interfaces_state(self): """ 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" - }) + result = self._dict( + {'name': self.get_value(starting + str(i)), 'state': 'up'} + ) else: - result = self._dict({ - "name" : self.get_value(starting + str(i)), - "state" : "down" - }) - elif self.get_value(starting + str(i)) == "" : - result = self._dict({ - "name" : "", - "state" : "" - }) + result = self._dict( + { + 'name': self.get_value(starting + str(i)), + 'state': 'down', + } + ) + elif self.get_value(starting + str(i)) == '': + result = self._dict({'name': '', 'state': ''}) # append result to list results.append(result) - + self._interfaces_state = results - + return self._interfaces_state - + _interfaces_speed = None - + @property def interfaces_speed(self): """ @@ -200,21 +204,24 @@ def interfaces_speed(self): """ 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))) - }) + result = self._dict( + { + 'name': self.get_value(starting + str(i)), + 'speed': int(self.get_value(starting_speed + str(i))), + } + ) results.append(result) - + self._interfaces_speed = results - + return self._interfaces_speed - + _interfaces_bytes = None + @property def interfaces_bytes(self): """ @@ -222,23 +229,25 @@ def interfaces_bytes(self): """ 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))), - }) + 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))), + } + ) results.append(result) self._interfaces_bytes = results - + return self._interfaces_bytes - + _interfaces_MAC = None - + @property def interfaces_MAC(self): """ @@ -246,49 +255,57 @@ def interfaces_MAC(self): """ 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(self.get_value(starting_mac + str(i))) + mac = binascii.b2a_hex( + self.get_value(starting_mac + str(i)).encode() + ).decode() # 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[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 - }) + 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, + } + ) results.append(result) - + self._interfaces_MAC = results - + return self._interfaces_MAC - + _interfaces_type = None - + @property def interfaces_type(self): """ 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))], - }) + result = self._dict( + { + 'name': self.get_value(starting + str(i)), + 'type': types[self.get_value(types_oid + str(i))], + } + ) results.append(result) - + self._interfaces_type = results - + return self._interfaces_type - + @property def interfaces_to_dict(self): """ @@ -296,20 +313,22 @@ def interfaces_to_dict(self): """ results = [] for i in range(0, len(self.get_interfaces())): - print '===== %d =====' % 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']) - }) + 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']), + } + ) results.append(result) return results - + @property def wireless_dbm(self): """ @@ -320,7 +339,7 @@ def wireless_dbm(self): for i in range(0, len(res[3])): dbm.append(int(res[3][i][0][1])) return dbm - + @property def wireless_links(self): ''' @@ -329,7 +348,7 @@ def wireless_links(self): 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]) - separated_by_meaning= [] + separated_by_meaning = [] dbm = [] tx_bytes = [] rx_bytes = [] @@ -337,10 +356,10 @@ def wireless_links(self): rx_packets = [] tx_rate = [] rx_rate = [] - + for i in range(0, len(results[3]), link_number): - separated_by_meaning.append(results[3][i:i+link_number]) - + separated_by_meaning.append(results[3][slice(i, i + link_number)]) + for i in range(0, len(separated_by_meaning[0])): dbm.append(int(separated_by_meaning[0][i][0][1])) tx_bytes.append(int(separated_by_meaning[1][i][0][1])) @@ -349,17 +368,19 @@ def wireless_links(self): rx_packets.append(int(separated_by_meaning[4][i][0][1])) tx_rate.append(int(separated_by_meaning[5][i][0][1])) rx_rate.append(int(separated_by_meaning[6][i][0][1])) - + 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] - }) + 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], + } + ) final.append(result) return final @@ -370,7 +391,7 @@ def RAM_total(self): """ 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): """ @@ -378,22 +399,24 @@ def RAM_free(self): """ 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 - }) \ No newline at end of file + 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, + } + ) diff --git a/netengine/backends/snmp/base.py b/netengine/backends/snmp/base.py index 4a629d3..64f21d7 100644 --- a/netengine/backends/snmp/base.py +++ b/netengine/backends/snmp/base.py @@ -1,20 +1,25 @@ try: from pysnmp.entity.rfc3413.oneliner import cmdgen except ImportError: - raise ImportError('pysnmp library is not installed, install it with "pip install pysnmp"') + raise ImportError( + 'pysnmp library is not installed, install it with "pip install pysnmp"' + ) + +import logging from netengine.backends import BaseBackend from netengine.exceptions import NetEngineError - __all__ = ['SNMP'] +logger = logging.getLogger(__name__) + class SNMP(BaseBackend): """ SNMP base backend """ - + _oid_to_retrieve = None def __init__(self, host, community='public', agent='my-agent', port=161): @@ -30,16 +35,8 @@ def __init__(self, host, community='public', agent='my-agent', port=161): self.port = port def __str__(self): - """ prints a human readable object description """ - return "" % self.host - - def __repr__(self): - """ returns unicode string represantation """ - return self.__str__() - - def __unicode__(self): - """ unicode __str__() for python2.7 """ - return unicode(self.__str__()) + """prints a human readable object description""" + return f'' @property def _command(self): @@ -52,10 +49,10 @@ def _oid(self, oid): """ returns valid oid value to be passed to getCmd() or nextCmd() """ - if type(oid) not in (str, unicode, tuple, list): + if type(oid) not in (str, tuple, list): raise AttributeError('get accepts only strings, tuples or lists') # allow string representations of oids with commas , - elif isinstance(oid, basestring): + elif isinstance(oid, str): # ignore spaces oid = oid.replace(' ', '').replace(',', '.') # convert lists and tuples into strings @@ -73,13 +70,13 @@ def get(self, oid): :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) """ - print 'DEBUG: SNMP GET %s' % self._oid(oid) + logger.info(f'DEBUG: SNMP GET {self._oid(oid)}') return self._command.getCmd(self.community, self.transport, self._oid(oid)) def next(self, oid): @@ -88,13 +85,13 @@ def next(self, oid): :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) """ - print 'DEBUG: SNMP NEXT %s' % self._oid(oid) + logger.info(f'DEBUG: SNMP NEXT {self._oid(oid)}') return self._command.nextCmd(self.community, self.transport, self._oid(oid)) def get_value(self, oid): @@ -107,19 +104,21 @@ def get_value(self, oid): return str(result[3][0][1]) # snmp stores results in several arrays except IndexError: 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 """ 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') - + + if self._oid_to_retrieve is None: + raise NetEngineError( + 'Please fix properly the _oid_to_retrieve string in OpenWRT or AirOS SNMP backend' + ) + indexes = self.next(self._oid_to_retrieve)[3] - + for i in range(len(indexes)): value_to_retr.append(int(indexes[i][0][1])) - + return value_to_retr diff --git a/netengine/backends/snmp/openwrt.py b/netengine/backends/snmp/openwrt.py index cd95637..2e21adb 100644 --- a/netengine/backends/snmp/openwrt.py +++ b/netengine/backends/snmp/openwrt.py @@ -6,10 +6,13 @@ import binascii +import logging from datetime import timedelta from netengine.backends.snmp import SNMP +logger = logging.getLogger(__name__) + class OpenWRT(SNMP): """ @@ -20,8 +23,8 @@ class OpenWRT(SNMP): _interface_dict = {} def __str__(self): - """ print a human readable object description """ - return u"" % self.host + """print a human readable object description""" + return f'' def validate(self): """ @@ -58,7 +61,7 @@ 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 + return int(self.get_value('1.3.6.1.2.1.1.3.0')) // 100 @property def uptime_tuple(self): @@ -67,7 +70,7 @@ def uptime_tuple(self): """ td = timedelta(seconds=self.uptime) - return td.days, td.seconds//3600, (td.seconds//60)%60 + return td.days, td.seconds // 3600, (td.seconds // 60) % 60 _interfaces = None @@ -80,12 +83,12 @@ def get_interfaces(self): 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) + value_to_get1 = value_to_get + str(i) if value_to_get1: interfaces.append(self.get_value(value_to_get1)) - self._interfaces = filter(None, interfaces) + self._interfaces = [_f for _f in interfaces if _f] return self._interfaces @@ -104,12 +107,20 @@ def interfaces_MAC(self): mac1.append(self.get_value('1.3.6.1.2.1.2.2.1.6.' + str(i))) mac_trans = [] for i in mac1: - mac_trans.append(':'.join(binascii.b2a_hex(i)[a:a+2] for a in range(0, 12, 2) if i != "")) + mac_string = binascii.b2a_hex(i.encode()).decode() + mac_trans.append( + ':'.join( + [ + mac_string[slice(i, i + 2)] + for i in range(0, 12, 2) + if i != '' + ] + ) + ) for i in range(0, len(self.get_interfaces())): - result = self._dict({ - "name" : self.get_interfaces()[i], - "mac_address" : mac_trans[i] - }) + result = self._dict( + {'name': self.get_interfaces()[i], 'mac_address': mac_trans[i]} + ) results.append(result) self._interfaces_MAC = results @@ -125,16 +136,18 @@ def interfaces_mtu(self): """ 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) 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))) - }) + result = self._dict( + { + 'name': self.get_value(starting + str(i)), + 'mtu': int(self.get_value(to + str(i))), + } + ) results.append(result) self._interfaces_mtu = results @@ -150,13 +163,15 @@ def interfaces_speed(self): """ 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 i = 1 - consecutive_fails = 0 # counter that indicates how many consecutive attempts failed + consecutive_fails = ( + 0 # counter that indicates how many consecutive attempts failed + ) while True: # break cycles if STOP_AFTER_FAILS reached if consecutive_fails == STOP_AFTER_FAILS: @@ -180,10 +195,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 @@ -202,29 +214,27 @@ def interfaces_state(self): """ 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" - }) + result = self._dict( + {'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" - }) + result = self._dict( + { + '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 @@ -240,16 +250,18 @@ def interfaces_bytes(self): """ 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))), - }) + 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))), + } + ) results.append(result) self._interfaces_bytes = results @@ -264,15 +276,21 @@ def interfaces_type(self): 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"} + types = { + '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))], - }) + result = self._dict( + { + 'name': self.get_value(starting + str(i)), + 'type': types[self.get_value(types_oid + str(i))], + } + ) results.append(result) self._interfaces_type = results @@ -291,9 +309,9 @@ def interface_addr_and_mask(self): 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 = {} @@ -305,10 +323,7 @@ def interface_addr_and_mask(self): 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 @@ -322,57 +337,48 @@ def interfaces_to_dict(self): results = [] for i in range(0, len(self.get_interfaces())): - print '====== %d ======' % i + logger.info(f'====== {i} ======') - print '... name ...' + logger.info('... name ...') name = self.interfaces_MAC[i]['name'] - print '... if_type ...' + logger.info('... if_type ...') if_type = self.interfaces_type[i]['type'] - print '... mac_address ...' + logger.info('... mac_address ...') mac_address = self.interfaces_MAC[i]['mac_address'] - print '... rx_bytes ...' + logger.info('... rx_bytes ...') rx_bytes = int(self.interfaces_bytes[i]['rx']) - print '... tx_bytes ...' + logger.info('... tx_bytes ...') tx_bytes = int(self.interfaces_bytes[i]['tx']) - print '... state ...' + logger.info('... state ...') state = self.interfaces_state[i]['state'] - print '... mtu ...' + logger.info('... mtu ...') mtu = int(self.interfaces_mtu[i]['mtu']) - print '... speed ...' + logger.info('... speed ...') speed = int(self.interfaces_speed[i]['speed']) - print '... ip address & subnet ...' + logger.info('... ip address & subnet ...') ip_and_netmask = self.interface_addr_and_mask - if name in ip_and_netmask.keys(): + if name in list(ip_and_netmask.keys()): 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 - }) - - # 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']) - # }) + 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, + } + ) results.append(result) return results @@ -381,20 +387,22 @@ 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")) + 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, - }) + 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, + } + ) diff --git a/netengine/exceptions.py b/netengine/exceptions.py index 6b09db6..0e1d3a5 100644 --- a/netengine/exceptions.py +++ b/netengine/exceptions.py @@ -4,5 +4,6 @@ class NetEngineError(Exception): - """ net engine related error """ - pass \ No newline at end of file + """net engine related error""" + + pass diff --git a/netengine/shortcuts.py b/netengine/shortcuts.py deleted file mode 100644 index 58a4592..0000000 --- a/netengine/shortcuts.py +++ /dev/null @@ -1,4 +0,0 @@ -try: - from collections import OrderedDict -except ImportError: - OrderedDict = dict # python < 2.7 diff --git a/requirements-test.txt b/requirements-test.txt index a6c88f5..fed2167 100644 --- a/requirements-test.txt +++ b/requirements-test.txt @@ -1,2 +1,5 @@ -nose -coverage +nose2~=0.10.0 +coverage~=5.5 +sphinx~=4.0.2 +openwisp-utils[qa]~=0.7.4 +pylinkvalidator~=0.3.0 diff --git a/requirements.txt b/requirements.txt index 4d6cfba..81d4504 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,2 @@ -netaddr -pysnmp -ipaddress -argparse -mock +netaddr~=0.8.0 +pysnmp~=4.4.12 diff --git a/run-qa-checks b/run-qa-checks new file mode 100755 index 0000000..329fdbb --- /dev/null +++ b/run-qa-checks @@ -0,0 +1,20 @@ +#!/bin/bash + +set -e + +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 diff --git a/runtests.py b/runtests.py index c253020..6493dbb 100755 --- a/runtests.py +++ b/runtests.py @@ -2,22 +2,17 @@ # -*- coding: utf-8 -*- import os + try: - import nose -except: + import nose2 +except ImportError: message = """nose package not installed, install test requirements with: pip install -r requirements-test.txt """ 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") - result = nose.run( - argv=[ - os.path.abspath(__file__), - "--with-cov", - "--cover-package=netengine", - tests_path - ] - ) + tests_path = os.path.join(os.path.abspath(os.path.dirname(file_path)), 'tests',) + nose2.discover() + result = nose2.main() diff --git a/setup.cfg b/setup.cfg new file mode 100644 index 0000000..5522a04 --- /dev/null +++ b/setup.cfg @@ -0,0 +1,20 @@ +[bdist_wheel] +universal=1 + +[flake8] +exclude = *.egg-info, + .git, + ./tests/*settings*.py, + docs/* +max-line-length = 110 +# W503: line break before or after operator +# W504: line break after or after operator +# W605: invalid escape sequence +ignore = W605, W503, W504 + +[isort] +line_length=88 +multi_line_output=3 +use_parentheses=True +include_trailing_comma=True +force_grid_wrap=0 diff --git a/setup.py b/setup.py index c2583d2..8d69b17 100755 --- a/setup.py +++ b/setup.py @@ -1,6 +1,6 @@ #!/usr/bin/env python -from setuptools import setup, find_packages +from setuptools import find_packages, setup from netengine import get_version @@ -12,7 +12,12 @@ def get_install_requires(): requirements = [] 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'): + if ( + line.startswith('#') + or line == '' + or line.startswith('http') + or line.startswith('git') + ): continue # add line to requirements requirements.append(line) @@ -22,7 +27,7 @@ def get_install_requires(): setup( name='netengine', version=get_version(), - description="Abstraction layer for extracting information from network devices.", + 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', @@ -40,5 +45,5 @@ def get_install_requires(): 'Programming Language :: Python', 'Topic :: System :: Networking', ], - test_suite='nose.collector' + test_suite='nose2.collector.collector', ) diff --git a/tests/__init__.py b/tests/__init__.py index 893702f..e69de29 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -1,3 +0,0 @@ -from .base import * -from .dummy import * -from .snmp import * diff --git a/tests/settings.py b/tests/settings.py index 65c5c40..da8f734 100644 --- a/tests/settings.py +++ b/tests/settings.py @@ -3,4 +3,4 @@ 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') is '1' +settings['disable_mocks'] = os.getenv('DISABLE_MOCKS', '0') == '1' diff --git a/tests/snmp/__init__.py b/tests/snmp/__init__.py deleted file mode 100644 index 2f933b1..0000000 --- a/tests/snmp/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -from .base import * -from .openwrt import * -from .airos import * \ No newline at end of file diff --git a/tests/base.py b/tests/test_base.py similarity index 85% rename from tests/base.py rename to tests/test_base.py index 225e62c..a3883db 100644 --- a/tests/base.py +++ b/tests/test_base.py @@ -1,93 +1,89 @@ import unittest -from netengine import get_version, __version__ -from netengine.backends import BaseBackend from netaddr import AddrFormatError +from netengine import __version__, get_version +from netengine.backends import BaseBackend __all__ = ['TestBaseBackend'] class TestBaseBackend(unittest.TestCase): - def test_version(self): get_version() __version__ - + def test_dict(self): device = BaseBackend() dictionary = device._dict({}) self.assertTrue(isinstance(dictionary, dict)) - + def test_base_backend(self): device = BaseBackend() - + self.assertTrue(device.__netengine__) - + with self.assertRaises(NotImplementedError): device.validate() - + with self.assertRaises(NotImplementedError): device.to_dict() - + with self.assertRaises(NotImplementedError): device.to_json() - + with self.assertRaises(NotImplementedError): str(device) - + with self.assertRaises(NotImplementedError): device.__repr__() - - with self.assertRaises(NotImplementedError): - device.__unicode__() - + 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.uptime - + with self.assertRaises(NotImplementedError): device.uptime_tuple - + 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 - + with self.assertRaises(NotImplementedError): device.olsr - + def test_get_manufacturer_unicode(self): device = BaseBackend() with self.assertRaises(AddrFormatError): - device.get_manufacturer(u"wrong MAC") + device.get_manufacturer('wrong MAC') diff --git a/tests/dummy.py b/tests/test_dummy.py similarity index 69% rename from tests/dummy.py rename to tests/test_dummy.py index ea86c4d..7c5120c 100644 --- a/tests/dummy.py +++ b/tests/test_dummy.py @@ -1,35 +1,29 @@ -import unittest import json -from netengine.backends import Dummy +import unittest +from netengine.backends import Dummy __all__ = ['TestDummyBackend'] class TestDummyBackend(unittest.TestCase): - def setUp(self): self.dummy = Dummy('10.40.0.1') - + def test_str(self): self.assertIn('Dummy NetEngine', str(self.dummy)) - + def test_validate(self): self.dummy.validate() - + 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)) - ) - + self.assertIn('Xensource, Inc.', str(self.dummy.get_manufacturer(dummy_addr))) + def test_to_json(self): json_string = self.dummy.to_json() - - self.assertTrue(isinstance(json_string, basestring)) - - dictionary = json.loads(json_string) - + self.assertTrue(isinstance(json_string, str)) + json.loads(json_string) diff --git a/tests/test_snmp/__init__.py b/tests/test_snmp/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/snmp/airos.py b/tests/test_snmp/test_airos.py similarity index 93% rename from tests/snmp/airos.py rename to tests/test_snmp/test_airos.py index 43aa1fe..4ea3227 100644 --- a/tests/snmp/airos.py +++ b/tests/test_snmp/test_airos.py @@ -1,18 +1,19 @@ import unittest -from netengine.backends.snmp import AirOS -from netengine.exceptions import NetEngineError +from unittest.mock import patch + from pysnmp.entity.rfc3413.oneliner import cmdgen from pysnmp.smi.error import NoSuchObjectError +from netengine.backends.snmp import AirOS +from netengine.exceptions import NetEngineError + from ..settings import settings from ..utils import MockOutputMixin, SpyMock - __all__ = ['TestSNMPAirOS'] class TestSNMPAirOS(unittest.TestCase, MockOutputMixin): - def setUp(self): self.host = settings['airos-snmp']['host'] self.community = settings['airos-snmp']['community'] @@ -25,7 +26,7 @@ def setUp(self): target=cmdgen.CommandGenerator, attribute='nextCmd', wrap_obj=self.device._command, - return_value=[0, 0, 0, [[[0, 1]]] * 5] + return_value=[0, 0, 0, [[[0, 1]]] * 5], ) self.getcmd_patcher = SpyMock._patch( target=cmdgen.CommandGenerator, @@ -36,46 +37,46 @@ def setUp(self): ), ) self.getcmd_patcher.start() - + def test_get_value_error(self): self.getcmd_patcher.stop() 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') self.assertRaises(NetEngineError, wrong.validate) - + def test_validate_positive_result(self): self.device.validate() - + def test_get(self): with self.assertRaises(AttributeError): 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(u'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]) + def test_properties(self): device = self.device - + device.os device.name device.model device.os device.uptime device.uptime_tuple - + def test_name(self): self.assertIsInstance(self.device.name, str) - + def test_os(self): self.assertIsInstance(self.device.os, tuple) - + def test_get_interfaces(self): with self.nextcmd_patcher: self.assertIsInstance(self.device.get_interfaces(), list) @@ -83,27 +84,27 @@ def test_get_interfaces(self): def test_get_interfaces_mtu(self): with self.nextcmd_patcher: self.assertIsInstance(self.device.interfaces_mtu, list) - + def test_interfaces_state(self): with self.nextcmd_patcher: self.assertIsInstance(self.device.interfaces_state, list) - + def test_interfaces_speed(self): with self.nextcmd_patcher: self.assertIsInstance(self.device.interfaces_speed, list) - + def test_interfaces_bytes(self): with self.nextcmd_patcher: self.assertIsInstance(self.device.interfaces_bytes, list) - + def test_interfaces_MAC(self): with self.nextcmd_patcher: self.assertIsInstance(self.device.interfaces_MAC, list) - + def test_interfaces_type(self): with self.nextcmd_patcher: self.assertIsInstance(self.device.interfaces_type, list) - + def test_interfaces_to_dict(self): with self.nextcmd_patcher: self.assertIsInstance(self.device.interfaces_to_dict, list) @@ -111,10 +112,10 @@ def test_interfaces_to_dict(self): def test_wireless_dbm(self): with self.nextcmd_patcher: self.assertIsInstance(self.device.wireless_dbm, list) - + def test_interfaces_number(self): self.assertIsInstance(self.device.interfaces_number, int) - + def test_wireless_to_dict(self): with self.nextcmd_patcher as np: SpyMock._update_patch( @@ -127,7 +128,7 @@ def test_wireless_to_dict(self): def test_RAM_free(self): self.assertIsInstance(self.device.RAM_free, int) - + def test_RAM_total(self): self.assertIsInstance(self.device.RAM_total, int) @@ -140,7 +141,7 @@ def test_to_dict(self): ), ) self.assertTrue(isinstance(self.device.to_dict(), dict)) - + def test_manufacturer_to_dict(self): with self.nextcmd_patcher as np: SpyMock._update_patch( @@ -150,22 +151,22 @@ def test_manufacturer_to_dict(self): ), ) self.assertIsNotNone(self.device.to_dict()['manufacturer']) - + def test_manufacturer(self): with self.nextcmd_patcher: self.assertIsNotNone(self.device.manufacturer) - + def test_model(self): self.assertIsInstance(self.device.model, str) - + def test_firmware(self): self.assertIsInstance(self.device.firmware, str) - + def test_uptime(self): self.assertIsInstance(self.device.uptime, int) - + def test_uptime_tuple(self): self.assertIsInstance(self.device.uptime_tuple, tuple) def tearDown(self): - self.getcmd_patcher.stop() + patch.stopall() diff --git a/tests/snmp/base.py b/tests/test_snmp/test_base.py similarity index 96% rename from tests/snmp/base.py rename to tests/test_snmp/test_base.py index 347f4e3..7c7a82c 100644 --- a/tests/snmp/base.py +++ b/tests/test_snmp/test_base.py @@ -5,25 +5,23 @@ 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): @@ -48,22 +46,22 @@ def test_not_implemented_methods(self): 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() diff --git a/tests/snmp/openwrt.py b/tests/test_snmp/test_openwrt.py similarity index 94% rename from tests/snmp/openwrt.py rename to tests/test_snmp/test_openwrt.py index 8a9a19c..f12588e 100644 --- a/tests/snmp/openwrt.py +++ b/tests/test_snmp/test_openwrt.py @@ -1,7 +1,10 @@ import unittest -from netengine.backends.snmp import OpenWRT +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 @@ -13,7 +16,7 @@ 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(self.host, self.community, self.port) + 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') @@ -21,7 +24,7 @@ def setUp(self): target=cmdgen.CommandGenerator, attribute='nextCmd', wrap_obj=self.device._command, - return_value=[0, 0, 0, [[[0, 1]]] * 5] + return_value=[0, 0, 0, [[[0, 1]]] * 5], ) self.getcmd_patcher = SpyMock._patch( target=cmdgen.CommandGenerator, @@ -95,7 +98,7 @@ def test_to_dict(self): 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): @@ -104,4 +107,4 @@ def test_manufacturer_to_dict(self): self.assertIsNotNone(self.device.to_dict()['manufacturer']) def tearDown(self): - self.getcmd_patcher.stop() + patch.stopall() diff --git a/tests/utils.py b/tests/utils.py index ee167f3..6c3b548 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -1,9 +1,10 @@ import json -import mock import os +from unittest import mock from .settings import settings + class SpyMock: @staticmethod def _patch(*args, **kwargs): @@ -30,9 +31,9 @@ def _load_mock_json(file): @staticmethod def _get_mocked_getcmd(data, input): oid = input[2] - result = data[oid].encode('ascii', 'ignore') + result = data[oid] if type(result) == list: - result = "\n".join(result[0:]) + result = '\n'.join(result[0:]) return [0, 0, 0, [[0, result]]] @staticmethod @@ -42,6 +43,6 @@ def _get_mocked_wireless_links(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.2.1.1.9.1.1': [0, 0, 0, [[[0, 1]]] * 5], } return return_data[oid]