adjustments

Author: erindru

.circleci/continue_config.yml

@@ -302,14 +302,14 @@ workflows:
           matrix:
             parameters:
               engine:
-                #- snowflake
-                #- databricks
-                #- redshift
-                #- bigquery
-                #- clickhouse-cloud
-                #- athena                
+                - snowflake
+                - databricks
+                - redshift
+                - bigquery
+                - clickhouse-cloud
+                - athena                
                 - fabric
-                #- gcp-postgres
+                - gcp-postgres
           #filters:
           #  branches:
           #    only:

Makefile

@@ -174,7 +174,7 @@ athena-test: guard-AWS_ACCESS_KEY_ID guard-AWS_SECRET_ACCESS_KEY guard-ATHENA_S3
 	pytest -n auto -m "athena" --retries 3 --junitxml=test-results/junit-athena.xml
 
 fabric-test: guard-FABRIC_HOST guard-FABRIC_CLIENT_ID guard-FABRIC_CLIENT_SECRET guard-FABRIC_DATABASE engine-fabric-install
-	pytest -n auto -m "fabric" --retries 3 --junitxml=test-results/junit-fabric.xml	
+	pytest -n auto -m "fabric" --retries 3 --junitxml=test-results/junit-fabric.xml
 
 gcp-postgres-test: guard-GCP_POSTGRES_INSTANCE_CONNECTION_STRING guard-GCP_POSTGRES_USER guard-GCP_POSTGRES_PASSWORD guard-GCP_POSTGRES_KEYFILE_JSON engine-gcppostgres-install
 	pytest -n auto -m "gcp_postgres" --retries 3 --junitxml=test-results/junit-gcp-postgres.xml

pyproject.toml

