| #! /usr/bin/python |
| |
| # Copyright 2015 Linaro Limited |
| # |
| # This program is free software; you can redistribute it and/or modify |
| # it under the terms of the GNU General Public License as published by |
| # the Free Software Foundation; either version 2 of the License, or |
| # (at your option) any later version. |
| # |
| # This program is distributed in the hope that it will be useful, |
| # but WITHOUT ANY WARRANTY; without even the implied warranty of |
| # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
| # GNU General Public License for more details. |
| # |
| # You should have received a copy of the GNU General Public License |
| # along with this program; if not, write to the Free Software |
| # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, |
| # MA 02110-1301, USA. |
| |
| import logging |
| import sys |
| import re |
| import pickle |
| import pexpect |
| |
| # Dummy switch driver, designed specifically for |
| # testing/validation. Just remembers what it's been told and gives the |
| # same data back on demand. |
| # |
| # To keep track of data in the dummy switch, this code will simply |
| # dump out and read back its internal state to/from a Python pickle |
| # file as needed. On first use, if no such file exists then the Dummy |
| # driver will simply generate a simple switch model: |
| # |
| # * N ports in access mode |
| # * 1 VLAN (tag 1) labelled DEFAULT |
| # |
| # The "hostname" given to the switch in VLANd is important, as it will |
| # determine both the number of ports allocated in this model and the |
| # name of the pickle file used for data storage. Call the switch |
| # "dummy-N" in your vland.cfg file to have N ports. If you want to use |
| # more than one dummy switch instance, ensure you give them different |
| # numbers, e.g. "dummy-25", "dummy-48", etc. |
| |
| if __name__ == '__main__': |
| import os |
| vlandpath = os.path.abspath(os.path.normpath(os.path.dirname(sys.argv[0]))) |
| sys.path.insert(0, vlandpath) |
| sys.path.insert(0, "%s/.." % vlandpath) |
| |
| from errors import InputError, PExpectError |
| from drivers.common import SwitchDriver, SwitchErrors |
| |
| class Dummy(SwitchDriver): |
| |
| connection = None |
| _username = None |
| _password = None |
| _enable_password = None |
| _dummy_vlans = {} |
| _dummy_ports = {} |
| _state_file = None |
| |
| _capabilities = [ |
| ] |
| |
| def __init__(self, switch_hostname, switch_telnetport=23, debug = False): |
| SwitchDriver.__init__(self, switch_hostname, debug) |
| self._systemdata = [] |
| self.errors = SwitchErrors() |
| self._state_file = "%s.pk" % switch_hostname |
| |
| ################################ |
| ### Switch-level API functions |
| ################################ |
| |
| # Save the current running config - we want config to remain |
| # across reboots |
| def switch_save_running_config(self): |
| pass |
| |
| # Restart the switch - we need to reload config to do a |
| # roll-back. Do NOT save running-config first if the switch asks - |
| # we're trying to dump recent changes, not save them. |
| def switch_restart(self): |
| pass |
| |
| # List the capabilities of the switch (and driver) - some things |
| # make no sense to abstract. Returns a dict of strings, each one |
| # describing an extra feature that that higher levels may care |
| # about |
| def switch_get_capabilities(self): |
| return self._capabilities |
| |
| ################################ |
| ### VLAN API functions |
| ################################ |
| |
| # Create a VLAN with the specified tag |
| def vlan_create(self, tag): |
| logging.debug("Creating VLAN %d", tag) |
| if not tag in self._dummy_vlans: |
| self._dummy_vlans[tag] = "VLAN%s" % tag |
| else: |
| # It's not an error if it already exists, but log anyway |
| logging.debug("VLAN %d already exists, name %s", |
| tag, self._dummy_vlans[tag]) |
| |
| # Destroy a VLAN with the specified tag |
| def vlan_destroy(self, tag): |
| logging.debug("Destroying VLAN %d", tag) |
| if tag in self._dummy_vlans: |
| del self._dummy_vlans[tag] |
| else: |
| # It's not an error if it doesn't exist, but log anyway |
| logging.debug("VLAN %d did not exist", tag) |
| |
| # Set the name of a VLAN |
| def vlan_set_name(self, tag, name): |
| logging.debug("Setting name of VLAN %d to %s", tag, name) |
| if not tag in self._dummy_vlans: |
| raise InputError("Tag %d does not exist") |
| self._dummy_vlans[tag] = "VLAN%s" % tag |
| |
| # Get a list of the VLAN tags currently registered on the switch |
| def vlan_get_list(self): |
| logging.debug("Grabbing list of VLANs") |
| return sorted(self._dummy_vlans.keys()) |
| |
| # For a given VLAN tag, ask the switch what the associated name is |
| def vlan_get_name(self, tag): |
| logging.debug("Grabbing the name of VLAN %d", tag) |
| if not tag in self._dummy_vlans: |
| raise InputError("Tag %d does not exist") |
| return self._dummy_vlans[tag] |
| |
| ################################ |
| ### Port API functions |
| ################################ |
| |
| # Set the mode of a port: access or trunk |
| def port_set_mode(self, port, mode): |
| logging.debug("Setting port %s to %s mode", port, mode) |
| if not port in self._dummy_ports: |
| raise InputError("Port %s does not exist" % port) |
| self._dummy_ports[port]['mode'] = mode |
| |
| # Get the mode of a port: access or trunk |
| def port_get_mode(self, port): |
| logging.debug("Getting mode of port %s", port) |
| if not port in self._dummy_ports: |
| raise InputError("Port %s does not exist" % port) |
| return self._dummy_ports[port]['mode'] |
| |
| # Set an access port to be in a specified VLAN (tag) |
| def port_set_access_vlan(self, port, tag): |
| logging.debug("Setting access port %s to VLAN %d", port, tag) |
| if not port in self._dummy_ports: |
| raise InputError("Port %s does not exist" % port) |
| if not tag in self._dummy_vlans: |
| raise InputError("VLAN %d does not exist" % tag) |
| self._dummy_ports[port]['access_vlan'] = tag |
| |
| # Add a trunk port to a specified VLAN (tag) |
| def port_add_trunk_to_vlan(self, port, tag): |
| logging.debug("Adding trunk port %s to VLAN %d", port, tag) |
| if not port in self._dummy_ports: |
| raise InputError("Port %s does not exist" % port) |
| if not tag in self._dummy_vlans: |
| raise InputError("VLAN %d does not exist" % tag) |
| self._dummy_ports[port]['trunk_vlans'].append(tag) |
| |
| # Remove a trunk port from a specified VLAN (tag) |
| def port_remove_trunk_from_vlan(self, port, tag): |
| logging.debug("Removing trunk port %s from VLAN %d", port, tag) |
| if not port in self._dummy_ports: |
| raise InputError("Port %s does not exist" % port) |
| if not tag in self._dummy_vlans: |
| raise InputError("VLAN %d does not exist" % tag) |
| self._dummy_ports[port]['trunk_vlans'].remove(tag) |
| |
| # Get the configured VLAN tag for an access port (tag) |
| def port_get_access_vlan(self, port): |
| logging.debug("Getting VLAN for access port %s", port) |
| if not port in self._dummy_ports: |
| raise InputError("Port %s does not exist" % port) |
| return self._dummy_ports[port]['access_vlan'] |
| |
| # Get the list of configured VLAN tags for a trunk port |
| def port_get_trunk_vlan_list(self, port): |
| logging.debug("Getting VLAN(s) for trunk port %s", port) |
| if not port in self._dummy_ports: |
| raise InputError("Port %s does not exist" % port) |
| return sorted(self._dummy_ports[port]['trunk_vlans']) |
| |
| ################################ |
| ### Internal functions |
| ################################ |
| |
| # Connect to the switch and log in |
| def _switch_connect(self): |
| # Open data file if it exists, otherwise initialise |
| try: |
| pkl_file = open(self._state_file, 'rb') |
| self._dummy_vlans = pickle.load(pkl_file) |
| self._dummy_ports = pickle.load(pkl_file) |
| pkl_file.close() |
| except: |
| # Create data here |
| self._dummy_vlans = {1: 'DEFAULT'} |
| match = re.match(r'dummy-(\d+)', self.hostname) |
| if match: |
| num_ports = int(match.group(1)) |
| else: |
| raise InputError("Unable to determine number of ports from switch name") |
| for i in range(1, num_ports+1): |
| port_name = "dm%2.2d" % int(i) |
| self._dummy_ports[port_name] = {} |
| self._dummy_ports[port_name]['mode'] = 'access' |
| self._dummy_ports[port_name]['access_vlan'] = 1 |
| self._dummy_ports[port_name]['trunk_vlans'] = [] |
| |
| # Now build a list of our ports, for later sanity checking |
| self._ports = self._get_port_names() |
| if len(self._ports) < 4: |
| raise IOError("Not enough ports detected - problem!") |
| |
| def _logout(self): |
| pkl_file = open(self._state_file, 'wb') |
| pickle.dump(self._dummy_vlans, pkl_file) |
| pickle.dump(self._dummy_ports, pkl_file) |
| pkl_file.close() |
| |
| def _get_port_names(self): |
| logging.debug("Grabbing list of ports") |
| interfaces = [] |
| for interface in sorted(self._dummy_ports.keys()): |
| interfaces.append(interface) |
| self._port_numbers[interface] = len(interfaces) |
| return interfaces |
| |
| if __name__ == "__main__": |
| |
| # Simple test harness - exercise the main working functions above to verify |
| # they work. This does *NOT* test really disruptive things like "save |
| # running-config" and "reload" - test those by hand. |
| |
| import optparse |
| |
| switch = 'dummy-48' |
| parser = optparse.OptionParser() |
| parser.add_option("--switch", |
| dest = "switch", |
| action = "store", |
| nargs = 1, |
| type = "string", |
| help = "specify switch to connect to for testing", |
| metavar = "<switch>") |
| (opts, args) = parser.parse_args() |
| if opts.switch: |
| switch = opts.switch |
| |
| logging.basicConfig(level = logging.DEBUG, |
| format = '%(asctime)s %(levelname)-8s %(message)s') |
| p = Dummy(switch, 23, debug=True) |
| p.switch_connect('admin', '', None) |
| |
| print "VLANs are:" |
| buf = p.vlan_get_list() |
| p.dump_list(buf) |
| |
| buf = p.vlan_get_name(1) |
| print "VLAN 1 is named \"%s\"" % buf |
| |
| print "Create VLAN 3" |
| p.vlan_create(3) |
| |
| print "Create VLAN 4" |
| p.vlan_create(4) |
| |
| buf = p.vlan_get_name(3) |
| print "VLAN 3 is named \"%s\"" % buf |
| |
| print "Set name of VLAN 3 to test333" |
| p.vlan_set_name(3, "test333") |
| |
| buf = p.vlan_get_name(3) |
| print "VLAN 3 is named \"%s\"" % buf |
| |
| print "VLANs are:" |
| buf = p.vlan_get_list() |
| p.dump_list(buf) |
| |
| print "Destroy VLAN 3" |
| p.vlan_destroy(3) |
| |
| print "VLANs are:" |
| buf = p.vlan_get_list() |
| p.dump_list(buf) |
| |
| buf = p.port_get_mode("dm10") |
| print "Port dm10 is in %s mode" % buf |
| |
| buf = p.port_get_mode("dm11") |
| print "Port dm11 is in %s mode" % buf |
| |
| # Test access stuff |
| print "Set dm09 to access mode" |
| p.port_set_mode("dm09", "access") |
| |
| print "Move dm9 to VLAN 4" |
| p.port_set_access_vlan("dm09", 4) |
| |
| buf = p.port_get_access_vlan("dm09") |
| print "Read from switch: dm09 is on VLAN %s" % buf |
| |
| print "Move dm09 back to VLAN 1" |
| p.port_set_access_vlan("dm09", 1) |
| |
| print "Create VLAN 2" |
| p.vlan_create(2) |
| |
| print "Create VLAN 3" |
| p.vlan_create(3) |
| |
| print "Create VLAN 4" |
| p.vlan_create(4) |
| |
| # Test access stuff |
| print "Set dm09 to trunk mode" |
| p.port_set_mode("dm09", "trunk") |
| print "Read from switch: which VLANs is dm09 on?" |
| buf = p.port_get_trunk_vlan_list("dm09") |
| p.dump_list(buf) |
| |
| # The adds below are NOOPs in effect on this switch - no filtering |
| # for "trunk" ports |
| print "Add dm09 to VLAN 2" |
| p.port_add_trunk_to_vlan("dm09", 2) |
| print "Add dm09 to VLAN 3" |
| p.port_add_trunk_to_vlan("dm09", 3) |
| print "Add dm09 to VLAN 4" |
| p.port_add_trunk_to_vlan("dm09", 4) |
| print "Read from switch: which VLANs is dm09 on?" |
| buf = p.port_get_trunk_vlan_list("dm09") |
| p.dump_list(buf) |
| |
| # And the same for removals here |
| p.port_remove_trunk_from_vlan("dm09", 3) |
| p.port_remove_trunk_from_vlan("dm09", 2) |
| p.port_remove_trunk_from_vlan("dm09", 4) |
| print "Read from switch: which VLANs is dm09 on?" |
| buf = p.port_get_trunk_vlan_list("dm09") |
| p.dump_list(buf) |
| |
| print 'Restarting switch, to explicitly reset config' |
| p.switch_restart() |
| |
| p.switch_save_running_config() |
| |
| p.switch_disconnect() |