KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use test_hosts method in autotest

Best Python code snippet using autotest_python

test_action_bestfit.py

Source:test_action_bestfit.py

1from manage_iq_base_action_test_case import ManageIQBaseActionTestCase2from lib.bestfit import BestFit3from st2common.runners.base_action import Action4import mock5class BestFitTestCase(ManageIQBaseActionTestCase):6    __test__ = True7    action_cls = BestFit8    def test_init(self):9        action = self.get_action_instance({})10        self.assertIsInstance(action, BestFit)11        self.assertIsInstance(action, Action)12    def test_filter_datastores_regex_match(self):13        action = self.get_action_instance({})14        test_dict = {'datastoreFilterRegEx': {"filters": ["(?i)(iso)"]}}15        test_name = "dev_isos"16        expected_output = False17        result_output = action._filter_datastores(test_name, test_dict)18        self.assertEqual(expected_output, result_output)19    def test_filter_datastores_regex_no_match(self):20        action = self.get_action_instance({})21        test_dict = {'datastoreFilterRegEx': {"filters": ["(?i)(iso)"]}}22        test_name = "testdatastore"23        expected_output = True24        result_output = action._filter_datastores(test_name, test_dict)25        self.assertEqual(expected_output, result_output)26    def test_filter_datastores_name_match(self):27        action = self.get_action_instance({})28        test_dict = {'datastoreFilterRegEx': {"filters": ["dev_isos"]}}29        test_name = "dev_isos"30        expected_output = False31        result_output = action._filter_datastores(test_name, test_dict)32        self.assertEqual(expected_output, result_output)33    def test_filter_datastores_name_no_match(self):34        action = self.get_action_instance({})35        test_dict = {'datastoreFilterRegEx': {"filters": ["dev_isos"]}}36        test_name = "testdatastore"37        expected_output = True38        result_output = action._filter_datastores(test_name, test_dict)39        self.assertEqual(expected_output, result_output)40    def test_check_storages(self):41        action = self.get_action_instance({})42        test_dict = {'datastoreFilterRegEx': {"filters": ["dev_isos"]}}43        test_storages = [{'storage_id': 1}]44        expected_result = {'id': 1,45                           'name': "abc",46                           'free_space': 100}47        mock_datastore = mock.Mock(id=expected_result['id'],48                                   free_space=expected_result['free_space'])49        name_property = mock.PropertyMock(return_value=expected_result['name'])50        type(mock_datastore).name = name_property51        mock_client = mock.MagicMock()52        mock_client.collections.data_stores.return_value = mock_datastore53        result_name, result_id = action._check_storages(mock_client, test_storages, test_dict)54        self.assertEquals(result_name, expected_result['name'])55        self.assertEquals(result_id, str(expected_result['id']))56    def test_find_storage(self):57        action = self.get_action_instance({})58        test_datastore_name = "test"59        expected_result = {'id': 1,60                           'name': test_datastore_name,61                           'free_space': 100}62        mock_datastore = mock.Mock(id=expected_result['id'],63                                   free_space=expected_result['free_space'])64        name_property = mock.PropertyMock(return_value=expected_result['name'])65        type(mock_datastore).name = name_property66        mock_data_stores = mock.Mock(all=[mock_datastore])67        mock_collections = mock.Mock(data_stores=mock_data_stores)68        mock_client = mock.Mock(collections=mock_collections)69        result_name, result_id = action._find_storage(mock_client, test_datastore_name)70        self.assertEquals(result_name, expected_result['name'])71        self.assertEquals(result_id, str(expected_result['id']))72    @mock.patch("lib.bestfit.BestFit._find_storage")73    def test__load_disks(self, mock__find_storage):74        action = self.get_action_instance({})75        test_json = {"all_disks":76                    [{"size_gb": "35",77                    "uuid": "6000C29c-8b35-69bc-dcb6-efd847c579e7",78                    "key": "2000",79                    "datastore": "abc"}]}80        test_datastore = {'id': 1,81                          'name': "abc",82                          'free_space': 100}83        expected_result = (test_datastore['name'], str(test_datastore['id']))84        mock__find_storage.return_value = expected_result85        mock_client = mock.MagicMock()86        result = action._load_disks(mock_client, test_json)87        self.assertEqual(result, expected_result)88    def test_check_hosts(self):89        action = self.get_action_instance({})90        test_dict = {'clusterName': "test_cluster",91                     'datastoreFilterRegEx': {"filters": ["dev_isos"]}}92        test_storages = [{'storage_id': 1}]93        test_datastore = {'id': 1,94                          'name': "abc",95                          'free_space': 100}96        test_hosts = [{'v_owning_cluster': "test_cluster",97                       'v_total_vms': 5,98                       'id': 1,99                       'name': "test_host_1",100                       'host_storages': test_storages}]101        expected_result = (True, {'clusterName': test_dict['clusterName'],102                                  'hostName': test_hosts[0]['name'],103                                  'hostID': str(test_hosts[0]['id']),104                                  'datastoreName': test_datastore['name'],105                                  'datastoreID': str(test_datastore['id'])})106        mock_datastore = mock.Mock(id=test_datastore['id'],107                                   free_space=test_datastore['free_space'])108        datastore_name_property = mock.PropertyMock(return_value=test_datastore['name'])109        type(mock_datastore).name = datastore_name_property110        mock_host = mock.Mock(v_owning_cluster=test_hosts[0]['v_owning_cluster'],111                              v_total_vms=test_hosts[0]['v_total_vms'],112                              id=test_hosts[0]['id'],113                              host_storages=test_hosts[0]['host_storages'],114                              power_state='on')115        host_name_property = mock.PropertyMock(return_value=test_hosts[0]['name'])116        type(mock_host).name = host_name_property117        mock_client = mock.MagicMock()118        mock_client.collections.data_stores.return_value = mock_datastore119        result = action._check_hosts(mock_client, [mock_host], test_dict)120        self.assertEquals(result, expected_result)121    def test_check_hosts_empty_cluster_raises(self):122        action = self.get_action_instance({})123        client = None124        hosts = []125        kwargs = {'clusterName': None}126        with self.assertRaises(ValueError):127            action._check_hosts(client, hosts, kwargs)128    def test_check_hosts_empty_hosts_fails(self):129        action = self.get_action_instance({})130        client = None131        hosts = []132        kwargs = {'clusterName': 'testCluster'}133        result = action._check_hosts(client, hosts, kwargs)134        self.assertEquals(result, (False, {'clusterName': 'testCluster',135                                           'hostName': None,136                                           'hostID': None,137                                           'datastoreName': None,138                                           'datastoreID': None}))139    def test_check_hosts_no_hosts_in_cluster_fails(self):140        action = self.get_action_instance({})141        client = None142        hosts = [143            mock.MagicMock(v_own_cluster='otherCluster'),144            mock.MagicMock(v_own_cluster='testCluster', power_state='off'),145        ]146        kwargs = {'clusterName': 'testCluster'}147        result = action._check_hosts(client, hosts, kwargs)148        self.assertEquals(result, (False, {'clusterName': 'testCluster',149                                           'hostName': None,150                                           'hostID': None,151                                           'datastoreName': None,152                                           'datastoreID': None}))153    @mock.patch("lib.bestfit.BestFit._check_storages")154    def test_check_hosts_no_datastore_fails(self, mock_check_storages):155        action = self.get_action_instance({})156        mock_check_storages.return_value = (None, None)157        client = None158        hosts = [159            mock.MagicMock(v_own_cluster='testCluster', power_state='on'),160        ]161        kwargs = {'clusterName': 'testCluster'}162        result = action._check_hosts(client, hosts, kwargs)163        self.assertEquals(result, (False, {'clusterName': 'testCluster',164                                           'hostName': None,165                                           'hostID': None,166                                           'datastoreName': None,167                                           'datastoreID': None}))168    @mock.patch("lib.bestfit.BestFit._load_disks")169    def test_check_hosts_disk_datastore(self, mock__load_disks):170        action = self.get_action_instance({})171        test_storages = [{'storage_id': 1}]172        test_datastore = {'id': 1,173                          'name': "abc",174                          'free_space': 100}175        test_dict = {'clusterName': "test_cluster",176                     'datastoreFilterRegEx': {"filters": ["dev_isos"]},177                     'disk_json': '[{"datastore":"abc"}]'}178        test_hosts = [{'v_owning_cluster': "test_cluster",179                       'v_total_vms': 5,180                       'id': 1,181                       'name': "test_host_1",182                       'host_storages': test_storages}]183        expected_result = (True, {'clusterName': test_dict['clusterName'],184                                  'hostName': test_hosts[0]['name'],185                                  'hostID': str(test_hosts[0]['id']),186                                  'datastoreName': test_datastore['name'],187                                  'datastoreID': str(test_datastore['id'])})188        mock_host = mock.Mock(v_owning_cluster=test_hosts[0]['v_owning_cluster'],189                              v_total_vms=test_hosts[0]['v_total_vms'],190                              id=test_hosts[0]['id'],191                              host_storages=test_hosts[0]['host_storages'],192                              power_state='on')193        host_name_property = mock.PropertyMock(return_value=test_hosts[0]['name'])194        type(mock_host).name = host_name_property195        mock_client = mock.MagicMock()196        mock__load_disks.return_value = (test_datastore['name'], str(test_datastore['id']))197        result = action._check_hosts(mock_client, [mock_host], test_dict)198        self.assertEquals(result, expected_result)199    @mock.patch("lib.bestfit.BestFit._check_hosts")200    def test_bestfit(self, mock_check_hosts):201        action = self.get_action_instance(self.config_blank)202        test_dict = {'username': 'user',203                     'password': 'pass',204                     'endpoint': 'endpoint'}205        test_storages = [{'storage_id': 1}]206        test_hosts = [{'v_owning_cluster': "test_cluster",207                       'v_total_vms': 5,208                       'id': 1,209                       'name': "test_host_1",210                       'host_storages': test_storages}]211        check_hosts_result = (True, "check_hosts result")212        mock_check_hosts.return_value = check_hosts_result213        mock_host = mock.Mock(v_owning_cluster=test_hosts[0]['v_owning_cluster'],214                              v_total_vms=test_hosts[0]['v_total_vms'],215                              id=test_hosts[0]['id'],216                              host_storages=test_hosts[0]['host_storages'])217        host_name_property = mock.PropertyMock(return_value=test_hosts[0]['name'])218        type(mock_host).name = host_name_property219        mock_client = mock.MagicMock()220        mock_client.collections.hosts.query_string.return_value = [mock_host]221        result = action.bestfit(mock_client, test_dict)...

