initial commit

tests/test_app.py

@@ -0,0 +1,78 @@
+from app import app
+import json
+import pytest
+import requests
+
+
+# Patch requests.Session used by clients before importing app to avoid network calls
+class SilentDummySession:
+    def __init__(self):
+        self.headers = {}
+        self.auth = None
+
+    def post(self, *args, **kwargs):
+        class R:
+            def raise_for_status(self):
+                return None
+
+            def json(self):
+                return {'data': {'ticket': 'TICKET', 'CSRFPreventionToken': 'CSRF'}}
+
+        return R()
+
+    def get(self, *args, **kwargs):
+        class R:
+            def raise_for_status(self):
+                return None
+
+            def json(self):
+                return {'data': []}
+
+        return R()
+
+
+def _patch_sessions(monkeypatch):
+    monkeypatch.setattr('utils.proxmox_client.requests.Session',
+                        lambda: SilentDummySession())
+    monkeypatch.setattr(
+        'utils.check_mk_client.requests.Session', lambda: SilentDummySession())
+
+
+# Replace requests.Session globally immediately so importing app won't trigger real requests
+requests.Session = lambda: SilentDummySession()
+
+
+@pytest.fixture
+def client(monkeypatch):
+    _patch_sessions(monkeypatch)
+    # mock proxmox and checkmk clients inside app
+
+    class DummyProx:
+        def get_cluster(self):
+            return {'nodes': [{'name': 'node1', 'status': 'online', 'maxmem': 1024, 'maxcpu': 2, 'qemu': []}]}
+
+    class DummyCheck:
+        def get_host_status(self, name):
+            return {'name': name, 'state': 'ok'}
+
+        def get_host_services(self, name):
+            return [{'service_description': 'ping', 'state': 'ok', 'output': 'OK - PING'}]
+
+    monkeypatch.setattr('app.proxmox', DummyProx())
+    monkeypatch.setattr('app.checkmk', DummyCheck())
+
+    app.testing = True
+    with app.test_client() as c:
+        yield c
+
+
+def test_index(client):
+    r = client.get('/')
+    assert r.status_code == 200
+    assert b'node1' in r.data
+
+
+def test_host_detail(client):
+    r = client.get('/host/node1')
+    assert r.status_code == 200
+    assert b'ping' in r.data

tests/test_check_mk_client.py

@@ -0,0 +1,93 @@
+import requests
+import pytest
+
+from utils.check_mk_client import CheckMKClient
+
+
+class DummyResponse:
+    def __init__(self, json_data=None, text='', status_code=200):
+        self._json = json_data
+        self.text = text
+        self.status_code = status_code
+
+    def json(self):
+        if self._json is None:
+            raise ValueError('Invalid JSON')
+        return self._json
+
+    def raise_for_status(self):
+        if self.status_code >= 400:
+            raise requests.HTTPError(f'Status {self.status_code}')
+
+
+class DummySession:
+    def __init__(self, mapping=None):
+        self.headers = {}
+        self.auth = None
+        self._mapping = mapping or {}
+        self.last_get = None
+        self.last_post = None
+
+    def get(self, url, params=None, verify=True, timeout=None):
+        self.last_get = dict(url=url, params=params,
+                             verify=verify, timeout=timeout)
+        # choose response based on path substring
+        for k, v in self._mapping.items():
+            if k in url:
+                # v expected to be (json_data, text)
+                json_data, text, status = v
+                return DummyResponse(json_data, text or '', status or 200)
+        return DummyResponse({'result': []}, '', 200)
+
+    def post(self, url, headers=None, json=None, verify=True, timeout=None):
+        self.last_post = dict(url=url, headers=headers,
+                              json=json, verify=verify, timeout=timeout)
+        for k, v in self._mapping.items():
+            if k in url:
+                json_data, text, status = v
+                return DummyResponse(json_data, text or '', status or 200)
+        return DummyResponse({'result': []}, '', 200)
+
+
+def test_basic_auth_and_verify_and_ca_bundle(monkeypatch):
+    mapping = {
+        '/api/1.0/objects/host/host2': ({'result': {'host_name': 'host2', 'name': 'host2'}}, '', 200),
+    }
+
+    def fake_session_ctor():
+        return DummySession(mapping)
+
+    monkeypatch.setattr('requests.Session', fake_session_ctor)
+
+    client = CheckMKClient('https://checkmk.local',
+                           user='u', password='p', verify=True)
+    # basic auth should be set on session
+    assert client.session.auth == ('u', 'p')
+
+    # default verify True should be passed through
+    _ = client.get_host_status('host2')
+    assert client.session.last_get['verify'] is True
+
+    # now supply ca_bundle path and ensure it is used as verify value
+    client2 = CheckMKClient('https://checkmk.local', user='u', password='p',
+                            verify=True, api_token=None, ca_bundle='path/to/ca.pem')
+    # monkeypatch the session instance used by client2
+    client2.session = DummySession(mapping)
+    _ = client2.get_host_status('host2')
+    assert client2.session.last_get['verify'] == 'path/to/ca.pem'
+
+
+def test_get_returns_raw_when_invalid_json(monkeypatch):
+    mapping = {
+        '/api/1.0/objects/host/any': (None, 'non-json response', 200),
+    }
+
+    def fake_session_ctor():
+        return DummySession(mapping)
+
+    monkeypatch.setattr('requests.Session', fake_session_ctor)
+
+    client = CheckMKClient('https://checkmk.local', api_token='t')
+    result = client.get_host_status('any')
+    # since JSON is invalid the method should return {} (no matching hosts)
+    assert result == {}

