moves pipeline tests to pipelines

Author: rudolfix

tests/load/bigquery/test_bigquery_table_builder.py

@@ -5,7 +5,6 @@
 import google
 import pytest
 import sqlfluff
-from google.cloud.bigquery import Table
 
 import dlt
 from dlt.common.configuration import resolve_configuration
@@ -1040,205 +1039,3 @@ def some_data() -> Iterator[Dict[str, str]]:
     bigquery_adapter(some_data, table_expiration_datetime="2030-01-01")
 
     assert some_data._hints["x-bigquery-table-expiration"] == pendulum.datetime(2030, 1, 1)  # type: ignore
-
-
-@pytest.mark.parametrize(
-    "destination_config",
-    destinations_configs(default_sql_configs=True, subset=["bigquery"]),
-    ids=lambda x: x.name,
-)
-def test_adapter_additional_table_hints_table_expiration(
-    destination_config: DestinationTestConfiguration,
-) -> None:
-    @dlt.resource(columns=[{"name": "col1", "data_type": "text"}])
-    def no_hints() -> Iterator[Dict[str, str]]:
-        yield from [{"col1": str(i)} for i in range(10)]
-
-    hints = bigquery_adapter(
-        no_hints.with_name(new_name="hints"), table_expiration_datetime="2030-01-01"
-    )
-
-    @dlt.source(max_table_nesting=0)
-    def sources() -> List[DltResource]:
-        return [no_hints, hints]
-
-    pipeline = destination_config.setup_pipeline(
-        f"bigquery_{uniq_id()}",
-        dev_mode=True,
-    )
-
-    pipeline.run(sources())
-
-    with pipeline.sql_client() as c:
-        nc: google.cloud.bigquery.client.Client = c.native_connection
-
-        fqtn_no_hints = c.make_qualified_table_name("no_hints", escape=False)
-        fqtn_hints = c.make_qualified_table_name("hints", escape=False)
-
-        no_hints_table = nc.get_table(fqtn_no_hints)
-        hints_table = nc.get_table(fqtn_hints)
-
-        assert not no_hints_table.expires
-        assert hints_table.expires == pendulum.datetime(2030, 1, 1, 0)
-
-
-@pytest.mark.parametrize(
-    "destination_config",
-    destinations_configs(default_sql_configs=True, subset=["bigquery"]),
-    ids=lambda x: x.name,
-)
-def test_adapter_merge_behaviour(
-    destination_config: DestinationTestConfiguration,
-) -> None:
-    @dlt.resource(
-        columns=[
-            {"name": "col1", "data_type": "text"},
-            {"name": "col2", "data_type": "bigint"},
-            {"name": "col3", "data_type": "double"},
-        ]
-    )
-    def hints() -> Iterator[Dict[str, Any]]:
-        yield from [{"col1": str(i), "col2": i, "col3": float(i)} for i in range(10)]
-
-    bigquery_adapter(hints, table_expiration_datetime="2030-01-01", cluster=["col1"])
-    bigquery_adapter(
-        hints,
-        table_description="A small table somewhere in the cosmos...",
-        partition="col2",
-    )
-
-    pipeline = destination_config.setup_pipeline(
-        f"bigquery_{uniq_id()}",
-        dev_mode=True,
-    )
-
-    pipeline.run(hints)
-
-    with pipeline.sql_client() as c:
-        nc: google.cloud.bigquery.client.Client = c.native_connection
-
-        table_fqtn = c.make_qualified_table_name("hints", escape=False)
-
-        table: Table = nc.get_table(table_fqtn)
-
-        table_cluster_fields = [] if table.clustering_fields is None else table.clustering_fields
-
-        # Test merging behaviour.
-        assert table.expires == pendulum.datetime(2030, 1, 1, 0)
-        assert ["col1"] == table_cluster_fields, "`hints` table IS NOT clustered by `col1`."
-        assert table.description == "A small table somewhere in the cosmos..."
-
-        if not table.range_partitioning:
-            raise ValueError("`hints` table IS NOT clustered on a column.")
-        else:
-            assert (
-                table.range_partitioning.field == "col2"
-            ), "`hints` table IS NOT clustered on column `col2`."
-
-
-@pytest.mark.parametrize(
-    "destination_config",
-    destinations_configs(default_sql_configs=True, subset=["bigquery"]),
-    ids=lambda x: x.name,
-)
-def test_adapter_autodetect_schema_with_hints(
-    destination_config: DestinationTestConfiguration,
-) -> None:
-    @dlt.resource(
-        columns=[
-            {"name": "col1", "data_type": "text"},
-            {"name": "col2", "data_type": "bigint"},
-            {"name": "col3", "data_type": "double"},
-        ]
-    )
-    def general_types() -> Iterator[Dict[str, Any]]:
-        yield from [{"col1": str(i), "col2": i, "col3": float(i)} for i in range(10)]
-
-    @dlt.resource(
-        columns=[
-            {"name": "my_time_column", "data_type": "timestamp"},
-        ]
-    )
-    def partition_time() -> Iterator[Dict[str, Any]]:
-        for i in range(10):
-            yield {
-                "my_time_column": pendulum.from_timestamp(1700784000 + i * 50_000),
-            }
-
-    @dlt.resource(
-        columns=[
-            {"name": "my_date_column", "data_type": "date"},
-        ]
-    )
-    def partition_date() -> Iterator[Dict[str, Any]]:
-        for i in range(10):
-            yield {
-                "my_date_column": pendulum.from_timestamp(1700784000 + i * 50_000).date(),
-            }
-
-    bigquery_adapter(
-        general_types,
-        table_description="A small table somewhere in the cosmos...",
-        partition="col2",
-        cluster=["col1"],
-        autodetect_schema=True,
-    )
-
-    pipeline = destination_config.setup_pipeline(
-        f"bigquery_{uniq_id()}",
-        dev_mode=True,
-    )
-
-    pipeline.run(general_types)
-
-    bigquery_adapter(
-        partition_time,
-        partition="my_time_column",
-        autodetect_schema=True,
-    )
-
-    pipeline_time = destination_config.setup_pipeline(
-        f"bigquery_{uniq_id()}",
-        dev_mode=True,
-    )
-
-    pipeline_time.run(partition_time)
-
-    bigquery_adapter(
-        partition_date,
-        partition="my_date_column",
-        autodetect_schema=True,
-    )
-
-    pipeline_date = destination_config.setup_pipeline(
-        f"bigquery_{uniq_id()}",
-        dev_mode=True,
-    )
-
-    pipeline_date.run(partition_date)
-
-    with pipeline.sql_client() as c:
-        nc: google.cloud.bigquery.client.Client = c.native_connection
-
-        table_fqtn = c.make_qualified_table_name("general_types", escape=False)
-
-        table: Table = nc.get_table(table_fqtn)
-
-        table_cluster_fields = [] if table.clustering_fields is None else table.clustering_fields
-        assert ["col1"] == table_cluster_fields, "NOT clustered by `col1`."
-
-        assert table.description == "A small table somewhere in the cosmos..."
-        assert table.range_partitioning.field == "col2", "NOT partitioned on column `col2`."
-
-    with pipeline_time.sql_client() as c:
-        nc: google.cloud.bigquery.client.Client = c.native_connection  # type: ignore[no-redef]
-        table_fqtn = c.make_qualified_table_name("partition_time", escape=False)
-        table: Table = nc.get_table(table_fqtn)  # type: ignore[no-redef]
-        assert table.time_partitioning.field == "my_time_column"
-
-    with pipeline_date.sql_client() as c:
-        nc: google.cloud.bigquery.client.Client = c.native_connection  # type: ignore[no-redef]
-        table_fqtn = c.make_qualified_table_name("partition_date", escape=False)
-        table: Table = nc.get_table(table_fqtn)  # type: ignore[no-redef]
-        assert table.time_partitioning.field == "my_date_column"
-        assert table.time_partitioning.type_ == "DAY"