test_topology.py

Source:test_topology.py

1from __future__ import absolute_import2from __future__ import division3from __future__ import print_function4from __future__ import unicode_literals5from fboss.fb_thrift_clients import FbossAgentClient, QsfpServiceClient6from fboss.thrift_clients import (7    PlainTextFbossAgentClientDontUseInFb as PlainTextFbossAgentClient)8from fboss.thrift_clients import QsfpServiceClient as PlainTextQsfpServiceClient9from fboss.netlink_manager.netlink_manager_client import NetlinkManagerClient10from fboss.system_tests.testutils.test_client import TestClient11from neteng.fboss.ttypes import FbossBaseError12from fboss.system_tests.facebook.utils.ensemble_requirements import (13    is_host_ping_reachable)14from thrift.transport.TTransport import TTransportException15from ServiceRouter import TServiceRouterException16import logging17import time18import unittest19class FbossTestException(BaseException):20    pass21class RoleType(object):22    SINGLE_SWITCH = "single"23    CHASSIS = "chassis"24class TestHost(object):25    """ FBOSS System Test Host Information26        For testing, a "host" is anything that can27        source and sink traffic that's connected to28        the switch we're testing29        The TestHost must be running the TestClient Thrift Service30        Example usage:31            config = FBOSSTestBaseConfig("switch142.awesomeco.com")32            h154 = TestHost("host154.awesomeco.com")33            h154.add_interface("eth0", "2381:db00:111:400a::154", 2)34            config.add_host(h154)35    """36    def __init__(self, dnsname, port=None):37        """ Init a TestHost38            @param dnsname: dnsname for host, e.g., 'fboss124.fb.com'39            @param port: the port of the TestClient thrift service40        """41        if port is None:42            port = TestClient.DEFAULT_PORT43        self.name = dnsname44        self.port = port    # the port of the TestClient thrift service45        # the interfaces we're to use in testing46        self.interface_map = {}47    def add_interface(self, intf_name, ip, switch_port=None):48        """ Add a testing interface on this host49            Tell the test that this host's interface is allowed for50            sourcing and syncing traffic.51            @param intf_name: a string (e.g., "eth0")52            @param ip: a v4 or v6 address as a string or None53            @param switch_port: an int or None - which switch port?54        """55        self.interface_map[intf_name] = {56            'ip': ip,57            'switch_port': switch_port  # 'None' means the config doesn't know58                                        # which switch port this host is59                                        # connected60        }61    def ips(self):62        for intf in self.interface_map.values():63            yield intf['ip']64    def intfs(self):65        return list(self.interface_map.keys())66    def thrift_client(self):67        return TestClient(self.name, self.port)68    def thrift_verify(self, retries, log=None):69        alive = False70        for i in range(retries):  # don't use @retry decorator - this is OSS71            try:72                with self.thrift_client() as client:73                    alive = client.status()74                    if alive:75                        break76                    if log:77                        log.debug("Failed to thrift verify %s: retry %d" %78                                    (self.name, i))79                    time.sleep(1)80            except Exception as e:81                msg = "Failed to thrift verify %s: " % self.name82                msg += " retry %d: Exception %s" % (i, str(e))83                if log:84                    log.debug(msg)85                else:86                    print(msg)87                time.sleep(1)88        if log:89            status = "ALIVE" if alive else "NOT alive"90            log.debug("Server %s is %s" % (self.name, status))91        if not alive and log:92            log.warning("Server %s failed to thrift verify" % self.name)93        return alive94    def __str__(self):95        return self.name96class FBOSSTestTopology(object):97    """ FBOSS System Test Config98        Contains all of the state for which hosts, switches,99        and tests to run.100        For now, this is just for single switch testing.101    """102    def __init__(self, switch, ensemble, min_hosts,103            port=None, qsfp_port=None):104        if port is None:105            port = PlainTextFbossAgentClient.DEFAULT_PORT106        if qsfp_port is None:107            qsfp_port = PlainTextQsfpServiceClient.DEFAULT_PORT108        self.port = port109        self.qsfp_port = qsfp_port110        self.log = logging.getLogger("__main__")111        self.switch = switch112        self.test_hosts = {}113        self.role_type = RoleType.SINGLE_SWITCH114        self.ensemble = ensemble115        self.min_hosts = min_hosts116    def add_host(self, host):117        """ This host is physically connected to this switch. """118        self.test_hosts[host.name] = host119    def remove_host(self, host_name):120        if host_name in self.test_hosts:121            del self.test_hosts[host_name]122        else:123            raise FbossTestException("host not in test topology: %s" % host_name)124    def number_of_hosts(self):125        return len(self.test_hosts)126    def min_hosts_or_skip(self, n_hosts):127        if len(self.test_hosts) < n_hosts:128            raise unittest.SkipTest("Test needs %d hosts, topology has %d" %129                                    (n_hosts, len(self.test_hosts)))130        if not self.verify_hosts(min_hosts=n_hosts):131            raise unittest.SkipTest("Not able to find %d good hosts" %132                                    (n_hosts))133    def verify_switch(self, retries=30, log=None, port=None):134        """ Verify the switch is THRIFT reachable """135        if not log:136            log = self.log137        port = port or self.port138        """ Verfiy that both userver and agent are reachable """139        # check if userver is reachable, tests can bring down userver140        # Example: T33276019141        if not is_host_ping_reachable(self.switch.name, log):142            return False143        """ Verify the switch is THRIFT reachable """144        for retry in range(retries):  # don't use @retry, this is OSS145            try:146                with FbossAgentClient(self.switch.name, port) as client:147                    client.keepalive()  # simple check, will pass if agent is up148                    # this actually checks that agent is in a ready state149                    client.getAllPortInfo()  # will throw FbossBaseError150                    return True151            except (FbossBaseError, TTransportException,152                    TServiceRouterException) as e:153                log.warning("Switch failed to thrift verify (try #%d): %s"154                                % (retry, str(e)))155                time.sleep(1)156        return False157    def verify_hosts(self, fail_on_error=False, min_hosts=0, retries=30,158                     log=None):159        """ Verify each host is thrift reachable160                if fail_on_error is false, just remove unreachable161                hosts from our setup with a warning.162            zero min_hosts means we don't need any servers to run these tests163        """164        if not log:165            log = self.log166        bad_hosts = []167        for host in self.test_hosts.values():168            if not host.thrift_verify(retries=retries, log=log):169                bad_hosts.append(host.name)170        if bad_hosts:171            if fail_on_error:172                raise FbossBaseError("fail_on_error set and hosts down: %s" %173                                     " ".join(bad_hosts))174            else:175                for host_name in bad_hosts:176                    self.log.warning("Removing thrift-unreachable host: %s " %177                                        host_name)178                    self.remove_host(host_name)179        return len(self.test_hosts) >= min_hosts180    def switch_thrift(self, port=None):181        port = port or self.port182        return FbossAgentClient(self.switch.name, port=port)183    def switch_thrift_plaintext(self, port=None):184        port = port or self.port185        return PlainTextFbossAgentClient(self.switch.name, port=port)186    def qsfp_thrift(self):187        return QsfpServiceClient(self.switch.name, self.qsfp_port)188    def netlink_manager_thrift(self):189        return NetlinkManagerClient(self.switch.name, NetlinkManagerClient.DEFAULT_PORT)190    def hosts(self):191        return list(self.test_hosts.values())192    def hosts_thrift(self, hostname):193        self._valid_testhost(hostname)194        return self.test_hosts[hostname].thrift_client()195    def _valid_testhost(self, hostname):196        """ Is this host in our test topology? """197        if hostname not in self.test_hosts:198            raise FbossTestException("host not in test topology: %s" % hostname)199    def host_ips(self, hostname):200        self._valid_testhost(hostname)201        return self.test_hosts[hostname].ips()202    def get_switch_port_id_from_ip(self, host_binary_ip):203        # TODO(ashwinp): Add support for ARP.204        with self.switch_thrift() as sw_client:205                ndp_entries = sw_client.getNdpTable()206                for ndp_entry in ndp_entries:207                    if ndp_entry.ip == host_binary_ip:208                        return ndp_entry.port209        return None210class FBOSSChassisTestTopology(object):211    """ FBOSS System Test Config212        Contains all of the state for which hosts, switches,213        and tests to run.214        This is specifically for devices that are composed of multiple215        subdevices, like in the case of backpack216    """217    def __init__(self, chassis, cards, port=None, qsfp_port=None):218        if port is None:219            port = PlainTextFbossAgentClient.DEFAULT_PORT220        if qsfp_port is None:221            qsfp_port = PlainTextQsfpServiceClient.DEFAULT_PORT222        self.port = port223        self.qsfp_port = qsfp_port224        self.log = logging.getLogger("__main__")225        self.chassis = chassis226        self.cards = self._generate_cards(cards)227        self.role_type = RoleType.CHASSIS228    def _generate_cards(self, cards):229        return [self._generate_card_topology(card) for card in cards]230    def _generate_card_topology(self, card):231        return FBOSSTestTopology(card, self.port, self.qsfp_port)232    def verify_switch(self):233        """ Verify the switch is THRIFT reachable """234        return all(card.verify_switch() for card in self.cards)235    def verify_hosts(self, min_hosts):236        return True237    def number_of_hosts(self):...

