# Copyright (c) 2012, Calxeda Inc. # # All rights reserved. # # Redistribution and use in source and binary forms, with or without # modification, are permitted provided that the following conditions are # met: # # * Redistributions of source code must retain the above copyright # notice, this list of conditions and the following disclaimer. # * Redistributions in binary form must reproduce the above copyright # notice, this list of conditions and the following disclaimer in the # documentation and/or other materials provided with the distribution. # * Neither the name of Calxeda Inc. nor the names of its contributors # may be used to endorse or promote products derived from this software # without specific prior written permission. # # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS # FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE # COPYRIGHT HOLDERS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, # INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, # BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS # OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND # ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR # TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF # THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH # DAMAGE. import random import time import unittest from cxmanage_api.fabric import Fabric from cxmanage_api.tftp import InternalTftp, ExternalTftp from cxmanage_api.firmware_package import FirmwarePackage from cxmanage_api.ubootenv import UbootEnv from cxmanage_api.cx_exceptions import CommandFailedError from cxmanage_test import TestSensor from cxmanage_test.node_test import DummyBMC from pyipmi import make_bmc NUM_NODES = 128 ADDRESSES = ["192.168.100.%i" % x for x in range(1, NUM_NODES + 1)] class FabricTest(unittest.TestCase): """ Test the various Fabric commands """ def setUp(self): # Set up the controller and add targets self.fabric = Fabric("192.168.100.1", node=DummyNode) self.nodes = [DummyNode(x) for x in ADDRESSES] self.fabric._nodes = dict((i, self.nodes[i]) for i in xrange(NUM_NODES)) def test_tftp(self): """ Test the tftp property """ tftp = InternalTftp() self.fabric.tftp = tftp self.assertTrue(self.fabric.tftp is tftp) for node in self.nodes: self.assertTrue(node.tftp is tftp) tftp = ExternalTftp("127.0.0.1") self.fabric.tftp = tftp self.assertTrue(self.fabric.tftp is tftp) for node in self.nodes: self.assertTrue(node.tftp is tftp) def test_get_mac_addresses(self): """ Test get_mac_addresses command """ self.fabric.get_mac_addresses() self.assertEqual(self.nodes[0].executed, ["get_fabric_macaddrs"]) for node in self.nodes[1:]: self.assertEqual(node.executed, []) def test_get_uplink_info(self): """ Test get_mac_addresses command """ self.fabric.get_uplink_info() self.assertEqual(self.nodes[0].executed, ["get_fabric_uplink_info"]) for node in self.nodes[1:]: self.assertEqual(node.executed, []) def test_get_uplink(self): """ Test get_uplink command """ self.assertEqual(self.fabric.get_uplink(iface=0), 0) def test_set_uplink(self): """ Test set_uplink command """ iface, uplink = 0, 0 self.fabric.set_uplink(iface=iface, uplink=uplink) self.assertEqual( [("fabric_config_set_uplink", iface, uplink)], self.nodes[0].bmc.executed ) def test_get_sensors(self): """ Test get_sensors command """ self.fabric.get_sensors() self.fabric.get_sensors("Node Power") for node in self.nodes: self.assertEqual(node.executed, ["get_sensors", "get_sensors"]) def test_get_firmware_info(self): """ Test get_firmware_info command """ self.fabric.get_firmware_info() for node in self.nodes: self.assertEqual(node.executed, ["get_firmware_info"]) def test_is_updatable(self): """ Test is_updatable command """ package = FirmwarePackage() self.fabric.is_updatable(package) for node in self.nodes: self.assertEqual(node.executed, [("is_updatable", package)]) def test_update_firmware(self): """ Test update_firmware command """ package = FirmwarePackage() self.fabric.update_firmware(package) for node in self.nodes: self.assertEqual(node.executed, [("update_firmware", package)]) def test_config_reset(self): """ Test config_reset command """ self.fabric.config_reset() for node in self.nodes: self.assertEqual(node.executed, ["config_reset"]) def test_set_boot_order(self): """ Test set_boot_order command """ boot_args = "disk0,pxe,retry" self.fabric.set_boot_order(boot_args) for node in self.nodes: self.assertEqual(node.executed, [("set_boot_order", boot_args)]) def test_get_boot_order(self): """ Test get_boot_order command """ self.fabric.get_boot_order() for node in self.nodes: self.assertEqual(node.executed, ["get_boot_order"]) def test_get_versions(self): """ Test get_versions command """ self.fabric.get_versions() for node in self.nodes: self.assertEqual(node.executed, ["get_versions"]) def test_get_ubootenv(self): """ Test get_ubootenv command """ self.fabric.get_ubootenv() for node in self.nodes: self.assertEqual(node.executed, ["get_ubootenv"]) def test_ipmitool_command(self): """ Test ipmitool_command command """ ipmitool_args = "power status" self.fabric.ipmitool_command(ipmitool_args) for node in self.nodes: self.assertEqual(node.executed, [("ipmitool_command", ipmitool_args)]) def test_get_server_ip(self): """ Test get_server_ip command """ self.fabric.get_server_ip("interface", "ipv6", "user", "password", "aggressive") for node in self.nodes: self.assertEqual(node.executed, [("get_server_ip", "interface", "ipv6", "user", "password", "aggressive")]) def test_failed_command(self): """ Test a failed command """ fail_nodes = [DummyFailNode(x) for x in ADDRESSES] self.fabric._nodes = dict((i, fail_nodes[i]) for i in xrange(NUM_NODES)) try: self.fabric.get_power() self.fail() except CommandFailedError: for node in fail_nodes: self.assertEqual(node.executed, ["get_power"]) def test_primary_node(self): """Test the primary_node property Currently it should always return node 0. """ self.assertEqual(self.fabric.primary_node, self.nodes[0]) def test_get_ipsrc(self): """Test the get_ipsrc method """ self.fabric.get_ipsrc() bmc = self.fabric.primary_node.bmc self.assertIn('fabric_config_get_ip_src', bmc.executed) def test_set_ipsrc(self): """Test the set_ipsrc method""" ipsrc = random.randint(1, 2) self.fabric.set_ipsrc(ipsrc) bmc = self.fabric.primary_node.bmc self.assertIn('fabric_config_set_ip_src', bmc.executed) # fabric_ipsrc is just part of DummyBMC - not a real bmc attribute # it's there to make sure the ipsrc_mode value gets passed to the bmc. self.assertEqual(bmc.fabric_ipsrc, ipsrc) def test_apply_factory_default_config(self): """Test the apply_factory_default_config method""" self.fabric.apply_factory_default_config() bmc = self.fabric.primary_node.bmc self.assertIn('fabric_config_factory_default', bmc.executed) def test_get_ipaddr_base(self): """Test the get_ipaddr_base method""" ipaddr_base = self.fabric.get_ipaddr_base() bmc = self.fabric.primary_node.bmc self.assertIn('fabric_config_get_ip_addr_base', bmc.executed) self.assertEqual(bmc.ipaddr_base, ipaddr_base) def test_update_config(self): """Test the update_config method """ self.fabric.update_config() bmc = self.fabric.primary_node.bmc self.assertIn('fabric_config_update_config', bmc.executed) def test_get_linkspeed(self): """Test the get_linkspeed method """ self.fabric.get_linkspeed() bmc = self.fabric.primary_node.bmc self.assertIn('fabric_config_get_linkspeed', bmc.executed) def test_set_linkspeed(self): """Test the set_linkspeed method""" valid_linkspeeds = [1, 2.5, 5, 7.5, 10] linkspeed = random.choice(valid_linkspeeds) self.fabric.set_linkspeed(linkspeed) bmc = self.fabric.primary_node.bmc self.assertIn('fabric_config_set_linkspeed', bmc.executed) # fabric_linkspeed is just part of DummyBMC - not a real bmc attribute # it's there to make sure the ipsrc_mode value gets passed to the bmc. self.assertEqual(bmc.fabric_linkspeed, linkspeed) def test_get_linkspeed_policy(self): """Test the get_linkspeed_policy method """ self.fabric.get_linkspeed_policy() bmc = self.fabric.primary_node.bmc self.assertIn('fabric_config_get_linkspeed_policy', bmc.executed) def test_set_linkspeed_policy(self): """Test the set_linkspeed_policy method""" ls_policy = random.randint(0, 1) self.fabric.set_linkspeed_policy(ls_policy) bmc = self.fabric.primary_node.bmc self.assertIn('fabric_config_set_linkspeed_policy', bmc.executed) # fabric_ls_policy is just part of DummyBMC - not a real bmc attribute # it's there to make sure the ipsrc_mode value gets passed to the bmc. self.assertEqual(bmc.fabric_ls_policy, ls_policy) def test_get_link_stats(self): """Test the get_link_stats() method.""" for i in range(0, 5): stats = self.fabric.get_link_stats(i) for nn, node in self.fabric.nodes.items(): self.assertIn(('get_link_stats', i), node.executed) def test_get_linkmap(self): """Test the get_linkmap method""" maps = self.fabric.get_linkmap() for nn, node in self.fabric.nodes.items(): self.assertIn('get_linkmap', node.executed) def test_get_routing_table(self): """Test the get_routing_table method""" maps = self.fabric.get_routing_table() for nn, node in self.fabric.nodes.items(): self.assertIn('get_routing_table', node.executed) def test_get_depth_chart(self): """Test the depth_chart method""" maps = self.fabric.get_depth_chart() for nn, node in self.fabric.nodes.items(): self.assertIn('get_depth_chart', node.executed) def test_get_link_users_factor(self): """Test the get_link_users_factor method """ self.fabric.get_link_users_factor() bmc = self.fabric.primary_node.bmc self.assertIn('fabric_config_get_link_users_factor', bmc.executed) def test_set_link_users_factor(self): """Test the set_link_users_factor method""" lu_factor = random.randint(5, 50) self.fabric.set_link_users_factor(lu_factor) bmc = self.fabric.primary_node.bmc self.assertIn('fabric_config_set_link_users_factor', bmc.executed) # fabric_lu_factor is just part of DummyBMC - not a real bmc attribute # it's there to make sure the ipsrc_mode value gets passed to the bmc. self.assertEqual(bmc.fabric_lu_factor, lu_factor) def test_add_macaddr (self): """Test the add_macaddr method""" valid_nodeids = [0, 1, 2, 3] t_nodeid = random.choice(valid_nodeids) valid_ifaces = [0, 1, 2] t_iface = random.choice(valid_ifaces) t_macaddr = "66:55:44:33:22:11" self.fabric.add_macaddr (t_nodeid, t_iface, t_macaddr) bmc = self.fabric.primary_node.bmc self.assertIn ('fabric_add_macaddr', bmc.executed) def test_rm_macaddr (self): """Test the rm_macaddr method""" valid_nodeids = [0, 1, 2, 3] t_nodeid = random.choice(valid_nodeids) valid_ifaces = [0, 1, 2] t_iface = random.choice(valid_ifaces) t_macaddr = "66:55:44:33:22:11" self.fabric.rm_macaddr (t_nodeid, t_iface, t_macaddr) bmc = self.fabric.primary_node.bmc self.assertIn ('fabric_rm_macaddr', bmc.executed) class DummyNode(object): """ Dummy node for the nodemanager tests """ def __init__(self, ip_address, username="admin", password="admin", tftp=None, *args, **kwargs): self.executed = [] self.ip_address = ip_address self.tftp = tftp self.bmc = make_bmc(DummyBMC, hostname=ip_address, username=username, password=password, verbose=False) def get_power(self): self.executed.append("get_power") return False def set_power(self, mode): self.executed.append(("set_power", mode)) def get_power_policy(self): self.executed.append("get_power_policy") return "always-off" def set_power_policy(self, mode): self.executed.append(("set_power_policy", mode)) def mc_reset(self): self.executed.append("mc_reset") def get_firmware_info(self): self.executed.append("get_firmware_info") def is_updatable(self, package, partition_arg="INACTIVE", priority=None): self.executed.append(("is_updatable", package)) def update_firmware(self, package, partition_arg="INACTIVE", priority=None): self.executed.append(("update_firmware", package)) time.sleep(random.randint(0, 2)) def get_sensors(self, name=""): self.executed.append("get_sensors") power_value = "%f (+/- 0) Watts" % random.uniform(0, 10) temp_value = "%f (+/- 0) degrees C" % random.uniform(30, 50) sensors = [ TestSensor("Node Power", power_value), TestSensor("Board Temp", temp_value) ] return [x for x in sensors if name.lower() in x.sensor_name.lower()] def config_reset(self): self.executed.append("config_reset") def set_boot_order(self, boot_args): self.executed.append(("set_boot_order", boot_args)) def get_boot_order(self): self.executed.append("get_boot_order") return ["disk", "pxe"] def get_versions(self): self.executed.append("get_versions") class Result: def __init__(self): self.header = "Calxeda SoC (0x0096CD)" self.hardware_version = "TestBoard X00" self.firmware_version = "v0.0.0" self.ecme_version = "v0.0.0" self.ecme_timestamp = "0" self.a9boot_version = "v0.0.0" self.uboot_version = "v0.0.0" return Result() def ipmitool_command(self, ipmitool_args): self.executed.append(("ipmitool_command", ipmitool_args)) return "Dummy output" def get_ubootenv(self): self.executed.append("get_ubootenv") ubootenv = UbootEnv() ubootenv.variables["bootcmd0"] = "run bootcmd_default" ubootenv.variables["bootcmd_default"] = "run bootcmd_sata" return ubootenv def get_fabric_ipinfo(self): return {} def get_server_ip(self, interface, ipv6, user, password, aggressive): self.executed.append(("get_server_ip", interface, ipv6, user, password, aggressive)) return "192.168.200.1" def get_fabric_macaddrs(self): self.executed.append("get_fabric_macaddrs") result = {} for node in range(NUM_NODES): result[node] = {} for port in range(3): address = "00:00:00:00:%02x:%02x" % (node, port) result[node][port] = address return result def get_fabric_uplink_info(self): self.executed.append('get_fabric_uplink_info') results = {} for n in range(1, NUM_NODES): results[n] = {'eth0': 0, 'eth1': 0, 'mgmt': 0} return results def get_link_stats(self, link=0): self.executed.append(('get_link_stats', link)) return { 'FS_LC%s_BYTE_CNT_0' % link: '0x0', 'FS_LC%s_BYTE_CNT_1' % link: '0x0', 'FS_LC%s_CFG_0' % link: '0x1030107f', 'FS_LC%s_CFG_1' % link: '0x104f', 'FS_LC%s_CM_RXDATA_0' % link: '0x0', 'FS_LC%s_CM_RXDATA_1' % link: '0x0', 'FS_LC%s_CM_TXDATA_0' % link: '0x0', 'FS_LC%s_CM_TXDATA_1' % link: '0x0', 'FS_LC%s_PKT_CNT_0' % link: '0x0', 'FS_LC%s_PKT_CNT_1' % link: '0x0', 'FS_LC%s_RDRPSCNT' % link: '0x0', 'FS_LC%s_RERRSCNT' % link: '0x0', 'FS_LC%sRMCSCNT' % link: '0x0', 'FS_LC%s_RPKTSCNT' % link: '0x0', 'FS_LC%s_RUCSCNT' % link: '0x0', 'FS_LC%s_SC_STAT' % link: '0x0', 'FS_LC%s_STATE' % link: '0x1033', 'FS_LC%s_TDRPSCNT' % link: '0x0', 'FS_LC%s_TPKTSCNT' % link: '0x1' } def get_linkmap(self): self.executed.append('get_linkmap') results = {} for n in range(0, NUM_NODES): results[n] = {n: {1: 2, 3: 1, 4: 3}} return results def get_routing_table(self): self.executed.append('get_routing_table') results = {} for n in range(0, NUM_NODES): results[n] = {n: {1: [0, 0, 0, 3, 0], 2: [0, 3, 0, 0, 2], 3: [0, 2, 0, 0, 3]}} return results def get_depth_chart(self): self.executed.append('get_depth_chart') results = {} for n in range(0, NUM_NODES): results[n] = {n: {1: {'shortest': (0, 0)}, 2: {'hops': [(3, 1)], 'shortest': (0, 0)}, 3: {'hops': [(2, 1)], 'shortest': (0, 0)}}} return results def get_uplink(self, iface): self.executed.append(('get_uplink', iface)) return 0 def set_uplink(self, uplink, iface): self.executed.append(('set_uplink', uplink, iface)) class DummyFailNode(DummyNode): """ Dummy node that should fail on some commands """ class DummyFailError(Exception): pass def get_power(self): self.executed.append("get_power") raise DummyFailNode.DummyFailError class DummyImage: def __init__(self, filename, image_type, *args): self.filename = filename self.type = image_type