Merge pull request #28 from napalm-automation/develop

Release 0.3.0

.travis.yml

@@ -1,10 +1,20 @@
 language: python
+
 python:
 - 2.7
+
 install:
-- pip install -r requirements.txt
+- pip install -r requirements-dev.txt
 - pip install .
-- pip install coveralls
+
+script:
+- py.test
+- pylama .
+
+after_success:
+   - coveralls
+   - if [ $TRAVIS_TAG ]; then curl -X POST https://readthedocs.org/build/napalm; fi
+
 deploy:
   provider: pypi
   user: dbarroso
@@ -13,14 +23,3 @@ deploy:
   on:
     tags: true
     branch: master
-script:
-- cd test/unit
-- nosetests --with-coverage --cover-package napalm_fortios -v TestFortiOSDriver:TestGetterFortiOSDriver.test_get_bgp_neighbors
-- nosetests --with-coverage --cover-package napalm_fortios -v TestFortiOSDriver:TestGetterFortiOSDriver.test_get_environment
-- nosetests --with-coverage --cover-package napalm_fortios -v TestFortiOSDriver:TestGetterFortiOSDriver.test_get_facts
-- nosetests --with-coverage --cover-package napalm_fortios -v TestFortiOSDriver:TestGetterFortiOSDriver.test_get_interfaces
-- nosetests --with-coverage --cover-package napalm_fortios -v TestFortiOSDriver:TestGetterFortiOSDriver.test_get_interfaces_counters
-- nosetests --with-coverage --cover-package napalm_fortios -v TestFortiOSDriver:TestGetterFortiOSDriver.test_get_config
-- cd ../..
-- coverage combine test/unit/.coverage
-after_success: coveralls

napalm_fortios/fortios.py

@@ -16,7 +16,8 @@
 from pyFG.exceptions import FailedCommit, CommandExecutionException
 from napalm_base.base import NetworkDriver
 from napalm_base.exceptions import ReplaceConfigException, MergeConfigException
-from napalm_base.utils.string_parsers import colon_separated_string_to_dict, convert_uptime_string_seconds
+from napalm_base.utils.string_parsers import colon_separated_string_to_dict,\
+                                             convert_uptime_string_seconds
 
 
 class FortiOSDriver(NetworkDriver):
@@ -30,7 +31,8 @@ def __init__(self, hostname, username, password, timeout=60, optional_args=None)
         else:
             self.vdom = None
 
-        self.device = FortiOS(hostname, username=username, password=password, timeout=timeout, vdom=self.vdom)
+        self.device = FortiOS(hostname, username=username, password=password,
+                              timeout=timeout, vdom=self.vdom)
         self.config_replace = False
 
     def open(self):
@@ -39,19 +41,27 @@ def open(self):
     def close(self):
         self.device.close()
 
+    def is_alive(self):
+            """Returns a flag with the state of the SSH connection."""
+            return {
+                'is_alive': self.device.ssh.get_transport().is_active()
+            }
+
     def execute_command_with_vdom(self, command, vdom=None):
         # If the user doesn't specify a particular vdom we use the default vdom for the object.
         vdom = vdom or self.vdom
 
         if vdom == 'global' and self.vdom is not None:
-            # If vdom is global we go to the global vdom, execute the commands and then back to the root.
-            # There is a catch, if the device doesn't have vdoms enabled we have to execute the command in the root
+            # If vdom is global we go to the global vdom, execute the commands
+            # and then back to the root. There is a catch, if the device doesn't
+            # have vdoms enabled we have to execute the command in the root
             command = 'conf global\n{command}\nend'.format(command=command)
 
             # We skip the lines telling us that we changed vdom
             return self.device.execute_command(command)[1:-2]
         elif vdom not in ['global', None]:
-            # If we have a vdom we change to the vdom, execute the commands and then exit back to the root
+            # If we have a vdom we change to the vdom, execute
+            # the commands and then exit back to the root
             command = 'conf vdom\nedit {vdom}\n{command}\nend'.format(vdom=vdom, command=command)
 
             # We skip the lines telling us that we changed vdom
@@ -132,10 +142,14 @@ def rollback(self):
 
         self.device.load_config(empty_candidate=True)
         self.load_replace_candidate(config=rollback_config)
-        self.device.candidate_config['vpn certificate local']['Fortinet_CA_SSLProxy'].del_param('private-key')
-        self.device.candidate_config['vpn certificate local']['Fortinet_CA_SSLProxy'].del_param('certificate')
-        self.device.candidate_config['vpn certificate local']['Fortinet_SSLProxy'].del_param('private-key')
-        self.device.candidate_config['vpn certificate local']['Fortinet_SSLProxy'].del_param('certificate')
+        self.device.candidate_config['vpn certificate local']['Fortinet_CA_SSLProxy'].\
+            del_param('private-key')
+        self.device.candidate_config['vpn certificate local']['Fortinet_CA_SSLProxy'].\
+            del_param('certificate')
+        self.device.candidate_config['vpn certificate local']['Fortinet_SSLProxy'].\
+            del_param('private-key')
+        self.device.candidate_config['vpn certificate local']['Fortinet_SSLProxy'].\
+            del_param('certificate')
         self.device.commit()
 
     def get_config(self, retrieve="all"):
