Merge pull request #41 from CMIP-REF/invoke-cli-helper

changelog/41.trivial.md

@@ -0,0 +1 @@
+Added the `invoke_cli` pytest fixture to provide a consistent interface for running CLI tests

conftest.py

@@ -9,7 +9,10 @@
 import esgpull
 import pandas as pd
 import pytest
+from click.testing import Result
+from typer.testing import CliRunner
 
+from ref import cli
 from ref.config import Config
 from ref.datasets.cmip6 import CMIP6DatasetAdapter
 
@@ -42,3 +45,32 @@ def config(tmp_path, monkeypatch) -> Config:
     cfg.save()
 
     return cfg
+
+
+@pytest.fixture
+def invoke_cli():
+    """
+    Invoke the CLI with the given arguments and verify the exit code
+    """
+
+    # We want to split stderr and stdout
+    # stderr == logging
+    # stdout == output from commands
+    runner = CliRunner(mix_stderr=False)
+
+    def _invoke_cli(args: list[str], expected_exit_code: int = 0) -> Result:
+        result = runner.invoke(
+            app=cli.app,
+            args=args,
+        )
+
+        if result.exit_code != expected_exit_code:
+            print(result.stdout)
+            print(result.stderr)
+
+            if result.exception:
+                raise result.exception
+            raise ValueError(f"Expected exit code {expected_exit_code}, got {result.exit_code}")
+        return result
+
+    return _invoke_cli

packages/ref/tests/unit/cli/test_config.py

@@ -1,51 +1,43 @@
 import os
 
-from typer.testing import CliRunner
 
-from ref.cli import app
+def test_without_subcommand(invoke_cli):
+    # exit code 2 denotes a user error
+    result = invoke_cli(["config"], expected_exit_code=2)
+    assert "Missing command." in result.stderr
 
-runner = CliRunner()
 
+def test_config_help(invoke_cli):
+    result = invoke_cli(["config", "--help"])
 
-def test_without_subcommand():
-    result = runner.invoke(app, ["config"])
-    assert result.exit_code == 2
-    assert "Missing command." in result.output
-
-
-def test_config_help():
-    result = runner.invoke(app, ["config", "--help"])
-    assert result.exit_code == 0
-
-    assert "View and update the REF configuration" in result.output
+    assert "View and update the REF configuration" in result.stdout
 
 
 class TestConfigList:
-    def test_config_list(self, config):
-        result = runner.invoke(app, ["config", "list"])
-        assert result.exit_code == 0
+    def test_config_list(self, config, invoke_cli):
+        result = invoke_cli(["config", "list"])
 
         config_dir = os.environ.get("REF_CONFIGURATION")
         assert f'data = "{config_dir}/data"\n' in result.output
         assert 'database_url = "sqlite://' in result.output
 
