Browse Source

icinga: rework helpers to use icinga2api instead of requests

keep-around/eebcbb94b84fc1aac139574ac66ba6b8bb6a2c17
singuliere 3 years ago
parent
commit
cb13139e01
No known key found for this signature in database GPG Key ID: 900857755EF189C2
  1. 9
      conftest.py
  2. 1
      requirements.in
  3. 1
      requirements.txt
  4. 59
      tests/icinga_helper.py

9
conftest.py

@ -0,0 +1,9 @@
# meant to be read by testinfra in molecule/*/tests
# when assuming molecule/*/molecule.yml sets PYTHONPATH=../..
import pytest
from tests.icinga_helper import IcingaHelper
def pytest_configure(config):
IcingaHelper.set_ansible_inventory(config.getoption("--ansible-inventory"))

1
requirements.in

@ -9,3 +9,4 @@ sphinx
testinfra
openstacksdk==0.22.0
netaddr==0.7.19
-e git+https://github.com/fmnisme/python-icinga2api.git@9a1a3cc7968d6c72bf49e97ef387b2824e6835e9#egg=icinga2api

1
requirements.txt

@ -4,6 +4,7 @@
#
# pip-compile --output-file requirements.txt requirements.in
#
-e git+https://github.com/fmnisme/python-icinga2api.git@9a1a3cc7968d6c72bf49e97ef387b2824e6835e9#egg=icinga2api
alabaster==0.7.10 # via sphinx
ansible-lint==3.4.23 # via molecule
ansible==2.7.5

59
tests/icinga_helper.py

@ -1,16 +1,22 @@
from icinga2api.client import Client
import re
import requests
import retry
import urllib3
import yaml
import yaml
import testinfra
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
class IcingaHelper(object):
def get_auth(self, host):
host = host.get_host('ansible://icinga-host', ansible_inventory=host.backend.ansible_inventory)
# set by ../conftest.py
@staticmethod
def set_ansible_inventory(inventory):
IcingaHelper.inventory = inventory
def get_auth(self):
host = testinfra.get_host('ansible://icinga-host', ansible_inventory=self.inventory)
with host.sudo():
f = host.file("/etc/icinga2/conf.d/api-users.conf")
return (
@ -23,41 +29,32 @@ class IcingaHelper(object):
return 'icinga.' + yaml.load(
open(vars_dir + '/domain.yml'))['domain']
def get_api_session(self, host):
s = requests.Session()
s.auth = self.get_auth(host)
s.headers.update({'Accept': 'application/json'})
s.verify = False
s.timeout = 5
return s
def get_client(self):
(user, password) = self.get_auth()
client = Client(
'https://{}:5665'.format(self.get_address()),
user, password,
ca_certificate=False,
timeout=5
)
return client
@retry.retry(AssertionError, tries=7)
def wait_for_service(self, session, api, name):
r = session.get(
'{api}/objects/services/{name}'.format(api=api, name=name)
)
r.raise_for_status()
answer = r.json()
assert int(answer['results'][0]['attrs']['state']) == 0
def wait_for_service(self, client, name):
answer = client.objects.get('Service', name)
assert int(answer['attrs']['state']) == 0
return True
def is_service_ok(self, host, name):
def is_service_ok(self, name):
#
# force the check to reduce the waiting time
#
session = self.get_api_session(host)
api = 'https://{}:5665/v1'.format(self.get_address())
r = session.post(
'{api}/actions/reschedule-check'.format(api=api),
json={
"type": "Service",
"filter": "service.__name==\"{}\"".format(name),
"force": True,
}
client = self.get_client()
answer = client.actions.reschedule_check(
'Service',
'service.__name=="{}"'.format(name),
)
r.raise_for_status()
answer = r.json()
assert len(answer['results']) == 1
assert int(answer['results'][0]['code']) == 200
return self.wait_for_service(session, api, name)
return self.wait_for_service(client, name)
Loading…
Cancel
Save