@@ -163,7 +177,8 @@ def get_config(self, retrieve="all"):
 
     def get_facts(self):
         system_status = self.get_command_with_vdom('get system status', vdom='global')
-        performance_status = self.get_command_with_vdom('get system performance status', vdom='global')
+        performance_status = self.get_command_with_vdom('get system performance status',
+                                                        vdom='global')
 
         interfaces = self.execute_command_with_vdom('get system interface | grep ==', vdom='global')
         interface_list = [x.split()[2] for x in interfaces if x.strip() is not '']
@@ -246,6 +261,51 @@ def _search_line_in_lines(search, lines):
             if search in l:
                 return l
 
+    def get_firewall_policies(self):
+        cmd = self.execute_command_with_vdom('show firewall policy')
+        policy = dict()
+        policy_id = None
+        default_policy = dict()
+        position = 1
+
+        for line in cmd:
+            policy_data = line.strip()
+            if policy_data.find("edit") == 0:
+                policy_id = policy_data.split()[1]
+                policy[policy_id] = dict()
+            if policy_id is not None:
+                if len(policy_data.split()) > 2:
+                    policy_setting = policy_data.split()[1]
+                    policy[policy_id][policy_setting] = policy_data.split()[2].replace("\"", "")
+
+        for key in policy:
+
+            enabled = 'status' in policy[key]
+
+            logtraffic = policy[key]['logtraffic'] if 'logtraffic' in policy[key] else False
+
+            action = 'permit' if 'action' in policy[key] else 'reject'
+
+            policy_item = dict()
+            default_policy[key] = list()
+            policy_item['position'] = position
+            policy_item['packet_hits'] = -1
+            policy_item['byte_hits'] = -1
+            policy_item['id'] = unicode(key)
+            policy_item['enabled'] = enabled
+            policy_item['schedule'] = unicode(policy[key]['schedule'])
+            policy_item['log'] = unicode(logtraffic)
+            policy_item['l3_src'] = unicode(policy[key]['srcaddr'])
+            policy_item['l3_dst'] = unicode(policy[key]['dstaddr'])
+            policy_item['service'] = unicode(policy[key]['service'])
+            policy_item['src_zone'] = unicode(policy[key]['srcintf'])
+            policy_item['dst_zone'] = unicode(policy[key]['dstintf'])
+            policy_item['action'] = unicode(action)
+            default_policy[key].append(policy_item)
+
+            position = position + 1
+        return default_policy
+
     def get_bgp_neighbors(self):
 
         families = ['ipv4', 'ipv6']
@@ -279,7 +339,8 @@ def get_bgp_neighbors(self):
                 neighbor_dict['address_family']['ipv4'] = dict()
                 neighbor_dict['address_family']['ipv6'] = dict()
 
-            detail_output = [x.lower() for x in self.execute_command_with_vdom(command_detail.format(neighbor))]
+            detail_output = [x.lower() for x in
+                             self.execute_command_with_vdom(command_detail.format(neighbor))]
             m = re.search('remote router id (.+?)\n', '\n'.join(detail_output))
             if m:
                 neighbor_dict['remote_id'] = unicode(m.group(1))
@@ -326,7 +387,8 @@ def get_interfaces_counters(self):
             elif (data[1].startswith('RX packets') or data[1].startswith('TX packets')) and if_name:
                 if_data = data[1].split(' ')
                 direction = if_data[0].lower()
-                interface_counters[if_name][direction + '_unicast_packets'] = int(if_data[1].split(':')[1])
+                interface_counters[if_name][direction + '_unicast_packets'] = \
+                    int(if_data[1].split(':')[1])
                 interface_counters[if_name][direction + '_errors'] = int(if_data[2].split(':')[1])
                 interface_counters[if_name][direction + '_discards'] = int(if_data[2].split(':')[1])
                 interface_counters[if_name][direction + '_multicast_packets'] = -1
@@ -340,9 +402,6 @@ def get_interfaces_counters(self):
                     interface_counters[if_name]['tx_octets'] = int(if_data[7].split(':')[1])
         return interface_counters
 
-    def get_lldp_neighbors(self):
-        return {}
-
     def get_environment(self):
 
         def parse_string(line):
@@ -380,15 +439,16 @@ def get_cpu(cpu_lines):
             return output
 
         def get_memory(memory_line):
-            total, used = int(memory_line[1]) >> 20, int(memory_line[2]) >> 20  # convert from byte to MB
+            total, used = int(memory_line[1]) >> 20, int(memory_line[2]) >> 20  # byte to MB
             return dict(available_ram=total, used_ram=used)
 
         def get_temperature(temperature_lines, detail_block):
             output = dict()
             for temp_line in temperature_lines:
                 if 'disabled' in temp_line:
                     sensor_name = search_disabled(temp_line)