functions.py

Source:functions.py

1import csv2import networkx as nx3import numpy as np4from sklearn.linear_model import LogisticRegression5from sklearn.ensemble import RandomForestRegressor6from sklearn.model_selection import train_test_split 7import matplotlib.pyplot as plt8import collections91011def get_train_data(G): 12    with open("train.csv", 'r') as f:13        train_data = f.read().splitlines()1415    train_hosts = list()16    y_train = list()17    for row in train_data:18        host, label = row.split(",")19        train_hosts.append(host)20        y_train.append(label.lower())21        # Create the training matrix. Each row corresponds to a web host.22    # Use the following 3 features for each web host (unweighted degrees)23    # (1) out-degree of node24    # (2) in-degree of node25    # (3) average degree of neighborhood of node26    X_train = np.zeros((len(train_hosts), 6))27    avg_neig_deg = nx.average_neighbor_degree(G, nodes=train_hosts)28    centrality = nx.degree_centrality(G) 29    hits = nx.hits(G)30    pagerank = nx.pagerank(G)31    for i in range(len(train_hosts)):32        X_train[i,0] = G.in_degree(train_hosts[i])33        X_train[i,1] = G.out_degree(train_hosts[i])34        X_train[i,2] = avg_neig_deg[train_hosts[i]]35        X_train[i,3] = centrality[train_hosts[i]]36        X_train[i,4] = hits[0][train_hosts[i]] + hits[1][train_hosts[i]]37        X_train[i,5] = pagerank[train_hosts[i]]38    return X_train,y_train 3940############################################################################4142def get_test_data(G): 43    with open("test.csv", 'r') as f:44        test_hosts = f.read().splitlines()45    # Create the test matrix. Use the same 3 features as above46    X_test = np.zeros((len(test_hosts), 6))47    avg_neig_deg = nx.average_neighbor_degree(G, nodes=test_hosts)48    centrality = nx.degree_centrality(G) 49    hits = nx.hits(G)50    pagerank = nx.pagerank(G)51    for i in range(len(test_hosts)):52        X_test[i,0] = G.in_degree(test_hosts[i])53        X_test[i,1] = G.out_degree(test_hosts[i])54        X_test[i,2] = avg_neig_deg[test_hosts[i]]55        X_test[i,3] = nx.degree_centrality(G)[test_hosts[i]]56        X_test[i,4] = hits[1][test_hosts[i]]+hits[0][test_hosts[i]]57        X_test[i,5] = pagerank[test_hosts[i]]58    return X_test5960############################################################################6162def create_prediction(clf,y_pred    ): 63    #y_pred = clf.predict_proba(X_test)64    with open("test.csv", 'r') as f:65        test_hosts = f.read().splitlines()6667    # Write predictions to a file68    with open('graph_baseline.csv', 'w') as csvfile:69        writer = csv.writer(csvfile, delimiter=',')70        lst = clf.classes_.tolist()71        lst.insert(0, "Host")72        writer.writerow(lst)73        for i,test_host in enumerate(test_hosts):74            lst = y_pred[i,:].tolist()75            lst.insert(0, test_host)76            writer.writerow(lst)77    78############################################################################7980def show_class_distribution(y_train):81    dict_classes = collections.Counter(y_train)82    classes = list(dict_classes.keys())83    occurence_classes = list(dict_classes.values())8485    percent = 100/sum(occurence_classes) * np.array(occurence_classes)86    plt.figure(figsize=(8,8))8788    patches, texts = plt.pie(occurence_classes, labels=None,89                      shadow=True, startangle=90)9091    labels = ['{0} - {1:1.2f} %'.format(i,j) for i,j in zip(classes, percent)]92    plt.title("Classes distribution")93    plt.legend(patches, labels, loc='upper left', bbox_to_anchor=(-0.1, 1.),94               fontsize=8)
...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.