<filename>tests/filter_test.py
# pylint: skip-file

import pytest
from server.dict_filter import DictFilter, FilterException

TEST_LOG_DB = [
    {
        'asctime': '2022-01-11 15:29:01,041',
        'name': 'FleetManagerClient',
        'levelname': 'INFO',
        'message': 'Message',
        'source_ip': '192.168.1.89'
    },
    {
        'asctime': '2022-01-11 15:29:01,587',
        'name': 'fleet-manager-client',
        'levelname': 'DEBUG',
        'message': 'Message',
        'source_ip': '127.0.0.1'
    },
    {
        'asctime': '2022-01-11 15:29:01,602',
        'name': 'fleet-manager-server',
        'levelname': 'WARNING',
        'message': 'Another message',
        'source_ip': '192.168.1.89'
    },
    {
        'asctime': '2022-01-11 15:29:11,073',
        'name': 'FleetManagerClient',
        'levelname': 'WARNING',
        'message': 'Message',
        'source_ip': '127.0.0.1'
    },
    {
        'asctime': '2022-01-11 15:29:11,084',
        'name': 'fleet-manager-client',
        'levelname': 'DEBUG',
        'message': 'Message',
        'source_ip': '192.168.1.89'
    },
    {
        'asctime': '2022-01-11 15:29:11,096',
        'name': 'fleet-manager-server',
        'levelname': 'DEBUG',
        'message': 'Message',
        'source_ip': '192.168.1.91'
    }
]

KEY_ALIAS = {
    'source_ip': [
        'ip'
    ],
    'levelname': [
        'level'
    ]
}

TEST_DATA = [
    ("source_ip=192.168.1.91",
    [
        {
            'asctime': '2022-01-11 15:29:11,096',
            'name': 'fleet-manager-server',
            'levelname': 'DEBUG',
            'message': 'Message',
            'source_ip': '192.168.1.91'
        }
    ]),
    ("ip=192.168.1.89",
    [
        {
            'asctime': '2022-01-11 15:29:01,041',
            'name': 'FleetManagerClient',
            'levelname': 'INFO',
            'message': 'Message',
            'source_ip': '192.168.1.89'
        },
        {
            'asctime': '2022-01-11 15:29:01,602',
            'name': 'fleet-manager-server',
            'levelname': 'WARNING',
            'message': 'Another message',
            'source_ip': '192.168.1.89'
        },
        {
        'asctime': '2022-01-11 15:29:11,084',
        'name': 'fleet-manager-client',
        'levelname': 'DEBUG',
        'message': 'Message',
        'source_ip': '192.168.1.89'
        }
    ]),
    ("ip=192.168.1.89;level=WARNING",
    [
        {
            'asctime': '2022-01-11 15:29:01,602',
            'name': 'fleet-manager-server',
            'levelname': 'WARNING',
            'message': 'Another message',
            'source_ip': '192.168.1.89'
        }
    ]),
    ("ip=19172.16.17.32;level=WARNING,DEBUG",
    [
        {
            'asctime': '2022-01-11 15:29:01,602',
            'name': 'fleet-manager-server',
            'levelname': 'WARNING',
            'message': 'Another message',
            'source_ip': '192.168.1.89'
        },
        {
            'asctime': '2022-01-11 15:29:11,084',
            'name': 'fleet-manager-client',
            'levelname': 'DEBUG',
            'message': 'Message',
            'source_ip': '192.168.1.89'
        }
    ]),
    ("levelname=DEBUG",
    [
        {
            'asctime': '2022-01-11 15:29:01,587',
            'name': 'fleet-manager-client',
            'levelname': 'DEBUG',
            'message': 'Message',
            'source_ip': '127.0.0.1'
        },
        {
            'asctime': '2022-01-11 15:29:11,084',
            'name': 'fleet-manager-client',
            'levelname': 'DEBUG',
            'message': 'Message',
            'source_ip': '192.168.1.89'
        },
        {
            'asctime': '2022-01-11 15:29:11,096',
            'name': 'fleet-manager-server',
            'levelname': 'DEBUG',
            'message': 'Message',
            'source_ip': '192.168.1.91'
        }
    ]),
]