-                    output[sensor_name] = {'is_alert': False, 'is_critical': False, 'temperature': 0.0}
+                    output[sensor_name] = {'is_alert': False, 'is_critical': False,
+                                           'temperature': 0.0}
                     continue
 
                 m = search_normal(temp_line)
@@ -400,7 +460,8 @@ def get_temperature(temperature_lines, detail_block):
                 index_line = detail_block.index(fullline)
                 sensor_block = detail_block[index_line:]
 
-                v = int(self._search_line_in_lines('upper_non_recoverable', sensor_block).split('=')[1])
+                v = int(self._search_line_in_lines('upper_non_recoverable',
+                                                   sensor_block).split('=')[1])
 
                 output[sensor_name] = dict(temperature=float(temp_value), is_alert=is_alert,
                                            is_critical=True if v > temp_value else False)
@@ -409,22 +470,29 @@ def get_temperature(temperature_lines, detail_block):
 
         out = dict()
 
-        sensors_block = [parse_string(x) for x in self.execute_command_with_vdom('execute sensor detail', vdom='global') if x]
+        sensors_block = [parse_string(x) for x in
+                         self.execute_command_with_vdom('execute sensor detail', vdom='global')
+                         if x]
 
         # temp
-        temp_lines = [x for x in sensors_block if any([True for y in ['dts', 'temp', 'adt7490'] if y in x])]
+        temp_lines = [x for x in sensors_block
+                      if any([True for y in ['dts', 'temp', 'adt7490'] if y in x])]
         out['temperature'] = get_temperature(temp_lines, sensors_block)
 
         # fans
         out['fans'] = get_fans([x for x in sensors_block if 'fan' in x and 'temp' not in x])
 
         # cpu
         out['cpu'] = get_cpu(
-            [x for x in self.execute_command_with_vdom('get system performance status | grep CPU', vdom='global')[1:] if x])
+            [x for x in
+             self.execute_command_with_vdom('get system performance status | grep CPU',
+                                            vdom='global')[1:] if x])
 
         # memory
         memory_command = 'diag hard sys mem | grep Mem:'
-        t = [x for x in re.split('\s+', self.execute_command_with_vdom(memory_command, vdom='global')[0]) if x]
+        t = [x for x in
+             re.split('\s+', self.execute_command_with_vdom(memory_command,
+                                                            vdom='global')[0]) if x]
         out['memory'] = get_memory(t)
 
         # power, not implemented

requirements-dev.txt

@@ -4,4 +4,5 @@ pytest-json
 pytest-pythonpath
 pylama
 flake8-import-order
+coveralls
 -r requirements.txt

requirements.txt

@@ -1,2 +1,3 @@
 napalm-base>=0.18.0
 pyFG
+future

setup.cfg

@@ -4,3 +4,12 @@ ignore = D203,C901
 
 [pylama:pep8]
 max_line_length = 100
+
+[tool:pytest]
+addopts = --cov=./ -vs
+json_report = report.json
+jsonapi = true
+
+[coverage:run]
+include =
+  napalm_fortios/*

setup.py

@@ -12,7 +12,7 @@
 
 setup(
     name="napalm-fortios",
-    version="0.2.1",
+    version="0.3.0",
     packages=find_packages(),
     author="David Barroso",
     author_email="dbarrosop@dravetech.com",

test/unit/TestFortiOSDriver.py

@@ -15,8 +15,7 @@
 import unittest
 
 from napalm_fortios.fortios import FortiOSDriver
-from napalm_base.test.base import TestConfigNetworkDriver, TestGettersNetworkDriver
-from pyFG.fortios import FortiConfig
+from napalm_base.test.base import TestConfigNetworkDriver
 
 
 class TestConfigFortiOSDriver(unittest.TestCase, TestConfigNetworkDriver):
@@ -30,39 +29,3 @@ def setUpClass(cls):
 
         cls.device = FortiOSDriver(hostname, username, password, timeout=60)
         cls.device.open()
-
-
-class TestGetterFortiOSDriver(unittest.TestCase, TestGettersNetworkDriver):
-
-    @classmethod
-    def setUpClass(cls):
-        cls.mock = True
-
-        hostname = '192.168.56.201'
-        username = 'vagrant'
-        password = 'vagrant'
-        cls.vendor = 'eos'
-
-        cls.device = FortiOSDriver(hostname, username, password, timeout=60)
-
-        if cls.mock:
-            cls.device.device = FakeFortiOSDevice()
-        else:
-            cls.device.open()
-
-
-class FakeFortiOSDevice:
-
-    @staticmethod
-    def read_txt_file(filename):
-        with open(filename.lower()) as data_file:
-            return data_file.read().splitlines()
-
-    def execute_command(self, command):
-        return self.read_txt_file(
-            'fortios/mock_data/{}.txt'.format(command.replace(' ', '_').replace('|', '').replace(':', '')))
-
-    def load_config(self, config_block):
-        self.running_config = FortiConfig('running')
-        self.running_config.parse_config_output(self.read_txt_file('fortios/mock_data/{}.txt'.format(
-            config_block.replace(' ', '_'))))