path: root/drivers/Dummy.py

#! /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()