How to use test_hosts method in autotest

Best Python code snippet using autotest_python

test_action_bestfit.py

Source:test_action_bestfit.py Github

copy

Full Screen

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)...

Full Screen

Full Screen

test_topology.py

Source:test_topology.py Github

copy

Full Screen

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):...

Full Screen

Full Screen

functions.py

Source:functions.py Github

copy

Full Screen

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) ...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run autotest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful