diff options
Diffstat (limited to 'roles/openshift_health_checker/test')
8 files changed, 642 insertions, 89 deletions
diff --git a/roles/openshift_health_checker/test/aos_version_test.py b/roles/openshift_health_checker/test/aos_version_test.py index 697805dd2..4100f6c70 100644 --- a/roles/openshift_health_checker/test/aos_version_test.py +++ b/roles/openshift_health_checker/test/aos_version_test.py @@ -18,7 +18,43 @@ expected_pkgs = { } -@pytest.mark.parametrize('pkgs, expect_not_found', [ +@pytest.mark.parametrize('pkgs,expected_pkgs_dict', [ + ( + # all found + [Package('spam', '3.2.1'), Package('eggs', '3.2.1')], + expected_pkgs, + ), + ( + # found with more specific version + [Package('spam', '3.2.1'), Package('eggs', '3.2.1.5')], + expected_pkgs, + ), + ( + [Package('ovs', '2.6'), Package('ovs', '2.4')], + { + "ovs": { + "name": "ovs", + "version": ["2.6", "2.7"], + "check_multi": False, + } + }, + ), + ( + [Package('ovs', '2.7')], + { + "ovs": { + "name": "ovs", + "version": ["2.6", "2.7"], + "check_multi": False, + } + }, + ), +]) +def test_check_precise_version_found(pkgs, expected_pkgs_dict): + aos_version._check_precise_version_found(pkgs, expected_pkgs_dict) + + +@pytest.mark.parametrize('pkgs,expect_not_found', [ ( [], { @@ -55,14 +91,6 @@ expected_pkgs = { }, # not the right version ), ( - [Package('spam', '3.2.1'), Package('eggs', '3.2.1')], - {}, # all found - ), - ( - [Package('spam', '3.2.1'), Package('eggs', '3.2.1.5')], - {}, # found with more specific version - ), - ( [Package('eggs', '1.2.3'), Package('eggs', '3.2.1.5')], { "spam": { @@ -73,64 +101,86 @@ expected_pkgs = { }, # eggs found with multiple versions ), ]) -def test_check_pkgs_for_precise_version(pkgs, expect_not_found): - if expect_not_found: - with pytest.raises(aos_version.PreciseVersionNotFound) as e: - aos_version._check_precise_version_found(pkgs, expected_pkgs) - - assert list(expect_not_found.values()) == e.value.problem_pkgs - else: +def test_check_precise_version_found_fail(pkgs, expect_not_found): + with pytest.raises(aos_version.PreciseVersionNotFound) as e: aos_version._check_precise_version_found(pkgs, expected_pkgs) + assert list(expect_not_found.values()) == e.value.problem_pkgs -@pytest.mark.parametrize('pkgs, expect_higher', [ +@pytest.mark.parametrize('pkgs,expected_pkgs_dict', [ ( [], - [], + expected_pkgs, ), ( + # more precise but not strictly higher [Package('spam', '3.2.1.9')], - [], # more precise but not strictly higher + expected_pkgs, ), ( + [Package('ovs', '2.7')], + { + "ovs": { + "name": "ovs", + "version": ["2.6", "2.7"], + "check_multi": False, + } + }, + ), +]) +def test_check_higher_version_found(pkgs, expected_pkgs_dict): + aos_version._check_higher_version_found(pkgs, expected_pkgs_dict) + + +@pytest.mark.parametrize('pkgs,expected_pkgs_dict,expect_higher', [ + ( [Package('spam', '3.3')], + expected_pkgs, ['spam-3.3'], # lower precision, but higher ), ( [Package('spam', '3.2.1'), Package('eggs', '3.3.2')], + expected_pkgs, ['eggs-3.3.2'], # one too high ), ( [Package('eggs', '1.2.3'), Package('eggs', '3.2.1.5'), Package('eggs', '3.4')], + expected_pkgs, ['eggs-3.4'], # multiple versions, one is higher ), ( [Package('eggs', '3.2.1'), Package('eggs', '3.4'), Package('eggs', '3.3')], + expected_pkgs, ['eggs-3.4'], # multiple versions, two are higher ), + ( + [Package('ovs', '2.8')], + { + "ovs": { + "name": "ovs", + "version": ["2.6", "2.7"], + "check_multi": False, + } + }, + ['ovs-2.8'], + ), ]) -def test_check_pkgs_for_greater_version(pkgs, expect_higher): - if expect_higher: - with pytest.raises(aos_version.FoundHigherVersion) as e: - aos_version._check_higher_version_found(pkgs, expected_pkgs) - assert set(expect_higher) == set(e.value.problem_pkgs) - else: - aos_version._check_higher_version_found(pkgs, expected_pkgs) +def test_check_higher_version_found_fail(pkgs, expected_pkgs_dict, expect_higher): + with pytest.raises(aos_version.FoundHigherVersion) as e: + aos_version._check_higher_version_found(pkgs, expected_pkgs_dict) + assert set(expect_higher) == set(e.value.problem_pkgs) -@pytest.mark.parametrize('pkgs, expect_to_flag_pkgs', [ - ( - [], - [], - ), - ( - [Package('spam', '3.2.1')], - [], - ), - ( - [Package('spam', '3.2.1'), Package('eggs', '3.2.2')], - [], - ), +@pytest.mark.parametrize('pkgs', [ + [], + [Package('spam', '3.2.1')], + [Package('spam', '3.2.1'), Package('eggs', '3.2.2')], +]) +def test_check_multi_minor_release(pkgs): + aos_version._check_multi_minor_release(pkgs, expected_pkgs) + + +@pytest.mark.parametrize('pkgs,expect_to_flag_pkgs', [ ( [Package('spam', '3.2.1'), Package('spam', '3.3.2')], ['spam'], @@ -140,10 +190,7 @@ def test_check_pkgs_for_greater_version(pkgs, expect_higher): ['eggs'], ), ]) -def test_check_pkgs_for_multi_release(pkgs, expect_to_flag_pkgs): - if expect_to_flag_pkgs: - with pytest.raises(aos_version.FoundMultiRelease) as e: - aos_version._check_multi_minor_release(pkgs, expected_pkgs) - assert set(expect_to_flag_pkgs) == set(e.value.problem_pkgs) - else: +def test_check_multi_minor_release_fail(pkgs, expect_to_flag_pkgs): + with pytest.raises(aos_version.FoundMultiRelease) as e: aos_version._check_multi_minor_release(pkgs, expected_pkgs) + assert set(expect_to_flag_pkgs) == set(e.value.problem_pkgs) diff --git a/roles/openshift_health_checker/test/docker_storage_test.py b/roles/openshift_health_checker/test/docker_storage_test.py index 876614b1d..99c529054 100644 --- a/roles/openshift_health_checker/test/docker_storage_test.py +++ b/roles/openshift_health_checker/test/docker_storage_test.py @@ -23,7 +23,8 @@ def test_is_active(is_containerized, group_names, is_active): assert DockerStorage.is_active(task_vars=task_vars) == is_active -non_atomic_task_vars = {"openshift": {"common": {"is_atomic": False}}} +def non_atomic_task_vars(): + return {"openshift": {"common": {"is_atomic": False}}} @pytest.mark.parametrize('docker_info, failed, expect_msg', [ @@ -56,7 +57,7 @@ non_atomic_task_vars = {"openshift": {"common": {"is_atomic": False}}} ( dict(info={ "Driver": "overlay2", - "DriverStatus": [] + "DriverStatus": [("Backing Filesystem", "xfs")], }), False, [], @@ -64,9 +65,30 @@ non_atomic_task_vars = {"openshift": {"common": {"is_atomic": False}}} ( dict(info={ "Driver": "overlay", + "DriverStatus": [("Backing Filesystem", "btrfs")], }), True, - ["unsupported Docker storage driver"], + ["storage is type 'btrfs'", "only supported with\n'xfs'"], + ), + ( + dict(info={ + "Driver": "overlay2", + "DriverStatus": [("Backing Filesystem", "xfs")], + "OperatingSystem": "Red Hat Enterprise Linux Server release 7.2 (Maipo)", + "KernelVersion": "3.10.0-327.22.2.el7.x86_64", + }), + True, + ["Docker reports kernel version 3.10.0-327"], + ), + ( + dict(info={ + "Driver": "overlay", + "DriverStatus": [("Backing Filesystem", "xfs")], + "OperatingSystem": "CentOS", + "KernelVersion": "3.10.0-514", + }), + False, + [], ), ( dict(info={ @@ -85,8 +107,9 @@ def test_check_storage_driver(docker_info, failed, expect_msg): return docker_info check = dummy_check(execute_module=execute_module) - check._check_dm_usage = lambda status, task_vars: dict() # stub out for this test - result = check.run(tmp=None, task_vars=non_atomic_task_vars) + check.check_dm_usage = lambda status, task_vars: dict() # stub out for this test + check.check_overlay_usage = lambda info, task_vars: dict() # stub out for this test + result = check.run(tmp=None, task_vars=non_atomic_task_vars()) if failed: assert result["failed"] @@ -146,8 +169,8 @@ not_enough_space = { ]) def test_dm_usage(task_vars, driver_status, vg_free, success, expect_msg): check = dummy_check() - check._get_vg_free = lambda pool, task_vars: vg_free - result = check._check_dm_usage(driver_status, task_vars) + check.get_vg_free = lambda pool, task_vars: vg_free + result = check.check_dm_usage(driver_status, task_vars) result_success = not result.get("failed") assert result_success is success @@ -195,10 +218,10 @@ def test_vg_free(pool, command_returns, raises, returns): check = dummy_check(execute_module=execute_module) if raises: with pytest.raises(OpenShiftCheckException) as err: - check._get_vg_free(pool, {}) + check.get_vg_free(pool, {}) assert raises in str(err.value) else: - ret = check._get_vg_free(pool, {}) + ret = check.get_vg_free(pool, {}) assert ret == returns @@ -209,7 +232,7 @@ def test_vg_free(pool, command_returns, raises, returns): ("12g", 12.0 * 1024**3), ]) def test_convert_to_bytes(string, expect_bytes): - got = DockerStorage._convert_to_bytes(string) + got = DockerStorage.convert_to_bytes(string) assert got == expect_bytes @@ -219,6 +242,70 @@ def test_convert_to_bytes(string, expect_bytes): ]) def test_convert_to_bytes_error(string): with pytest.raises(ValueError) as err: - DockerStorage._convert_to_bytes(string) + DockerStorage.convert_to_bytes(string) assert "Cannot convert" in str(err.value) assert string in str(err.value) + + +ansible_mounts_enough = [{ + 'mount': '/var/lib/docker', + 'size_available': 50 * 10**9, + 'size_total': 50 * 10**9, +}] +ansible_mounts_not_enough = [{ + 'mount': '/var/lib/docker', + 'size_available': 0, + 'size_total': 50 * 10**9, +}] +ansible_mounts_missing_fields = [dict(mount='/var/lib/docker')] +ansible_mounts_zero_size = [{ + 'mount': '/var/lib/docker', + 'size_available': 0, + 'size_total': 0, +}] + + +@pytest.mark.parametrize('ansible_mounts, threshold, expect_fail, expect_msg', [ + ( + ansible_mounts_enough, + None, + False, + [], + ), + ( + ansible_mounts_not_enough, + None, + True, + ["usage percentage", "higher than threshold"], + ), + ( + ansible_mounts_not_enough, + "bogus percent", + True, + ["is not a percentage"], + ), + ( + ansible_mounts_missing_fields, + None, + True, + ["Ansible bug"], + ), + ( + ansible_mounts_zero_size, + None, + True, + ["Ansible bug"], + ), +]) +def test_overlay_usage(ansible_mounts, threshold, expect_fail, expect_msg): + check = dummy_check() + task_vars = non_atomic_task_vars() + task_vars["ansible_mounts"] = ansible_mounts + if threshold is not None: + task_vars["max_overlay_usage_percent"] = threshold + docker_info = dict(DockerRootDir="/var/lib/docker", Driver="overlay") + result = check.check_overlay_usage(docker_info, task_vars) + + assert expect_fail == bool(result.get("failed")) + for msg in expect_msg: + assert msg in result["msg"] diff --git a/roles/openshift_health_checker/test/etcd_traffic_test.py b/roles/openshift_health_checker/test/etcd_traffic_test.py new file mode 100644 index 000000000..287175e29 --- /dev/null +++ b/roles/openshift_health_checker/test/etcd_traffic_test.py @@ -0,0 +1,80 @@ +import pytest + +from openshift_checks.etcd_traffic import EtcdTraffic + + +@pytest.mark.parametrize('group_names,version,is_active', [ + (['masters'], "3.5", False), + (['masters'], "3.6", False), + (['nodes'], "3.4", False), + (['etcd'], "3.4", True), + (['etcd'], "3.5", True), + (['etcd'], "3.1", False), + (['masters', 'nodes'], "3.5", False), + (['masters', 'etcd'], "3.5", True), + ([], "3.4", False), +]) +def test_is_active(group_names, version, is_active): + task_vars = dict( + group_names=group_names, + openshift=dict( + common=dict(short_version=version), + ), + ) + assert EtcdTraffic.is_active(task_vars=task_vars) == is_active + + +@pytest.mark.parametrize('group_names,matched,failed,extra_words', [ + (["masters"], True, True, ["Higher than normal", "traffic"]), + (["masters", "etcd"], False, False, []), + (["etcd"], False, False, []), +]) +def test_log_matches_high_traffic_msg(group_names, matched, failed, extra_words): + def execute_module(module_name, args, task_vars): + return { + "matched": matched, + "failed": failed, + } + + task_vars = dict( + group_names=group_names, + openshift=dict( + common=dict(service_type="origin", is_containerized=False), + ) + ) + + check = EtcdTraffic(execute_module=execute_module) + result = check.run(tmp=None, task_vars=task_vars) + + for word in extra_words: + assert word in result.get("msg", "") + + assert result.get("failed", False) == failed + + +@pytest.mark.parametrize('is_containerized,expected_unit_value', [ + (False, "etcd"), + (True, "etcd_container"), +]) +def test_systemd_unit_matches_deployment_type(is_containerized, expected_unit_value): + task_vars = dict( + openshift=dict( + common=dict(is_containerized=is_containerized), + ) + ) + + def execute_module(module_name, args, task_vars): + assert module_name == "search_journalctl" + matchers = args["log_matchers"] + + for matcher in matchers: + assert matcher["unit"] == expected_unit_value + + return {"failed": False} + + check = EtcdTraffic(execute_module=execute_module) + check.run(tmp=None, task_vars=task_vars) + + +def fake_execute_module(*args): + raise AssertionError('this function should not be called') diff --git a/roles/openshift_health_checker/test/kibana_test.py b/roles/openshift_health_checker/test/kibana_test.py index 19140a1b6..40a5d19d8 100644 --- a/roles/openshift_health_checker/test/kibana_test.py +++ b/roles/openshift_health_checker/test/kibana_test.py @@ -169,7 +169,7 @@ def test_get_kibana_url(route, expect_url, expect_error): ), ]) def test_verify_url_internal_failure(exec_result, expect): - check = Kibana(execute_module=lambda module_name, args, task_vars: dict(failed=True, msg=exec_result)) + check = Kibana(execute_module=lambda module_name, args, tmp, task_vars: dict(failed=True, msg=exec_result)) check._get_kibana_url = lambda task_vars: ('url', None) error = check._check_kibana_route({}) diff --git a/roles/openshift_health_checker/test/logging_check_test.py b/roles/openshift_health_checker/test/logging_check_test.py index b6db34fe3..4f71fbf52 100644 --- a/roles/openshift_health_checker/test/logging_check_test.py +++ b/roles/openshift_health_checker/test/logging_check_test.py @@ -50,6 +50,16 @@ plain_kibana_pod = { } } +plain_kibana_pod_no_containerstatus = { + "metadata": { + "labels": {"component": "kibana", "deploymentconfig": "logging-kibana"}, + "name": "logging-kibana-1", + }, + "status": { + "conditions": [{"status": "True", "type": "Ready"}], + } +} + fluentd_pod_node1 = { "metadata": { "labels": {"component": "fluentd", "deploymentconfig": "logging-fluentd"}, @@ -80,7 +90,7 @@ plain_curator_pod = { ("Permission denied", "Unexpected error using `oc`"), ]) def test_oc_failure(problem, expect): - def execute_module(module_name, args, task_vars): + def execute_module(module_name, args, tmp, task_vars): if module_name == "ocutil": return dict(failed=True, result=problem) return dict(changed=False) @@ -118,7 +128,7 @@ def test_is_active(groups, logging_deployed, is_active): ( 'No resources found.', None, - 'There are no pods in the logging namespace', + 'No pods were found for the "es"', ), ( json.dumps({'items': [plain_kibana_pod, plain_es_pod, plain_curator_pod, fluentd_pod_node1]}), @@ -135,3 +145,23 @@ def test_get_pods_for_component(pod_output, expect_pods, expect_error): {} ) assert_error(error, expect_error) + + +@pytest.mark.parametrize('name, pods, expected_pods', [ + ( + 'test single pod found, scheduled, but no containerStatuses field', + [plain_kibana_pod_no_containerstatus], + [plain_kibana_pod_no_containerstatus], + ), + ( + 'set of pods has at least one pod with containerStatuses (scheduled); should still fail', + [plain_kibana_pod_no_containerstatus, plain_kibana_pod], + [plain_kibana_pod_no_containerstatus], + ), + +], ids=lambda argvals: argvals[0]) +def test_get_not_running_pods_no_container_status(name, pods, expected_pods): + check = canned_loggingcheck(lambda exec_module, namespace, cmd, args, task_vars: '') + result = check.not_running_pods(pods) + + assert result == expected_pods diff --git a/roles/openshift_health_checker/test/logging_index_time_test.py b/roles/openshift_health_checker/test/logging_index_time_test.py new file mode 100644 index 000000000..79e7c7d4c --- /dev/null +++ b/roles/openshift_health_checker/test/logging_index_time_test.py @@ -0,0 +1,182 @@ +import json + +import pytest + +from openshift_checks.logging.logging_index_time import LoggingIndexTime, OpenShiftCheckException + + +SAMPLE_UUID = "unique-test-uuid" + + +def canned_loggingindextime(exec_oc=None): + """Create a check object with a canned exec_oc method""" + check = LoggingIndexTime("dummy") # fails if a module is actually invoked + if exec_oc: + check.exec_oc = exec_oc + return check + + +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_loggingindextime(None) + pods = check.running_pods(pods) + assert pods == expect_pods + + +@pytest.mark.parametrize('name, json_response, uuid, timeout, extra_words', [ + ( + 'valid count in response', + { + "count": 1, + }, + SAMPLE_UUID, + 0.001, + [], + ), +], ids=lambda argval: argval[0]) +def test_wait_until_cmd_or_err_succeeds(name, json_response, uuid, timeout, extra_words): + def exec_oc(execute_module, ns, exec_cmd, args, task_vars): + return json.dumps(json_response) + + check = canned_loggingindextime(exec_oc) + check.wait_until_cmd_or_err(plain_running_elasticsearch_pod, uuid, timeout, None) + + +@pytest.mark.parametrize('name, json_response, uuid, timeout, extra_words', [ + ( + 'invalid json response', + { + "invalid_field": 1, + }, + SAMPLE_UUID, + 0.001, + ["invalid response", "Elasticsearch"], + ), + ( + 'empty response', + {}, + SAMPLE_UUID, + 0.001, + ["invalid response", "Elasticsearch"], + ), + ( + 'valid response but invalid match count', + { + "count": 0, + }, + SAMPLE_UUID, + 0.005, + ["expecting match", SAMPLE_UUID, "0.005s"], + ) +], ids=lambda argval: argval[0]) +def test_wait_until_cmd_or_err(name, json_response, uuid, timeout, extra_words): + def exec_oc(execute_module, ns, exec_cmd, args, task_vars): + return json.dumps(json_response) + + check = canned_loggingindextime(exec_oc) + with pytest.raises(OpenShiftCheckException) as error: + check.wait_until_cmd_or_err(plain_running_elasticsearch_pod, uuid, timeout, None) + + for word in extra_words: + assert word in str(error) + + +@pytest.mark.parametrize('name, json_response, uuid, extra_words', [ + ( + 'correct response code, found unique id is returned', + { + "statusCode": 404, + }, + "sample unique id", + ["sample unique id"], + ), +], ids=lambda argval: argval[0]) +def test_curl_kibana_with_uuid(name, json_response, uuid, extra_words): + def exec_oc(execute_module, ns, exec_cmd, args, task_vars): + return json.dumps(json_response) + + check = canned_loggingindextime(exec_oc) + check.generate_uuid = lambda: uuid + + result = check.curl_kibana_with_uuid(plain_running_kibana_pod, None) + + for word in extra_words: + assert word in result + + +@pytest.mark.parametrize('name, json_response, uuid, extra_words', [ + ( + 'invalid json response', + { + "invalid_field": "invalid", + }, + SAMPLE_UUID, + ["invalid response returned", 'Missing "statusCode" key'], + ), + ( + 'wrong error code in response', + { + "statusCode": 500, + }, + SAMPLE_UUID, + ["Expecting error code", "500"], + ), +], ids=lambda argval: argval[0]) +def test_failed_curl_kibana_with_uuid(name, json_response, uuid, extra_words): + def exec_oc(execute_module, ns, exec_cmd, args, task_vars): + return json.dumps(json_response) + + check = canned_loggingindextime(exec_oc) + check.generate_uuid = lambda: uuid + + with pytest.raises(OpenShiftCheckException) as error: + check.curl_kibana_with_uuid(plain_running_kibana_pod, None) + + for word in extra_words: + assert word in str(error) diff --git a/roles/openshift_health_checker/test/package_version_test.py b/roles/openshift_health_checker/test/package_version_test.py index 91eace512..1bb6371ae 100644 --- a/roles/openshift_health_checker/test/package_version_test.py +++ b/roles/openshift_health_checker/test/package_version_test.py @@ -72,36 +72,6 @@ def test_package_version(openshift_release): assert result is return_value -@pytest.mark.parametrize('deployment_type,openshift_release,expected_ovs_version', [ - ("openshift-enterprise", "3.5", "2.6"), - ("origin", "3.6", "2.6"), - ("openshift-enterprise", "3.4", "2.4"), - ("origin", "3.3", "2.4"), -]) -def test_ovs_package_version(deployment_type, openshift_release, expected_ovs_version): - task_vars = dict( - openshift=dict(common=dict(service_type='origin')), - openshift_release=openshift_release, - openshift_image_tag='v' + openshift_release, - openshift_deployment_type=deployment_type, - ) - return_value = object() - - def execute_module(module_name=None, module_args=None, tmp=None, task_vars=None): - assert module_name == 'aos_version' - assert "package_list" in module_args - - for pkg in module_args["package_list"]: - if pkg["name"] == "openvswitch": - assert pkg["version"] == expected_ovs_version - - return return_value - - check = PackageVersion(execute_module=execute_module) - result = check.run(tmp=None, task_vars=task_vars) - assert result is return_value - - @pytest.mark.parametrize('deployment_type,openshift_release,expected_docker_version', [ ("origin", "3.5", "1.12"), ("openshift-enterprise", "3.4", "1.12"), diff --git a/roles/openshift_health_checker/test/search_journalctl_test.py b/roles/openshift_health_checker/test/search_journalctl_test.py new file mode 100644 index 000000000..724928aa1 --- /dev/null +++ b/roles/openshift_health_checker/test/search_journalctl_test.py @@ -0,0 +1,157 @@ +import pytest +import search_journalctl + + +def canned_search_journalctl(get_log_output=None): + """Create a search_journalctl object with canned get_log_output method""" + module = search_journalctl + if get_log_output: + module.get_log_output = get_log_output + return module + + +DEFAULT_TIMESTAMP = 1496341364 + + +def get_timestamp(modifier=0): + return DEFAULT_TIMESTAMP + modifier + + +def get_timestamp_microseconds(modifier=0): + return get_timestamp(modifier) * 1000000 + + +def create_test_log_object(stamp, msg): + return '{{"__REALTIME_TIMESTAMP": "{}", "MESSAGE": "{}"}}'.format(stamp, msg) + + +@pytest.mark.parametrize('name,matchers,log_input,expected_matches,expected_errors', [ + ( + 'test with valid params', + [ + { + "start_regexp": r"Sample Logs Beginning", + "regexp": r"test log message", + "unit": "test", + }, + ], + [ + create_test_log_object(get_timestamp_microseconds(), "test log message"), + create_test_log_object(get_timestamp_microseconds(), "Sample Logs Beginning"), + ], + ["test log message"], + [], + ), + ( + 'test with invalid json in log input', + [ + { + "start_regexp": r"Sample Logs Beginning", + "regexp": r"test log message", + "unit": "test-unit", + }, + ], + [ + '{__REALTIME_TIMESTAMP: ' + str(get_timestamp_microseconds()) + ', "MESSAGE": "test log message"}', + ], + [], + [ + ["invalid json", "test-unit", "test log message"], + ], + ), + ( + 'test with invalid regexp', + [ + { + "start_regexp": r"Sample Logs Beginning", + "regexp": r"test [ log message", + "unit": "test", + }, + ], + [ + create_test_log_object(get_timestamp_microseconds(), "test log message"), + create_test_log_object(get_timestamp_microseconds(), "sample log message"), + create_test_log_object(get_timestamp_microseconds(), "fake log message"), + create_test_log_object(get_timestamp_microseconds(), "dummy log message"), + create_test_log_object(get_timestamp_microseconds(), "Sample Logs Beginning"), + ], + [], + [ + ["invalid regular expression"], + ], + ), +], ids=lambda argval: argval[0]) +def test_get_log_matches(name, matchers, log_input, expected_matches, expected_errors): + def get_log_output(matcher): + return log_input + + module = canned_search_journalctl(get_log_output) + matched_regexp, errors = module.get_log_matches(matchers, 500, 60 * 60) + + assert set(matched_regexp) == set(expected_matches) + assert len(expected_errors) == len(errors) + + for idx, partial_err_set in enumerate(expected_errors): + for partial_err_msg in partial_err_set: + assert partial_err_msg in errors[idx] + + +@pytest.mark.parametrize('name,matcher,log_count_lim,stamp_lim_seconds,log_input,expected_match', [ + ( + 'test with matching log message, but out of bounds of log_count_lim', + { + "start_regexp": r"Sample Logs Beginning", + "regexp": r"dummy log message", + "unit": "test", + }, + 3, + get_timestamp(-100 * 60 * 60), + [ + create_test_log_object(get_timestamp_microseconds(), "test log message"), + create_test_log_object(get_timestamp_microseconds(), "sample log message"), + create_test_log_object(get_timestamp_microseconds(), "fake log message"), + create_test_log_object(get_timestamp_microseconds(), "dummy log message"), + create_test_log_object(get_timestamp_microseconds(), "Sample Logs Beginning"), + ], + None, + ), + ( + 'test with matching log message, but with timestamp too old', + { + "start_regexp": r"Sample Logs Beginning", + "regexp": r"dummy log message", + "unit": "test", + }, + 100, + get_timestamp(-10), + [ + create_test_log_object(get_timestamp_microseconds(), "test log message"), + create_test_log_object(get_timestamp_microseconds(), "sample log message"), + create_test_log_object(get_timestamp_microseconds(), "fake log message"), + create_test_log_object(get_timestamp_microseconds(-1000), "dummy log message"), + create_test_log_object(get_timestamp_microseconds(-1000), "Sample Logs Beginning"), + ], + None, + ), + ( + 'test with matching log message, and timestamp within time limit', + { + "start_regexp": r"Sample Logs Beginning", + "regexp": r"dummy log message", + "unit": "test", + }, + 100, + get_timestamp(-1010), + [ + create_test_log_object(get_timestamp_microseconds(), "test log message"), + create_test_log_object(get_timestamp_microseconds(), "sample log message"), + create_test_log_object(get_timestamp_microseconds(), "fake log message"), + create_test_log_object(get_timestamp_microseconds(-1000), "dummy log message"), + create_test_log_object(get_timestamp_microseconds(-1000), "Sample Logs Beginning"), + ], + create_test_log_object(get_timestamp_microseconds(-1000), "dummy log message"), + ), +], ids=lambda argval: argval[0]) +def test_find_matches_skips_logs(name, matcher, log_count_lim, stamp_lim_seconds, log_input, expected_match): + match = search_journalctl.find_matches(log_input, matcher, log_count_lim, stamp_lim_seconds) + assert match == expected_match |