TEST_DATA_WITH_BAD_INPUT = [
    ('level=',
    TEST_LOG_DB),
    ('ip=192.168.1.89;level=',
    [
        {
            'asctime': '2022-01-11 15:29:01,041',
            'name': 'FleetManagerClient',
            'levelname': 'INFO',
            'message': 'Message',
            'source_ip': '192.168.1.89'
        },
        {
            'asctime': '2022-01-11 15:29:01,602',
            'name': 'fleet-manager-server',
            'levelname': 'WARNING',
            'message': 'Another message',
            'source_ip': '192.168.1.89'
        },
        {
        'asctime': '2022-01-11 15:29:11,084',
        'name': 'fleet-manager-client',
        'levelname': 'DEBUG',
        'message': 'Message',
        'source_ip': '192.168.1.89'
        }
    ]),
    ('level=;ip=192.168.1.89',
    [
        {
            'asctime': '2022-01-11 15:29:01,041',
            'name': 'FleetManagerClient',
            'levelname': 'INFO',
            'message': 'Message',
            'source_ip': '192.168.1.89'
        },
        {
            'asctime': '2022-01-11 15:29:01,602',
            'name': 'fleet-manager-server',
            'levelname': 'WARNING',
            'message': 'Another message',
            'source_ip': '192.168.1.89'
        },
        {
        'asctime': '2022-01-11 15:29:11,084',
        'name': 'fleet-manager-client',
        'levelname': 'DEBUG',
        'message': 'Message',
        'source_ip': '192.168.1.89'
        }
    ]),
]

TEST_DATA_WITH_ERROR_INPUT = [
    ('',
    FilterException,
    'No filter string defined'),
    (None,
    FilterException,
    'No filter string defined'),
    ('not-a-key=1',
    FilterException,
    'Key: "not-a-key" could not be translated or found'),
]

@pytest.mark.parametrize("test_string, expected_result", TEST_DATA)
def test_filter(test_string, expected_result):
    with DictFilter(filter_string=test_string, key_alias=KEY_ALIAS) as dict_filter:
        filtered_db = dict_filter.filter(TEST_LOG_DB)

        assert len(filtered_db) == len(expected_result)
        assert filtered_db == expected_result

@pytest.mark.parametrize("test_string, expected_result", TEST_DATA_WITH_BAD_INPUT)
def test_filter_with_bad_input(test_string, expected_result):
    with DictFilter(filter_string=test_string, key_alias=KEY_ALIAS) as dict_filter:
        filtered_db = dict_filter.filter(TEST_LOG_DB)

        assert len(filtered_db) == len(expected_result)
        assert filtered_db == expected_result

@pytest.mark.parametrize("test_string, exception, exception_message", TEST_DATA_WITH_ERROR_INPUT)
def test_filter_with_error_input(test_string, exception, exception_message):
    with pytest.raises(exception) as e_info:
        with DictFilter(filter_string=test_string, key_alias=KEY_ALIAS) as dict_filter:
            dict_filter.filter(TEST_LOG_DB)

    assert str(e_info.value) == exception_message
<filename>tests/filter_test.py
# pylint: skip-file

import pytest
from server.dict_filter import DictFilter, FilterException

TEST_LOG_DB = [
    {
        'asctime': '2022-01-11 15:29:01,041',
        'name': 'FleetManagerClient',
        'levelname': 'INFO',
        'message': 'Message',
        'source_ip': '192.168.1.89'
    },
    {
        'asctime': '2022-01-11 15:29:01,587',
        'name': 'fleet-manager-client',
        'levelname': 'DEBUG',
        'message': 'Message',
        'source_ip': '127.0.0.1'
    },
    {
        'asctime': '2022-01-11 15:29:01,602',
        'name': 'fleet-manager-server',
        'levelname': 'WARNING',
        'message': 'Another message',
        'source_ip': '192.168.1.89'
    },
    {
        'asctime': '2022-01-11 15:29:11,073',
        'name': 'FleetManagerClient',
        'levelname': 'WARNING',
        'message': 'Message',
        'source_ip': '127.0.0.1'
    },
    {
        'asctime': '2022-01-11 15:29:11,084',
        'name': 'fleet-manager-client',
        'levelname': 'DEBUG',
        'message': 'Message',
        'source_ip': '192.168.1.89'
    },
    {
        'asctime': '2022-01-11 15:29:11,096',
        'name': 'fleet-manager-server',
        'levelname': 'DEBUG',
        'message': 'Message',
        'source_ip': '192.168.1.91'
    }
]

KEY_ALIAS = {
    'source_ip': [
        'ip'
    ],
    'levelname': [
        'level'
    ]
}

