How to use update_host method in tempest

Best Python code snippet using tempest_python

host.py

Source:host.py Github

copy

Full Screen

1import os2import sys3import json4import logging5from pyzabbix import ZabbixAPI, ZabbixAPIException6DIR = '/root/app/zabbix_autoupdater/'7with open(DIR + 'config/config.json') as f:8 config = json.load(f)9sys.path.append('..')10LOG_PATH = '/var/log/zabbix-scripts/'11LOG_NAME = 'autoupdater'12if not os.path.exists(LOG_PATH):13 os.makedirs(LOG_PATH)14logging.basicConfig(format=u'%(levelname)-8s [%(asctime)s] %(message)s',15 level=logging.INFO,16 filename=u'{}{}.log'.format(LOG_PATH, LOG_NAME))17zabbix_server = config['zabbix_server']18zabbix_username = config['zabbix_username']19zabbix_password = config['zabbix_password']20class Host:21 zapi = None22 def __init__(self):23 self.zapi = ZabbixAPI(zabbix_server)24 self.zapi.login(zabbix_username, zabbix_password)25 def create(self, host, name, proxy_hostid, global_group_id, local_group_id, template1):26 if self.zapi:27 params = {'host': host,28 'name': name,29 'proxy_hostid ': str(proxy_hostid),30 'interfaces': [31 {32 'type': 2,33 'main': 1,34 'useip': 1,35 'ip': host,36 'dns': '',37 'port': '161'38 }39 ],40 'groups': [41 {42 'groupid': str(global_group_id)43 },44 {45 'groupid': str(local_group_id)46 }47 ],48 'templates': [49 {50 'templateid': str(template1)51 }52 # {53 # 'templateid': str(template2)54 # }55 ]56 }57 try:58 create_host_answer = self.zapi.do_request(59 method="host.create", params=params)60 if create_host_answer:61 logging.info('Хост {} - создан!'.format(name))62 return create_host_answer63 else:64 logging.error('Не могу создать хост - {}'.format(host))65 except ZabbixAPIException:66 logging.error('Ошибка при создании хоста - {}'.format(host))67 return False68 else:69 logging.info('Невозможно подключиться к Заббиксу')70 def delete(self, host_id):71 params = [str(host_id)]72 if self.zapi:73 try:74 del_host = self.zapi.do_request(75 method="host.delete", params=params)76 if del_host:77 logging.info('Хост {} - Удален!'.format(host_id))78 return del_host79 else:80 logging.error('Нет такого хоста - {}'.format(host_id))81 except:82 logging.error('Ошибка при удалении хоста - {}'.format(host_id))83 return False84 else:85 logging.info('Невозможно подключиться к Заббиксу')86 def get_id(self, host):87 params = {'filter': {'host': [host]}}88 if self.zapi:89 try:90 get_id = self.zapi.do_request(method="host.get", params=params)[91 'result'][0]['hostid']92 if get_id:93 logging.info('Получен ID хоста {} - {}!'.format(host, get_id))94 return get_id95 else:96 logging.error('Нет такого хоста - {}'.format(host))97 except IndexError:98 logging.error(99 'Ошибка при получении ID хоста - {}'.format(host))100 return False101 else:102 logging.info('Невозможно подключиться к Заббиксу')103 def get_name(self, host):104 params = {'filter': {'name': [host]}}105 if self.zapi:106 try:107 get_id = self.zapi.do_request(method="host.get", params=params)[108 'result'][0]['hostid']109 if get_id:110 logging.info('Получен name ID хоста - {}!'.format(host))111 return get_id112 else:113 logging.error('Нет такого хоста - {}'.format(host))114 except IndexError:115 logging.error(116 'Ошибка при получении name ID хоста - {}'.format(host))117 return False118 else:119 logging.info('Невозможно подключиться к Заббиксу')120 def update(self, id, update_item, param):121 params = {'hostid': str(id), str(update_item): str(param)}122 if self.zapi:123 try:124 update_host = self.zapi.do_request(125 method="host.update", params=params)126 if update_host:127 logging.info('У хоста {} обновлен параметр {} на {}'.format(128 id, update_item, param))129 return True130 else:131 logging.error('Нет такого хоста - {}'.format(id))132 except:133 logging.error('Ошибка при обновлении хоста - {}'.format(id))134 return False135 else:136 logging.info('Невозможно подключиться к Заббиксу')137 def update_two_param(self, id, update_item, param, param2):138 params = {'hostid': str(id), str(update_item): [param, param2]}139 if self.zapi:140 try:141 update_host = self.zapi.do_request(142 method="host.update", params=params)143 if update_host:144 logging.info('У хоста {} обновлен параметр {} на {}'.format(145 id, update_item, param))146 logging.info('У хоста {} обновлен параметр {} на {}'.format(147 id, update_item, param2))148 else:149 logging.error('Нет такого хоста - {}'.format(id))150 except:151 logging.error('Ошибка при обновлении хоста - {}'.format(id))152 return False153 else:154 logging.info('Невозможно подключиться к Заббиксу')155 def update_many_param(self, hostid, update_item, *parameters):156 # т.к. импортирует кортеж из списка157 params = {'hostid': hostid, update_item: parameters[0]}158 if self.zapi:159 try:160 update_host = self.zapi.do_request(161 method="host.update", params=params)162 if update_host:163 logging.info('У хоста {} обновлен параметр {} на {}'.format(164 hostid, update_item, params['groups']))165 else:166 logging.error('Нет такого хоста - {}'.format(hostid))167 except:168 logging.error(169 'Ошибка при обновлении хоста - {}'.format(hostid))170 return False171 else:172 logging.info('Невозможно подключиться к Заббиксу')173 def update_inventory(self, hostid, update_item, param):174 params = {'hostid': hostid, update_item: param}175 if self.zapi:176 try:177 update_host = self.zapi.do_request(method="host.update", params=params)178 if update_host:179 logging.info('У хоста {} обновлен параметр {} на {}'.format(180 hostid, update_item, params['groups']))181 else:182 logging.error('Нет такого хоста - {}'.format(hostid))183 except:184 logging.error(185 'Ошибка при обновлении инвертарных данных хоста - {}'.format(hostid))186 return False187 else:188 logging.info('Невозможно подключиться к Заббиксу')189 def get_proxyid(self, host):190 params = {'filter': {'host': [host]}}191 if self.zapi:192 try:193 get_id = self.zapi.do_request(method="host.get", params=params)194 if get_id:195 logging.info('Получен ID хоста - {}!'.format(host))196 return get_id['result'][0]['proxy_hostid']197 else:198 logging.error('Нет такого хоста - {}'.format(host))199 except IndexError:200 logging.error(201 'Ошибка при получении ID хоста - {}'.format(host))202 return False203 else:204 logging.info('Невозможно подключиться к Заббиксу')205 def get_hostid_of_proxy(self, proxy_id):206 params = {"output": ["hostid"], "proxyids": str(proxy_id)}207 if self.zapi:208 try:209 get_id = self.zapi.do_request(method="host.get", params=params)210 if get_id:211 logging.info(212 'Получен ID хостов от прокси - {}!'.format(proxy_id))213 return get_id['result']214 else:215 logging.error('У прокси {} нет хостов'.format(proxy_id))216 except IndexError:217 logging.error(218 'Ошибка при получении хостов от прокси - {}'.format(proxy_id))219 return False220 else:221 logging.info('Невозможно подключиться к Заббиксу')222 def get_hostid_of_template(self, template_id):223 params = {"output": ["hostid"], "templateids": str(template_id)}224 if self.zapi:225 try:226 get_id = self.zapi.do_request(method="host.get", params=params)227 if get_id:228 logging.info(229 'Получен ID хостов от шаблона - {}!'.format(template_id))230 return get_id['result']231 else:232 logging.error('У шаблона {} нет хостов'.format(template_id))233 except IndexError:234 logging.error(235 'Ошибка при получении хостов с шаблоном - {}'.format(template_id))236 return False237 else:238 logging.info('Невозможно подключиться к Заббиксу')239 def add_macros(self, host_id, macros, value):240 # {241 # "jsonrpc": "2.0",242 # "method": "host.update",243 # "params": {244 # "hostid": "10126",245 # "macros": {246 # "add": {247 # "{$MACRO1}": "value",248 # "{$MACRO2}": "5"249 # },250 # "delete": [251 # "{$MACRO4}"252 # ],253 # "update": {254 # "{$MACRO3}": "new_value"255 # }256 # }257 # },258 # "auth": "038e1d7b1735c6a5436ee9eae095879e",259 # "id": 1260 # }261 params = {262 'hostid': str(host_id),263 'macros': {264 'add': {265 macros: str(value)266 }267 }268 }269 if self.zapi:270 try:271 update_host = self.zapi.do_request(272 method="host.update", params=params)273 if update_host:274 logging.info('Хосту {} добавлен макрос {} со значением {}'.format(275 host_id, macros, value))276 return True277 else:278 logging.error('Нет такого хоста - {}'.format(host_id))279 except ZabbixAPIException:280 logging.error('Ошибка при обновлении хоста - {}'.format(host_id))281 return False282 else:283 logging.info('Невозможно подключиться к Заббиксу')284 def del_macros(self, host_id, macros, value):285 params = {286 'hostid': str(host_id),287 'macros': {288 'delete': {289 macros: str(value)290 }291 }292 }293 if self.zapi:294 try:295 update_host = self.zapi.do_request(296 method="host.update", params=params)297 if update_host:298 logging.info('Хосту {} добавлен макрос {} со значением {}'.format(299 host_id, macros, value))300 return True301 else:302 logging.error('Нет такого хоста - {}'.format(host_id))303 except ZabbixAPIException:304 logging.error('Ошибка при обновлении хоста - {}'.format(host_id))305 return False306 else:307 logging.info('Невозможно подключиться к Заббиксу')308 def del_macros(self, host_id, macros):309 params = {310 'hostid': str(host_id),311 'macros': {312 'delete': {313 macros314 }315 }316 }317 if self.zapi:318 try:319 update_host = self.zapi.do_request(320 method="host.update", params=params)321 if update_host:322 logging.info('У хоста {} удален макрос {}'.format(323 host_id, macros))324 return True325 else:326 logging.error('Нет такого хоста - {}'.format(host_id))327 except ZabbixAPIException:328 logging.error('Ошибка при обновлении хоста - {}'.format(host_id))329 return False330 else:331 logging.info('Невозможно подключиться к Заббиксу')332 def update_macros(self, host_id, macros, value):333 params = {334 'hostid': str(host_id),335 'macros': {336 'update': {337 macros: str(value)338 }339 }340 }341 if self.zapi:342 try:343 update_host = self.zapi.do_request(344 method="host.update", params=params)345 if update_host:346 logging.info(' У хоста {} обновлен макрос {} со значением {}'.format(347 host_id, macros, value))348 return True349 else:350 logging.error('Нет такого хоста - {}'.format(host_id))351 except ZabbixAPIException:352 logging.error('Ошибка при обновлении хоста - {}'.format(host_id))353 return False354 else:...

Full Screen

Full Screen

test_escalations_hosts.py

Source:test_escalations_hosts.py Github

copy

Full Screen

...28from nose.plugins.attrib import attr29import time30# These tests need to be run separately and not in parallel with other tests.31# Because it disables the infrastructure for brief periods32def update_host(apiclient, state, host_id):33 """34 Function to Enable/Disable Host35 """36 host_status = Host.update(37 apiclient,38 id=host_id,39 allocationstate=state40 )41 return host_status.resourcestate42def update_cluster(apiclient, state, cluster_id, managed_state):43 """44 Function to Enable/Disable cluster45 """46 cluster_status = Cluster.update(47 apiclient,48 id=cluster_id,49 allocationstate=state,50 managedstate=managed_state51 )52 return cluster_status.managedstate, cluster_status.allocationstate53def update_pod(apiclient, state, pod_id):54 """55 Function to Enable/Disable pod56 """57 pod_status = Pod.update(58 apiclient,59 id=pod_id,60 allocationstate=state61 )62 return pod_status.allocationstate63def update_zone(apiclient, state, zone):64 """65 Function to Enable/Disable zone66 """67 zone_status = zone.update(68 apiclient,69 allocationstate=state70 )71 return zone_status.allocationstate72def check_db(self, host_state):73 """74 Function to check capacity_state in op_host_capacity table75 """76 capacity_state = self.dbclient.execute(77 "select capacity_state from op_host_capacity where host_id='%s';" %78 self.host_db_id[0][0])79 self.assertEqual(80 capacity_state[0][0],81 host_state +82 "d",83 "Invalid db query response for capacity_state %s" %84 self.host_db_id[0][0])85 return capacity_state[0][0]86class TestHosts(cloudstackTestCase):87 """88 Testing Hosts89 """90 @classmethod91 def setUpClass(cls):92 cls.testClient = super(TestHosts, cls).getClsTestClient()93 cls.testdata = cls.testClient.getParsedTestDataConfig()94 cls.apiclient = cls.testClient.getApiClient()95 cls.dbclient = cls.testClient.getDbConnection()96 cls._cleanup = []97 # get zone, domain etc98 cls.zone = Zone(get_zone(cls.apiclient, cls.testClient.getZoneForTests()).__dict__)99 cls.domain = get_domain(cls.apiclient)100 cls.pod = get_pod(cls.apiclient, cls.zone.id)101 # list hosts102 hosts = list_hosts(cls.apiclient, type="Routing")103 if len(hosts) > 0:104 cls.my_host_id = hosts[0].id105 cls.host_db_id = cls.dbclient.execute(106 "select id from host where uuid='%s';" %107 cls.my_host_id)108 cls.my_cluster_id = hosts[0].clusterid109 else:110 raise unittest.SkipTest("There is no host available in the setup")111 @classmethod112 def tearDownClass(cls):113 cleanup_resources(cls.apiclient, cls._cleanup)114 return115 def setUp(self):116 self.cleanup = []117 return118 def tearDown(self):119 # Clean up120 cleanup_resources(self.apiclient, self.cleanup)121 return122 @attr(tags=["advanced", "basic"], required_hardware="false")123 def test_01_op_host_capacity_disable_cluster(self):124 """125 Disable the host and it's cluster,126 make sure that capacity_state is not affected by enabling/disabling127 of cluster in the op_host_capacity table128 """129 # disable the host and check op_host_capacity table130 host_state = "Disable"131 host_resourcestate = update_host(132 self.apiclient,133 host_state,134 self.my_host_id)135 self.assertEqual(136 host_resourcestate,137 host_state + "d",138 "Host state not correct"139 )140 check_db(self, host_state)141 # disable the cluster and check op_host_capacity table142 cluster_state = "Disabled"143 managed_state = "Managed"144 cluster_managedstate, cluster_allocationstate = update_cluster(145 self.apiclient, cluster_state, self.my_cluster_id, managed_state)146 self.assertEqual(147 cluster_allocationstate,148 cluster_state,149 "Not able to enable/disable the cluster"150 )151 self.assertEqual(152 cluster_managedstate,153 managed_state,154 "Not able to managed/unmanage the cluster"155 )156 check_db(self, host_state)157 # enable the cluster and check op_host_capacity table158 cluster_state = "Enabled"159 cluster_managedstate, cluster_allocationstate = update_cluster(160 self.apiclient, cluster_state, self.my_cluster_id, managed_state)161 self.assertEqual(162 cluster_allocationstate,163 cluster_state,164 "Not able to enable/disable the cluster"165 )166 self.assertEqual(167 cluster_managedstate,168 managed_state,169 "Not able to managed/unmanage the cluster"170 )171 check_db(self, host_state)172 # enable the host and check op_host_capacity table173 host_state = "Enable"174 host_resourcestate = update_host(175 self.apiclient,176 host_state,177 self.my_host_id)178 self.assertEqual(179 host_resourcestate,180 host_state + "d",181 "Host state not correct"182 )183 check_db(self, host_state)184 return185 @attr(tags=["advanced", "basic"], required_hardware="false")186 def test_02_op_host_capacity_disable_pod(self):187 """188 Disable the host and it's pod,189 make sure that capacity_state is not affected by enabling/disabling190 of pod in the op_host_capacity table191 """192 # disable the host and check op_host_capacity table193 host_state = "Disable"194 host_resourcestate = update_host(195 self.apiclient,196 host_state,197 self.my_host_id)198 self.assertEqual(199 host_resourcestate,200 host_state + "d",201 "Host state not correct"202 )203 check_db(self, host_state)204 # disable the pod and check op_host_capacity table205 pod_state = "Disabled"206 pod_allocationstate = update_pod(207 self.apiclient,208 pod_state,209 self.pod.id)210 self.assertEqual(211 pod_allocationstate,212 pod_state,213 "Not able to enable/disable the pod"214 )215 check_db(self, host_state)216 # enable the pod and check op_host_capacity table217 pod_state = "Enabled"218 pod_allocationstate = update_pod(219 self.apiclient,220 pod_state,221 self.pod.id)222 self.assertEqual(223 pod_allocationstate,224 pod_state,225 "Not able to enable/disable the pod"226 )227 check_db(self, host_state)228 # enable the host and check op_host_capacity table229 host_state = "Enable"230 host_resourcestate = update_host(231 self.apiclient,232 host_state,233 self.my_host_id)234 self.assertEqual(235 host_resourcestate,236 host_state + "d",237 "Host state not correct"238 )239 check_db(self, host_state)240 return241 @attr(tags=["advanced", "basic", "tag1"], required_hardware="false")242 def test_03_op_host_capacity_disable_zone(self):243 """244 Disable the host and it's zone,245 make sure that capacity_state is not affected by enabling/disabling246 of zone in the op_host_capacity table247 """248 # disable the host and check op_host_capacity table249 host_state = "Disable"250 host_resourcestate = update_host(251 self.apiclient,252 host_state,253 self.my_host_id)254 self.assertEqual(255 host_resourcestate,256 host_state + "d",257 "Host state not correct"258 )259 check_db(self, host_state)260 # disbale the zone and check op_host_capacity table261 zone_state = "Disabled"262 zone_allocationstate = update_zone(263 self.apiclient,264 zone_state,265 self.zone)266 self.assertEqual(267 zone_allocationstate,268 zone_state,269 "Not able to enable/disable the zone"270 )271 check_db(self, host_state)272 # enable the zone and check op_host_capacity table273 zone_state = "Enabled"274 zone_allocationstate = update_zone(275 self.apiclient,276 zone_state,277 self.zone)278 self.assertEqual(279 zone_allocationstate,280 zone_state,281 "Not able to enable/disable the zone"282 )283 check_db(self, host_state)284 # enable the host and check op_host_capacity table285 host_state = "Enable"286 host_resourcestate = update_host(287 self.apiclient,288 host_state,289 self.my_host_id)290 self.assertEqual(291 host_resourcestate,292 host_state + "d",293 "Host state not correct"294 )295 check_db(self, host_state)296 return297 @attr(tags=["advanced", "basic"], required_hardware="false")298 def test_04_disable_host_unmanage_cluster_check_hosts_status(self):299 """300 Disable the host then unmanage the cluster,301 make sure that the host goes to Disconnected state302 """303 # disable host304 host_state = "Disable"305 host_resourcestate = update_host(306 self.apiclient,307 host_state,308 self.my_host_id)309 self.assertEqual(310 host_resourcestate,311 host_state + "d",312 "Host state not correct"313 )314 # unmanage cluster315 cluster_state = "Enabled"316 managed_state = "Unmanaged"317 cluster_managedstate, cluster_allocationstate = update_cluster(318 self.apiclient, cluster_state, self.my_cluster_id, managed_state)319 self.assertEqual(320 cluster_allocationstate,321 cluster_state,322 "Not able to enable/disable the cluster"323 )324 self.assertEqual(325 cluster_managedstate,326 managed_state,327 "Not able to managed/unmanage the cluster"328 )329 # check host state now330 time.sleep(30)331 host_list = list_hosts(self.apiclient, id=self.my_host_id)332 self.assertEqual(333 host_list[0].state,334 "Disconnected",335 " Host is not in Disconnected state after unmanaging cluster"336 )337 # manage the cluster again and let the hosts come back to Up state.338 managed_state = "Managed"339 cluster_managedstate, cluster_allocationstate = update_cluster(340 self.apiclient, cluster_state, self.my_cluster_id, managed_state)341 self.assertEqual(342 cluster_allocationstate,343 cluster_state,344 "Not able to enable/disable the cluster"345 )346 self.assertEqual(347 cluster_managedstate,348 managed_state,349 "Not able to managed/unmanage the cluster"350 )351 # check host state now352 time.sleep(90)353 host_list = list_hosts(self.apiclient, id=self.my_host_id)354 self.assertEqual(355 host_list[0].state,356 "Up",357 " Host is not in Up state after managing cluster"358 )359 # enable the host360 host_state = "Enable"361 host_resourcestate = update_host(362 self.apiclient,363 host_state,364 self.my_host_id)365 self.assertEqual(366 host_resourcestate,367 host_state + "d",368 "Host state not correct"369 )...

Full Screen

Full Screen

model_host_tests.py

Source:model_host_tests.py Github

copy

Full Screen

...19 query_host = Host.query.filter_by(id=h.id).first()20 self.assertEqual(query_host.port, 22)21 self.assertIsInstance(query_host.status, Status)22 self.assertEqual(query_host.status.title, Status.PREPARE)23 def test_update_host(self):24 h = Host()25 h.create()26 update_host = Host.query.filter_by(id=h.id).first()27 new_status = Status.query.filter_by(title=Status.BUSINESS).first()28 update_host.update(29 ip=update_host.ip,30 port=update_host.port,31 domain='test.domain',32 pwd=update_host.pwd,33 db_pwd=update_host.db_pwd,34 memo=update_host.memo,35 status=new_status.id)36 self.assertEqual('test.domain', update_host.domain)37 self.assertEqual(new_status.title, update_host.status.title)...

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 tempest automation tests on LambdaTest cloud grid

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

... View sample repo... View DocumentationSign up Free
›
_

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

...
Sign up with GoogleSign up with Email
Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful