feat(test): treat model file paths as test selectors

docs/concepts/tests.md

@@ -463,6 +463,20 @@ You can also run tests that match a pattern or substring using a glob pathname e
 $ sqlmesh test tests/test_*
 ```
 
+Passing the path of a model file runs the tests for that model, which is useful for commit hooks and other tools that work with changed files rather than test names:
+
+```
+$ sqlmesh test models/full_model.sql
+```
+
+Model files and test files can be mixed, and the results are unioned. A test selected by more than one argument still runs only once, so the following runs each of `full_model`'s tests a single time even though both arguments cover them:
+
+```
+$ sqlmesh test models/full_model.sql tests/test_full_model.yaml
+```
+
+An argument that is neither a known model file nor a known test file is an error, so a mistyped or stale path fails instead of quietly running no tests. A model that simply has no tests is not an error.
+
 You can pass `--local` to run tests without loading state from the configured state connection:
 
 ``` bash

docs/reference/cli.md

@@ -623,6 +623,10 @@ Usage: sqlmesh test [OPTIONS] [TESTS]...
 
   Run model unit tests.
 
+  TESTS are test files, `file.yaml::test_name` selectors, or model files, in
+  which case the tests for those models are run. They are unioned, and a test
+  selected more than once still only runs once.
+
 Options:
   -k TEXT              Only run tests that match the pattern of substring.
   -v, --verbose        Verbose output.

sqlmesh/cli/main.py

@@ -835,7 +835,12 @@ def test(
     select_model: t.List[str],
     tests: t.List[str],
 ) -> None:
-    """Run model unit tests."""
+    """Run model unit tests.
+
+    TESTS are test files, `file.yaml::test_name` selectors, or model files, in which case the
+    tests for those models are run. They are unioned, and a test selected more than once still
+    only runs once.
+    """
     model_names = (
         obj._new_selector().expand_model_selections(select_model) if select_model else None
     )
@@ -845,6 +850,7 @@ def test(
         verbosity=Verbosity(verbose),
         preserve_fixtures=preserve_fixtures,
         model_names=model_names,
+        raise_on_unknown_paths=True,
     )
     if not result.wasSuccessful():
         exit(1)

sqlmesh/core/context.py

@@ -36,6 +36,7 @@
 import abc
 import collections
 import logging
+import os.path
 import sys
 import time
 import traceback
@@ -119,7 +120,7 @@
     filter_tests_by_patterns,
 )
 from sqlmesh.core.user import User
-from sqlmesh.utils import CorrelationId, UniqueKeyDict, Verbosity
+from sqlmesh.utils import CorrelationId, UniqueKeyDict, Verbosity, unique
 from sqlmesh.utils.concurrency import concurrent_apply_to_values
 from sqlmesh.utils.dag import DAG
 from sqlmesh.utils.date import (
@@ -2407,17 +2408,26 @@ def test(
         preserve_fixtures: bool = False,
         stream: t.Optional[t.TextIO] = None,
         model_names: t.Optional[t.Collection[str]] = None,
+        raise_on_unknown_paths: bool = False,
     ) -> ModelTextTestResult:
         """Discover and run model tests"""
         if verbosity >= Verbosity.VERBOSE:
             import pandas as pd
 
             pd.set_option("display.max_columns", None)
 
-        baseline_meta = self.select_tests(tests=tests, patterns=match_patterns, model_names=None)
+        baseline_meta = self.select_tests(
+            tests=tests,
+            patterns=match_patterns,
+            model_names=None,
+            raise_on_unknown_paths=raise_on_unknown_paths,
+        )
         if model_names is not None:
             test_meta = self.select_tests(
-                tests=tests, patterns=match_patterns, model_names=model_names
+                tests=tests,
+                patterns=match_patterns,
+                model_names=model_names,
+                raise_on_unknown_paths=raise_on_unknown_paths,
             )
             tests_skipped = len(baseline_meta) - len(test_meta)
         else:
@@ -3611,30 +3621,112 @@ def lint_models(
 
         return all_violations
 
+    def _tests_by_absolute_model_path(self) -> t.Dict[str, t.List[ModelTestMetadata]]:
+        """Map each model file to the tests that target the model(s) defined in it."""
+        tests_by_model_name: t.Dict[str, t.List[ModelTestMetadata]] = collections.defaultdict(list)
+        for metadata in self._model_test_metadata:
+            if metadata.model_name:
+                tests_by_model_name[
+                    normalize_model_name(
+                        metadata.model_name,
+                        default_catalog=self.default_catalog,
+                        dialect=self.default_dialect,
+                    )
+                ].append(metadata)
+
+        # A path is made absolute rather than resolved, so this costs no syscalls per model.
+        tests_by_path: t.Dict[str, t.List[ModelTestMetadata]] = {}
+        for fqn, model in self._models.items():
+            if model._path is not None:
+                tests_by_path.setdefault(os.path.abspath(model._path), []).extend(
+                    tests_by_model_name.get(fqn, [])
+                )
+
+        return tests_by_path
+
+    def _select_tests_by_test_path(self, selector: str) -> t.Optional[t.List[ModelTestMetadata]]:
+        """Resolve a selector against the test files, or return None if it matches none of them.
+
+        The selector is a test file path or a `path::test_name`. Paths are matched as given
+        first, so an unchanged selector never pays for normalization.
+        """
+        if "::" in selector:
+            metadata = self._model_test_metadata_fully_qualified_name_index.get(selector)
+            if metadata is None:
+                path, _, test_name = selector.rpartition("::")
+                metadata = self._model_test_metadata_fully_qualified_name_index.get(
+                    f"{os.path.abspath(path)}::{test_name}"
+                )
+            return [metadata] if metadata is not None else None
+
+        for candidate in (Path(selector), Path(os.path.abspath(selector))):
+            matched = self._model_test_metadata_path_index.get(candidate)
+            if matched is not None:
+                return list(matched)
+
+        return None
+
+    def _unknown_test_selector_error(self, selector: str) -> str:
+        """Explains why a selector matched nothing.
+
+        A `path::test_name` whose file is a known test file failed on the test name, not the
+        path, so the message says so rather than claiming the file is unknown.
+        """
+        if "::" in selector:
+            path, _, _ = selector.rpartition("::")
+            if any(
+                candidate in self._model_test_metadata_path_index
+                for candidate in (Path(path), Path(os.path.abspath(path)))
+            ):
+                return f"'{selector}' is not a known test in '{path}'."
+
+        return f"'{selector}' is not a known model or test file."
+
     def select_tests(
         self,
         tests: t.Optional[t.List[str]] = None,
         patterns: t.Optional[t.List[str]] = None,
         model_names: t.Optional[t.Collection[str]] = None,
+        raise_on_unknown_paths: bool = False,
     ) -> t.List[ModelTestMetadata]:
-        """Filter pre-loaded test metadata based on tests and patterns."""
+        """Filter pre-loaded test metadata based on tests and patterns.
+
+        Args:
+            tests: Test selectors. Each one is a test file path, a `path::test_name`, or the path
+                of a model file, in which case that model's tests are selected. Selectors are
+                unioned and the result is deduplicated, so a model file and a test file that
+                resolve to the same test run it once rather than twice.
+            patterns: Patterns matched against fully qualified test names.
+            model_names: If given, narrows the selection to tests targeting these models.
+            raise_on_unknown_paths: Whether to raise when a selector matches neither a known test
+                nor a known model file. Off by default so that callers which probe arbitrary
+                documents, such as the LSP, keep getting an empty result instead of an error.
+        """
 
         test_meta = self._model_test_metadata
 
         if tests:
-            filtered_tests = []
+            filtered_tests: t.List[ModelTestMetadata] = []
+            # Built at most once, and only if a selector turns out not to be a test file.
+            tests_by_model_path: t.Optional[t.Dict[str, t.List[ModelTestMetadata]]] = None
+
             for test in tests:
-                if "::" in test:
-                    if test in self._model_test_metadata_fully_qualified_name_index:
-                        filtered_tests.append(
-                            self._model_test_metadata_fully_qualified_name_index[test]
-                        )
-                else:
-                    test_path = Path(test)
-                    if test_path in self._model_test_metadata_path_index:
-                        filtered_tests.extend(self._model_test_metadata_path_index[test_path])
+                matched = self._select_tests_by_test_path(test)
+                if matched is None and "::" not in test:
+                    if tests_by_model_path is None:
+                        tests_by_model_path = self._tests_by_absolute_model_path()
+                    # A known model with no tests matches an empty list, which is not the same
+                    # as a selector that resolves to nothing at all.
+                    matched = tests_by_model_path.get(os.path.abspath(test))
+                if matched is None:
+                    if raise_on_unknown_paths:
+                        raise SQLMeshError(self._unknown_test_selector_error(test))
+                    continue
+                filtered_tests.extend(matched)
 
-            test_meta = filtered_tests
+            # Selectors can overlap, e.g. a model file and the test file holding its tests, so
+            # the union is deduplicated to avoid running the same test more than once.
+            test_meta = unique(filtered_tests)
 
         if patterns:
             test_meta = filter_tests_by_patterns(test_meta, patterns)

tests/cli/test_cli.py

@@ -2666,6 +2666,27 @@ def test_format_does_not_open_state_connection(
     mock.assert_not_called()
 
 
+def test_test_accepts_model_paths(runner: CliRunner, tmp_path: Path) -> None:
+    create_example_project(tmp_path)
+
+    result = runner.invoke(
+        cli, ["--paths", str(tmp_path), "test", str(tmp_path / "models" / "full_model.sql")]
+    )
+    assert result.exit_code == 0, f"Test failed: {result.output}\nException: {result.exception}"
+    assert "Ran 1 test" in result.output
+
+
+def test_test_unknown_path_fails(runner: CliRunner, tmp_path: Path) -> None:
+    """A staged file that resolves to nothing must fail rather than silently run no tests."""
+    create_example_project(tmp_path)
+
+    result = runner.invoke(
+        cli, ["--paths", str(tmp_path), "test", str(tmp_path / "models" / "nope.sql")]
+    )
+    assert result.exit_code != 0
+    assert "is not a known model or test file" in result.output
+
+
 def test_test_local_runs_project_unit_tests(runner: CliRunner, tmp_path: Path, mocker) -> None:
     """A real unit test from the project's YAML runs under `--local` without touching state."""
     create_example_project(tmp_path)
@@ -2776,3 +2797,77 @@ def test_test_local_multi_repo_partial(runner: CliRunner, copy_to_temp_path, moc
     )
     assert "Successfully Ran 1 tests" in output, "the repo_2 test should still run"
     mock.assert_not_called()
+
+
+def test_test_local_with_model_paths(runner: CliRunner, tmp_path: Path, mocker) -> None:
+    """`--local` and model path selectors compose, which is the pre-commit hook case in #6020."""
+    create_example_project(tmp_path)
+    mock = _patch_state_access(mocker)
+
+    result = runner.invoke(
+        cli,
+        ["--paths", str(tmp_path), "test", "--local", str(tmp_path / "models" / "full_model.sql")],
+    )
+
+    assert result.exit_code == 0, f"Test failed: {result.output}\nException: {result.exception}"
+    assert "Successfully Ran 1 tests" in " ".join(result.output.split())
+    mock.assert_not_called()
+
+    # An unresolvable path still fails loudly, without reaching state.
+    result = runner.invoke(
+        cli, ["--paths", str(tmp_path), "test", "--local", str(tmp_path / "models" / "nope.sql")]
+    )
+    assert result.exit_code != 0
+    assert "is not a known model or test file" in result.output
+    mock.assert_not_called()
+
+
+def test_test_local_with_python_model_paths(runner: CliRunner, tmp_path: Path, mocker) -> None:
+    """The `--local` + path-selector combination works for Python models too."""
+    create_example_project(tmp_path)
+
+    (tmp_path / "models" / "py_model.py").write_text(
+        """
+import pandas as pd  # noqa: TID253
+from sqlmesh import model, ExecutionContext
+import typing as t
+
+@model(
+  name="sqlmesh_example.py_model",
+  columns={"id": "int"},
+)
+def execute(context: ExecutionContext, **kwargs: t.Any) -> pd.DataFrame:
+  return pd.DataFrame([{"id": 1}])
+""",
+        encoding="utf-8",
+    )
+    (tmp_path / "tests" / "test_py_model.yaml").write_text(
+        """
+test_py_model:
+  model: sqlmesh_example.py_model
+  outputs:
+    query:
+      rows:
+      - id: 1
+""",
+        encoding="utf-8",
+    )
+
+    mock = _patch_state_access(mocker)
+
+    result = runner.invoke(
+        cli,
+        ["--paths", str(tmp_path), "test", "--local", str(tmp_path / "models" / "py_model.py")],
+    )
+
+    assert result.exit_code == 0, f"Test failed: {result.output}\nException: {result.exception}"
+    assert "Successfully Ran 1 tests" in " ".join(result.output.split())
+    mock.assert_not_called()
+
+    # A Python file that is not a model is still an error rather than a silent no-op.
+    result = runner.invoke(
+        cli, ["--paths", str(tmp_path), "test", "--local", str(tmp_path / "models" / "nope.py")]
+    )
+    assert result.exit_code != 0
+    assert "is not a known model or test file" in result.output
+    mock.assert_not_called()