#!/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)