class NetDevicesTest(unittest.TestCase):

    def setUp(self):
        self.nd = NetDevices()
        self.nodename = self.nd.keys()[0]
        self.nodeobj = self.nd.values()[0]

    def testBasics(self):
        """Basic test of NetDevices functionality."""
        self.assertEqual(len(self.nd), 1)
        self.assertEqual(self.nodeobj.nodeName, self.nodename)
        self.assertEqual(self.nodeobj.manufacturer, 'JUNIPER')

    def testAclsdb(self):
        """Test acls.db handling."""
        self.assert_('181j' in self.nodeobj.acls)

    def testAutoacls(self):
        """Test autoacls.py handling."""
        self.assert_('115j' in self.nodeobj.acls)

    def testFind(self):
        """Test the find() method."""
        self.assertEqual(self.nd.find(self.nodename), self.nodeobj)
        nodebasename = self.nodename[:self.nodename.index('.')]
        self.assertEqual(self.nd.find(nodebasename), self.nodeobj)
        self.assertRaises(KeyError, lambda: self.nd.find(self.nodename[0:3]))

    def __init__(self, devices=None, commands=None, creds=None,
                 incremental=None, max_conns=10, verbose=False,
                 timeout=DEFAULT_TIMEOUT, production_only=True,
                 allow_fallback=True, with_errors=True, force_cli=False,
                 with_acls=False, command_interval=0):
        if devices is None:
            raise exceptions.ImproperlyConfigured('You must specify some `devices` to interact with!')

        self.devices = devices
        self.commands = self.commands or (commands or []) # Always fallback to []
        self.creds = creds
        self.incremental = incremental
        self.max_conns = max_conns
        self.verbose = verbose
        self.timeout = timeout if timeout != self.timeout else self.timeout
        self.nd = NetDevices(production_only=production_only, with_acls=with_acls)
        self.allow_fallback = allow_fallback
        self.with_errors = with_errors
        self.force_cli = force_cli
        self.command_interval = command_interval
        self.curr_conns = 0
        self.jobs = []

        # Always fallback to {} for these
        self.errors = self.errors if self.errors is not None else {}
        self.results = self.results if self.results is not None else {}

        #self.deferrals = []
        self.supported_platforms = self._validate_platforms()
        self._setup_jobs()

class CheckNetDevices(unittest.TestCase):
    def setUp(self):
        self.router = NetDevices()['test1-abc.net.aol.com']
        self.when = datetime(2006, 7, 24, 20, tzinfo=UTC)

    def testNetDevicesBounce(self):
        """Test integration of bounce windows with NetDevices."""
        self.assertEquals(self.router.bounce.status(self.when), 'red')

    def testAllowability(self):
        """Test allowability checks."""
        self.failIf(self.router.allowable('load-acl', self.when))
        morning = datetime(2006, 7, 25, 9, tzinfo=UTC)        # 5 am EDT
        self.assert_(self.router.allowable('load-acl', morning))
        self.assertEquals(self.router.next_ok('load-acl', self.when), morning)
        self.assertEquals(self.router.next_ok('load-acl', morning), morning)

def get_bulk_acls():
    """
    Returns a dict of acls with an applied count over settings.AUTOLOAD_BULK_THRESH
    """
    from trigger.netdevices import NetDevices
    nd = NetDevices()
    all_acls = defaultdict(int)
    for dev in nd.all():
        for acl in dev.acls:
            all_acls[acl] += 1

    bulk_acls = {}
    for acl, count in all_acls.items():
        if count >= settings.AUTOLOAD_BULK_THRESH and acl != '':
            bulk_acls[acl] = count

    return bulk_acls

 def setUp(self):
     self.nd = NetDevices()
     _setup_aclsdb(self.nd)
     self.q = queue.Queue(verbose=False)
     self.acl = ACL_NAME
     self.acl_list = [self.acl]
     self.device = self.nd.find(DEVICE_NAME)
     self.device_name = DEVICE_NAME
     self.device_list = [self.device_name]
     self.user = USERNAME