@@ -259,6 +259,7 @@ markers = [
     "mssql: test for MSSQL",
     "mysql: test for MySQL",
     "postgres: test for Postgres",
+    "gcp_postgres: test for Postgres on GCP",
     "redshift: test for Redshift",
     "snowflake: test for Snowflake",
     "spark: test for Spark",
@@ -267,7 +268,7 @@ markers = [
 ]
 addopts = "-n 0 --dist=loadgroup"
 asyncio_default_fixture_loop_scope = "session"
-log_cli = false # Set this to true to enable logging during tests
+log_cli = true # Set this to true to enable logging during tests
 log_cli_format = "%(asctime)s.%(msecs)03d %(filename)s:%(lineno)d %(levelname)s %(message)s"
 log_cli_level = "INFO"
 filterwarnings = [

sqlmesh/core/config/loader.py

@@ -83,6 +83,7 @@ def load_config_from_paths(
     personal_paths: t.Optional[t.List[Path]] = None,
     config_name: str = "config",
     load_from_env: bool = True,
+    variables: t.Optional[t.Dict[str, t.Any]] = None,
     **kwargs: t.Any,
 ) -> C:
     project_paths = project_paths or []
@@ -116,7 +117,7 @@ def load_config_from_paths(
                     "YAML configs do not support multiple configs. Use Python instead.",
                 )
             yaml_config_path = path.resolve()
-            non_python_configs.append(load_config_from_yaml(path))
+            non_python_configs.append(load_config_from_yaml(path, variables))
         elif extension == "py":
             try:
                 python_config = load_config_from_python_module(
@@ -194,8 +195,10 @@ def load_config_from_paths(
     return non_python_config
 
 
-def load_config_from_yaml(path: Path) -> t.Dict[str, t.Any]:
-    content = yaml_load(path)
+def load_config_from_yaml(
+    path: Path, variables: t.Optional[t.Dict[str, t.Any]] = None
+) -> t.Dict[str, t.Any]:
+    content = yaml_load(path, variables=variables)
     if not isinstance(content, dict):
         raise ConfigError(
             f"Invalid YAML configuration: expected a dictionary but got {type(content).__name__}. "

tests/core/engine_adapter/integration/__init__.py

@@ -193,6 +193,7 @@ def __init__(
         engine_adapter: EngineAdapter,
         mark: str,
         gateway: str,
+        tmp_path: pathlib.Path,
         is_remote: bool = False,
         columns_to_types: t.Optional[t.Dict[str, t.Union[str, exp.DataType]]] = None,
     ):
@@ -210,6 +211,7 @@ def __init__(
         self._catalogs: t.List[
             str
         ] = []  # keep track of any catalogs created via self.create_catalog() so we can drop them at the end
+        self.tmp_path = tmp_path
 
     @property
     def test_type(self) -> str:
@@ -655,6 +657,7 @@ def create_context(
                 private_sqlmesh_dir / "config.yml",
                 private_sqlmesh_dir / "config.yaml",
             ],
+            variables={"tmp_path": str(path or self.tmp_path)},
         )
         if config_mutator:
             config_mutator(self.gateway, config)

tests/core/engine_adapter/integration/config.yaml

@@ -5,7 +5,7 @@ gateways:
       type: duckdb
       catalogs:
         memory: ':memory:'
-        testing: 'testing.duckdb'
+        testing: "{{ var('tmp_path') }}/testing.duckdb"
 
   # Databases with docker images available
   inttest_trino_hive:

tests/core/engine_adapter/integration/conftest.py

@@ -27,15 +27,15 @@
 logger = logging.getLogger(__name__)
 
 
-@pytest.fixture(scope="session")
-def config() -> Config:
+@pytest.fixture
+def config(tmp_path: pathlib.Path) -> Config:
     return load_config_from_paths(
         Config,
         project_paths=[
-            pathlib.Path("examples/wursthall/config.yaml"),
             pathlib.Path(os.path.join(os.path.dirname(__file__), "config.yaml")),
         ],
         personal_paths=[pathlib.Path("~/.sqlmesh/config.yaml").expanduser()],
+        variables={"tmp_path": str(tmp_path)},
     )
 
 
@@ -89,7 +89,9 @@ def _create(engine_name: str, gateway: str) -> EngineAdapter:
 
 @pytest.fixture
 def create_test_context(
-    request: FixtureRequest, create_engine_adapter: t.Callable[[str, str], EngineAdapter]
+    request: FixtureRequest,
+    create_engine_adapter: t.Callable[[str, str], EngineAdapter],
+    tmp_path: pathlib.Path,
 ) -> t.Callable[[IntegrationTestEngine, str, str, str], t.Iterable[TestContext]]:
     def _create(
         engine: IntegrationTestEngine, gateway: str, test_type: str, table_format: str
@@ -103,6 +105,7 @@ def _create(
             engine_adapter,
             f"{engine.engine}_{table_format}",
             gateway,
+            tmp_path=tmp_path,
             is_remote=is_remote,
         )
 

tests/core/engine_adapter/integration/test_integration.py

@@ -1,7 +1,6 @@
 # type: ignore
 from __future__ import annotations
 
-import os
 import pathlib
 import re
 import sys
@@ -19,7 +18,6 @@
 
 from sqlmesh import Config, Context
 from sqlmesh.cli.project_init import init_example_project
-from sqlmesh.core.config import load_config_from_paths
 from sqlmesh.core.config.connection import ConnectionConfig
 import sqlmesh.core.dialect as d
 from sqlmesh.core.environment import EnvironmentSuffixTarget
@@ -1936,49 +1934,16 @@ def test_transaction(ctx: TestContext):
     ctx.compare_with_current(table, input_data)
 
 
-def test_sushi(ctx: TestContext, tmp_path_factory: pytest.TempPathFactory):
+def test_sushi(ctx: TestContext, tmp_path: pathlib.Path):
     if ctx.mark == "athena_hive":
         pytest.skip(
             "Sushi end-to-end tests only need to run once for Athena because sushi needs a hybrid of both Hive and Iceberg"
         )
 
-    tmp_path = tmp_path_factory.mktemp(f"sushi_{ctx.test_id}")
-
     sushi_test_schema = ctx.add_test_suffix("sushi")
     sushi_state_schema = ctx.add_test_suffix("sushi_state")
     raw_test_schema = ctx.add_test_suffix("raw")
 
-    config = load_config_from_paths(
-        Config,
-        project_paths=[
-            pathlib.Path(os.path.join(os.path.dirname(__file__), "config.yaml")),
-        ],
-        personal_paths=[pathlib.Path("~/.sqlmesh/config.yaml").expanduser()],
-    )
-    before_all = [
-        f"CREATE SCHEMA IF NOT EXISTS {raw_test_schema}",
-        f"DROP VIEW IF EXISTS {raw_test_schema}.demographics",
-        f"CREATE VIEW {raw_test_schema}.demographics AS (SELECT 1 AS customer_id, '00000' AS zip)",
-    ]
-    config.before_all = [
-        quote_identifiers(
-            parse_one(e, dialect=config.model_defaults.dialect),
-            dialect=config.model_defaults.dialect,
-        ).sql(dialect=config.model_defaults.dialect)
-        for e in before_all
-    ]
-
-    # To enable parallelism in integration tests
-    config.gateways = {ctx.gateway: config.gateways[ctx.gateway]}
-    current_gateway_config = config.gateways[ctx.gateway]
-    current_gateway_config.state_schema = sushi_state_schema
-
-    if ctx.dialect == "athena":
-        # Ensure that this test is using the same s3_warehouse_location as TestContext (which includes the testrun_id)
-        current_gateway_config.connection.s3_warehouse_location = (
-            ctx.engine_adapter.s3_warehouse_location
-        )
-
     # Copy sushi example to tmpdir
     shutil.copytree(pathlib.Path("./examples/sushi"), tmp_path, dirs_exist_ok=True)
 
@@ -2000,7 +1965,23 @@ def test_sushi(ctx: TestContext, tmp_path_factory: pytest.TempPathFactory):
                     contents = contents.replace(search, replace)
                 f.write_text(contents)
 
-    context = Context(paths=tmp_path, config=config, gateway=ctx.gateway)
+    before_all = [
+        f"CREATE SCHEMA IF NOT EXISTS {raw_test_schema}",
+        f"DROP VIEW IF EXISTS {raw_test_schema}.demographics",
+        f"CREATE VIEW {raw_test_schema}.demographics AS (SELECT 1 AS customer_id, '00000' AS zip)",
+    ]
+
+    def _mutate_config(gateway: str, config: Config) -> None:
+        config.gateways[gateway].state_schema = sushi_state_schema
+        config.before_all = [
+            quote_identifiers(
+                parse_one(e, dialect=config.model_defaults.dialect),
+                dialect=config.model_defaults.dialect,
+            ).sql(dialect=config.model_defaults.dialect)
+            for e in before_all
+        ]
+
+    context = ctx.create_context(_mutate_config, path=tmp_path, ephemeral_state_connection=False)
 
     end = now()
     start = to_date(end - timedelta(days=7))
@@ -2355,9 +2336,7 @@ def validate_no_comments(
         ctx._schemas.append(schema)
 
 
-def test_init_project(ctx: TestContext, tmp_path_factory: pytest.TempPathFactory):
-    tmp_path = tmp_path_factory.mktemp(f"init_project_{ctx.test_id}")
-
+def test_init_project(ctx: TestContext, tmp_path: pathlib.Path):
     schema_name = ctx.add_test_suffix(TEST_SCHEMA)
     state_schema = ctx.add_test_suffix("sqlmesh_state")
 
@@ -2383,33 +2362,15 @@ def _normalize_snowflake(name: str, prefix_regex: str = "(sqlmesh__)(.*)"):
 
     init_example_project(tmp_path, ctx.engine_type, schema_name=schema_name)
 
-    config = load_config_from_paths(
-        Config,
-        project_paths=[
-            pathlib.Path(os.path.join(os.path.dirname(__file__), "config.yaml")),
-        ],
-        personal_paths=[pathlib.Path("~/.sqlmesh/config.yaml").expanduser()],
-    )
-
-    # ensure default dialect comes from init_example_project and not ~/.sqlmesh/config.yaml
-    if config.model_defaults.dialect != ctx.dialect:
-        config.model_defaults = config.model_defaults.copy(update={"dialect": ctx.dialect})
-
-    # To enable parallelism in integration tests
-    config.gateways = {ctx.gateway: config.gateways[ctx.gateway]}
-    current_gateway_config = config.gateways[ctx.gateway]
-
-    if ctx.dialect == "athena":
-        # Ensure that this test is using the same s3_warehouse_location as TestContext (which includes the testrun_id)
-        current_gateway_config.connection.s3_warehouse_location = (
-            ctx.engine_adapter.s3_warehouse_location
-        )
+    def _mutate_config(gateway: str, config: Config):
+        # ensure default dialect comes from init_example_project and not ~/.sqlmesh/config.yaml
+        if config.model_defaults.dialect != ctx.dialect:
+            config.model_defaults = config.model_defaults.copy(update={"dialect": ctx.dialect})
 
-    # Ensure the state schema is unique to this test
-    config.gateways[ctx.gateway].state_schema = state_schema
+        # Ensure the state schema is unique to this test (since we deliberately use the warehouse as the state connection)
+        config.gateways[gateway].state_schema = state_schema
 
-    context = Context(paths=tmp_path, config=config, gateway=ctx.gateway)
-    ctx.engine_adapter = context.engine_adapter
+    context = ctx.create_context(_mutate_config, path=tmp_path, ephemeral_state_connection=False)
 
     if ctx.default_table_format:
         # if the default table format is explicitly set, ensure its being used