Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_LoginToServer(self):
try:
ZabbixAPI(url='http://127.0.0.1',
user='Admin',
password='zabbix')
except ZabbixAPIException:
self.fail('Can\'t login to Zabbix')
hosts = zapi.host.get(filter={"host": host_name}, selectInterfaces=["interfaceid"])
if hosts:
host_id = hosts[0]["hostid"]
print("Found host id {0}".format(host_id))
try:
item = zapi.item.create(
hostid=host_id,
name='Used disk space on $1 in %',
key_='vfs.fs.size[/,pused]',
type=0,
value_type=3,
interfaceid=hosts[0]["interfaces"][0]["interfaceid"],
delay=30
)
except ZabbixAPIException as e:
print(e)
sys.exit()
print("Added item with itemid {0} to host: {1}".format(item["itemids"][0], host_name))
else:
print("No hosts found")
def import_usermacro(zabbix, yml, usermacro2hostmacroid, host2hostid, template2templateid):
"Import user macro from YAML. Return created object, None on error, True if object already exists"
result = None
try:
try:
t = (str(host2hostid[yml['hostid']]),yml['macro']) # host?
except KeyError:
t = (str(template2templateid[yml['hostid']]),yml['macro']) # template?
if t in usermacro2hostmacroid:
if usermacro2hostmacroid[t]['value'] == str(yml['value']): return True # skip already existing objects
else: # update existing hostmacro
result = zabbix.usermacro.update(hostmacroid=usermacro2hostmacroid[t]['hostmacroid'], value=str(yml['value']))
else: # create new hostmacro
result = zabbix.usermacro.create(hostid=host2hostid[yml['hostid']], macro=yml['macro'], value=str(yml['value']))
except ZabbixAPIException as e:
if 'already exist' in str(e):
result = True
else:
result = None
logging.exception(e)
return result
def wrapperd_f(*args, **kwargs):
try:
return func(*args, **kwargs)
except (requests.ConnectionError, ZabbixAPIException) as e:
logging.error('Zabbix Error: {}'.format(e))
return fail_result
return wrapperd_f
'valueMaps': {
'createMissing': True,
'updateExisting': True
},
}
if os.path.isdir(path):
#path = path/*.xml
files = glob.glob(path+'/*.xml')
for file in files:
print(file)
with open(file, 'r') as f:
template = f.read()
try:
zapi.confimport('xml', template, rules)
except ZabbixAPIException as e:
print(e)
print('')
elif os.path.isfile(path):
files = glob.glob(path)
for file in files:
with open(file, 'r') as f:
template = f.read()
try:
zapi.confimport('xml', template, rules)
except ZabbixAPIException as e:
print(e)
else:
print('I need a xml file')
conf = config[DOMAIN]
if conf[CONF_SSL]:
schema = 'https'
else:
schema = 'http'
url = urljoin('{}://{}'.format(schema, conf[CONF_HOST]), conf[CONF_PATH])
username = conf.get(CONF_USERNAME, None)
password = conf.get(CONF_PASSWORD, None)
zapi = ZabbixAPI(url)
try:
zapi.login(username, password)
_LOGGER.info("Connected to Zabbix API Version %s", zapi.api_version())
except ZabbixAPIException as login_exception:
_LOGGER.error("Unable to login to the Zabbix API: %s", login_exception)
return False
hass.data[DOMAIN] = zapi
return True
def delete_user(self, user):
try:
self.api.user.delete(user.backend_id)
except (pyzabbix.ZabbixAPIException, RequestException) as e:
raise ZabbixBackendError(e)
def get_host_sla_triggerid(self, api, hostid):
try:
return api.trigger.get(hostids=hostid, output=['triggerid'],
# XXX a hardcoded description, consider refactoring
filter={'description': 'Missing data about the VM'})[0]['triggerid']
except IndexError:
raise pyzabbix.ZabbixAPIException('No template with id: %s' % hostid)
hostid=zabbix_server_id,
status=ENABLED
)
self.item_id = response['itemids'][0]
response = self.zapi.trigger.create(
hostid=zabbix_server_id,
description=description,
expression='{Zabbix server:test.alerta.diff()}>0',
type=GENERATE_MULTIPLE_EVENTS,
priority=INFORMATION,
status=ENABLED,
manual_close=ALLOW_MANUAL_CLOSE
)
self.trigger_id = response['triggerids'][0]
except ZabbixAPIException:
triggers = self.zapi.trigger.get(hostids=zabbix_server_id)
self.trigger_id = [t for t in triggers if t['description'] == description][0]['triggerid']
self.item_id = self.zapi.item.get(triggerids=self.trigger_id)[0]['itemid']
def zabbix_send(value):
cfg = protobix.ZabbixAgentConfig()
cfg.server_active = trapper
zbx = protobix.DataContainer(cfg)
zbx.data_type = 'items'
zbx.add_item(host='Zabbix server', key='test.alerta', value=value)
response = zbx.send()
print(response)
print('sending test items')
now = int(time.time())