class TestNetDevicesWithoutAcls(unittest.TestCase):
    """
    Test NetDevices with ``settings.WITH_ACLs`` set to ``False``.
    """
    def setUp(self):
        self.nd = NetDevices(with_acls=False)
        self.nodename = self.nd.keys()[0]
        self.device = self.nd.values()[0]

    def test_aclsdb(self):
        """Test acls.db handling."""
        self.assertFalse('test1-abc-only' in self.device.explicit_acls)

    def test_autoacls(self):
        """Test autoacls.py handling."""
        expected = set()
        self.assertEqual(expected, self.device.implicit_acls)

    def tearDown(self):
        _reset_netdevices()

 def __init__(self, devices=None, max_conns=10, verbose=False, timeout=30,
              production_only=True):
     self.curr_connections = 0
     self.reactor_running  = False
     self.devices = devices or []
     self.verbose = verbose
     self.max_conns = max_conns
     self.nd = NetDevices(production_only=production_only)
     self.jobs = []
     self.errors = {}
     self.data = {}
     self.deferrals = self._setup_jobs()
     self.timeout = timeout # in seconds

class TestAclsDB(unittest.TestCase):
    def setUp(self):
        self.nd = NetDevices()
        self.acl = ACL_NAME
        self.device = self.nd.find(DEVICE_NAME)
        self.implicit_acls = set(['115j', 'router-protect.core'])

    def test_01_add_acl_success(self):
        """Test associate ACL to device success"""
        exp = 'added acl %s to %s' % (self.acl, self.device)
        self.assertEqual(exp, adb.add_acl(self.device, self.acl))

    def test_02_add_acl_failure(self):
        """Test associate ACL to device failure"""
        exp = exceptions.ACLSetError
        self.assertRaises(exp, adb.add_acl, self.device, self.acl)

    def test_03_remove_acl_success(self):
        """Test remove ACL from device success"""
        exp = 'removed acl %s from %s' % (self.acl, self.device)
        self.assertEqual(exp, adb.remove_acl(self.device, self.acl))

    def test_04_remove_acl_failure(self):
        """Test remove ACL from device failure"""
        exp = exceptions.ACLSetError
        self.assertRaises(exp, adb.remove_acl, self.device, self.acl)

    def test_05_get_acl_dict(self):
        """Test get dict of associations"""
        exp = {'all': self.implicit_acls, 'explicit': set(),
               'implicit': self.implicit_acls}
        self.assertEqual(exp, adb.get_acl_dict(self.device))

    def test_06_get_acl_set_success(self):
        """Test get set of associations success"""
        exp = self.implicit_acls
        self.assertEqual(exp, adb.get_acl_set(self.device))

    def test_07_get_acl_set_failure(self):
        """Test get set of associations failure"""
        exp = exceptions.InvalidACLSet
        acl_set = 'bogus'
        self.assertRaises(exp, adb.get_acl_set, self.device, acl_set)

    def tearDown(self):
        NetDevices._Singleton = None

    def __init__(self, devices=None, commands=None, incremental=None,
                 max_conns=10, verbose=False, timeout=30,
                 production_only=True, allow_fallback=True):
        if devices is None:
            raise exceptions.ImproperlyConfigured('You must specify some ``devices`` to interact with!')

        self.devices = devices
        self.commands = self.commands or (commands or []) # Always fallback to []
        self.incremental = incremental
        self.max_conns = max_conns
        self.verbose = verbose
        self.timeout = timeout # in seconds
        self.nd = NetDevices(production_only=production_only)
        self.allow_fallback = allow_fallback
        self.curr_conns = 0
        self.jobs = []
        self.errors = {}
        self.results = {}
        self.deferrals = self._setup_jobs()
        self.supported_platforms = self._validate_platforms()

