diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index dfe554a..fe03822 100644 --- a/.gitlab-ci.yml +++ b/.gitlab-ci.yml @@ -24,3 +24,4 @@ lint: - pip install -r requirements.txt - pip install -r requirements-dev.txt - make --no-print-directory --quiet lint + - make --no-print-directory --quiet test diff --git a/ldpy.py b/ldpy.py index 902e6cb..071ea48 100755 --- a/ldpy.py +++ b/ldpy.py @@ -17,6 +17,8 @@ import sys import client # pylint: disable=import-error +LAST_E_CHOICES = range(1, 21) + logger = logging.getLogger(os.path.splitext(os.path.basename(sys.argv[0]))[0]) @@ -61,7 +63,8 @@ def parse_args(args=sys.argv[1:]): description=sys.modules[__name__].__doc__, formatter_class=CustomFormatter ) - parser.add_argument( + group = parser.add_argument_group("log setting") + group.add_argument( "-d", "--debug", action="store_true", @@ -69,19 +72,46 @@ def parse_args(args=sys.argv[1:]): help="Log activity in console", ) - parser.add_argument( + group = parser.add_mutually_exclusive_group() + group.add_argument( "-m", "--mapping", action="store_true", help="Show the stream mapping", ) + group.add_argument( + "-l", + "--last", + choices=LAST_E_CHOICES, + const="3", + help="Last entries of the stream", + metavar="INT", + nargs="?", + type=int, + ) return parser.parse_args(args) +def get_last_entries(entries_nb): + """Get stream last n entries.""" + if entries_nb in LAST_E_CHOICES: + logger.debug("Wait before getting '%s' entries!", entries_nb) + query = {"size": entries_nb} + last_entries = client.opnsrch_clt.search(body=query) + + else: + logger.critical("'%s' is not in '%s'", entries_nb, LAST_E_CHOICES) + raise ValueError + + return last_entries + + def get_map_props(): """Get stream mapping. + 'demo' stream did not have an LDP_ALIAS + LDP indices are rolling and LDP do not implement the call to get indices related to an alias. """ @@ -108,8 +138,14 @@ def main(options): if options.mapping: response = get_map_props() logger.debug("Mapping for '%s' stream: %s", client.LDP_STREAM_NAME, response) + elif options.last: + response = get_last_entries(options.last) + logger.debug( + "Last '%s' entries for '%s' stream:", options.last, client.LDP_STREAM_NAME + ) + logger.debug(response) else: - raise RuntimeError + raise NotImplementedError return response diff --git a/tests_ldpy.py b/tests_ldpy.py index 9dee03b..8393937 100644 --- a/tests_ldpy.py +++ b/tests_ldpy.py @@ -9,7 +9,7 @@ Author: frederic zind 2022-09-24 Licenses: GNU AGPL v3: http://www.gnu.org/licenses/ """ import logging -from pytest import raises +from pytest import mark, raises import ldpy @@ -19,12 +19,15 @@ 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 @@ -54,21 +57,41 @@ def test_logging_default(): # Testing parse_args() # ### def test_parse_args(): - """Defaults arguments are here.""" + """Mapping argument is set.""" cli_options = ldpy.parse_args(["--debug", "--mapping"]) assert cli_options.debug assert cli_options.mapping -# ### -# Testing get_map_props() -# ### -def test_get_map_props(): - """This is an integration test. +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 - It need credentials to guery LDP stream. - """ - pass + +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 get_last_entries() +# ### +@mark.parametrize("entry_np", [0, 21]) +def test_get_last_entries_out_of_range(entry_np): + """Value is out of range for the last entries.""" + with raises(ValueError): + ldpy.get_last_entries(entry_np) + + +@mark.parametrize("entry_np", range(1, 21)) +def test_get_last_entries_in_range(entry_np): + """Value is in range for the last entries.""" + response = ldpy.get_last_entries(entry_np) + assert len(response["hits"]["hits"]) == entry_np # ### @@ -78,7 +101,7 @@ def test_main_without_option(): """Called without option.""" options = FakeOptions([]) - with raises(RuntimeError): + with raises(NotImplementedError): ldpy.main(options) @@ -91,3 +114,12 @@ def test_main_demo_with_mapping(): response = ldpy.main(options) assert isinstance(response, list) + + +def test_main_demo_with_last_const(): + """Called with last option.""" + options = FakeOptions(["last"]) + response = ldpy.main(options) + + assert isinstance(response, dict) + assert isinstance(response["hits"]["hits"], list)