LDPy/tests_ldpy.py

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)