-    def test_config_list_custom_missing(self, config):
-        result = runner.invoke(
-            app,
+    def test_config_list_custom_missing(self, config, invoke_cli):
+        result = invoke_cli(
             [
                 "--configuration-directory",
                 "missing",
                 "config",
                 "list",
             ],
+            expected_exit_code=1,
         )
-        assert result.exit_code == 1, result.output
+
+        assert "Configuration file not found" in result.stdout
 
 
 class TestConfigUpdate:
-    def test_config_update(self):
-        result = runner.invoke(app, ["config", "update"])
-        assert result.exit_code == 0
+    def test_config_update(self, invoke_cli):
+        result = invoke_cli(["config", "update"])
 
         # TODO: actually implement this functionality
-        assert "config" in result.output
+        assert "config" in result.stdout

packages/ref/tests/unit/cli/test_datasets.py

@@ -1,77 +1,61 @@
 from pathlib import Path
 
-from typer.testing import CliRunner
-
-from ref.cli import app
 from ref.datasets.cmip6 import CMIP6DatasetAdapter
 from ref.models import Dataset
 from ref.models.dataset import CMIP6Dataset, CMIP6File
 
-runner = CliRunner()
-
 
-def test_ingest_help():
-    result = runner.invoke(app, ["datasets", "ingest", "--help"])
-    assert result.exit_code == 0
+def test_ingest_help(invoke_cli):
+    result = invoke_cli(["datasets", "ingest", "--help"])
 
-    assert "Ingest a dataset" in result.output
+    assert "Ingest a dataset" in result.stdout
 
 
 class TestDatasetsList:
-    def test_list(self, db_seeded):
-        result = runner.invoke(app, ["datasets", "list"])
-        assert result.exit_code == 0, result.output
-        assert "experi…" in result.output
+    def test_list(self, db_seeded, invoke_cli):
+        result = invoke_cli(["datasets", "list"])
+        assert "experi…" in result.stdout
 
-    def test_list_limit(self, db_seeded):
-        result = runner.invoke(app, ["datasets", "list", "--limit", "1", "--column", "instance_id"])
-        assert result.exit_code == 0, result.output
-        assert len(result.output.strip().split("\n")) == 3  # header + spacer + 1 row
+    def test_list_limit(self, db_seeded, invoke_cli):
+        result = invoke_cli(["datasets", "list", "--limit", "1", "--column", "instance_id"])
+        assert len(result.stdout.strip().split("\n")) == 3  # header + spacer + 1 row
 
-    def test_list_column(self, db_seeded):
-        result = runner.invoke(app, ["datasets", "list", "--column", "variable_id"])
-        assert result.exit_code == 0, result.output
-        assert "variable_id" in result.output
-        assert "grid" not in result.output
+    def test_list_column(self, db_seeded, invoke_cli):
+        result = invoke_cli(["datasets", "list", "--column", "variable_id"])
+        assert "variable_id" in result.stdout
+        assert "grid" not in result.stdout
 
-    def test_list_column_invalid(self, db_seeded):
-        result = runner.invoke(app, ["datasets", "list", "--column", "wrong"])
-        assert result.exit_code == 1
+    def test_list_column_invalid(self, db_seeded, invoke_cli):
+        invoke_cli(["datasets", "list", "--column", "wrong"], expected_exit_code=1)
 
 
 class TestDatasetsListColumns:
-    def test_list(self, db_seeded):
-        result = runner.invoke(app, ["datasets", "list-columns"])
-        assert result.exit_code == 0, result.output
-        assert result.output.strip() == "\n".join(
+    def test_list(self, db_seeded, invoke_cli):
+        result = invoke_cli(["datasets", "list-columns"])
+        assert result.stdout.strip() == "\n".join(
             sorted(CMIP6DatasetAdapter().load_catalog(db_seeded, include_files=False).columns.to_list())
         )
 
-    def test_list_include_files(self, db_seeded):
-        result = runner.invoke(app, ["datasets", "list-columns", "--include-files"])
-        assert result.exit_code == 0, result.output
-        assert result.output.strip() == "\n".join(
+    def test_list_include_files(self, db_seeded, invoke_cli):
+        result = invoke_cli(["datasets", "list-columns", "--include-files"])
+        assert result.stdout.strip() == "\n".join(
             sorted(CMIP6DatasetAdapter().load_catalog(db_seeded, include_files=True).columns.to_list())
         )
-        assert "start_time" in result.output
+        assert "start_time" in result.stdout
 
 
 class TestIngest:
     data_dir = Path("CMIP6") / "ScenarioMIP" / "CSIRO" / "ACCESS-ESM1-5" / "ssp126" / "r1i1p1f1"
 
-    def test_ingest(self, esgf_data_dir, db):
-        result = runner.invoke(
-            app, ["datasets", "ingest", str(esgf_data_dir / self.data_dir), "--source-type", "cmip6"]
-        )
-        assert result.exit_code == 0, result.output
+    def test_ingest(self, esgf_data_dir, db, invoke_cli):
+        invoke_cli(["datasets", "ingest", str(esgf_data_dir / self.data_dir), "--source-type", "cmip6"])
 
         assert db.session.query(Dataset).count() == 5
         assert db.session.query(CMIP6Dataset).count() == 5
         assert db.session.query(CMIP6File).count() == 9
 
-    def test_ingest_and_solve(self, esgf_data_dir, db):
-        result = runner.invoke(
-            app,
+    def test_ingest_and_solve(self, esgf_data_dir, db, invoke_cli):
+        result = invoke_cli(
             [
                 "--log-level",
                 "info",
@@ -83,12 +67,10 @@ def test_ingest_and_solve(self, esgf_data_dir, db):
                 "--solve",
             ],
         )
-        assert result.exit_code == 0, result.output
-        assert "Solving for metrics that require recalculation." in result.output
+        assert "Solving for metrics that require recalculation." in result.stderr
 
-    def test_ingest_multiple_times(self, esgf_data_dir, db):
-        result = runner.invoke(
-            app,
+    def test_ingest_multiple_times(self, esgf_data_dir, db, invoke_cli):
+        invoke_cli(
             [
                 "datasets",
                 "ingest",
@@ -97,13 +79,11 @@ def test_ingest_multiple_times(self, esgf_data_dir, db):
                 "cmip6",
             ],
         )
-        assert result.exit_code == 0, result.output
 
         assert db.session.query(Dataset).count() == 1
         assert db.session.query(CMIP6File).count() == 2
 
-        result = runner.invoke(
-            app,
+        invoke_cli(
             [
                 "datasets",
                 "ingest",
@@ -112,12 +92,10 @@ def test_ingest_multiple_times(self, esgf_data_dir, db):
                 "cmip6",
             ],
         )
-        assert result.exit_code == 0, result.output
 
         assert db.session.query(Dataset).count() == 1
 
-        result = runner.invoke(
-            app,
+        invoke_cli(
             [
                 "datasets",
                 "ingest",
@@ -126,24 +104,36 @@ def test_ingest_multiple_times(self, esgf_data_dir, db):
                 "cmip6",
             ],
         )
-        assert result.exit_code == 0, result.output
 
         assert db.session.query(Dataset).count() == 2
 
-    def test_ingest_missing(self, esgf_data_dir, db):
-        result = runner.invoke(
-            app, ["datasets", "ingest", str(esgf_data_dir / "missing"), "--source-type", "cmip6"]
+    def test_ingest_missing(self, esgf_data_dir, db, invoke_cli):
+        result = invoke_cli(
+            [
+                "datasets",
+                "ingest",
+                str(esgf_data_dir / "missing"),
+                "--source-type",
+                "cmip6",
+            ],
+            expected_exit_code=1,
         )
         assert isinstance(result.exception, FileNotFoundError)
         assert result.exception.filename == esgf_data_dir / "missing"
 
-        assert f'File or directory {esgf_data_dir / "missing"} does not exist' in result.output
+        assert f'File or directory {esgf_data_dir / "missing"} does not exist' in result.stderr
 
-    def test_ingest_dryrun(self, esgf_data_dir, db):
-        result = runner.invoke(
-            app, ["datasets", "ingest", str(esgf_data_dir), "--source-type", "cmip6", "--dry-run"]
+    def test_ingest_dryrun(self, esgf_data_dir, db, invoke_cli):
+        invoke_cli(
+            [
+                "datasets",
+                "ingest",
+                str(esgf_data_dir),
+                "--source-type",
+                "cmip6",
+                "--dry-run",
+            ]
         )
-        assert result.exit_code == 0, result.output
 
         # Check that no data was loaded
         assert db.session.query(Dataset).count() == 0

packages/ref/tests/unit/cli/test_root.py

@@ -1,69 +1,55 @@
-from typer.testing import CliRunner
-
 from ref import __core_version__, __version__
-from ref.cli import app
-
-runner = CliRunner(
-    mix_stderr=False,
-)
 
 
-def test_without_subcommand():
-    result = runner.invoke(app, [])
-    assert result.exit_code == 0
+def test_without_subcommand(invoke_cli):
+    result = invoke_cli([])
     assert "Usage:" in result.stdout
     assert "ref [OPTIONS] COMMAND [ARGS]" in result.stdout
     assert "ref: A CLI for the CMIP Rapid Evaluation Framework" in result.stdout
 
 
-def test_version():
-    result = runner.invoke(app, ["--version"])
-    assert result.exit_code == 0
-    assert f"ref: {__version__}\nref-core: {__core_version__}" in result.output
+def test_version(invoke_cli):
+    result = invoke_cli(["--version"])
+    assert f"ref: {__version__}\nref-core: {__core_version__}" in result.stdout
 
 
-def test_verbose():
+def test_verbose(invoke_cli):
     exp_log = "| DEBUG    | ref.config:default:178 - Loading default configuration from"
-    result = runner.invoke(
-        app,
+    result = invoke_cli(
         ["--verbose", "config", "list"],
     )
     assert exp_log in result.stderr
 
-    result = runner.invoke(
-        app,
+    result = invoke_cli(
         ["config", "list"],
     )
     # Only info and higher messages logged
     assert exp_log not in result.stderr
 
 
-def test_config_directory_custom(config):
+def test_config_directory_custom(config, invoke_cli):
     config.paths.tmp = "test-value"
     config.save()
 
-    result = runner.invoke(
-        app,
+    result = invoke_cli(
         [
             "--configuration-directory",
             str(config._config_file.parent),
             "config",
             "list",
         ],
     )
-    assert result.exit_code == 0
     assert 'tmp = "test-value"\n' in result.output
 
 
-def test_config_directory_append(config):
+def test_config_directory_append(config, invoke_cli):
     # configuration directory must be passed before command
-    result = runner.invoke(
-        app,
+    invoke_cli(
         [
             "config",
             "list",
             "--configuration-directory",
             str(config._config_file.parent),
         ],
+        expected_exit_code=2,
     )
-    assert result.exit_code == 2