update snowflake connection to handle new closed state properly, rena…

…me "warehouse" config key to "snowflake_warehouse", t ests

plugins/snowflake/dbt/adapters/snowflake/connections.py

@@ -35,8 +35,8 @@ def type(self):
         return 'snowflake'
 
     def _connection_keys(self):
-        return ('account', 'user', 'database', 'schema', 'warehouse', 'role',
-                'client_session_keep_alive')
+        return ('account', 'user', 'database', 'schema', 'snowflake_warehouse',
+                 'role', 'client_session_keep_alive')
 
     def auth_args(self):
         # Pull all of the optional authentication args for the connector,

plugins/snowflake/dbt/adapters/snowflake/impl.py

@@ -13,7 +13,7 @@ class SnowflakeAdapter(SQLAdapter):
 
     AdapterSpecificConfigs = frozenset(
         {"transient", "cluster_by", "automatic_clustering", "secure",
-         "copy_grants", "warehouse"}
+         "copy_grants", "snowflake_warehouse"}
     )
 
     @classmethod
@@ -62,7 +62,7 @@ def _use_warehouse(self, warehouse: str):
 
     def pre_model_hook(self, config: Mapping[str, Any]) -> Optional[str]:
         default_warehouse = self.config.credentials.warehouse
-        warehouse = config.get('warehouse', default_warehouse)
+        warehouse = config.get('snowflake_warehouse', default_warehouse)
         if warehouse == default_warehouse or warehouse is None:
             return None
         previous = self._get_warehouse()

test/integration/050_warehouse_test/models/invalid_warehouse.sql

@@ -1,2 +1,2 @@
-{{ config(warehouse='DBT_TEST_DOES_NOT_EXIST') }}
+{{ config(snowflake_warehouse='DBT_TEST_DOES_NOT_EXIST') }}
 select current_warehouse() as warehouse

test/integration/050_warehouse_test/models/override_warehouse.sql

@@ -1,2 +1,2 @@
-{{ config(warehouse='DBT_TEST_ALT', materialized='table') }}
+{{ config(snowflake_warehouse='DBT_TEST_ALT', materialized='table') }}
 select current_warehouse() as warehouse

test/integration/050_warehouse_test/project-config-models/expected_warehouse.sql

@@ -0,0 +1,2 @@
+{{ config(materialized='table') }}
+select 'DBT_TEST_ALT' as warehouse

test/integration/050_warehouse_test/project-config-models/override_warehouse.sql

@@ -0,0 +1,2 @@
+{{ config(materialized='table') }}
+select current_warehouse() as warehouse

test/integration/050_warehouse_test/test_warehouses.py

@@ -2,7 +2,7 @@
 import os
 
 
-class TestDebug(DBTIntegrationTest):
+class TestModelWarehouse(DBTIntegrationTest):
     @property
     def schema(self):
         return 'dbt_warehouse_050'
@@ -26,3 +26,36 @@ def test_snowflake_override_ok(self):
     @use_profile('snowflake')
     def test_snowflake_override_noexist(self):
         self.run_dbt(['run', '--models', 'invalid_warehouse'], expect_pass=False)
+
+
+class TestConfigWarehouse(DBTIntegrationTest):
+    @property
+    def schema(self):
+        return 'dbt_warehouse_050'
+
+    @property
+    def project_config(self):
+        return {
+            'source-paths': ['project-config-models'],
+            'models': {
+                'test': {
+                    'snowflake_warehouse': 'DBT_TEST_ALT',
+                },
+            },
+        }
+
+    @staticmethod
+    def dir(value):
+        return os.path.normpath(value)
+
+    @property
+    def models(self):
+        return self.dir('models')
+
+    @use_profile('snowflake')
+    def test_snowflake_override_ok(self):
+        self.run_dbt([
+            'run',
+            '--models', 'override_warehouse', 'expected_warehouse',
+        ])
+        self.assertManyRelationsEqual([['OVERRIDE_WAREHOUSE'], ['EXPECTED_WAREHOUSE']])

test/integration/base.py

@@ -623,7 +623,7 @@ def run_sql_common(self, sql, fetch, conn):
                 else:
                     return
             except BaseException as e:
-                if conn.handle and conn.handle.closed != 0:
+                if conn.handle and not conn.handle.is_closed():
                     conn.handle.rollback()
                 print(sql)
                 print(e)

test/unit/test_snowflake_adapter.py

@@ -181,7 +181,7 @@ def _strip_transactions(self):
 
     def test_pre_post_hooks_warehouse(self):
         with self.current_warehouse('warehouse'):
-            config = {'warehouse': 'other_warehouse'}
+            config = {'snowflake_warehouse': 'other_warehouse'}
             result = self.adapter.pre_model_hook(config)
             self.assertIsNotNone(result)
             calls = [