class NetDevicesTest(unittest.TestCase):
    def setUp(self):
        self.nd = NetDevices(with_acls=False)
        print self.nd.values()
        self.nodename = self.nd.keys()[0]
        self.nodeobj = self.nd.values()[0]

    def testBasics(self):
        """Basic test of NetDevices functionality."""
        self.assertEqual(len(self.nd), 3)
        self.assertEqual(self.nodeobj.nodeName, self.nodename)
        self.assertEqual(self.nodeobj.manufacturer, "JUNIPER")

    def testFind(self):
        """Test the find() method."""
        self.assertEqual(self.nd.find(self.nodename), self.nodeobj)
        nodebasename = self.nodename[: self.nodename.index(".")]
        self.assertEqual(self.nd.find(nodebasename), self.nodeobj)
        self.assertRaises(KeyError, lambda: self.nd.find(self.nodename[0:3]))

#.........这里部分代码省略.........
    # Defaults to all supported platforms
    platforms = settings.SUPPORTED_PLATFORMS

    # The commands to run (defaults to [])
    commands = None

    # The timeout for commands to return results. We are setting this to 0
    # so that if it's not overloaded in a subclass, the timeout value passed to
    # the constructor will be preferred, especially if it is set to ``None``
    # which Twisted uses to disable timeouts completely.
    timeout = 0

    # How results are stored (defaults to {})
    results = None

    # How errors are stored (defaults to {})
    errors = None

    def __init__(self, devices=None, commands=None, creds=None,
                 incremental=None, max_conns=10, verbose=False,
                 timeout=DEFAULT_TIMEOUT, production_only=True,
                 allow_fallback=True, with_errors=True, force_cli=False,
                 with_acls=False, command_interval=0):
        if devices is None:
            raise exceptions.ImproperlyConfigured('You must specify some `devices` to interact with!')

        self.devices = devices
        self.commands = self.commands or (commands or []) # Always fallback to []
        self.creds = creds
        self.incremental = incremental
        self.max_conns = max_conns
        self.verbose = verbose
        self.timeout = timeout if timeout != self.timeout else self.timeout
        self.nd = NetDevices(production_only=production_only, with_acls=with_acls)
        self.allow_fallback = allow_fallback
        self.with_errors = with_errors
        self.force_cli = force_cli
        self.command_interval = command_interval
        self.curr_conns = 0
        self.jobs = []

        # Always fallback to {} for these
        self.errors = self.errors if self.errors is not None else {}
        self.results = self.results if self.results is not None else {}

        #self.deferrals = []
        self.supported_platforms = self._validate_platforms()
        self._setup_jobs()

    def _validate_platforms(self):
        """
        Determine the set of supported platforms for this instance by making
        sure the specified vendors/platforms for the class match up.
        """
        supported_platforms = {}
        for vendor in self.vendors:
            if vendor in self.platforms:
                types = self.platforms[vendor]
                if not types:
                    raise exceptions.MissingPlatform('No platforms specified for %r' % vendor)
                else:
                    #self.supported_platforms[vendor] = types
                    supported_platforms[vendor] = types
            else:
                raise exceptions.ImproperlyConfigured('Platforms for vendor %r not found. Please provide it at either the class level or using the arguments.' % vendor)

 def setUp(self):
     self.nd = NetDevices()
     self.acl = ACL_NAME
     self.device = self.nd.find(DEVICE_NAME)
     self.implicit_acls = set(['115j', 'router-protect.core'])