TEST_DATA = [
    ("source_ip=192.168.1.91",
    [
        {
            'asctime': '2022-01-11 15:29:11,096',
            'name': 'fleet-manager-server',
            'levelname': 'DEBUG',
            'message': 'Message',
            'source_ip': '192.168.1.91'
        }
    ]),
    ("ip=192.168.1.89",
    [
        {
            'asctime': '2022-01-11 15:29:01,041',
            'name': 'FleetManagerClient',
            'levelname': 'INFO',
            'message': 'Message',
            'source_ip': '192.168.1.89'
        },
        {
            'asctime': '2022-01-11 15:29:01,602',
            'name': 'fleet-manager-server',
            'levelname': 'WARNING',
            'message': 'Another message',
            'source_ip': '192.168.1.89'
        },
        {
        'asctime': '2022-01-11 15:29:11,084',
        'name': 'fleet-manager-client',
        'levelname': 'DEBUG',
        'message': 'Message',
        'source_ip': '192.168.1.89'
        }
    ]),
    ("ip=192.168.1.89;level=WARNING",
    [
        {
            'asctime': '2022-01-11 15:29:01,602',
            'name': 'fleet-manager-server',
            'levelname': 'WARNING',
            'message': 'Another message',
            'source_ip': '192.168.1.89'
        }
    ]),
    ("ip=19172.16.17.32;level=WARNING,DEBUG",
    [
        {
            'asctime': '2022-01-11 15:29:01,602',
            'name': 'fleet-manager-server',
            'levelname': 'WARNING',
            'message': 'Another message',
            'source_ip': '192.168.1.89'
        },
        {
            'asctime': '2022-01-11 15:29:11,084',
            'name': 'fleet-manager-client',
            'levelname': 'DEBUG',
            'message': 'Message',
            'source_ip': '192.168.1.89'
        }
    ]),
    ("levelname=DEBUG",
    [
        {
            'asctime': '2022-01-11 15:29:01,587',
            'name': 'fleet-manager-client',
            'levelname': 'DEBUG',
            'message': 'Message',
            'source_ip': '127.0.0.1'
        },
        {
            'asctime': '2022-01-11 15:29:11,084',
            'name': 'fleet-manager-client',
            'levelname': 'DEBUG',
            'message': 'Message',
            'source_ip': '192.168.1.89'
        },
        {
            'asctime': '2022-01-11 15:29:11,096',
            'name': 'fleet-manager-server',
            'levelname': 'DEBUG',
            'message': 'Message',
            'source_ip': '192.168.1.91'
        }
    ]),
]

TEST_DATA_WITH_BAD_INPUT = [
    ('level=',
    TEST_LOG_DB),
    ('ip=192.168.1.89;level=',
    [
        {
            'asctime': '2022-01-11 15:29:01,041',
            'name': 'FleetManagerClient',
            'levelname': 'INFO',
            'message': 'Message',
            'source_ip': '192.168.1.89'
        },
        {
            'asctime': '2022-01-11 15:29:01,602',
            'name': 'fleet-manager-server',
            'levelname': 'WARNING',
            'message': 'Another message',
            'source_ip': '192.168.1.89'
        },
        {
        'asctime': '2022-01-11 15:29:11,084',
        'name': 'fleet-manager-client',
        'levelname': 'DEBUG',
        'message': 'Message',
        'source_ip': '192.168.1.89'
        }
    ]),
    ('level=;ip=192.168.1.89',
    [
        {
            'asctime': '2022-01-11 15:29:01,041',
            'name': 'FleetManagerClient',
            'levelname': 'INFO',
            'message': 'Message',
            'source_ip': '192.168.1.89'
        },
        {
            'asctime': '2022-01-11 15:29:01,602',
            'name': 'fleet-manager-server',
            'levelname': 'WARNING',
            'message': 'Another message',
            'source_ip': '192.168.1.89'
        },
        {
        'asctime': '2022-01-11 15:29:11,084',
        'name': 'fleet-manager-client',
        'levelname': 'DEBUG',
        'message': 'Message',
        'source_ip': '192.168.1.89'
        }
    ]),
]

TEST_DATA_WITH_ERROR_INPUT = [
    ('',
    FilterException,
    'No filter string defined'),
    (None,
    FilterException,
    'No filter string defined'),
    ('not-a-key=1',
    FilterException,
    'Key: "not-a-key" could not be translated or found'),
]

@pytest.mark.parametrize("test_string, expected_result", TEST_DATA)
def test_filter(test_string, expected_result):
    with DictFilter(filter_string=test_string, key_alias=KEY_ALIAS) as dict_filter:
        filtered_db = dict_filter.filter(TEST_LOG_DB)

        assert len(filtered_db) == len(expected_result)
        assert filtered_db == expected_result

@pytest.mark.parametrize("test_string, expected_result", TEST_DATA_WITH_BAD_INPUT)
def test_filter_with_bad_input(test_string, expected_result):
    with DictFilter(filter_string=test_string, key_alias=KEY_ALIAS) as dict_filter:
        filtered_db = dict_filter.filter(TEST_LOG_DB)

        assert len(filtered_db) == len(expected_result)
        assert filtered_db == expected_result

@pytest.mark.parametrize("test_string, exception, exception_message", TEST_DATA_WITH_ERROR_INPUT)
def test_filter_with_error_input(test_string, exception, exception_message):
    with pytest.raises(exception) as e_info:
        with DictFilter(filter_string=test_string, key_alias=KEY_ALIAS) as dict_filter:
            dict_filter.filter(TEST_LOG_DB)

    assert str(e_info.value) == exception_message
# pylint: skip-file