sqlmesh.core.test

  1from __future__ import annotations
  2
  3import pathlib
  4import typing as t
  5import unittest
  6
  7from sqlmesh.core.engine_adapter import EngineAdapter
  8from sqlmesh.core.model import Model
  9from sqlmesh.core.test.definition import ModelTest as ModelTest, generate_test as generate_test
 10from sqlmesh.core.test.discovery import (
 11    ModelTestMetadata as ModelTestMetadata,
 12    filter_tests_by_patterns as filter_tests_by_patterns,
 13    get_all_model_tests as get_all_model_tests,
 14    load_model_test_file as load_model_test_file,
 15)
 16from sqlmesh.core.test.result import ModelTextTestResult as ModelTextTestResult
 17from sqlmesh.utils import UniqueKeyDict
 18
 19if t.TYPE_CHECKING:
 20    from sqlmesh.core.config.loader import C
 21
 22
 23def run_tests(
 24    model_test_metadata: list[ModelTestMetadata],
 25    models: UniqueKeyDict[str, Model],
 26    config: C,
 27    gateway: t.Optional[str] = None,
 28    dialect: str | None = None,
 29    verbosity: int = 1,
 30    preserve_fixtures: bool = False,
 31    stream: t.TextIO | None = None,
 32    default_catalog: str | None = None,
 33    default_catalog_dialect: str = "",
 34) -> ModelTextTestResult:
 35    """Create a test suite of ModelTest objects and run it.
 36
 37    Args:
 38        model_test_metadata: A list of ModelTestMetadata named tuples.
 39        models: All models to use for expansion and mapping of physical locations.
 40        verbosity: The verbosity level.
 41        preserve_fixtures: Preserve the fixture tables in the testing database, useful for debugging.
 42    """
 43    testing_adapter_by_gateway: t.Dict[str, EngineAdapter] = {}
 44    default_gateway = gateway or config.default_gateway
 45
 46    try:
 47        tests = []
 48        for metadata in model_test_metadata:
 49            body = metadata.body
 50            gateway = body.get("gateway") or default_gateway
 51            testing_engine_adapter = testing_adapter_by_gateway.get(gateway)
 52            if not testing_engine_adapter:
 53                testing_engine_adapter = config.get_test_connection(
 54                    gateway,
 55                    default_catalog,
 56                    default_catalog_dialect,
 57                ).create_engine_adapter(register_comments_override=False)
 58                testing_adapter_by_gateway[gateway] = testing_engine_adapter
 59
 60            tests.append(
 61                ModelTest.create_test(
 62                    body=body,
 63                    test_name=metadata.test_name,
 64                    models=models,
 65                    engine_adapter=testing_engine_adapter,
 66                    dialect=dialect,
 67                    path=metadata.path,
 68                    default_catalog=default_catalog,
 69                    preserve_fixtures=preserve_fixtures,
 70                )
 71            )
 72
 73        result = t.cast(
 74            ModelTextTestResult,
 75            unittest.TextTestRunner(
 76                stream=stream, verbosity=verbosity, resultclass=ModelTextTestResult
 77            ).run(unittest.TestSuite(tests)),
 78        )
 79    finally:
 80        for testing_engine_adapter in testing_adapter_by_gateway.values():
 81            testing_engine_adapter.close()
 82
 83    return result
 84
 85
 86def run_model_tests(
 87    tests: list[str],
 88    models: UniqueKeyDict[str, Model],
 89    config: C,
 90    gateway: t.Optional[str] = None,
 91    dialect: str | None = None,
 92    verbosity: int = 1,
 93    patterns: list[str] | None = None,
 94    preserve_fixtures: bool = False,
 95    stream: t.TextIO | None = None,
 96    default_catalog: t.Optional[str] = None,
 97    default_catalog_dialect: str = "",
 98) -> ModelTextTestResult:
 99    """Load and run tests.
100
101    Args:
102        tests: A list of tests to run, e.g. [tests/test_orders.yaml::test_single_order]
103        models: All models to use for expansion and mapping of physical locations.
104        verbosity: The verbosity level.
105        patterns: A list of patterns to match against.
106        preserve_fixtures: Preserve the fixture tables in the testing database, useful for debugging.
107    """
108    loaded_tests = []
109    for test in tests:
110        filename, test_name = test.split("::", maxsplit=1) if "::" in test else (test, "")
111        path = pathlib.Path(filename)
112
113        if test_name:
114            loaded_tests.append(load_model_test_file(path)[test_name])
115        else:
116            loaded_tests.extend(load_model_test_file(path).values())
117
118    if patterns:
119        loaded_tests = filter_tests_by_patterns(loaded_tests, patterns)
120
121    return run_tests(
122        loaded_tests,
123        models,
124        config,
125        gateway=gateway,
126        dialect=dialect,
127        verbosity=verbosity,
128        preserve_fixtures=preserve_fixtures,
129        stream=stream,
130        default_catalog=default_catalog,
131        default_catalog_dialect=default_catalog_dialect,
132    )