class Commando(object):
    """
    I run commands on devices but am not much use unless you subclass me and
    configure vendor-specific parse/generate methods.
    """
    def __init__(self, devices=None, max_conns=10, verbose=False, timeout=30,
                 production_only=True):
        self.curr_connections = 0
        self.reactor_running  = False
        self.devices = devices or []
        self.verbose = verbose
        self.max_conns = max_conns
        self.nd = NetDevices(production_only=production_only)
        self.jobs = []
        self.errors = {}
        self.data = {}
        self.deferrals = self._setup_jobs()
        self.timeout = timeout # in seconds

    def _decrement_connections(self, data):
        """
        Self-explanatory. Called by _add_worker() as both callback/errback
        so we can accurately refill the jobs queue, which relies on the
        current connection count.
        """
        self.curr_connections -= 1
        return True

    def set_data(self, device, data):
        """
        Another method for storing results. If you'd rather just change the
        default method for storing results, overload this. All default
        parse/generate methods call this."""
        self.data[device] = data
        return True

    #=======================================
    # Vendor-specific parse/generate methods
    #=======================================

    def _normalize_manufacturer(self, manufacturer):
        """Normalize the manufacturer name into a method"""
        return manufacturer.replace(' ', '_').lower()

    def _lookup(self, device, prefix):
        """Base lookup method."""
        manuf = self._normalize_manufacturer(device.manufacturer)
        try:
            func = getattr(self, prefix + manuf)
        except AttributeError:
            return 'base prefix' + prefix  (device)
            #return self._base_generate_cmd(device)

        return func(device)

    def _parse_lookup(self, device):
        """Base parse method."""
        manuf = self._normalize_manufacturer
        try:
            func = getattr(self, 'parse_' + manuf)
        except AttributeError:
            return self._base_generate_cmd(device)

        return func(device)

    # Yes there is probably a better way to do this in the long-run instead of
    # individual parse/generate methods for each vendor, but this works for now.
    def _base_parse(self, data, device):
        """
        Parse output from a device. Overload this to customize this default
        behavior.
        """
        self.set_data(device, data)
        return True

    def _base_generate_cmd(self, dev=None):
        """
        Generate commands to be run on a device. If you don't overload this, it
        returns an empty list.
        """
        return []

    # TODO (jathan): Find a way to dynamically generate/call these methods
    # TODO (jathan): Methods should be prefixed with their action, not vendor

    # IOS (Cisco)
    ios_parse = _base_parse
    generate_ios_cmd = _base_generate_cmd

    # Brocade
    brocade_parse = _base_parse
    generate_brocade_cmd = _base_generate_cmd

    # Foundry
    foundry_parse = _base_parse
    generate_foundry_cmd = _base_generate_cmd

    # Juniper (JUNOS)
    junos_parse = _base_parse
    generate_junos_cmd = _base_generate_cmd
#.........这里部分代码省略.........

 def setUp(self):
     self.router = NetDevices()['iwg1-r3.router.aol.com']
     self.when = datetime(2006, 7, 24, 20, tzinfo=UTC)

 def setUp(self):
     self.nd = NetDevices()
     self.nodename = self.nd.keys()[0]
     self.device = self.nd.values()[0]
     self.device.explicit_acls = set(['test1-abc-only'])

__all__ = ['mock_redis']

# misc
from . import misc
from misc import *
__all__.extend(misc.__all__)

if __name__ == '__main__':
    os.environ['NETDEVICES_SOURCE'] = 'data/netdevices.xml'

    mock_redis.install()
    import redis
    from trigger.netdevices import NetDevices
    from trigger.acl.db import AclsDB

    r = redis.Redis()
    a = AclsDB()
    nd = NetDevices()

    dev = nd.find('test1-abc')

    print r.keys('*')
    print a.add_acl(dev, 'bacon')
    print r.keys('*')

    _k = 'acls:explicit:'
    key = _k + dev.nodeName

    print r.smembers(key)

 def setUp(self):
     self.nd = NetDevices()
     self.device = self.nd[DEVICE_NAME]
     self.device2 = self.nd[DEVICE2_NAME]
     self.nodename = self.device.nodeName
     self.device.explicit_acls = set(['test1-abc-only'])

