177 lines
4.3 KiB
Python
177 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
|
# coding: utf8
|
|
|
|
"""test_ldpy.
|
|
|
|
Testing `ldpy.py`
|
|
|
|
Author: frederic zind 2022-09-24
|
|
Licenses: GNU AGPL v3: http://www.gnu.org/licenses/
|
|
"""
|
|
import logging
|
|
import pytest
|
|
from pytest import mark, raises
|
|
|
|
import ldpy
|
|
from ldpy import LAST_E_CHOICES, STD_FIELDS
|
|
|
|
# Min and Max values for the request retrieving the last entries.
|
|
LAST_E_MIN = LAST_E_CHOICES[0] - 1
|
|
LAST_E_MAX = len(LAST_E_CHOICES) + 1
|
|
META_FIELDS = ["_id", "_source", "_shards", "hits"]
|
|
DEMO_STREAM_FIELDS = ["rating_num", "message", "title", "id", "category"]
|
|
|
|
|
|
# Faking options from argparse
|
|
class FakeOptions: # pylint: disable=too-few-public-methods
|
|
"""Options object mock."""
|
|
|
|
debug = False
|
|
last = None
|
|
mapping = False
|
|
|
|
def __init__(self, options):
|
|
"""Built loggers options."""
|
|
if "debug" in options:
|
|
self.debug = True
|
|
if "last" in options:
|
|
self.last = 3
|
|
if "mapping" in options:
|
|
self.mapping = True
|
|
|
|
|
|
# ###
|
|
# Testing setup_logging()
|
|
# ###
|
|
def test_logging_debug():
|
|
"""Logger level is DEBUG. Set up with --debug argument."""
|
|
options = FakeOptions(["debug"])
|
|
ldpy.setup_logging(options)
|
|
|
|
assert isinstance((ldpy.logger), logging.Logger)
|
|
assert ldpy.logger.level == 10
|
|
|
|
|
|
def test_logging_default():
|
|
"""Logger level is INFO. Set up without arguments."""
|
|
options = FakeOptions([])
|
|
ldpy.setup_logging(options)
|
|
|
|
assert isinstance((ldpy.logger), logging.Logger)
|
|
assert ldpy.logger.level == 20
|
|
|
|
|
|
# ###
|
|
# Testing parse_args()
|
|
# ###
|
|
def test_parse_args():
|
|
"""Mapping argument is set."""
|
|
cli_options = ldpy.parse_args(["--debug", "--mapping"])
|
|
assert cli_options.debug
|
|
assert cli_options.mapping
|
|
|
|
|
|
def test_parse_args_last_set():
|
|
"""Last argument is set to a non-default value."""
|
|
cli_options = ldpy.parse_args(["--debug", "--last", "2"])
|
|
assert cli_options.debug
|
|
assert cli_options.last == 2
|
|
|
|
|
|
def test_parse_args_last_const():
|
|
"""Last argument is set to a non-default value."""
|
|
cli_options = ldpy.parse_args(["--debug", "--last"])
|
|
assert cli_options.debug
|
|
assert cli_options.last == 3
|
|
|
|
|
|
# ###
|
|
# Testing strip_std_field()
|
|
# ###
|
|
@mark.parametrize("fields", [META_FIELDS, STD_FIELDS])
|
|
def test_strip_std_field(fields):
|
|
"""Remove fields present in all LDP stream."""
|
|
payload = [
|
|
"category",
|
|
"gl2_source_input",
|
|
"gl2_source_node",
|
|
"id",
|
|
"message",
|
|
"rating_num",
|
|
"source",
|
|
"streams",
|
|
"timestamp",
|
|
"title",
|
|
"X-OVH-CONTENT-SIZE",
|
|
"X-OVH-DELIVERY-DATE",
|
|
"X-OVH-INPUT",
|
|
]
|
|
stripped_entry = ldpy.strip_std_field(payload)
|
|
|
|
assert isinstance(stripped_entry, list)
|
|
|
|
for field in fields:
|
|
assert field not in stripped_entry
|
|
|
|
|
|
# ###
|
|
# Testing request_last_entries()
|
|
# ###
|
|
@mark.parametrize("opt_last", [LAST_E_MIN, LAST_E_MAX])
|
|
def test_request_last_entries_out_of_range(opt_last):
|
|
"""Value is out of range for the last entries."""
|
|
with raises(ValueError):
|
|
ldpy.request_last_entries(opt_last)
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
@mark.parametrize(
|
|
"opt_last, fields",
|
|
[
|
|
(LAST_E_CHOICES[0], DEMO_STREAM_FIELDS),
|
|
(len(LAST_E_CHOICES), DEMO_STREAM_FIELDS),
|
|
],
|
|
)
|
|
def test_request_last_entries_in_range(opt_last, fields):
|
|
"""Value is in range for the last entries."""
|
|
response_id = opt_last - 1
|
|
response = ldpy.request_last_entries(opt_last, fields)
|
|
|
|
assert len(response) == opt_last
|
|
for field in STD_FIELDS:
|
|
assert field not in response[0]["_source"]
|
|
assert field not in response[response_id]["_source"]
|
|
|
|
|
|
# ###
|
|
# Testing main()
|
|
# ###
|
|
def test_main_without_option():
|
|
"""Called without option."""
|
|
options = FakeOptions([])
|
|
|
|
with raises(NotImplementedError):
|
|
ldpy.main(options)
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_main_demo_with_mapping():
|
|
"""Called with mapping option.
|
|
|
|
`main()` just transfers `request_map_props()` return: this test is not really useful
|
|
"""
|
|
options = FakeOptions(["mapping"])
|
|
response = ldpy.main(options)
|
|
|
|
assert isinstance(response, list)
|
|
|
|
|
|
@pytest.mark.vcr()
|
|
def test_main_demo_with_last_const():
|
|
"""Called with last option."""
|
|
options = FakeOptions(["last"])
|
|
response = ldpy.main(options)
|
|
|
|
assert isinstance(response, list)
|
|
assert isinstance(response[0], dict)
|