add unit tests

roles/openshift_health_checker/test/podlogs_test.py

@@ -0,0 +1,190 @@
+import pytest
+import json
+
+
+from openshift_checks.logging.podlogs import PodLogQueryTime, OpenShiftCheckException
+
+
+SAMPLE_UID = "3ea53ca5-f60a-40e5-8828-983240aee8a1"
+
+
+def canned_podlogquerytime(exec_oc=None):
+    """Create a check object with a canned exec_oc method"""
+    check = PodLogQueryTime("dummy")  # fails if a module is actually invoked
+    if exec_oc:
+        check.oc_cmd = exec_oc
+    return check
+
+
+def assert_error(error, expect_error):
+    if expect_error:
+        assert error
+        assert expect_error in error
+    else:
+        assert not error
+
+
+plain_running_elasticsearch_pod = {
+    "metadata": {
+        "labels": {"component": "es", "deploymentconfig": "logging-es-data-master"},
+        "name": "logging-es-data-master-1",
+    },
+    "status": {
+        "containerStatuses": [{"ready": True}, {"ready": True}],
+        "phase": "Running",
+    }
+}
+plain_running_kibana_pod = {
+    "metadata": {
+        "labels": {"component": "kibana", "deploymentconfig": "logging-kibana"},
+        "name": "logging-kibana-1",
+    },
+    "status": {
+        "containerStatuses": [{"ready": True}, {"ready": True}],
+        "phase": "Running",
+    }
+}
+not_running_kibana_pod = {
+    "metadata": {
+        "labels": {"component": "kibana", "deploymentconfig": "logging-kibana"},
+        "name": "logging-kibana-2",
+    },
+    "status": {
+        "containerStatuses": [{"ready": True}, {"ready": False}],
+        "conditions": [{"status": "True", "type": "Ready"}],
+        "phase": "pending",
+    }
+}
+
+
+@pytest.mark.parametrize('pods, expect_pods', [
+    (
+        [not_running_kibana_pod],
+        [],
+    ),
+    (
+        [plain_running_kibana_pod],
+        [plain_running_kibana_pod],
+    ),
+    (
+        [],
+        [],
+    )
+])
+def test_check_running_pods(pods, expect_pods):
+    check = canned_podlogquerytime(None)
+    pods = check.running_pods(pods)
+    assert pods == expect_pods
+
+
+@pytest.mark.parametrize('json_response, uuid, timeout, expect_exception, extra_words', [
+    # test with invalid json response
+    (
+        {
+            "invalid_field": 1,
+        },
+        SAMPLE_UID,
+        0.001,
+        KeyError,
+        ["count"],
+    ),
+    # test with empty response
+    (
+        {},
+        SAMPLE_UID,
+        0.001,
+        KeyError,
+        ["count"],
+    ),
+    # test with valid count in response
+    (
+        {
+            "count": 1,
+        },
+        SAMPLE_UID,
+        0.001,
+        None,
+        [],
+    ),
+    # test with valid response but invalid count
+    (
+        {
+            "count": 0,
+        },
+        SAMPLE_UID,
+        0.005,
+        OpenShiftCheckException,
+        ["expecting match", SAMPLE_UID, "0.005s"],
+    )
+])
+def test_wait_until_cmd_or_err(json_response, uuid, timeout, expect_exception, extra_words):
+    def exec_oc(exec_cmd, args, task_vars):
+        return json.dumps(json_response)
+
+    check = canned_podlogquerytime(exec_oc)
+
+    error = ""
+    if expect_exception:
+        with pytest.raises(expect_exception) as error:
+            check.wait_until_cmd_or_err(plain_running_elasticsearch_pod, uuid, timeout, None)
+    else:
+         check.wait_until_cmd_or_err(plain_running_elasticsearch_pod, uuid, timeout, None)
+
+    if not expect_exception:
+        assert not error
+
+    for word in extra_words:
+        assert word in str(error)
+
+
+@pytest.mark.parametrize('json_response, uuid, expect_exception, extra_words', [
+    # test with invalid json response
+    (
+        {
+            "invalid_field": "invalid",
+        },
+        SAMPLE_UID,
+        OpenShiftCheckException,
+        ["expecting server error", "404", "None"],
+    ),
+    # test with wrong error code
+    (
+        {
+            "statusCode": 500,
+        },
+        SAMPLE_UID,
+        OpenShiftCheckException,
+        ["expecting server error", "500"],
+    ),
+    (
+        {
+            "statusCode": 404,
+        },
+        "sample unique id",
+        None,
+        ["sample unique id"],
+    ),
+])
+def test_curl_kibana_with_uuid(json_response, uuid, expect_exception, extra_words):
+    def exec_oc(exec_cmd, args, task_vars):
+        return json.dumps(json_response)
+
+    check = canned_podlogquerytime(exec_oc)
+    check.generate_uuid = lambda: uuid
+
+    error = ""
+    result = ""
+    if expect_exception:
+        with pytest.raises(expect_exception) as error:
+            check.curl_kibana_with_uuid(plain_running_kibana_pod, None)
+    else:
+         result = check.curl_kibana_with_uuid(plain_running_kibana_pod, None)
+
+    if not expect_exception:
+        assert not error
+
+    for word in extra_words:
+        if error:
+            assert word in str(error)
+        elif result:
+            assert word in result