class TestNetDeviceObject(unittest.TestCase):
    """
    Test NetDevice object methods.
    """
    def setUp(self):
        self.nd = NetDevices()
        self.nodename = self.nd.keys()[0]
        self.device = self.nd.values()[0]

    def test_stringify(self):
        """Test casting NetDevice to string"""
        expected = DEVICE_NAME
        self.assertEqual(expected, str(self.device))

    def test_bounce(self):
        """Test .bounce property"""
        expected = changemgmt.BounceWindow
        self.assertTrue(isinstance(self.device.bounce, expected))

    def test_shortName(self):
        """Test .shortName property"""
        expected = self.nodename.split('.', 1)[0]
        self.assertEqual(expected, self.device.shortName)

    def test_allowable(self):
        """Test allowable() method"""
        # This is already tested in test_changemgmt.py, so this is a stub.
        pass

    def test_next_ok(self):
        """Test next_ok() method"""
        # This is already tested in test_changemgmt.py, so this is a stub.
        pass

    def test_identity(self):
        """Exercise NetDevice identity tests."""
        # It's a router...
        self.assertTrue(self.device.is_router())
        # And therefore none of these other things...
        self.assertFalse(self.device.is_switch())
        self.assertFalse(self.device.is_firewall())
        self.assertFalse(self.device.is_netscaler())
        self.assertFalse(self.device.is_netscreen())
        self.assertFalse(self.device.is_ioslike())
        self.assertFalse(self.device.is_brocade_vdx())

    def test_hash_ssh(self):
        """Exercise NetDevice ssh test."""
        # TODO (jathan): Mock SSH connections so we can test actual connectivity
        # Device won't be reachable, so this should always fail
        self.assertFalse(self.device.has_ssh())
        # Since there's no SSH, no aync
        self.assertFalse(self.device.can_ssh_pty())

    def test_reachability(self):
        """Exercise NetDevice ssh test."""
        # TODO (jathan): Mock SSH connections so we can test actual connectivity
        self.assertFalse(self.device.is_reachable())

    def test_dump(self):
        """Test the dump() method."""
        with captured_output() as (out, err):
            self.device.dump()
        expected = NETDEVICE_DUMP_EXPECTED
        output = out.getvalue()
        self.assertEqual(expected, output)

    def tearDown(self):
        _reset_netdevices()

