Module: check_mk Branch: master Commit: ad44d82d4fa4334923d546f38dd1561abde66a65 URL: http://git.mathias-kettner.de/git/?p=check_mk.git;a=commit;h=ad44d82d4fa433… Author: Lars Michelsen <lm(a)mathias-kettner.de> Date: Fri Jul 29 14:08:27 2016 +0200 tests: added helper functions to use the Web API and livestatus queries; Added basic tests for the WATO Web API --- tests/testlib/__init__.py | 115 ++++++++++++++++++++++++++++++++++++++++- tests/web/test_wato_webapi.py | 61 ++++++++++++++++++++-- 2 files changed, 170 insertions(+), 6 deletions(-) diff --git a/tests/testlib/__init__.py b/tests/testlib/__init__.py index 9bf649d..7eec516 100644 --- a/tests/testlib/__init__.py +++ b/tests/testlib/__init__.py @@ -9,6 +9,7 @@ import re import requests import pipes import subprocess +import livestatus try: import simplejson as json @@ -143,7 +144,14 @@ class Site(object): self.url = "http://127.0.0.1/%s/check_mk/" % self.id + @property + def live(self): + return livestatus.SingleSiteConnection("tcp:127.0.0.1:9123") + + def execute(self, cmd, *args, **kwargs): + assert type(cmd) == list, "The command must be given as list" + cmd = [ "sudo", "su", "-l", self.id, "-c", pipes.quote(" ".join([ pipes.quote(p) for p in cmd ])) ] cmd_txt = " ".join(cmd) @@ -200,6 +208,11 @@ class Site(object): return omd(["status", "--bare", self.id]) == 0 + def set_config(self, key, val): + assert omd(["config", self.id, "set", key, val]) == 0 + + + @pytest.fixture(scope="session") def site(request): @@ -219,7 +232,14 @@ def site(request): site = Site(site_id=site_id(), version=site_version(), edition=site_edition()) site.cleanup_if_wrong_version() + + existed = site.exists() site.create() + + if not existed: + site.set_config("LIVESTATUS_TCP", "on") + site.set_config("LIVESTATUS_TCP_PORT", "9123") # Need multiple? Make it configurable! + site.start() def fin(): @@ -347,7 +367,21 @@ class WebSession(requests.Session): # Web-API for managing hosts, services etc. # + def _automation_credentials(self): + p = self.site.execute(["cat", "%s/var/check_mk/web/cmkautomation/automation.secret" % + self.site.root ], + stdout=subprocess.PIPE) + secret = p.communicate()[0].rstrip() + + return { + "_username" : "cmkautomation", + "_secret" : secret, + } + + def _api_request(self, url, data): + data.update(self._automation_credentials()) + req = self.post(url, data) response = json.loads(req.text) @@ -361,11 +395,90 @@ class WebSession(requests.Session): result = self._api_request(self.url + "webapi.py?action=add_host", { "request": json.dumps({ "hostname" : hostname, + "folder" : folder, "attributes" : attributes or {}, }), }) - assert result["result"] == None + assert result == None + + host = self.get_host(hostname) + + assert host["hostname"] == hostname + assert host["path"] == "" + assert host["attributes"] == attributes + + + def get_host(self, hostname): + result = self._api_request(self.url + "webapi.py?action=get_host", { + "request": json.dumps({ + "hostname" : hostname, + }), + }) + + assert type(result) == dict + assert "hostname" in result + assert "path" in result + assert "attributes" in result + + return result + + + def get_all_hosts(self, effective_attributes=0): + result = self._api_request(self.url + "webapi.py?action=get_all_hosts", { + "request": json.dumps({ + "effective_attributes": effective_attributes, + }), + }) + + assert type(result) == dict + return result + + + def delete_host(self, hostname): + result = self._api_request(self.url + "webapi.py?action=delete_host", { + "request": json.dumps({ + "hostname" : hostname, + }), + }) + + assert result == None + + hosts = self.get_all_hosts(hostname) + assert hostname not in hosts + + + def discover_services(self, hostname, mode=None): + request = { + "hostname" : hostname, + } + + if mode != None: + request["mode"] = mode + + result = self._api_request(self.url + "webapi.py?action=discover_services", { + "request": json.dumps(request), + }) + + assert type(result) == unicode + assert result.startswith("Service discovery successful") + + + def activate_changes(self, mode=None, foreign_changes=None): + request = {} + + if mode != None: + request["mode"] = mode + + if foreign_changes != None: + request["foreign_changes"] = foreign_changes + + result = self._api_request(self.url + "webapi.py?action=activate_changes", { + "request": json.dumps(request), + }) + + assert result == None + @pytest.fixture(scope="module") diff --git a/tests/web/test_wato_webapi.py b/tests/web/test_wato_webapi.py index c49490f..207ad31 100644 --- a/tests/web/test_wato_webapi.py +++ b/tests/web/test_wato_webapi.py @@ -4,12 +4,63 @@ import pytest from testlib import site, web -def test_01_global_settings(site, web): +def test_global_settings(site, web): r = web.get(site.url + "wato.py") assert "Global Settings" in r.text -def test_02_add_host(web): - web.add_host("test-host", attributes={ - "ipaddress": "127.0.0.1", - }) +def test_add_host(web): + try: + # Also tests get_host + web.add_host("test-host", attributes={ + "ipaddress": "127.0.0.1", + }) + finally: + web.delete_host("test-host") + + +def test_get_all_hosts_basic(web): + try: + web.add_host("test-host-list", attributes={ + "ipaddress": "127.0.0.1", + }) + + hosts = web.get_all_hosts() + assert "test-host-list" in hosts + finally: + web.delete_host("test-host-list") + + +def test_delete_host(web): + try: + web.add_host("test-host-delete", attributes={ + "ipaddress": "127.0.0.1", + }) + finally: + web.delete_host("test-host-delete") + + +def test_discover_servics(web): + try: + web.add_host("test-host-discovery", attributes={ + "ipaddress": "127.0.0.1", + }) + + web.discover_services("test-host-discovery") + finally: + web.delete_host("test-host-discovery") + + +def test_activate_changes(web, site): + try: + web.add_host("test-host-activate", attributes={ + "ipaddress": "127.0.0.1", + }) + + web.activate_changes() + + result = site.live.query("GET hosts\nColumns: name\nFilter: name = test-host-activate\n") + assert result == [["test-host-activate"]] + finally: + web.delete_host("test-host-activate") + web.activate_changes()