test: use dataclass cases for function discovery regressions

Author: kvlonge

tests/integration/src/sqlbuild/adapters/snowflake/_test_types.py

@@ -123,3 +123,10 @@ class SnowflakeCostTestCase:
     expected_query_count: int
     expected_resource_name: str
     expected_minimum_credits: int
+
+
+@dataclass(frozen=True)
+class SnowflakeFunctionDiscoveryTestCase:
+    description: str
+    function_name: str
+    expected_function_count: int

tests/integration/src/sqlbuild/adapters/snowflake/test_function_discovery.py

@@ -1,51 +1,64 @@
-"""Real Snowflake catalog discovery without an implicit connection database."""
+"""Real Snowflake function discovery against an explicit catalog."""
 
 from typing import Any
 
 import pytest
 
+from sqlbuild.adapter.contract.models import FunctionInfo
 from sqlbuild.adapters.snowflake.classes.snowflake_adapter import SnowflakeAdapter
+from tests.integration.src.sqlbuild.adapters.snowflake._test_types import (
+    SnowflakeFunctionDiscoveryTestCase,
+)
 from tests.integration.src.sqlbuild.adapters.snowflake.helpers import (
     build_snowflake_connection_config,
 )
 
 
-def test_given_no_current_database_when_discovering_udf_then_finds_explicit_target(
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        SnowflakeFunctionDiscoveryTestCase(
+            description="connection without database finds the explicit target",
+            function_name="normalize_order",
+            expected_function_count=1,
+        )
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_connection_without_database_when_discovering_udf_then_finds_explicit_target(
+    test_case: SnowflakeFunctionDiscoveryTestCase,
     adapter: SnowflakeAdapter,
     connection: Any,
     snowflake_database: str,
     snowflake_schema: str,
 ) -> None:
-    function_name = "normalize_order"
-    target = ".".join(
+    target: str = ".".join(
         adapter.render_identifier(part)
-        for part in (snowflake_database, snowflake_schema, function_name)
+        for part in (snowflake_database, snowflake_schema, test_case.function_name)
     )
     adapter.execute(
         connection=connection,
-        sql=f"CREATE FUNCTION {target}(value VARCHAR) RETURNS VARCHAR LANGUAGE SQL AS 'UPPER(value)'",
+        sql=f"CREATE FUNCTION {target}(value VARCHAR) RETURNS VARCHAR AS 'UPPER(value)'",
     )
-    config = build_snowflake_connection_config()
+    config: dict[str, object] = build_snowflake_connection_config()
     config.pop("database", None)
     config.pop("schema", None)
-    discovery_connection = adapter.connect(config)
+    discovery_connection: Any = adapter.connect(config)
     try:
-        current = adapter.execute(connection=discovery_connection, sql="SELECT CURRENT_DATABASE()")
-        if current.fetchone()[0] is not None:
-            pytest.skip("The test user supplies a default database")
-        functions = adapter.list_functions(
+        functions: tuple[FunctionInfo, ...] = adapter.list_functions(
             connection=discovery_connection,
             database=snowflake_database,
             schemas=(snowflake_schema,),
-            names=(function_name,),
+            names=(test_case.function_name,),
         )
-        assert len(functions) == 1
-        assert functions[0].name == function_name
-        assert functions[0].schema == snowflake_schema.lower()
-        assert functions[0].database == snowflake_database
     finally:
         adapter.close(discovery_connection)
 
+    assert len(functions) == test_case.expected_function_count
+    assert functions[0].name == test_case.function_name
+    assert functions[0].schema == snowflake_schema.lower()
+    assert functions[0].database == snowflake_database
+
 
 if __name__ == "__main__":
     pytest.main([__file__, "-vv"])

tests/unit/src/sqlbuild/adapters/snowflake/_test_types.py

@@ -1,7 +1,7 @@
 from dataclasses import dataclass
 from datetime import datetime
 
-from sqlbuild.adapter.contract.models import SchemaDiffResult
+from sqlbuild.adapter.contract.models import FunctionInfo, SchemaDiffResult
 from sqlbuild.compiler.lineage.types import InferredNullability
 from sqlbuild.cost.types import CostStatus
 
@@ -204,3 +204,15 @@ class SnowflakeRelationAgeMetadataTestCase:
     last_altered: datetime | None
     expected_created_at: datetime | None
     expected_last_altered_at: datetime | None
+
+
+@dataclass(frozen=True)
+class SnowflakeFunctionDiscoveryTestCase:
+    description: str
+    database: str | None
+    schemas: tuple[str, ...] | None
+    names: tuple[str, ...] | None
+    rows: tuple[tuple[object, ...], ...]
+    expected_relation: str
+    expected_params: tuple[str, ...]
+    expected_functions: tuple[FunctionInfo, ...]

tests/unit/src/sqlbuild/adapters/snowflake/test_function_discovery.py

@@ -6,64 +6,73 @@
 
 from sqlbuild.adapter.contract.models import FunctionInfo
 from sqlbuild.adapters.snowflake.classes.snowflake_adapter import SnowflakeAdapter
+from tests.unit.src.sqlbuild.adapters.snowflake._test_types import (
+    SnowflakeFunctionDiscoveryTestCase,
+)
 from tests.unit.src.sqlbuild.adapters.snowflake.helpers import (
     FakeSnowflakeMetadataConnection,
     FakeSnowflakeMetadataCursor,
 )
 
 
-def test_given_explicit_database_when_listing_functions_then_queries_requested_catalog() -> None:
-    cursor = FakeSnowflakeMetadataCursor(rows=[("NORMALIZE_ORDER", "PUBLIC", "function")])
-    connection = FakeSnowflakeMetadataConnection(cursor)
-
-    result = SnowflakeAdapter().list_functions(
-        connection=cast(Any, connection),
-        database="analytics",
-        schemas=("public",),
-        names=("normalize_order",),
-    )
-
-    assert cursor.executed_sql is not None
-    assert 'FROM "ANALYTICS".information_schema.functions' in cursor.executed_sql
-    assert cursor.executed_params == ("PUBLIC", "NORMALIZE_ORDER", "analytics")
-    assert result == (
-        FunctionInfo(
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        SnowflakeFunctionDiscoveryTestCase(
+            description="explicit database qualifies the information schema",
             database="analytics",
-            schema="public",
-            name="normalize_order",
-            function_type="function",
+            schemas=("public",),
+            names=("normalize_order",),
+            rows=(("NORMALIZE_ORDER", "PUBLIC", "function"),),
+            expected_relation='FROM "ANALYTICS".information_schema.functions',
+            expected_params=("PUBLIC", "NORMALIZE_ORDER", "analytics"),
+            expected_functions=(
+                FunctionInfo(
+                    database="analytics",
+                    schema="public",
+                    name="normalize_order",
+                    function_type="function",
+                ),
+            ),
         ),
-    )
-    assert cursor.closed
-
-
-def test_given_database_with_quote_when_listing_functions_then_escapes_identifier() -> None:
-    cursor = FakeSnowflakeMetadataCursor()
-
-    SnowflakeAdapter().list_functions(
-        connection=cast(Any, FakeSnowflakeMetadataConnection(cursor)),
-        database='order"catalog',
-        schemas=None,
-    )
-
-    assert cursor.executed_sql is not None
-    assert 'FROM "ORDER""CATALOG".information_schema.functions' in cursor.executed_sql
-    assert cursor.closed
-
-
-def test_given_no_database_when_listing_functions_then_uses_session_catalog() -> None:
-    cursor = FakeSnowflakeMetadataCursor()
+        SnowflakeFunctionDiscoveryTestCase(
+            description="database containing a quote is escaped",
+            database='order"catalog',
+            schemas=None,
+            names=None,
+            rows=(),
+            expected_relation='FROM "ORDER""CATALOG".information_schema.functions',
+            expected_params=('order"catalog',),
+            expected_functions=(),
+        ),
+        SnowflakeFunctionDiscoveryTestCase(
+            description="omitted database uses the session catalog",
+            database=None,
+            schemas=None,
+            names=None,
+            rows=(),
+            expected_relation="FROM information_schema.functions",
+            expected_params=(),
+            expected_functions=(),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_database_when_listing_functions_then_queries_matching_catalog(
+    test_case: SnowflakeFunctionDiscoveryTestCase,
+) -> None:
+    cursor: FakeSnowflakeMetadataCursor = FakeSnowflakeMetadataCursor(rows=list(test_case.rows))
 
-    result = SnowflakeAdapter().list_functions(
+    result: tuple[FunctionInfo, ...] = SnowflakeAdapter().list_functions(
         connection=cast(Any, FakeSnowflakeMetadataConnection(cursor)),
-        database=None,
-        schemas=None,
+        database=test_case.database,
+        schemas=test_case.schemas,
+        names=test_case.names,
     )
 
-    assert cursor.executed_sql is not None
-    assert "FROM information_schema.functions" in cursor.executed_sql
-    assert cursor.executed_params == ()
-    assert result == ()
+    assert test_case.expected_relation in str(cursor.executed_sql)
+    assert cursor.executed_params == test_case.expected_params
+    assert result == test_case.expected_functions
     assert cursor.closed