Some text" in html
+ assert "Second paragraph." in html
+
+
+@pytest.mark.unit
+def test_strip_html_removes_tags_and_collapses_ws():
+ txt = "
Hello World
\n X "
+ assert docs_mod._strip_html(txt) == "Hello World X"
+
+
+@pytest.mark.unit
+def test_read_markdown_file_with_front_matter(tmp_path: Path):
+ p = tmp_path / "doc.md"
+ p.write_text(
+ "---\ntitle: Hello\n---\nBody text\n",
+ encoding="utf-8",
+ )
+ fm, body = docs_mod._read_markdown_file(p)
+ assert fm == {"title": "Hello"}
+ assert body.strip() == "Body text"
+
+
+@pytest.mark.unit
+def test_read_markdown_file_no_front_matter(tmp_path: Path):
+ p = tmp_path / "doc.md"
+ p.write_text("Plain body", encoding="utf-8")
+ fm, body = docs_mod._read_markdown_file(p)
+ assert fm == {}
+ assert body == "Plain body"
+
+
+@pytest.mark.unit
+def test_reverse_deps_builds_inverse_graph():
+ nodes = {
+ "a": SimpleNamespace(name="a", deps=["b", "c"]),
+ "b": SimpleNamespace(name="b", deps=["c"]),
+ "c": SimpleNamespace(name="c", deps=[]),
+ }
+ rev = docs_mod._reverse_deps(nodes) # type: ignore[arg-type]
+ # c is used by a and b
+ assert rev["c"] == ["a", "b"]
+ # b is used by a
+ assert rev["b"] == ["a"]
+ # a is not used by anyone
+ assert rev["a"] == []
+
+
+@pytest.mark.unit
+def test_materialization_legend_has_incremental():
+ legend = docs_mod._materialization_legend()
+ assert "incremental" in legend
+ assert legend["incremental"]["label"] == "incremental"
+
+
+# ---------------------------------------------------------------------------
+# read_docs_metadata
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+def test_read_docs_metadata_merges_yaml_and_markdown(tmp_path: Path):
+ # project.yml with docs
+ (tmp_path / "project.yml").write_text(
+ textwrap.dedent(
+ """
+ docs:
+ models:
+ my_model:
+ description: "YAML desc"
+ columns:
+ col1: "YAML col1"
+ lineage:
+ col1:
+ from:
+ - table: src.t
+ column: x
+ transformed: true
+ """
+ ),
+ encoding="utf-8",
+ )
+
+ # Markdown model override (should win over YAML description)
+ md_dir = tmp_path / "docs" / "models"
+ md_dir.mkdir(parents=True)
+ (md_dir / "my_model.md").write_text("Markdown desc", encoding="utf-8")
+
+ # Markdown column override
+ col_dir = tmp_path / "docs" / "columns" / "my_model"
+ col_dir.mkdir(parents=True)
+ (col_dir / "col1.md").write_text("MD col1", encoding="utf-8")
+
+ meta = docs_mod.read_docs_metadata(tmp_path)
+
+ # model present
+ assert "my_model" in meta["models"]
+ # description_html from MD wins
+ assert "Markdown desc" in meta["models"]["my_model"]["description_html"]
+ # YAML lineage preserved
+ assert "lineage" in meta["models"]["my_model"]
+ # column override present
+ assert meta["columns"]["my_model"]["col1"].startswith("MD col1")
+
+
+# ---------------------------------------------------------------------------
+# _apply_descriptions_to_models
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+def test_apply_descriptions_to_models_applies_short_and_column_desc():
+ models = [
+ docs_mod.ModelDoc(
+ name="m1",
+ kind="sql",
+ path="models/m1.sql",
+ relation="db.sc.m1",
+ deps=[],
+ materialized="table",
+ )
+ ]
+ docs_meta = {
+ "models": {
+ "m1": {
+ "description_html": "
Hello world
",
+ "columns": {"col1": "Col 1
"},
+ }
+ },
+ "columns": {
+ "db.sc.m1": {
+ "col2": "Col 2
",
+ }
+ },
+ }
+ cols_by_table = {
+ "db.sc.m1": [
+ docs_mod.ColumnInfo("col1", "STRING", True),
+ docs_mod.ColumnInfo("col2", "INT", False),
+ ]
+ }
+
+ docs_mod._apply_descriptions_to_models(models, docs_meta, cols_by_table, with_schema=True)
+
+ assert models[0].description_html == "Hello world
"
+ assert models[0].description_short == "Hello world"
+ # column 1 desc from model-level
+ assert cols_by_table["db.sc.m1"][0].description_html == "Col 1
"
+ # column 2 desc from relation-level
+ assert cols_by_table["db.sc.m1"][1].description_html == "Col 2
"
+
+
+# ---------------------------------------------------------------------------
+# render_site (with patched jinja + registry)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+def test_render_site_writes_index_and_model_pages(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+ fake_nodes_raw = {
+ "model_a": SimpleNamespace(
+ name="model_a",
+ kind="sql",
+ path=tmp_path / "models" / "model_a.sql",
+ deps=["model_b"],
+ meta={"materialized": "view"},
+ ),
+ "model_b": SimpleNamespace(
+ name="model_b",
+ kind="python",
+ path=tmp_path / "models" / "model_b.py",
+ deps=[],
+ meta={},
+ ),
+ }
+
+ monkeypatch.setattr(
+ docs_mod,
+ "REGISTRY",
+ SimpleNamespace(
+ nodes=fake_nodes_raw,
+ macros={},
+ get_project_dir=lambda: tmp_path,
+ ),
+ raising=True,
+ )
+
+ monkeypatch.setattr(docs_mod, "_init_jinja", lambda: _FakeEnv(), raising=True)
+ fake_nodes = cast(dict[str, Node], fake_nodes_raw)
+
+ docs_mod.render_site(tmp_path, fake_nodes, executor=None, with_schema=False)
+
+ index_file = tmp_path / "index.html"
+ assert index_file.exists()
+ assert "INDEX" in index_file.read_text(encoding="utf-8")
+
+ model_a_file = tmp_path / "model_a.html"
+ model_b_file = tmp_path / "model_b.html"
+ assert model_a_file.exists()
+ assert model_b_file.exists()
+
+ assert "MODEL model_a" in model_a_file.read_text(encoding="utf-8")
+ assert "MODEL model_b" in model_b_file.read_text(encoding="utf-8")
+
+
+# ---------------------------------------------------------------------------
+# _collect_columns engine stubs
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+def test_collect_columns_prefers_spark():
+ class FakeCol:
+ def __init__(self, name: str):
+ self.name = name
+ self.dataType = "INT"
+ self.nullable = True
+
+ class FakeTable:
+ def __init__(self, name: str):
+ self.name = name
+ self.database = None
+ self.catalog = None
+
+ class FakeSparkCatalog:
+ def listTables(self):
+ return [FakeTable("T1")]
+
+ def listColumns(self, ident, database=None):
+ return [FakeCol("C1"), FakeCol("C2")]
+
+ class FakeSpark:
+ catalog = FakeSparkCatalog()
+
+ cols = docs_mod._collect_columns(SimpleNamespace(spark=FakeSpark()))
+ assert "T1" in cols
+ assert [c.name for c in cols["T1"]] == ["C1", "C2"]
+
+
+@pytest.mark.unit
+def test_collect_columns_with_unknown_executor_returns_empty():
+ cols = docs_mod._collect_columns(object())
+ assert cols == {}
+
+
+# ---------------------- _columns_duckdb ----------------------
+
+
+@pytest.mark.unit
+def test_columns_duckdb_collects_tables_and_cols():
+ class FakeCursor:
+ def __init__(self, rows):
+ self._rows = rows
+
+ def fetchall(self):
+ return self._rows
+
+ class FakeConn:
+ def __init__(self, rows):
+ self._rows = rows
+
+ def execute(self, _sql: str):
+ return FakeCursor(self._rows)
+
+ rows = [
+ # table_name, column_name, data_type, is_nullable
+ ("my_table", "id", "INTEGER", "NO"),
+ ("my_table", "name", "TEXT", "YES"),
+ ("other", "x", "BOOLEAN", "YES"),
+ ]
+ fake_con = FakeConn(rows)
+
+ cols = docs_mod._columns_duckdb(fake_con)
+
+ assert set(cols.keys()) == {"my_table", "other"}
+ mt = cols["my_table"]
+ assert [c.name for c in mt] == ["id", "name"]
+ assert mt[0].dtype == "INTEGER"
+ assert mt[0].nullable is False
+ assert mt[1].nullable is True
+
+
+# ---------------------- _columns_postgres ----------------------
+
+
+@pytest.mark.unit
+def test_columns_postgres_collects_from_engine():
+ class FakeResult:
+ def __init__(self, rows):
+ self._rows = rows
+
+ def fetchall(self):
+ return self._rows
+
+ class FakeConn:
+ def __init__(self, rows):
+ self._rows = rows
+
+ def execute(self, _stmt):
+ return FakeResult(self._rows)
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, exc_type, exc, tb):
+ return False
+
+ class FakeEngine:
+ def __init__(self, rows):
+ self._rows = rows
+
+ def begin(self):
+ return FakeConn(self._rows)
+
+ rows = [
+ # table_name, column_name, data_type, is_nullable
+ ("public_tbl", "id", "integer", "YES"),
+ ("public_tbl", "email", "text", "NO"),
+ ]
+ fake_engine = FakeEngine(rows)
+
+ cols = docs_mod._columns_postgres(fake_engine)
+
+ assert "public_tbl" in cols
+ tcols = cols["public_tbl"]
+ assert [c.name for c in tcols] == ["id", "email"]
+ assert tcols[0].dtype == "integer"
+ # in deiner Implementierung: nullable == "YES"
+ assert tcols[0].nullable is True
+ assert tcols[1].nullable is False
+
+
+# ---------------------- _columns_snowflake ----------------------
+
+
+@pytest.mark.unit
+def test_columns_snowflake_collects_from_session():
+ # snowflake .collect() liefert list[Row], aber wir können dicts nehmen
+ class FakeDF:
+ def __init__(self, rows):
+ self._rows = rows
+
+ def collect(self):
+ return self._rows
+
+ class FakeSession:
+ def __init__(self, rows):
+ self._rows = rows
+
+ def sql(self, _sql: str):
+ return FakeDF(self._rows)
+
+ rows = [
+ {
+ "TABLE_NAME": "T1",
+ "COLUMN_NAME": "ID",
+ "DATA_TYPE": "NUMBER",
+ "IS_NULLABLE": "NO",
+ },
+ {
+ "TABLE_NAME": "T1",
+ "COLUMN_NAME": "NAME",
+ "DATA_TYPE": "TEXT",
+ "IS_NULLABLE": "YES",
+ },
+ {
+ "TABLE_NAME": "T2",
+ "COLUMN_NAME": "TS",
+ "DATA_TYPE": "TIMESTAMP_NTZ",
+ "IS_NULLABLE": "YES",
+ },
+ ]
+ fake_session = FakeSession(rows)
+
+ cols = docs_mod._columns_snowflake(fake_session)
+
+ assert set(cols.keys()) == {"T1", "T2"}
+ t1 = cols["T1"]
+ assert [c.name for c in t1] == ["ID", "NAME"]
+ assert t1[0].dtype == "NUMBER"
+ assert t1[0].nullable is False
+ assert t1[1].nullable is True
+
+ t2 = cols["T2"]
+ assert t2[0].name == "TS"
+ assert t2[0].dtype == "TIMESTAMP_NTZ"
+ assert t2[0].nullable is True
+
+
+@pytest.mark.unit
+def test_build_macro_list_collects_and_sorts(monkeypatch: pytest.MonkeyPatch, tmp_path: Path):
+ # Arrange
+ project_dir = tmp_path
+ macros_dir = project_dir / "macros"
+ macros_dir.mkdir(parents=True, exist_ok=True)
+
+ m1 = macros_dir / "util_dates.sql"
+ m1.write_text("-- sql macro 1", encoding="utf-8")
+
+ m2 = macros_dir / "cleanup.sql"
+ m2.write_text("-- sql macro 2", encoding="utf-8")
+
+ m3 = macros_dir / "py_macro.py"
+ m3.write_text("def run(): ...", encoding="utf-8")
+
+ fake_registry = SimpleNamespace(
+ macros={
+ "util_dates": m1,
+ "cleanup": m2,
+ "py_macro": m3,
+ }
+ )
+ monkeypatch.setattr(docs_mod, "REGISTRY", fake_registry, raising=True)
+
+ # Act
+ res = docs_mod._build_macro_list(project_dir)
+
+ # Assert
+ names = [x["name"] for x in res]
+ assert names == ["py_macro", "cleanup", "util_dates"]
+
+ paths = {x["name"]: x["path"] for x in res}
+ assert paths["cleanup"] == "macros/cleanup.sql"
+ assert paths["util_dates"] == "macros/util_dates.sql"
+ assert paths["py_macro"] == "macros/py_macro.py"
+
+ kinds = {x["name"]: x["kind"] for x in res}
+ assert kinds["py_macro"] == "python"
+ assert kinds["cleanup"] == "sql"
+ assert kinds["util_dates"] == "sql"
+
+
+@pytest.mark.unit
+def test_infer_and_attach_lineage_sql_branch_is_used(monkeypatch: pytest.MonkeyPatch):
+ m = docs_mod.ModelDoc(
+ name="model_sql",
+ kind="sql",
+ path="models/model_sql.sql",
+ relation="project.dataset.model_sql",
+ deps=["src_table"],
+ materialized="table",
+ )
+ models = [m]
+
+ cols_by_table = {
+ "project.dataset.model_sql": [
+ docs_mod.ColumnInfo("id", "INT", True),
+ docs_mod.ColumnInfo("name", "STRING", True),
+ ]
+ }
+
+ class FakeExecutor:
+ def render_sql(self, node, jenv, ref_resolver=None, source_resolver=None):
+ return "select id, name from src_table"
+
+ def _resolve_ref(self, name, env):
+ return f"resolved_ref_{name}"
+
+ def _resolve_source(self, source, table):
+ return f"resolved_source_{source}_{table}"
+
+ fake_executor = FakeExecutor()
+
+ fake_node = SimpleNamespace(
+ name="model_sql",
+ kind="sql",
+ path=Path("models/model_sql.sql"),
+ deps=["src_table"],
+ meta={},
+ )
+ fake_registry = SimpleNamespace(
+ nodes={"model_sql": fake_node},
+ env=SimpleNamespace(),
+ )
+ monkeypatch.setattr(docs_mod, "REGISTRY", fake_registry, raising=True)
+
+ monkeypatch.setattr(
+ docs_mod,
+ "infer_sql_lineage",
+ lambda rendered: {"id": [{"from_relation": "src_table", "from_column": "id"}]},
+ raising=True,
+ )
+ monkeypatch.setattr(
+ docs_mod,
+ "parse_sql_lineage_overrides",
+ lambda rendered: {},
+ raising=True,
+ )
+ monkeypatch.setattr(
+ docs_mod,
+ "merge_lineage",
+ lambda base, overrides: base,
+ raising=True,
+ )
+ docs_meta: dict[str, object] = {}
+
+ # ACT
+ docs_mod._infer_and_attach_lineage(
+ models,
+ fake_executor,
+ docs_meta,
+ cols_by_table,
+ with_schema=True,
+ )
+
+ # ASSERT
+ col_id = cols_by_table["project.dataset.model_sql"][0]
+ assert col_id.name == "id"
+ assert col_id.lineage == [{"from_relation": "src_table", "from_column": "id"}]
+
+
+@pytest.mark.unit
+def test_infer_and_attach_lineage_yaml_override_branch_is_used(monkeypatch: pytest.MonkeyPatch):
+ m = docs_mod.ModelDoc(
+ name="model_yml",
+ kind="sql",
+ path="models/model_yml.sql",
+ relation="project.dataset.model_yml",
+ deps=[],
+ materialized="table",
+ )
+ models = [m]
+
+ cols_by_table = {
+ "project.dataset.model_yml": [
+ docs_mod.ColumnInfo("total", "NUMBER", True),
+ docs_mod.ColumnInfo("cnt", "NUMBER", True),
+ ]
+ }
+
+ docs_meta = {
+ "models": {
+ "model_yml": {
+ "description_html": None,
+ "columns": {},
+ "lineage": {
+ "total": {
+ "from": [
+ {"table": "project.dataset.orders", "column": "amount"},
+ ],
+ "transformed": True,
+ }
+ },
+ }
+ },
+ "columns": {},
+ }
+
+ monkeypatch.setattr(docs_mod, "infer_sql_lineage", lambda *_a, **_k: {}, raising=True)
+ monkeypatch.setattr(docs_mod, "parse_sql_lineage_overrides", lambda *_a, **_k: {}, raising=True)
+ monkeypatch.setattr(
+ docs_mod,
+ "merge_lineage",
+ lambda base, overrides: (overrides or base),
+ raising=True,
+ )
+
+ # ACT
+ docs_mod._infer_and_attach_lineage(
+ models,
+ executor=None,
+ docs_meta=docs_meta,
+ cols_by_table=cols_by_table,
+ with_schema=True,
+ )
+
+ # ASSERT
+ col_total = cols_by_table["project.dataset.model_yml"][0]
+ assert col_total.name == "total"
+ assert col_total.lineage == [
+ {
+ "from_relation": "project.dataset.orders",
+ "from_column": "amount",
+ "transformed": True,
+ }
+ ]
diff --git a/tests/unit/executors/test_bigquery_bf_exec_unit.py b/tests/unit/executors/test_bigquery_bf_exec_unit.py
new file mode 100644
index 0000000..9275053
--- /dev/null
+++ b/tests/unit/executors/test_bigquery_bf_exec_unit.py
@@ -0,0 +1,410 @@
+# tests/unit/executors/test_bigquery_bf_exec_unit.py
+from __future__ import annotations
+
+import sys
+import types
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any, ClassVar
+
+import pytest
+from tests.common.mock.bigquery import (
+ FakeBadRequest,
+ FakeClient,
+ FakeField,
+ FakeJob,
+ FakeNotFound,
+ install_fake_bigquery,
+)
+
+import fastflowtransform.executors._bigquery_mixin as bq_mix_mod
+import fastflowtransform.executors.bigquery_bf_exec as bq_exec_mod
+from fastflowtransform.core import Node
+
+# ---------------------- BigFrames-Fakes ------------------------------------
+
+
+class _FakeBigQueryOptions:
+ def __init__(self, *a, **kw):
+ self.kw = kw
+
+
+class _FakeBFSession:
+ def __init__(self, *a, **kw):
+ self.created_with = (a, kw)
+ self.read_calls: list[str] = []
+
+ def read_gbq(self, table_id: str) -> Any:
+ self.read_calls.append(table_id)
+ return SimpleNamespace(
+ columns=["id", "name"],
+ to_gbq=None,
+ materialize=None,
+ )
+
+
+@pytest.fixture
+def bq_exec(monkeypatch):
+ _ = install_fake_bigquery(monkeypatch, [bq_exec_mod, bq_mix_mod])
+
+ fake_bigframes = types.ModuleType("bigframes")
+ fake_conf = types.ModuleType("bigframes._config")
+ fake_conf_bq = types.ModuleType("bigframes._config.bigquery_options")
+
+ fake_conf_bq.BigQueryOptions = _FakeBigQueryOptions # type: ignore[attr-defined]
+ fake_bigframes.Session = _FakeBFSession # type: ignore[attr-defined]
+
+ sys.modules.setdefault("bigframes", fake_bigframes)
+ sys.modules.setdefault("bigframes._config", fake_conf)
+ sys.modules["bigframes._config.bigquery_options"] = fake_conf_bq
+
+ monkeypatch.setattr(bq_exec_mod, "bigframes", fake_bigframes, raising=True)
+
+ ex = bq_exec_mod.BigQueryBFExecutor(project="p1", dataset="ds1", location="EU")
+
+ assert isinstance(ex.client, FakeClient)
+ assert isinstance(ex.session, _FakeBFSession)
+
+ ex.client.add_dataset("p1.ds1")
+
+ monkeypatch.setattr(bq_exec_mod, "NotFound", FakeNotFound, raising=True)
+ monkeypatch.setattr(bq_exec_mod, "BadRequest", FakeBadRequest, raising=True)
+
+ return ex
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_read_relation_happy(bq_exec):
+ df = bq_exec._read_relation("tbl", Node(name="n", kind="sql", path=Path(".")), deps=["x"])
+ assert hasattr(df, "columns")
+ assert df.columns == ["id", "name"]
+ assert bq_exec.session.read_calls == ["p1.ds1.tbl"]
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_read_relation_not_found_raises_nice_message(bq_exec):
+ def boom(_table_id: str):
+ raise FakeNotFound("nope")
+
+ bq_exec.session.read_gbq = boom # type: ignore[assignment]
+ bq_exec.client.add_table("p1.ds1", "existing1")
+
+ with pytest.raises(RuntimeError) as exc:
+ bq_exec._read_relation(
+ "missing",
+ Node(name="n", kind="sql", path=Path(".")),
+ deps=["dep1"],
+ )
+ msg = str(exc.value)
+ assert "Dependency table not found" in msg
+ assert "dep1" in msg
+ assert "existing1" in msg
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_materialize_relation_prefers_to_gbq(bq_exec):
+ called: dict[str, Any] = {}
+
+ class DF:
+ def to_gbq(self, table_id, if_exists="replace"):
+ called["table_id"] = table_id
+ called["if_exists"] = if_exists
+
+ bq_exec._materialize_relation("out_tbl", DF(), Node(name="m", kind="python", path=Path(".")))
+ assert called["table_id"] == "p1.ds1.out_tbl"
+ assert called["if_exists"] == "replace"
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_materialize_relation_fallback_to_materialize(bq_exec):
+ called: dict[str, Any] = {}
+
+ class DF:
+ def materialize(self, table, mode="overwrite"):
+ called["table"] = table
+ called["mode"] = mode
+
+ bq_exec._materialize_relation("out_tbl", DF(), Node(name="m", kind="python", path=Path(".")))
+ assert called["table"] == "p1.ds1.out_tbl"
+ assert called["mode"] == "overwrite"
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_materialize_relation_raises_if_no_supported_method(bq_exec):
+ class DF:
+ columns: ClassVar[list[str]] = ["x"]
+
+ with pytest.raises(RuntimeError):
+ bq_exec._materialize_relation(
+ "out_tbl", DF(), Node(name="m", kind="python", path=Path("."))
+ )
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_validate_required_single_frame_ok(bq_exec):
+ frame = SimpleNamespace(columns=["a", "b", "c"])
+ bq_exec._validate_required("m1", frame, {"ds1.tbl": {"a", "b"}})
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_validate_required_single_frame_missing(bq_exec):
+ frame = SimpleNamespace(columns=["a"])
+ with pytest.raises(ValueError) as exc:
+ bq_exec._validate_required("m1", frame, {"ds1.tbl": {"a", "b"}})
+ assert "missing" in str(exc.value)
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_validate_required_multi_input_ok(bq_exec):
+ f1 = SimpleNamespace(columns=["id", "name"])
+ f2 = SimpleNamespace(columns=["user_id", "order_id"])
+ bq_exec._validate_required(
+ "m1",
+ {"ds1.users": f1, "ds1.orders": f2},
+ {"ds1.users": {"id"}, "ds1.orders": {"order_id"}},
+ )
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_is_frame_detection(bq_exec):
+ class WithToGbq:
+ def to_gbq(self): ...
+
+ class WithMaterialize:
+ def materialize(self): ...
+
+ class WithColumns:
+ columns: ClassVar[list[str]] = ["x"]
+
+ assert bq_exec._is_frame(WithToGbq()) is True
+ assert bq_exec._is_frame(WithMaterialize()) is True
+ assert bq_exec._is_frame(WithColumns()) is True
+ assert bq_exec._is_frame(object()) is False
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_columns_of_prefers_columns_attr(bq_exec):
+ frame = SimpleNamespace(columns=["a", "b"])
+ assert bq_exec._columns_of(frame) == ["a", "b"]
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_columns_of_falls_back_to_schema_names(bq_exec):
+ class F:
+ schema = SimpleNamespace(names=["x", "y"])
+
+ assert bq_exec._columns_of(F()) == ["x", "y"]
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_format_relation_for_ref(bq_exec):
+ rel = bq_exec._format_relation_for_ref("m1")
+ assert "p1" in rel
+ assert "ds1" in rel
+ assert "m1" in rel
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_format_source_reference(bq_exec):
+ cfg = {"identifier": "src_tbl", "project": "other_proj", "dataset": "raw"}
+ ref = bq_exec._format_source_reference(cfg, "src", "tbl")
+ assert "other_proj" in ref
+ assert "raw" in ref
+ assert "src_tbl" in ref
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_create_or_replace_view_calls_client_query(bq_exec):
+ bq_exec.client.queries.clear()
+ bq_exec._create_or_replace_view(
+ "p1.ds1.v_view", "SELECT 1", Node(name="x", kind="sql", path=Path("."))
+ )
+ assert any("CREATE OR REPLACE VIEW" in q[0] for q in bq_exec.client.queries)
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_create_or_replace_table_calls_client_query(bq_exec):
+ bq_exec.client.queries.clear()
+ bq_exec._create_or_replace_table(
+ "p1.ds1.t_out", "SELECT 1", Node(name="x", kind="sql", path=Path("."))
+ )
+ assert any("CREATE OR REPLACE TABLE" in q[0] for q in bq_exec.client.queries)
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_create_or_replace_view_from_table_calls_client_query(bq_exec):
+ bq_exec.client.queries.clear()
+ bq_exec._create_or_replace_view_from_table(
+ "v_users", "ds1.users", Node(name="x", kind="python", path=Path("."))
+ )
+ assert any("CREATE OR REPLACE VIEW" in q[0] for q in bq_exec.client.queries)
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_exists_relation_true(bq_exec, monkeypatch):
+ class _Job:
+ def __init__(self):
+ self._rows = [(1,)]
+
+ def result(self):
+ return self._rows
+
+ monkeypatch.setattr(
+ bq_exec.client,
+ "query",
+ lambda sql, location=None, job_config=None: _Job(),
+ raising=True,
+ )
+
+ ok = bq_exec.exists_relation("some_table")
+ assert ok is True
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_create_table_as_cleans_select(bq_exec):
+ bq_exec.client.queries.clear()
+ bq_exec.create_table_as("dst_tbl", "SELECT 1;")
+ sql = bq_exec.client.queries[-1][0]
+ assert "CREATE TABLE" in sql
+ assert "SELECT 1" in sql
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_incremental_insert_cleans_select(bq_exec):
+ bq_exec.client.queries.clear()
+ bq_exec.incremental_insert("dst_tbl", "SELECT 1;")
+ sql = bq_exec.client.queries[-1][0]
+ assert "INSERT INTO" in sql
+ assert "SELECT 1" in sql
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_incremental_merge_executes_two_statements(bq_exec):
+ bq_exec.client.queries.clear()
+ bq_exec.incremental_merge("dst_tbl", "SELECT 1 AS id", ["id"])
+ assert len(bq_exec.client.queries) == 2
+ assert "DELETE FROM" in bq_exec.client.queries[0][0]
+ assert "INSERT INTO" in bq_exec.client.queries[1][0]
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_alter_table_sync_schema_adds_missing_columns(bq_exec):
+ def fake_query(sql: str, location: str | None = None, job_config: Any | None = None):
+ if "WHERE 1=0" in sql:
+ return FakeJob(schema=[FakeField("id"), FakeField("new_col", "INT64")])
+ if sql.startswith("ALTER TABLE"):
+ bq_exec.client.queries.append((sql, location, job_config))
+ return FakeJob()
+ return FakeJob()
+
+ bq_exec.client.query = fake_query # type: ignore[assignment]
+ bq_exec.client.get_table = lambda ref: SimpleNamespace(schema=[FakeField("id")]) # type: ignore[assignment]
+
+ bq_exec.alter_table_sync_schema("existing", "SELECT 1 AS id, 2 AS new_col")
+ assert any("ADD COLUMN new_col" in q[0] for q in bq_exec.client.queries)
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_on_node_built_best_effort(monkeypatch, bq_exec):
+ called = {"ensure": 0, "upsert": 0}
+
+ def fake_ensure(ex):
+ called["ensure"] += 1
+
+ def fake_upsert(ex, name, rel, fp, eng):
+ called["upsert"] += 1
+
+ monkeypatch.setattr(bq_exec_mod, "ensure_meta_table", fake_ensure)
+ monkeypatch.setattr(bq_exec_mod, "upsert_meta", fake_upsert)
+
+ bq_exec.on_node_built(Node(name="m", kind="sql", path=Path(".")), "p1.ds1.m", "fp123")
+
+ assert called["ensure"] == 1
+ assert called["upsert"] == 1
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_bf_apply_sql_materialization_calls_super(monkeypatch, bq_exec):
+ monkeypatch.setattr(bq_exec, "_ensure_dataset", lambda: None, raising=True)
+
+ import fastflowtransform.executors.bigquery_bf_exec as bq_bf_mod # noqa PLC0415
+
+ called: dict[str, str] = {}
+
+ monkeypatch.setattr(
+ bq_bf_mod.BaseExecutor,
+ "_apply_sql_materialization",
+ lambda self, node, target_sql, select_body, materialization: called.update(
+ {
+ "node": node.name,
+ "target_sql": target_sql,
+ "select_body": select_body,
+ "materialization": materialization,
+ }
+ ),
+ raising=True,
+ )
+
+ node = Node(name="m_apply", kind="sql", path=Path("."))
+
+ bq_exec._apply_sql_materialization(
+ node,
+ target_sql="`p1`.`ds1`.`t_out`",
+ select_body="SELECT 1",
+ materialization="table",
+ )
+
+ assert called["node"] == "m_apply"
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_apply_sql_materialization_wraps_badrequest(monkeypatch, bq_exec):
+ monkeypatch.setattr(
+ bq_exec_mod.BaseExecutor,
+ "_apply_sql_materialization",
+ lambda *a, **k: (_ for _ in ()).throw(FakeBadRequest("bad SQL")),
+ raising=True,
+ )
+
+ node = Node(name="m_err", kind="sql", path=Path("."))
+
+ with pytest.raises(RuntimeError) as exc:
+ bq_exec._apply_sql_materialization(
+ node,
+ target_sql="`p1`.`ds1`.`broken`",
+ select_body="SELECT bad",
+ materialization="table",
+ )
+ msg = str(exc.value)
+ assert "BigQuery SQL failed for" in msg
+ assert "SELECT bad" in msg
+ assert "broken" in msg
diff --git a/tests/unit/executors/test_bigquery_exec_unit.py b/tests/unit/executors/test_bigquery_exec_unit.py
new file mode 100644
index 0000000..f0d5455
--- /dev/null
+++ b/tests/unit/executors/test_bigquery_exec_unit.py
@@ -0,0 +1,339 @@
+# tests/unit/executors/test_bigquery_exec_unit.py
+from __future__ import annotations
+
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any, cast
+
+import pandas as pd
+import pytest
+from tests.common.mock.bigquery import (
+ FakeBadRequest,
+ FakeClient,
+ FakeField,
+ FakeJob,
+ FakeNotFound,
+ FakeWriteDisposition,
+ install_fake_bigquery,
+)
+
+import fastflowtransform.executors._bigquery_mixin as bq_mix_mod
+import fastflowtransform.executors.bigquery_exec as bq_exec_mod
+from fastflowtransform.core import Node
+
+
+@pytest.fixture
+def bq_exec(monkeypatch):
+ _ = install_fake_bigquery(monkeypatch, [bq_exec_mod, bq_mix_mod])
+
+ fake_client = FakeClient(project="p1", location="EU")
+
+ ex = bq_exec_mod.BigQueryExecutor(
+ project="p1",
+ dataset="ds1",
+ location="EU",
+ client=cast(Any, fake_client),
+ )
+
+ fake_client.add_dataset("p1.ds1")
+
+ monkeypatch.setattr(bq_exec_mod, "NotFound", FakeNotFound, raising=True)
+ monkeypatch.setattr(bq_exec_mod, "BadRequest", FakeBadRequest, raising=True)
+
+ return ex
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_read_relation_happy(bq_exec, monkeypatch):
+ def fake_query(sql: str, location: str | None = None, job_config: Any | None = None):
+ return FakeJob(
+ rows=[(1, "A")],
+ schema=[FakeField("id"), FakeField("name")],
+ )
+
+ bq_exec.client.query = fake_query # type: ignore[assignment]
+
+ df = bq_exec._read_relation("tbl", Node(name="n", kind="sql", path=Path(".")), deps=["x"])
+ assert list(df.columns) == ["id", "name"]
+ assert df.to_dict(orient="records") == [{"id": 1, "name": "A"}]
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_read_relation_not_found_raises_nice_message(bq_exec, monkeypatch):
+ def boom(sql: str, location: str | None = None, job_config: Any | None = None):
+ raise FakeNotFound("nope")
+
+ bq_exec.client.query = boom # type: ignore[assignment]
+
+ bq_exec.client.add_table("p1.ds1", "existing1")
+
+ with pytest.raises(RuntimeError) as exc:
+ bq_exec._read_relation(
+ "missing",
+ Node(name="n", kind="sql", path=Path(".")),
+ deps=["dep1"],
+ )
+
+ msg = str(exc.value)
+ assert "Dependency table not found" in msg
+ assert "dep1" in msg
+ assert "existing1" in msg
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_materialize_relation_happy(bq_exec, monkeypatch):
+ df = pd.DataFrame({"id": [1], "name": ["A"]})
+
+ called: dict[str, Any] = {}
+
+ def fake_load(df_in, table_id, job_config=None, location=None):
+ called["table_id"] = table_id
+ called["location"] = location
+ called["job_config"] = job_config
+ return FakeJob()
+
+ bq_exec.client.load_table_from_dataframe = fake_load # type: ignore[assignment]
+
+ bq_exec._materialize_relation("out_tbl", df, Node(name="m", kind="python", path=Path(".")))
+
+ assert called["table_id"] == "p1.ds1.out_tbl"
+ assert called["location"] == "EU"
+ assert (
+ getattr(called["job_config"], "write_disposition", None)
+ == FakeWriteDisposition.WRITE_TRUNCATE
+ )
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_materialize_relation_wraps_badrequest(bq_exec, monkeypatch):
+ df = pd.DataFrame({"id": [1]})
+
+ def bad_load(*_a, **_k):
+ raise FakeBadRequest("boom")
+
+ bq_exec.client.load_table_from_dataframe = bad_load # type: ignore[assignment]
+
+ with pytest.raises(RuntimeError) as exc:
+ bq_exec._materialize_relation("out_tbl", df, Node(name="m", kind="python", path=Path(".")))
+ msg = str(exc.value)
+ assert "BigQuery write failed" in msg
+ assert "out_tbl" in msg
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_create_view_over_table_calls_bq(bq_exec):
+ bq_exec.client.queries.clear()
+ node = Node(name="m", kind="python", path=Path("."))
+ bq_exec._create_view_over_table("v_users", "ds1.users", node)
+ assert any("CREATE OR REPLACE VIEW" in q[0] for q in bq_exec.client.queries)
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_frame_name(bq_exec):
+ assert bq_exec._frame_name() == "pandas"
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_format_relation_for_ref(bq_exec):
+ rel = bq_exec._format_relation_for_ref("m1")
+ assert "p1" in rel
+ assert "ds1" in rel
+ assert "m1" in rel
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_format_source_reference(bq_exec):
+ cfg = {
+ "identifier": "src_tbl",
+ "project": "other_proj",
+ "dataset": "raw",
+ }
+ ref = bq_exec._format_source_reference(cfg, "src", "tbl")
+ assert "other_proj" in ref
+ assert "raw" in ref
+ assert "src_tbl" in ref
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_apply_sql_materialization_calls_super_and_ensures_dataset(monkeypatch, bq_exec):
+ monkeypatch.setattr(bq_exec, "_ensure_dataset", lambda: None, raising=True)
+
+ import fastflowtransform.executors.bigquery_exec as bq_exec_mod # noqa PLC0415
+
+ called: dict[str, str] = {}
+
+ monkeypatch.setattr(
+ bq_exec_mod.BaseExecutor,
+ "_apply_sql_materialization",
+ lambda self, node, target_sql, select_body, materialization: called.update(
+ {
+ "node": node.name,
+ "target_sql": target_sql,
+ "select_body": select_body,
+ "materialization": materialization,
+ }
+ ),
+ raising=True,
+ )
+
+ node = Node(name="m_apply", kind="sql", path=Path("."))
+ bq_exec._apply_sql_materialization(
+ node,
+ target_sql="`p1`.`ds1`.`t_out`",
+ select_body="SELECT 1",
+ materialization="table",
+ )
+
+ assert called["node"] == "m_apply"
+ assert called["target_sql"] == "`p1`.`ds1`.`t_out`"
+ assert called["select_body"] == "SELECT 1"
+ assert called["materialization"] == "table"
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_apply_sql_materialization_wraps_badrequest(monkeypatch, bq_exec):
+ monkeypatch.setattr(
+ bq_exec_mod.BaseExecutor,
+ "_apply_sql_materialization",
+ lambda *a, **k: (_ for _ in ()).throw(FakeBadRequest("bq exploded")),
+ raising=True,
+ )
+
+ node = Node(name="m_bad", kind="sql", path=Path("."))
+
+ with pytest.raises(RuntimeError) as exc:
+ bq_exec._apply_sql_materialization(
+ node,
+ target_sql="`p1`.`ds1`.`t_out`",
+ select_body="SELECT 1",
+ materialization="table",
+ )
+ msg = str(exc.value)
+ assert "BigQuery SQL failed for" in msg
+ assert "SELECT 1" in msg
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_create_or_replace_view_calls_client_query(bq_exec):
+ bq_exec.client.queries.clear()
+ bq_exec._create_or_replace_view(
+ "p1.ds1.v_view", "SELECT 1", Node(name="x", kind="sql", path=Path("."))
+ )
+ assert any("CREATE OR REPLACE VIEW" in q[0] for q in bq_exec.client.queries)
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_create_or_replace_table_calls_client_query(bq_exec):
+ bq_exec.client.queries.clear()
+ bq_exec._create_or_replace_table(
+ "p1.ds1.t_out", "SELECT 1", Node(name="x", kind="sql", path=Path("."))
+ )
+ assert any("CREATE OR REPLACE TABLE" in q[0] for q in bq_exec.client.queries)
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_create_or_replace_view_from_table_calls_client_query(bq_exec):
+ bq_exec.client.queries.clear()
+ bq_exec._create_or_replace_view_from_table(
+ "v_users", "ds1.users", Node(name="x", kind="python", path=Path("."))
+ )
+ assert any("CREATE OR REPLACE VIEW" in q[0] for q in bq_exec.client.queries)
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_on_node_built_best_effort(monkeypatch, bq_exec):
+ called = {"ensure": 0, "upsert": 0}
+
+ def fake_ensure(ex):
+ called["ensure"] += 1
+
+ def fake_upsert(ex, name, rel, fp, eng):
+ called["upsert"] += 1
+
+ monkeypatch.setattr(bq_exec_mod, "ensure_meta_table", fake_ensure)
+ monkeypatch.setattr(bq_exec_mod, "upsert_meta", fake_upsert)
+
+ bq_exec.on_node_built(Node(name="m", kind="sql", path=Path(".")), "p1.ds1.m", "fp123")
+
+ assert called["ensure"] == 1
+ assert called["upsert"] == 1
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_exists_relation_true(bq_exec, monkeypatch):
+ class _Job:
+ def __init__(self):
+ self._rows = [(1,)]
+
+ def result(self):
+ return self._rows
+
+ monkeypatch.setattr(
+ bq_exec.client,
+ "query",
+ lambda sql, location=None, job_config=None: _Job(),
+ raising=True,
+ )
+
+ assert bq_exec.exists_relation("some_table") is True
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_create_table_as_cleans_select(bq_exec):
+ bq_exec.client.queries.clear()
+ bq_exec.create_table_as("dst_tbl", "SELECT 1;")
+ sql = bq_exec.client.queries[-1][0]
+ assert "CREATE TABLE" in sql
+ assert "SELECT 1" in sql
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_incremental_insert_cleans_select(bq_exec):
+ bq_exec.client.queries.clear()
+ bq_exec.incremental_insert("dst_tbl", "SELECT 1;")
+ sql = bq_exec.client.queries[-1][0]
+ assert "INSERT INTO" in sql
+ assert "SELECT 1" in sql
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_incremental_merge_executes_two_statements(bq_exec):
+ bq_exec.client.queries.clear()
+ bq_exec.incremental_merge("dst_tbl", "SELECT 1 AS id", ["id"])
+ assert len(bq_exec.client.queries) == 2
+ assert "DELETE FROM" in bq_exec.client.queries[0][0]
+ assert "INSERT INTO" in bq_exec.client.queries[1][0]
+
+
+@pytest.mark.unit
+@pytest.mark.bigquery
+def test_alter_table_sync_schema_adds_missing_columns(bq_exec, monkeypatch):
+ bq_exec.client.get_table = lambda ref: SimpleNamespace(schema=[FakeField("id")]) # type: ignore[assignment]
+
+ bq_exec.client.queries.clear()
+ bq_exec.alter_table_sync_schema("existing", "SELECT 1 AS id, 2 AS new_col")
+
+ assert any("ADD COLUMN new_col" in q[0] for q in bq_exec.client.queries)
diff --git a/tests/unit/executors/test_duckdb_exec_unit.py b/tests/unit/executors/test_duckdb_exec_unit.py
new file mode 100644
index 0000000..6aa8475
--- /dev/null
+++ b/tests/unit/executors/test_duckdb_exec_unit.py
@@ -0,0 +1,244 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import pandas as pd
+import pytest
+
+from fastflowtransform.core import Node
+from fastflowtransform.executors.duckdb_exec import DuckExecutor, _q
+
+
+@pytest.fixture
+def duck_exec() -> DuckExecutor:
+ # real in-memory db is fine for unit tests
+ return DuckExecutor(":memory:")
+
+
+def _node(name: str = "m", kind: str = "python") -> Node:
+ return Node(name=name, kind=kind, path=Path("."))
+
+
+# ---------------------------------------------------------------------------
+# _read_relation
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+@pytest.mark.duckdb
+def test_read_relation_happy(duck_exec: DuckExecutor):
+ duck_exec.con.execute("create table my_tbl (id int, name varchar)")
+ duck_exec.con.execute("insert into my_tbl values (1, 'A'), (2, 'B')")
+
+ df = duck_exec._read_relation("my_tbl", _node(), deps=["up1"])
+ assert list(df.columns) == ["id", "name"]
+ assert df.to_dict(orient="records") == [
+ {"id": 1, "name": "A"},
+ {"id": 2, "name": "B"},
+ ]
+
+
+@pytest.mark.unit
+@pytest.mark.duckdb
+def test_read_relation_missing_raises_nice_error(duck_exec: DuckExecutor):
+ # no table created
+ with pytest.raises(RuntimeError) as exc:
+ duck_exec._read_relation("does_not_exist", _node(), deps=["m1", "m2"])
+ msg = str(exc.value)
+ assert "Dependency table not found" in msg
+ assert "m1" in msg
+ assert "Existing tables" in msg # from the executor
+
+
+# ---------------------------------------------------------------------------
+# _materialize_relation
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+@pytest.mark.duckdb
+def test_materialize_relation_registers_and_creates_table(duck_exec: DuckExecutor):
+ df = pd.DataFrame({"id": [1, 2], "name": ["A", "B"]})
+
+ duck_exec._materialize_relation("out_tbl", df, _node())
+
+ out = duck_exec.con.execute("select * from out_tbl order by id").fetchall()
+ assert out == [(1, "A"), (2, "B")]
+
+ # temp table should be gone / unregistered - we just assert that running again doesn't error
+ duck_exec._materialize_relation("out_tbl", df, _node())
+
+
+# ---------------------------------------------------------------------------
+# view over table
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+@pytest.mark.duckdb
+def test_create_or_replace_view_from_table(duck_exec: DuckExecutor):
+ duck_exec.con.execute("create table src_tbl (id int)")
+ duck_exec.con.execute("insert into src_tbl values (10)")
+ duck_exec._create_or_replace_view_from_table("v_src", "src_tbl", _node())
+
+ rows = duck_exec.con.execute("select * from v_src").fetchall()
+ assert rows == [(10,)]
+
+
+# ---------------------------------------------------------------------------
+# formatting helpers
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+@pytest.mark.duckdb
+def test_format_relation_for_ref(duck_exec: DuckExecutor):
+ rel = duck_exec._format_relation_for_ref("my_model")
+ # relation_for("my_model") → "my_model"
+ assert rel == _q("my_model")
+
+
+@pytest.mark.unit
+@pytest.mark.duckdb
+def test_format_source_reference_ok(duck_exec: DuckExecutor):
+ cfg = {
+ "catalog": "c1",
+ "schema": "s1",
+ "identifier": "src_tbl",
+ }
+ ref = duck_exec._format_source_reference(cfg, "src", "tbl")
+ # should be quoted catalog.schema.identifier
+ assert ref == '"c1"."s1"."src_tbl"'
+
+
+@pytest.mark.unit
+@pytest.mark.duckdb
+def test_format_source_reference_missing_identifier_raises(duck_exec: DuckExecutor):
+ cfg = {
+ "catalog": "c1",
+ "schema": "s1",
+ # no identifier!
+ }
+ with pytest.raises(KeyError):
+ duck_exec._format_source_reference(cfg, "src", "tbl")
+
+
+@pytest.mark.unit
+@pytest.mark.duckdb
+def test_format_source_reference_path_not_supported(duck_exec: DuckExecutor):
+ cfg = {"location": "/some/path.csv"}
+ with pytest.raises(NotImplementedError):
+ duck_exec._format_source_reference(cfg, "src", "tbl")
+
+
+# ---------------------------------------------------------------------------
+# on_node_built - best effort
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+@pytest.mark.duckdb
+def test_on_node_built_best_effort(duck_exec: DuckExecutor, monkeypatch: pytest.MonkeyPatch):
+ called = {"ensure": 0, "upsert": 0}
+
+ def fake_ensure(ex: Any) -> None:
+ called["ensure"] += 1
+
+ def fake_upsert(ex: Any, name: str, rel: str, fp: str, eng: str) -> None:
+ called["upsert"] += 1
+
+ # patch the functions used in on_node_built
+ import fastflowtransform.executors.duckdb_exec as duck_mod # noqa PLC0415
+
+ monkeypatch.setattr(duck_mod, "ensure_meta_table", fake_ensure, raising=True)
+ monkeypatch.setattr(duck_mod, "upsert_meta", fake_upsert, raising=True)
+
+ duck_exec.on_node_built(_node("m1"), "out_tbl", "fp123")
+
+ assert called["ensure"] == 1
+ assert called["upsert"] == 1
+
+
+# ---------------------------------------------------------------------------
+# exists_relation
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+@pytest.mark.duckdb
+def test_exists_relation_true(duck_exec: DuckExecutor):
+ duck_exec.con.execute("create table t1 (id int)")
+ assert duck_exec.exists_relation("t1") is True
+
+
+@pytest.mark.unit
+@pytest.mark.duckdb
+def test_exists_relation_false(duck_exec: DuckExecutor):
+ assert duck_exec.exists_relation("nope") is False
+
+
+# ---------------------------------------------------------------------------
+# create_table_as / incremental_insert / incremental_merge
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+@pytest.mark.duckdb
+def test_create_table_as_cleans_select(duck_exec: DuckExecutor):
+ duck_exec.create_table_as("t_created", "select 1 as id;")
+ rows = duck_exec.con.execute("select * from t_created").fetchall()
+ assert rows == [(1,)]
+
+
+@pytest.mark.unit
+@pytest.mark.duckdb
+def test_incremental_insert_appends_rows(duck_exec: DuckExecutor):
+ duck_exec.con.execute("create table tgt (id int)")
+ duck_exec.incremental_insert("tgt", "select 1 as id;")
+ duck_exec.incremental_insert("tgt", "select 2 as id;")
+ rows = duck_exec.con.execute("select * from tgt order by id").fetchall()
+ assert rows == [(1,), (2,)]
+
+
+@pytest.mark.unit
+@pytest.mark.duckdb
+def test_incremental_merge_deletes_then_inserts(duck_exec: DuckExecutor):
+ # target with PK-like col
+ duck_exec.con.execute("create table tgt (id int, val varchar)")
+ duck_exec.con.execute("insert into tgt values (1, 'old'), (2, 'keep')")
+
+ # src produces (1, 'new'), (3, 'new3')
+ duck_exec.incremental_merge(
+ "tgt",
+ "select 1 as id, 'new' as val union all select 3 as id, 'new3' as val",
+ unique_key=["id"],
+ )
+
+ rows = duck_exec.con.execute("select id, val from tgt order by id").fetchall()
+ # id=1 should be updated to 'new' (via delete+insert)
+ # id=2 stays
+ # id=3 inserted
+ assert rows == [(1, "new"), (2, "keep"), (3, "new3")]
+
+
+# ---------------------------------------------------------------------------
+# alter_table_sync_schema
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+@pytest.mark.duckdb
+def test_alter_table_sync_schema_adds_missing_columns(duck_exec: DuckExecutor):
+ # create target with only "id"
+ duck_exec.con.execute("create table my_tbl (id int)")
+
+ # select has id + new_col
+ duck_exec.alter_table_sync_schema("my_tbl", "select 1 as id, 2 as new_col")
+
+ # inspect schema
+ info = duck_exec.con.execute("pragma table_info('my_tbl')").fetchall()
+ # format: (cid, name, type, notnull, dflt_value, pk)
+ col_names = [r[1] for r in info]
+ assert "id" in col_names
+ assert "new_col" in col_names
diff --git a/tests/unit/executors/test_shims_unit.py b/tests/unit/executors/test_shims_unit.py
new file mode 100644
index 0000000..79b880a
--- /dev/null
+++ b/tests/unit/executors/test_shims_unit.py
@@ -0,0 +1,249 @@
+# tests/unit/executors/test_shims_unit.py
+from __future__ import annotations
+
+from collections.abc import Sequence
+from types import SimpleNamespace
+from typing import Any, cast
+
+import pytest
+from google.cloud.bigquery import Client
+from sqlalchemy import text as sa_text
+from sqlalchemy.engine import Engine
+
+from fastflowtransform.executors._shims import (
+ BigQueryConnShim,
+ SAConnShim,
+ _rewrite_pg_create_or_replace_table,
+)
+
+# ---------------------------------------------------------------------------
+# Fakes / helpers
+# ---------------------------------------------------------------------------
+
+
+class _FakeConn:
+ """Collects executed statements for assertions."""
+
+ def __init__(self) -> None:
+ self.executed: list[tuple[str, dict[str, Any] | None]] = []
+
+ # SQLAlchemy-style execute
+ def execute(self, stmt: Any, params: dict[str, Any] | None = None) -> Any:
+ # store string form for easier asserts
+ sql_str = stmt.text if hasattr(stmt, "text") else str(stmt)
+ self.executed.append((sql_str, params))
+ # return something fetchable
+ return SimpleNamespace(fetchone=lambda: None, fetchall=lambda: [])
+
+ # context manager API
+ def __enter__(self) -> _FakeConn:
+ return self
+
+ def __exit__(self, exc_type, exc, tb) -> None:
+ return None
+
+ # so SAConnShim can call conn.begin() in the fallback (we never do here)
+ def begin(self) -> _FakeConn:
+ return self
+
+
+class _FakeEngine:
+ """Engine that always returns the same connection."""
+
+ def __init__(self) -> None:
+ self.conn = _FakeConn()
+
+ def begin(self) -> _FakeConn:
+ return self.conn
+
+
+# ---------------------------------------------------------------------------
+# _rewrite_pg_create_or_replace_table
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+def test_rewrite_pg_create_or_replace_table_simple():
+ sql = "CREATE OR REPLACE TABLE public.t AS SELECT 1"
+ out = _rewrite_pg_create_or_replace_table(sql)
+ assert "DROP TABLE IF EXISTS public.t CASCADE;" in out
+ assert "CREATE TABLE public.t AS SELECT 1" in out
+
+
+@pytest.mark.unit
+def test_rewrite_pg_create_or_replace_table_with_schema_and_quotes():
+ sql = ' create or replace table "raw"."users" as select * from src '
+ out = _rewrite_pg_create_or_replace_table(sql)
+ # two statements
+ assert 'DROP TABLE IF EXISTS "raw"."users" CASCADE;' in out
+ assert 'CREATE TABLE "raw"."users" AS select * from src' in out
+
+
+@pytest.mark.unit
+def test_rewrite_pg_create_or_replace_table_untouched_for_other_sql():
+ sql = "SELECT 1"
+ out = _rewrite_pg_create_or_replace_table(sql)
+ assert out == sql
+
+
+# ---------------------------------------------------------------------------
+# SAConnShim
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+def test_sa_shim_executes_plain_sql_without_schema():
+ eng = _FakeEngine()
+ shim = SAConnShim(cast(Engine, eng), schema=None)
+
+ shim.execute("SELECT 1")
+
+ executed = eng.conn.executed
+ assert len(executed) == 1
+ sql, params = executed[0]
+ assert sql.strip().upper().startswith("SELECT 1")
+ assert params is None
+
+
+@pytest.mark.unit
+def test_sa_shim_sets_search_path_when_schema_given():
+ eng = _FakeEngine()
+ shim = SAConnShim(cast(Engine, eng), schema="public")
+
+ shim.execute("SELECT 42")
+
+ executed = eng.conn.executed
+ # 1) SET LOCAL ... 2) SELECT 42
+ assert len(executed) == 2
+ assert 'SET LOCAL search_path = "public"' in executed[0][0]
+ assert "SELECT 42" in executed[1][0]
+
+
+@pytest.mark.unit
+def test_sa_shim_rewrites_cor_table_into_two_statements():
+ eng = _FakeEngine()
+ shim = SAConnShim(cast(Engine, eng), schema=None)
+
+ shim.execute("CREATE OR REPLACE TABLE my_tbl AS SELECT 1")
+
+ executed = eng.conn.executed
+ # should have been split into DROP + CREATE
+ assert len(executed) == 2
+ assert "DROP TABLE IF EXISTS my_tbl CASCADE" in executed[0][0]
+ assert "CREATE TABLE my_tbl AS SELECT 1" in executed[1][0]
+
+
+@pytest.mark.unit
+def test_sa_shim_executes_iterable_sequentially():
+ eng = _FakeEngine()
+ shim = SAConnShim(cast(Engine, eng), schema=None)
+
+ shim.execute(["SELECT 1", "SELECT 2"])
+
+ executed = [sql for (sql, _) in eng.conn.executed]
+ assert executed == ["SELECT 1", "SELECT 2"]
+
+
+@pytest.mark.unit
+def test_sa_shim_executes_tuple_with_params_on_last_statement():
+ eng = _FakeEngine()
+ shim = SAConnShim(cast(Engine, eng), schema=None)
+
+ shim.execute(("SELECT :x", {"x": 10}))
+
+ executed = eng.conn.executed
+ assert len(executed) == 1
+ sql, params = executed[0]
+ assert "SELECT :x" in sql
+ assert params == {"x": 10}
+
+
+@pytest.mark.unit
+def test_sa_shim_executes_sqlalchemy_clauseelement():
+ eng = _FakeEngine()
+ shim = SAConnShim(cast(Engine, eng), schema=None)
+
+ stmt = sa_text("SELECT 1")
+ shim.execute(stmt)
+
+ executed = eng.conn.executed
+ assert len(executed) == 1
+ assert "SELECT 1" in executed[0][0]
+
+
+# ---------------------------------------------------------------------------
+# BigQueryConnShim
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+def test_bq_shim_executes_single_sql():
+ calls: dict[str, Any] = {}
+
+ class FakeClient:
+ def query(self, sql: str, location: str | None = None):
+ calls["sql"] = sql
+ calls["location"] = location
+ return "JOB-1"
+
+ fake = FakeClient()
+ shim = BigQueryConnShim(cast(Client, fake), location="EU")
+ res = shim.execute("SELECT 1")
+
+ assert res == "JOB-1"
+ assert calls["sql"] == "SELECT 1"
+ assert calls["location"] == "EU"
+
+
+@pytest.mark.unit
+def test_bq_shim_executes_sequence_and_returns_last_job():
+ seen: list[str] = []
+
+ class FakeJob:
+ def result(self) -> None:
+ return None
+
+ class FakeClient:
+ def query(self, sql: str, location: str | None = None):
+ seen.append(sql)
+ return FakeJob()
+
+ fake = FakeClient()
+ shim = BigQueryConnShim(cast(Client, fake), location="EU")
+ res = shim.execute(["SELECT 1", "SELECT 2", "SELECT 3"])
+
+ # should have executed all
+ assert seen == ["SELECT 1", "SELECT 2", "SELECT 3"]
+ # and returned the last job
+ assert isinstance(res, FakeJob)
+
+
+def test_bq_shim_raises_on_unsupported_type():
+ fake_client = SimpleNamespace(query=lambda *a, **k: None)
+
+ # für den Typchecker so tun, als wäre es ein echter Client
+ shim = BigQueryConnShim(client=cast(Client, fake_client))
+
+ with pytest.raises(TypeError):
+ shim.execute(123)
+
+
+# ---------------------------------------------------------------------------
+# Mixed / defensive
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+def test_sa_shim_iterable_of_mixed_types():
+ """Ensure iterable with strings and ClauseElements is executed in order."""
+ eng = _FakeEngine()
+ shim = SAConnShim(cast(Engine, eng))
+
+ stmts: Sequence[Any] = ["SELECT 1", sa_text("SELECT 2"), "SELECT 3"]
+ shim.execute(stmts)
+
+ executed = [sql for (sql, _) in eng.conn.executed]
+ # sql text may contain trailing semicolons/spaces from TextClause
+ assert "SELECT 1" in executed[0]
+ assert "SELECT 2" in executed[1]
+ assert "SELECT 3" in executed[2]
diff --git a/tests/unit/executors/test_snowflake_snowpark_exec.py b/tests/unit/executors/test_snowflake_snowpark_exec.py
new file mode 100644
index 0000000..c19bc61
--- /dev/null
+++ b/tests/unit/executors/test_snowflake_snowpark_exec.py
@@ -0,0 +1,521 @@
+# tests/unit/executors/test_snowflake_snowpark_exec_unit.py
+from __future__ import annotations
+
+import sys
+import types
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+import snowflake.snowpark as sf
+
+import fastflowtransform.executors.snowflake_snowpark_exec as sf_mod
+from fastflowtransform.executors.snowflake_snowpark_exec import _SFResult
+
+# ---------------------------------------------------------------------------
+# 1) Install a fake snowflake.snowpark BEFORE importing the executor module
+# ---------------------------------------------------------------------------
+
+fake_sf_pkg = sys.modules.setdefault("snowflake", types.ModuleType("snowflake"))
+fake_sf_snowpark_mod = types.ModuleType("snowflake.snowpark")
+
+
+class _FakeSchema:
+ """Minimal schema object exposing .names like real Snowpark dataframes do."""
+
+ def __init__(self, names: list[str]):
+ self.names = names
+
+
+class FakeSnowparkDataFrame:
+ """
+ Very small stand-in for snowflake.snowpark.DataFrame.
+ It needs:
+ - .schema.names
+ - .collect()
+ - .write.save_as_table(...)
+ """
+
+ def __init__(self, session: FakeSession, sql: str | None = None, cols: list[str] | None = None):
+ self._session = session
+ self._sql = sql
+ self.schema = _FakeSchema(cols or [])
+ self._rows: list[dict[str, Any]] = []
+
+ # allow tests to inject rows
+ def _with_rows(self, rows: list[dict[str, Any]]) -> FakeSnowparkDataFrame:
+ self._rows = rows
+ return self
+
+ def collect(self) -> list[dict[str, Any]]:
+ # return rows that were set, otherwise empty
+ return self._rows
+
+ class _Writer:
+ def __init__(self, outer: FakeSnowparkDataFrame):
+ self._outer = outer
+
+ def save_as_table(self, name: str, mode: str = "overwrite") -> None:
+ # just record the call on the session
+ self._outer._session.write_calls.append(("save_as_table", name, mode))
+
+ @property
+ def write(self) -> FakeSnowparkDataFrame._Writer: # type: ignore[name-defined]
+ return FakeSnowparkDataFrame._Writer(self)
+
+
+class FakeSession:
+ """
+ Minimal session mock:
+ - .sql(sql) -> FakeSnowparkDataFrame
+ - .table(name) -> FakeSnowparkDataFrame
+ - remembers SQL calls
+ """
+
+ def __init__(self, cfg: dict[str, Any]):
+ self.cfg = cfg
+ self.sql_calls: list[str] = []
+ self.table_calls: list[str] = []
+ self.write_calls: list[tuple[str, str, str]] = []
+
+ # --- behaviours that depend on the SQL string ---
+ def _df_for_sql(self, sql: str) -> FakeSnowparkDataFrame:
+ """
+ Create a DF whose columns/rows depend on the SQL text.
+ We only simulate the bits the executor actually uses.
+ """
+ # 1) exists_relation → information_schema.tables → return 1 row
+ if "information_schema.tables" in sql.lower():
+ return FakeSnowparkDataFrame(self, sql, [])._with_rows([{"1": 1}])
+
+ # 2) alter_table_sync_schema → information_schema.columns
+ if "information_schema.columns" in sql.lower():
+ # Pretend there is already an "id" column
+ return FakeSnowparkDataFrame(self, sql, [])._with_rows([{"COLUMN_NAME": "ID"}])
+
+ # 3) probe SELECT ... WHERE 1=0 → provide schema names
+ if "where 1=0" in sql.lower():
+ # executor expects to read probe.schema.names
+ return FakeSnowparkDataFrame(self, sql, ["ID", "NEW_COL"])
+
+ # 4) everything else → empty df, no rows
+ return FakeSnowparkDataFrame(self, sql, [])
+
+ def sql(self, sql: str) -> FakeSnowparkDataFrame:
+ self.sql_calls.append(sql)
+ return self._df_for_sql(sql)
+
+ def table(self, name: str) -> FakeSnowparkDataFrame:
+ self.table_calls.append(name)
+ # pretend table has 2 columns
+ return FakeSnowparkDataFrame(self, name, ["ID", "NAME"])
+
+
+class _FakeSessionBuilder:
+ def __init__(self):
+ self._cfg: dict[str, Any] = {}
+
+ def configs(self, cfg: dict[str, Any]) -> _FakeSessionBuilder:
+ self._cfg = cfg
+ return self
+
+ def create(self) -> FakeSession:
+ return FakeSession(self._cfg)
+
+
+# expose as snowflake.snowpark.Session
+fake_sf_snowpark_mod.Session = SimpleNamespace(builder=_FakeSessionBuilder()) # type: ignore[attr-defined]
+# expose DataFrame type so isinstance(..., SNDF) works
+fake_sf_snowpark_mod.DataFrame = FakeSnowparkDataFrame # type: ignore[attr-defined]
+
+fake_sf_snowpark_mod.DataFrame = FakeSnowparkDataFrame # type: ignore[attr-defined]
+sys.modules["snowflake.snowpark"] = fake_sf_snowpark_mod
+
+# ---------------------------------------------------------------------------
+# 2) Now we can safely import the module under test
+# ---------------------------------------------------------------------------
+import fastflowtransform.executors.snowflake_snowpark_exec as sf_exec_mod # noqa: E402
+from fastflowtransform.core import Node # noqa: E402
+from fastflowtransform.executors.snowflake_snowpark_exec import ( # noqa: E402
+ SnowflakeSnowparkExecutor,
+ _SFCursorShim,
+)
+
+
+@pytest.fixture
+def sf_exec(monkeypatch):
+ """
+ Build an executor with a fake Snowpark session.
+ """
+ # make sure the module uses our just-registered fake
+ monkeypatch.setattr(sf_exec_mod, "Session", fake_sf_snowpark_mod.Session, raising=True)
+
+ cfg = {
+ "account": "acc",
+ "user": "usr",
+ "password": "pwd",
+ "warehouse": "wh",
+ "database": "DB1",
+ "schema": "SC1",
+ }
+ ex = SnowflakeSnowparkExecutor(cfg)
+ # sanity: we actually got our fake session
+ assert isinstance(ex.session, FakeSession)
+ return ex
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_init_sets_db_schema_and_con(sf_exec):
+ assert sf_exec.database == "DB1"
+ assert sf_exec.schema == "SC1"
+ # con must be present
+ assert isinstance(sf_exec.con, _SFCursorShim)
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_q_and_qualified(sf_exec):
+ assert sf_exec._q("x") == '"x"'
+ assert sf_exec._qualified("TBL") == '"DB1"."SC1"."TBL"'
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_read_relation_calls_session_table(sf_exec):
+ node = Node(name="n", kind="sql", path=Path("."))
+ df = sf_exec._read_relation("MY_TBL", node, deps=[])
+ assert isinstance(df, FakeSnowparkDataFrame)
+ # session should have been asked for the fully qualified name
+ assert sf_exec.session.table_calls == ['"DB1"."SC1"."MY_TBL"']
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_materialize_relation_happy(sf_exec, monkeypatch):
+ called: dict[str, str] = {}
+
+ fake_df = SimpleNamespace(
+ write=SimpleNamespace(
+ save_as_table=lambda tbl, mode="overwrite": (
+ called.setdefault("table", tbl),
+ called.setdefault("mode", mode),
+ )
+ )
+ )
+
+ monkeypatch.setattr(
+ sf_exec,
+ "_is_frame",
+ lambda obj: obj is fake_df,
+ raising=True,
+ )
+
+ node = Node(name="m", kind="python", path=Path("."))
+
+ # ACT
+ sf_exec._materialize_relation("OUT_TBL", fake_df, node)
+
+ # ASSERT
+ assert called["table"] == '"DB1"."SC1"."OUT_TBL"'
+ assert called["mode"] == "overwrite"
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_materialize_relation_raises_on_non_frame(sf_exec):
+ node = Node(name="m", kind="python", path=Path("."))
+ with pytest.raises(TypeError):
+ sf_exec._materialize_relation("OUT_TBL", object(), node)
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_create_view_over_table_issues_sql(sf_exec):
+ node = Node(name="x", kind="sql", path=Path("."))
+ sf_exec._create_view_over_table("V_USERS", "USERS", node)
+ assert any("CREATE OR REPLACE VIEW" in s for s in sf_exec.session.sql_calls)
+ sql = sf_exec.session.sql_calls[-1]
+ assert '"DB1"."SC1"."V_USERS"' in sql
+ assert '"DB1"."SC1"."USERS"' in sql
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_validate_required_single_df_ok(sf_exec):
+ SNDF = sf_mod.SNDF
+ df = SNDF(sf_exec.session) # type: ignore[call-arg]
+ df.schema = SimpleNamespace(names=["ID", "NAME", "AGE"]) # type: ignore[attr-defined]
+
+ sf_exec._validate_required(
+ "model1",
+ df,
+ {"DB1.SC1.USERS": {"ID", "NAME"}},
+ )
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_validate_required_single_df_missing(sf_exec):
+ SNDF = sf_mod.SNDF
+ df = SNDF(sf_exec.session) # type: ignore[call-arg]
+ df.schema = SimpleNamespace(names=["ID"]) # type: ignore[attr-defined]
+
+ with pytest.raises(ValueError) as exc:
+ sf_exec._validate_required(
+ "model1",
+ df,
+ {"DB1.SC1.USERS": {"ID", "NAME"}},
+ )
+
+ msg = str(exc.value)
+ assert "missing columns" in msg
+ assert "NAME" in msg
+ assert "ID" in msg
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_validate_required_multi_input_ok(sf_exec):
+ df1 = FakeSnowparkDataFrame(sf_exec.session, cols=["ID", "NAME"])
+ df2 = FakeSnowparkDataFrame(sf_exec.session, cols=["USER_ID", "ORDER_ID"])
+ sf_exec._validate_required(
+ "model2",
+ {"DB1.SC1.USERS": df1, "DB1.SC1.ORDERS": df2},
+ {"DB1.SC1.USERS": {"ID"}, "DB1.SC1.ORDERS": {"ORDER_ID"}},
+ ) # no raise
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_validate_required_multi_input_missing_key(sf_exec):
+ df1 = FakeSnowparkDataFrame(sf_exec.session, cols=["ID", "NAME"])
+ with pytest.raises(ValueError) as exc:
+ sf_exec._validate_required(
+ "model2",
+ {"DB1.SC1.USERS": df1},
+ {"DB1.SC1.USERS": {"ID"}, "DB1.SC1.ORDERS": {"ORDER_ID"}},
+ )
+ assert "missing dependency key 'DB1.SC1.ORDERS'" in str(exc.value)
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_columns_of(sf_exec):
+ df = FakeSnowparkDataFrame(sf_exec.session, cols=["A", "B"])
+ assert sf_exec._columns_of(df) == ["A", "B"]
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_is_frame(sf_exec):
+ df = sf.DataFrame(sf_exec.session)
+ df.schema = SimpleNamespace(names=["ID"]) # type: ignore[attr-defined]
+
+ assert sf_exec._is_frame(df) is True
+ assert sf_exec._is_frame(object()) is False
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_frame_name(sf_exec):
+ assert sf_exec._frame_name() == "Snowpark"
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_format_relation_for_ref(sf_exec):
+ r = sf_exec._format_relation_for_ref("my_model")
+ assert '"DB1"' in r and '"SC1"' in r and "my_model" in r
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_format_source_reference_happy(sf_exec):
+ cfg = {"identifier": "SRC_TBL", "database": "DBX", "schema": "RAW"}
+ ref = sf_exec._format_source_reference(cfg, "src", "tbl")
+ assert ref == '"DBX"."RAW"."SRC_TBL"'
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_format_source_reference_raises_on_location(sf_exec):
+ cfg = {"identifier": "X", "location": "s3://foo"}
+ with pytest.raises(NotImplementedError):
+ sf_exec._format_source_reference(cfg, "src", "tbl")
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_format_source_reference_raises_on_missing_identifier(sf_exec):
+ with pytest.raises(KeyError):
+ sf_exec._format_source_reference({}, "src", "tbl")
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_create_or_replace_view_calls_session_sql(sf_exec):
+ node = Node(name="x", kind="sql", path=Path("."))
+ sf_exec._create_or_replace_view('"DB1"."SC1"."V1"', "SELECT 1", node)
+ assert any("CREATE OR REPLACE VIEW" in s for s in sf_exec.session.sql_calls)
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_create_or_replace_table_calls_session_sql(sf_exec):
+ node = Node(name="x", kind="sql", path=Path("."))
+ sf_exec._create_or_replace_table('"DB1"."SC1"."T1"', "SELECT 1", node)
+ assert any("CREATE OR REPLACE TABLE" in s for s in sf_exec.session.sql_calls)
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_create_or_replace_view_from_table_calls_session_sql(sf_exec):
+ node = Node(name="x", kind="sql", path=Path("."))
+ sf_exec._create_or_replace_view_from_table("V1", "T1", node)
+ sql = sf_exec.session.sql_calls[-1]
+ assert "CREATE OR REPLACE VIEW" in sql
+ assert '"DB1"."SC1"."V1"' in sql
+ assert '"DB1"."SC1"."T1"' in sql
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_on_node_built_best_effort(monkeypatch, sf_exec):
+ called = {"ensure": 0, "upsert": 0}
+
+ def fake_ensure(ex):
+ called["ensure"] += 1
+
+ def fake_upsert(ex, name, rel, fp, eng):
+ called["upsert"] += 1
+
+ monkeypatch.setattr(sf_exec_mod, "ensure_meta_table", fake_ensure)
+ monkeypatch.setattr(sf_exec_mod, "upsert_meta", fake_upsert)
+
+ sf_exec.on_node_built(Node(name="m", kind="sql", path=Path(".")), '"DB1"."SC1"."M"', "fp123")
+
+ assert called["ensure"] == 1
+ assert called["upsert"] == 1
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_exists_relation_true(sf_exec, monkeypatch):
+ # our fake session already returns one row for information_schema.tables
+ ok = sf_exec.exists_relation("SOME_TBL")
+ assert ok is True
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_exists_relation_false_on_error(sf_exec, monkeypatch):
+ def boom(sql: str):
+ raise RuntimeError("bad")
+
+ monkeypatch.setattr(sf_exec.session, "sql", boom, raising=True)
+ ok = sf_exec.exists_relation("SOME_TBL")
+ assert ok is False
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_create_table_as_strips_semicolon(sf_exec):
+ sf_exec.session.sql_calls.clear()
+ sf_exec.create_table_as("DST", "SELECT 1;")
+ sql = sf_exec.session.sql_calls[-1]
+ assert "CREATE OR REPLACE TABLE" in sql
+ assert "SELECT 1" in sql
+ assert not sql.strip().endswith(";")
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_incremental_insert_strips_semicolon(sf_exec):
+ sf_exec.session.sql_calls.clear()
+ sf_exec.incremental_insert("DST", "SELECT 1;")
+ sql = sf_exec.session.sql_calls[-1]
+ assert "INSERT INTO" in sql
+ assert "SELECT 1" in sql
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_incremental_merge_builds_two_statements(sf_exec):
+ sf_exec.session.sql_calls.clear()
+ sf_exec.incremental_merge("DST", "SELECT 1 AS id", ["id"])
+ sql = sf_exec.session.sql_calls[-1]
+ # both DELETE and INSERT statements should be in there
+ assert "DELETE FROM" in sql
+ assert "INSERT INTO" in sql
+ assert "WITH src AS" in sql
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_alter_table_sync_schema_adds_missing(sf_exec):
+ sf_exec.session.sql_calls.clear()
+ sf_exec.alter_table_sync_schema("EXISTING", "SELECT 1 AS ID, 2 AS NEW_COL")
+ # last sql should be ALTER TABLE ... ADD COLUMN ...
+ assert any("ALTER TABLE" in s and "ADD COLUMN" in s for s in sf_exec.session.sql_calls)
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_alter_table_sync_schema_noop_on_unknown_mode(sf_exec):
+ sf_exec.session.sql_calls.clear()
+ sf_exec.alter_table_sync_schema("EXISTING", "SELECT 1", mode="replace_all") # unknown
+ # no new calls
+ assert sf_exec.session.sql_calls == []
+
+
+@pytest.mark.unit
+@pytest.mark.snowflake
+def test_sfcursorshim_execute_returns_rows(sf_exec):
+ class FakeRow:
+ """Mimic Snowpark Row: has attributes *and* asDict()."""
+
+ def __init__(self, data: dict[str, Any]):
+ self._data = data
+ # make attributes accessible: r.A, r.B, ...
+ for k, v in data.items():
+ setattr(self, k, v)
+
+ def asDict(self) -> dict[str, Any]:
+ return self._data
+
+ class DFWithRows(FakeSnowparkDataFrame):
+ def __init__(self, session, sql, cols, rows):
+ super().__init__(session, sql, cols)
+ self._rows = rows
+
+ def collect(self) -> list[Any]: # type: ignore[override]
+ return self._rows
+
+ def fake_sql(sql: str):
+ return DFWithRows(
+ sf_exec.session,
+ sql,
+ ["A", "B"],
+ [
+ FakeRow({"A": 1, "B": "x"}),
+ FakeRow({"A": 2, "B": "y"}),
+ ],
+ )
+
+ # executor auf unseren fake umbiegen
+ sf_exec.session.sql = fake_sql # type: ignore[assignment]
+
+ # ACT
+ res = sf_exec.con.execute("SELECT 1")
+
+ # ASSERT
+ assert isinstance(res, _SFResult)
+ assert res.fetchall() == [(1, "x"), (2, "y")]
+ assert res.fetchone() == (1, "x")
diff --git a/tests/unit/test_core_python_tags.py b/tests/unit/test_core_python_tags.py
index 0f84375..95a0f94 100644
--- a/tests/unit/test_core_python_tags.py
+++ b/tests/unit/test_core_python_tags.py
@@ -14,7 +14,7 @@ def test_python_model_tags_propagate_to_node(tmp_path: Path, monkeypatch) -> Non
model_file.write_text(
(
"from fastflowtransform import model\n\n"
- "@model(name='py_tagged', tags=['example', 'demo'], meta={'materialized': 'view'})\n"
+ "@model(name='py_tagged', tags=['example', 'demo'], materialized='view')\n"
"def build(df=None):\n"
" return df\n"
),