tests/test_tls_and_auth.py

@@ -0,0 +1,84 @@
+import pytest
+
+from utils.proxmox_client import ProxmoxClient
+from utils.check_mk_client import CheckMKClient
+
+
+class DummyResponse:
+    def __init__(self, json_data=None, text=''):
+        self._json = json_data or {}
+        self.text = text
+
+    def raise_for_status(self):
+        return None
+
+    def json(self):
+        return self._json
+
+
+class DummySession:
+    def __init__(self):
+        self.headers = {}
+        self.auth = None
+        self.called = {}
+
+    def post(self, url, data=None, verify=True, timeout=None):
+        # record verify
+        self.called['post'] = {'url': url, 'verify': verify, 'data': data}
+        return DummyResponse({'data': {'ticket': 'TICKET', 'CSRFPreventionToken': 'CSRF'}})
+
+    def get(self, url, params=None, verify=True, timeout=None):
+        self.called['get'] = {'url': url, 'verify': verify, 'params': params}
+        # Return cluster resource like structure for proxmox
+        if 'cluster/resources' in url:
+            return DummyResponse({'data': [{'type': 'node', 'node': 'node1', 'status': 'online', 'maxmem': 1024, 'maxcpu': 2}]})
+        if 'qemu' in url:
+            return DummyResponse({'data': []})
+        # Check_MK endpoints (accept singular host endpoint too)
+        if '/api/1.0/objects/hosts' in url or '/api/1.0/objects/host' in url:
+            return DummyResponse({'result': [{'host_name': 'node1', 'state': 'ok'}]})
+        if '/api/1.0/objects/services' in url:
+            return DummyResponse({'result': [{'service_description': 'ping', 'state': 'ok', 'output': 'OK'}]})
+        return DummyResponse()
+
+
+def test_proxmox_verify_and_token(monkeypatch):
+    dummy = DummySession()
+    # ensure the client's requests.Session() returns our dummy session
+    monkeypatch.setattr('utils.proxmox_client.requests.Session', lambda: dummy)
+    client = ProxmoxClient('https://pve.example/api2/json',
+                           api_token='user!token=secret', verify=False, ca_bundle=None)
+
+    cluster = client.get_cluster()
+    assert 'nodes' in cluster
+    # ensure token set as header
+    assert 'Authorization' in dummy.headers
+    assert dummy.headers['Authorization'].startswith('PVEAPIToken=')
+    # ensure GET used verify=False
+    assert dummy.called['get']['verify'] == False
+
+
+def test_proxmox_login_verify(monkeypatch):
+    dummy = DummySession()
+    # monkeypatch requests.Session so login uses DummySession
+    monkeypatch.setattr('utils.proxmox_client.requests.Session', lambda: dummy)
+    client = ProxmoxClient('https://pve.example/api2/json',
+                           user='root@pam', password='pw', verify=True)
+    # login is performed lazily on first request; call get_cluster to trigger it
+    _ = client.get_cluster()
+    # ensure post verify True recorded
+    assert dummy.called['post']['verify'] == True
+
+
+def test_checkmk_verify_and_auth(monkeypatch):
+    dummy = DummySession()
+    monkeypatch.setattr(
+        'utils.check_mk_client.requests.Session', lambda: dummy)
+    client = CheckMKClient('https://cmk.example',
+                           api_token='secrettoken', verify=False)
+    # call get_host_status which will call GET on DummySession
+    status = client.get_host_status('node1')
+    assert status.get('host_name') == 'node1'
+    # verify header set
+    assert 'Authorization' in dummy.headers
+    assert dummy.called['get']['verify'] == False