class TestAclQueue(unittest.TestCase):
    def setUp(self):
        self.nd = NetDevices()
        _setup_aclsdb(self.nd)
        self.q = queue.Queue(verbose=False)
        self.acl = ACL_NAME
        self.acl_list = [self.acl]
        self.device = self.nd.find(DEVICE_NAME)
        self.device_name = DEVICE_NAME
        self.device_list = [self.device_name]
        self.user = USERNAME

    #
    # Integrated queue tests
    #

    def test_01_insert_integrated_success(self):
        """Test insert success into integrated queue"""
        self.assertTrue(self.q.insert(self.acl, self.device_list) is None)

    def test_02_insert_integrated_failure_device(self):
        """Test insert invalid device"""
        self.assertRaises(exceptions.TriggerError, self.q.insert, self.acl, ['bogus'])

    def test_03_insert_integrated_failure_acl(self):
        """Test insert devices w/ no ACL association"""
        self.assertRaises(exceptions.TriggerError, self.q.insert, 'bogus',
                          self.device_list)

    def test_04_list_integrated_success(self):
        """Test listing integrated queue"""
        self.q.insert(self.acl, self.device_list)
        expected = [(u'test1-abc.net.aol.com', u'foo')]
        self.assertEqual(sorted(expected), sorted(self.q.list()))

    def test_05_complete_integrated(self):
        """Test mark task complete"""
        self.q.complete(self.device_name, self.acl_list)
        expected = []
        self.assertEqual(sorted(expected), sorted(self.q.list()))

    def test_06_delete_integrated_with_devices(self):
        """Test delete ACL from queue providing devices"""
        self.q.insert(self.acl, self.device_list)
        self.assertTrue(self.q.delete(self.acl, self.device_list))

    def test_07_delete_integrated_no_devices(self):
        """Test delete ACL from queue without providing devices"""
        self.q.insert(self.acl, self.device_list)
        self.assertTrue(self.q.delete(self.acl))

    def test_08_remove_integrated_success(self):
        """Test remove (set as loaded) ACL from integrated queue"""
        self.q.insert(self.acl, self.device_list)
        self.q.remove(self.acl, self.device_list)
        expected = []
        self.assertEqual(sorted(expected), sorted(self.q.list()))

    def test_10_remove_integrated_failure(self):
        """Test remove (set as loaded) failure"""
        self.assertRaises(exceptions.ACLQueueError, self.q.remove, '', self.device_list)

    #
    # Manual queue tests
    #

    def test_11_insert_manual_success(self):
        """Test insert success into manual queue"""
        self.assertTrue(self.q.insert('manual task', None) is None)

    def test_12_list_manual_success(self):
        """Test list success of manual queue"""
        self.q.insert('manual task', None)
        expected = ('manual task', self.user)
        result = self.q.list('manual')
        actual = result[0][:2] # First tuple, items 0-1
        self.assertEqual(sorted(expected), sorted(actual))

    def test_13_delete_manual_success(self):
        """Test delete from manual queue"""
        self.q.delete('manual task')
        expected = []
        self.assertEqual(sorted(expected), sorted(self.q.list('manual')))

    #
    # Generic tests
    #

    def test_14_delete_failure(self):
        """Test delete of task not in queue"""
        self.assertFalse(self.q.delete('bogus'))

    def test_15_list_invalid(self):
        """Test list of invalid queue name"""
        self.assertFalse(self.q.list('bogus'))

    # Teardown

    def test_ZZ_cleanup_db(self):
        """Cleanup the temp database file"""
#.........这里部分代码省略.........

class TestNetDevicesWithAcls(unittest.TestCase):
    """
    Test NetDevices with ``settings.WITH_ACLs set`` to ``True``.
    """
    def setUp(self):
        self.nd = NetDevices()
        self.nodename = self.nd.keys()[0]
        self.device = self.nd.values()[0]
        self.device.explicit_acls = set(['test1-abc-only'])

    def test_basics(self):
        """Basic test of NetDevices functionality."""
        self.assertEqual(len(self.nd), 1)
        self.assertEqual(self.device.nodeName, self.nodename)
        self.assertEqual(self.device.manufacturer, 'JUNIPER')

    def test_aclsdb(self):
        """Test acls.db handling."""
        self.assertTrue('test1-abc-only' in self.device.explicit_acls)

    def test_autoacls(self):
        """Test autoacls.py handling."""
        self.assertTrue('router-protect.core' in self.device.implicit_acls)

    def test_find(self):
        """Test the find() method."""
        self.assertEqual(self.nd.find(self.nodename), self.device)
        nodebasename = self.nodename[:self.nodename.index('.')]
        self.assertEqual(self.nd.find(nodebasename), self.device)
        self.assertRaises(KeyError, lambda: self.nd.find(self.nodename[0:3]))

    def test_all(self):
        """Test the all() method."""
        expected = [self.device]
        self.assertEqual(expected, self.nd.all())

    def test_search(self):
        """Test the search() method."""
        expected = [self.device]
        self.assertEqual(expected, self.nd.search(self.nodename))
        self.assertEqual(expected, self.nd.search('17', field='onCallID'))
        self.assertEqual(expected, self.nd.search('juniper', field='vendor'))

    def test_match(self):
        """Test the match() method."""
        expected = [self.device]
        self.assertEqual(expected, self.nd.match(nodename=self.nodename))
        self.assertEqual(expected, self.nd.match(vendor='juniper'))
        self.assertNotEqual(expected, self.nd.match(vendor='cisco'))

    def tearDown(self):
        NetDevices._Singleton = None