From fe1801da3b828e0ef41a218cdfd45b7eeaed1ab2 Mon Sep 17 00:00:00 2001 From: James Frost Date: Fri, 1 Nov 2024 10:03:55 +0000 Subject: [PATCH 1/3] Split main into functions --- src/CSET/__init__.py | 106 +++++++++++++++++++++++-------------------- 1 file changed, 57 insertions(+), 49 deletions(-) diff --git a/src/CSET/__init__.py b/src/CSET/__init__.py index 89e47dfdd..a8a73b176 100644 --- a/src/CSET/__init__.py +++ b/src/CSET/__init__.py @@ -31,6 +31,39 @@ def main(): Handles argument parsing, setting up logging, top level error capturing, and execution of the desired subcommand. """ + parser = setup_argument_parser() + cli_args = sys.argv[1:] + shlex.split(os.getenv("CSET_ADDOPTS", "")) + args, unparsed_args = parser.parse_known_args(cli_args) + setup_logging(args.verbose) + + # Down here so runs after logging is setup. + logging.debug("CLI Arguments: %s", cli_args) + + if args.subparser is None: + print("Please choose a command.", file=sys.stderr) + parser.print_usage() + sys.exit(127) + + try: + # Execute the specified subcommand. + args.func(args, unparsed_args) + except ArgumentError as err: + # Error message for when needed template variables are missing. + print(err, file=sys.stderr) + parser.print_usage() + sys.exit(127) + except Exception as err: + # Provide slightly nicer error messages for unhandled exceptions. + print(err, file=sys.stderr) + # Display the time and full traceback when debug logging. + logging.debug("An unhandled exception occurred.") + if logging.root.isEnabledFor(logging.DEBUG): + raise + sys.exit(1) + + +def setup_argument_parser() -> argparse.ArgumentParser: + """Create argument parser for CSET CLI.""" parser = argparse.ArgumentParser( prog="cset", description="Convective Scale Evaluation Tool" ) @@ -138,50 +171,17 @@ def main(): ) parser_recipe_id.set_defaults(func=_recipe_id_command) - cli_args = sys.argv[1:] + shlex.split(os.getenv("CSET_ADDOPTS", "")) - args, unparsed_args = parser.parse_known_args(cli_args) - - # Setup logging. - logging.captureWarnings(True) - loglevel = calculate_loglevel(args) - logger = logging.getLogger() - logger.setLevel(min(loglevel, logging.INFO)) - stderr_log = logging.StreamHandler() - stderr_log.addFilter(lambda record: record.levelno >= loglevel) - stderr_log.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s")) - logger.addHandler(stderr_log) - - # Down here so runs after logging is setup. - logging.debug("CLI Arguments: %s", cli_args) + return parser - if args.subparser is None: - print("Please choose a command.", file=sys.stderr) - parser.print_usage() - sys.exit(127) - - try: - # Execute the specified subcommand. - args.func(args, unparsed_args) - except ArgumentError as err: - # Error message for when needed template variables are missing. - print(err, file=sys.stderr) - parser.print_usage() - sys.exit(127) - except Exception as err: - # Provide slightly nicer error messages for unhandled exceptions. - print(err, file=sys.stderr) - # Display the time and full traceback when debug logging. - logging.debug("An unhandled exception occurred.") - if logging.root.isEnabledFor(logging.DEBUG): - raise - sys.exit(1) - -def calculate_loglevel(args) -> int: - """Calculate the logging level to apply. +def setup_logging(verbosity: int): + """Configure logging level, format and output stream. Level is based on verbose argument and the LOGLEVEL environment variable. """ + logging.captureWarnings(True) + + # Calculate logging level. try: name_to_level = logging.getLevelNamesMapping() except AttributeError: @@ -189,17 +189,25 @@ def calculate_loglevel(args) -> int: # implementation detail for older versions. name_to_level = logging._nameToLevel # Level from CLI flags. - if args.verbose >= 2: - loglevel = logging.DEBUG - elif args.verbose == 1: - loglevel = logging.INFO + if verbosity >= 2: + cli_loglevel = logging.DEBUG + elif verbosity == 1: + cli_loglevel = logging.INFO else: - loglevel = logging.WARNING - return min( - loglevel, - # Level from environment variable. - name_to_level.get(os.getenv("LOGLEVEL"), logging.ERROR), - ) + cli_loglevel = logging.WARNING + # Level from environment variable. + env_loglevel = name_to_level.get(os.getenv("LOGLEVEL"), logging.ERROR) + loglevel = min(cli_loglevel, env_loglevel) + + # Configure the root logger. + logger = logging.getLogger() + # Record everything at least INFO for the log file. + logger.setLevel(min(loglevel, logging.INFO)) + stderr_log = logging.StreamHandler() + # Filter stderr log to just what is requested. + stderr_log.addFilter(lambda record: record.levelno >= loglevel) + stderr_log.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s")) + logger.addHandler(stderr_log) def _bake_command(args, unparsed_args): From 90a8d32594da57b500015137636e9ba69ca8efad Mon Sep 17 00:00:00 2001 From: James Frost Date: Thu, 5 Dec 2024 18:03:18 +0000 Subject: [PATCH 2/3] Convert CLI tests to use internal interface This is significantly faster and provides more accurate test coverage than running the commands as subprocessses. --- src/CSET/__init__.py | 4 +- tests/test_cli.py | 231 +++++++++++++++++++++++++++---------------- 2 files changed, 149 insertions(+), 86 deletions(-) diff --git a/src/CSET/__init__.py b/src/CSET/__init__.py index a8a73b176..f6eabb6da 100644 --- a/src/CSET/__init__.py +++ b/src/CSET/__init__.py @@ -25,14 +25,14 @@ from CSET._common import ArgumentError -def main(): +def main(raw_cli_args: list[str] = sys.argv): """CLI entrypoint. Handles argument parsing, setting up logging, top level error capturing, and execution of the desired subcommand. """ parser = setup_argument_parser() - cli_args = sys.argv[1:] + shlex.split(os.getenv("CSET_ADDOPTS", "")) + cli_args = raw_cli_args[1:] + shlex.split(os.getenv("CSET_ADDOPTS", "")) args, unparsed_args = parser.parse_known_args(cli_args) setup_logging(args.verbose) diff --git a/tests/test_cli.py b/tests/test_cli.py index f422316e6..6daa6bf92 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -17,45 +17,132 @@ In many ways these are integration tests. """ -import os +import logging import subprocess from pathlib import Path from uuid import uuid4 import pytest +import CSET +import CSET.operators -def test_command_line_help(): - """Check that help commands work.""" - subprocess.run(["cset", "--help"], check=True) + +def test_command_line_invocation(): + """Check invocation via different entrypoints. + + Uses subprocess to validate the external interface. + """ + # Invoke via cset command subprocess.run(["cset", "--version"], check=True) + # Invoke via __main__.py + subprocess.run(["python3", "-m", "CSET", "--version"], check=True) + + +# Every other test should not use the command line interface, but rather stay +# within python to ensure coverage measurement. +def test_argument_parser(): + """Tests the argument parser behaves appropriately.""" + parser = CSET.setup_argument_parser() + # Test verbose flag. + args = parser.parse_args(["recipe-id", "-r", "recipe.yaml"]) + assert args.verbose == 0 + args = parser.parse_args(["-v", "recipe-id", "-r", "recipe.yaml"]) + assert args.verbose == 1 + args = parser.parse_args(["-vv", "recipe-id", "-r", "recipe.yaml"]) + assert args.verbose == 2 + + +def test_setup_logging(): + """Tests the logging setup at various verbosity levels.""" + root_logger = logging.getLogger() + # Log level gets pinned at a minimum of INFO for file output. + CSET.setup_logging(0) + assert root_logger.level == logging.INFO + # -v + CSET.setup_logging(1) + assert root_logger.level == logging.INFO + # -vv + CSET.setup_logging(2) + assert root_logger.level == logging.DEBUG + + +def test_main_no_subparser(capsys): + """Appropriate error when no subparser is given.""" + with pytest.raises(SystemExit) as sysexit: + CSET.main(["cset"]) + assert sysexit.value.code == 127 + assert capsys.readouterr().err == "Please choose a command.\n" + + +def test_main_unhandled_exception_normal(capsys): + """Appropriate error when an unhandled exception occurs.""" + with pytest.raises(SystemExit) as sysexit: + CSET.main( + [ + "cset", + "bake", + "--recipe=/non-existent/recipe.yaml", + "--input-dir=/dev/null", + "--output-dir=/dev/null", + ] + ) + assert sysexit.value.code == 1 + assert ( + capsys.readouterr().err + == "[Errno 2] No such file or directory: '/non-existent/recipe.yaml'\n" + ) - # Gain coverage of __main__.py - subprocess.run(["python3", "-m", "CSET", "-h"], check=True) - # Test verbose options. This is really just to up the coverage number. - subprocess.run(["cset", "-v"]) - subprocess.run(["cset", "-vv"]) +def test_main_unhandled_exception_debug(caplog, capsys): + """Appropriate error when an unhandled exception occurs under debug mode.""" + with pytest.raises(FileNotFoundError): + CSET.main( + [ + "cset", + "-vv", + "bake", + "--recipe=/non-existent/recipe.yaml", + "--input-dir=/dev/null", + "--output-dir=/dev/null", + ] + ) + assert ( + "[Errno 2] No such file or directory: '/non-existent/recipe.yaml'\n" + in capsys.readouterr().err + ) + log_record = caplog.records[-1] + assert log_record.message == "An unhandled exception occurred." + assert log_record.levelno == logging.DEBUG -def test_bake_recipe_execution(tmp_path): +def test_bake_recipe_execution(monkeypatch): """Test running CSET recipe from the command line.""" - subprocess.run( + bake_ran = False + + def _bake_test(args, unparsed_args): + nonlocal bake_ran + bake_ran = True + assert args.input_dir == Path("/dev/null") + assert args.output_dir == Path("/dev/null") + assert args.recipe == Path("tests/test_data/noop_recipe.yaml") + + monkeypatch.setattr(CSET, "_bake_command", _bake_test) + CSET.main( [ "cset", "bake", - f"--input-dir={os.devnull}", - f"--output-dir={tmp_path}", + "--input-dir=/dev/null", + "--output-dir=/dev/null", "--recipe=tests/test_data/noop_recipe.yaml", - ], - check=True, + ] ) -def test_bake_invalid_args(): +def test_bake_invalid_args(capsys): """Invalid arguments give non-zero exit code.""" - with pytest.raises(subprocess.CalledProcessError): - subprocess.run( + with pytest.raises(SystemExit) as sysexit: + CSET.main( [ "cset", "bake", @@ -63,17 +150,20 @@ def test_bake_invalid_args(): "--input-dir=/tmp", "--output-dir=/tmp", "--not-a-real-option", - ], - check=True, + ] ) + assert sysexit.value.code == 127 + assert capsys.readouterr().err == "Unknown argument: --not-a-real-option\n" -def test_bake_invalid_args_input_dir(): +def test_bake_invalid_args_input_dir(capsys): """Missing required input-dir argument for bake.""" - with pytest.raises(subprocess.CalledProcessError): - subprocess.run( - ["cset", "bake", "--recipe=foo", "--output-dir=/tmp"], check=True - ) + with pytest.raises(SystemExit) as sysexit: + CSET.main(["cset", "bake", "--recipe=foo", "--output-dir=/tmp"]) + assert sysexit.value.code == 2 + assert capsys.readouterr().err.endswith( + "cset bake: error: the following arguments are required: -i/--input-dir\n" + ) def test_graph_creation(tmp_path: Path): @@ -84,116 +174,89 @@ def test_graph_creation(tmp_path: Path): # Run with output path specified output_file = tmp_path / f"{uuid4()}.svg" - subprocess.run( - ( + CSET.main( + [ "cset", "graph", "-o", str(output_file), "--recipe=tests/test_data/noop_recipe.yaml", - ), - check=True, + ] ) assert output_file.is_file() output_file.unlink() def test_graph_details(tmp_path: Path): - """Generate a graph with details with details.""" + """Generate a graph with details.""" output_file = tmp_path / f"{uuid4()}.svg" - subprocess.run( - ( + CSET.main( + [ "cset", "graph", "--details", "-o", str(output_file), "--recipe=tests/test_data/noop_recipe.yaml", - ), - check=True, + ] ) assert output_file.is_file() def test_cookbook_cwd(tmp_working_dir): """Unpacking the recipes into the current working directory.""" - subprocess.run(["cset", "cookbook", "CAPE_ratio_plot.yaml"], check=True) + CSET.main(["cset", "cookbook", "CAPE_ratio_plot.yaml"]) assert Path("CAPE_ratio_plot.yaml").is_file() def test_cookbook_path(tmp_path: Path): """Unpacking the recipes into a specified directory.""" - subprocess.run( - ["cset", "cookbook", "--output-dir", tmp_path, "CAPE_ratio_plot.yaml"], - check=True, + CSET.main( + ["cset", "cookbook", "--output-dir", str(tmp_path), "CAPE_ratio_plot.yaml"] ) assert (tmp_path / "CAPE_ratio_plot.yaml").is_file() -def test_cookbook_list_available_recipes(): +def test_cookbook_list_available_recipes(capsys): """List all available recipes.""" - proc = subprocess.run( - ["cset", "cookbook", "--details"], capture_output=True, check=True - ) + CSET.main(["cset", "cookbook", "--details"]) + stdout = capsys.readouterr().out # Check start. - assert proc.stdout.startswith(b"Available recipes:\n") + assert stdout.startswith("Available recipes:\n") # Check has some recipes. - assert len(proc.stdout.splitlines()) > 3 + assert len(stdout.splitlines()) > 3 -def test_cookbook_detail_recipe(): +def test_cookbook_detail_recipe(capsys): """Show detail of a recipe.""" - proc = subprocess.run( - [ - "cset", - "cookbook", - "--details", - "CAPE_ratio_plot.yaml", - ], - capture_output=True, - check=True, - ) - assert proc.stdout.startswith(b"\n\tCAPE_ratio_plot.yaml\n") + CSET.main(["cset", "cookbook", "--details", "CAPE_ratio_plot.yaml"]) + assert capsys.readouterr().out.startswith("\n\tCAPE_ratio_plot.yaml\n") def test_cookbook_non_existent_recipe(tmp_path): """Non-existent recipe give non-zero exit code.""" - with pytest.raises(subprocess.CalledProcessError): - subprocess.run( - ["cset", "cookbook", "--output-dir", tmp_path, "non-existent.yaml"], - check=True, + with pytest.raises(SystemExit) as sysexit: + CSET.main( + ["cset", "cookbook", "--output-dir", str(tmp_path), "non-existent.yaml"] ) + assert sysexit.value.code == 1 -def test_recipe_id(): +def test_recipe_id(capsys): """Get recipe ID for a recipe.""" - p = subprocess.run( - ["cset", "recipe-id", "-r", "tests/test_data/noop_recipe.yaml"], - check=True, - capture_output=True, - ) - assert p.stdout == b"noop\n" + CSET.main(["cset", "recipe-id", "-r", "tests/test_data/noop_recipe.yaml"]) + assert capsys.readouterr().out == "noop\n" -def test_recipe_id_no_title(): +def test_recipe_id_no_title(capsys): """Get recipe id for recipe without a title.""" - p = subprocess.run( - ["cset", "recipe-id", "-r", "tests/test_data/ensemble_air_temp.yaml"], - check=True, - capture_output=True, - ) + CSET.main(["cset", "recipe-id", "-r", "tests/test_data/ensemble_air_temp.yaml"]) # UUID output + newline. - assert len(p.stdout) == 37 + assert len(capsys.readouterr().out) == 37 -def test_cset_addopts(): +def test_cset_addopts(capsys, monkeypatch): """Lists in CSET_ADDOPTS environment variable don't crash the parser.""" - environment = dict(os.environ) - environment["CSET_ADDOPTS"] = "--LIST='[1, 2, 3]'" - p = subprocess.run( - ["cset", "recipe-id", "-r", "tests/test_data/addopts_test_recipe.yaml"], - check=True, - capture_output=True, - env=environment, - ) - assert p.stdout == b"list_1_2_3\n" + monkeypatch.setenv("CSET_ADDOPTS", "--LIST='[1, 2, 3]'") + CSET.main(["cset", "recipe-id", "-r", "tests/test_data/addopts_test_recipe.yaml"]) + assert capsys.readouterr().out == "list_1_2_3\n" From 29affb6f6b29bc0b1c033463e70cdfa25ed1ae67 Mon Sep 17 00:00:00 2001 From: James Frost Date: Thu, 12 Dec 2024 09:33:28 +0000 Subject: [PATCH 3/3] Add comment explaining argument parsing --- src/CSET/__init__.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/CSET/__init__.py b/src/CSET/__init__.py index f6eabb6da..15a39f95f 100644 --- a/src/CSET/__init__.py +++ b/src/CSET/__init__.py @@ -31,9 +31,12 @@ def main(raw_cli_args: list[str] = sys.argv): Handles argument parsing, setting up logging, top level error capturing, and execution of the desired subcommand. """ + # Read arguments from the command line and CSET_ADDOPTS environment variable + # into an args object. parser = setup_argument_parser() cli_args = raw_cli_args[1:] + shlex.split(os.getenv("CSET_ADDOPTS", "")) args, unparsed_args = parser.parse_known_args(cli_args) + setup_logging(args.verbose) # Down here so runs after logging is setup.