FIX: map vertica airflow connection schema for vertica database

cosmos/profiles/vertica/user_pass.py

@@ -1,4 +1,14 @@
-"Maps Airflow Vertica connections using username + password authentication to dbt profiles."
+"""
+Maps Airflow Vertica connections using username + password authentication to dbt profiles.
+
+NOTE: Use Airflow connection `schema` for vertica `database` to keep it consistent with other connection types and profiles.
+Also Vertica Airflow provider hook assumes this:
+https://github.com/apache/airflow/blob/395ac463494dba1478a05a32900218988495889c/airflow/providers/vertica/hooks/vertica.py#L72
+This seems to be a common approach also for Postgres, Redshift and Exasol since there is no `database` field in Airflow connection
+and `schema` is not required for the database connection.
+Posgres Hook:
+https://github.com/apache/airflow/blob/0953e0f844fa5db81c2b461ec2433de1935260b3/airflow/providers/postgres/hooks/postgres.py#L138
+"""
 from __future__ import annotations
 
 from typing import Any
@@ -31,8 +41,7 @@ class VerticaUserPasswordProfileMapping(BaseProfileMapping):
         "username": "login",
         "password": "password",
         "port": "port",
-        "schema": "schema",
-        "database": "extra.database",
+        "database": "schema",
         "autocommit": "extra.autocommit",
         "backup_server_node": "extra.backup_server_node",
         "binary_transfer": "extra.binary_transfer",

tests/profiles/vertica/test_vertica_user_pass.py

@@ -23,8 +23,7 @@ def mock_vertica_conn():  # type: ignore
         login="my_user",
         password="my_password",
         port=5433,
-        schema="my_schema",
-        extra='{"database": "my_database"}',
+        schema="my_database",
     )
 
     with patch("airflow.hooks.base.BaseHook.get_connection", return_value=conn):
@@ -43,8 +42,7 @@ def mock_vertica_conn_custom_port():  # type: ignore
         login="my_user",
         password="my_password",
         port=7472,
-        schema="my_schema",
-        extra='{"database": "my_database"}',
+        schema="my_database",
     )
 
     with patch("airflow.hooks.base.BaseHook.get_connection", return_value=conn):
@@ -69,8 +67,7 @@ def test_connection_claiming() -> None:
         "host": "my_host",
         "login": "my_user",
         "password": "my_password",
-        "schema": "my_schema",
-        "extra": '{"database": "my_database"}',
+        "schema": "my_database",
     }
 
     # if we're missing any of the values, it shouldn't claim
@@ -82,20 +79,20 @@ def test_connection_claiming() -> None:
         print("testing with", values)
 
         with patch("airflow.hooks.base.BaseHook.get_connection", return_value=conn):
-            profile_mapping = VerticaUserPasswordProfileMapping(conn)
+            profile_mapping = VerticaUserPasswordProfileMapping(conn, {"schema": "my_schema"})
             assert not profile_mapping.can_claim_connection()
 
-    # also test when there's no database
+    # also test when there's no schema
     conn = Connection(**potential_values)  # type: ignore
     conn.extra = ""
     with patch("airflow.hooks.base.BaseHook.get_connection", return_value=conn):
-        profile_mapping = VerticaUserPasswordProfileMapping(conn)
+        profile_mapping = VerticaUserPasswordProfileMapping(conn, {})
         assert not profile_mapping.can_claim_connection()
 
     # if we have them all, it should claim
     conn = Connection(**potential_values)  # type: ignore
     with patch("airflow.hooks.base.BaseHook.get_connection", return_value=conn):
-        profile_mapping = VerticaUserPasswordProfileMapping(conn)
+        profile_mapping = VerticaUserPasswordProfileMapping(conn, {"schema": "my_schema"})
         assert profile_mapping.can_claim_connection()
 
 
@@ -107,7 +104,7 @@ def test_profile_mapping_selected(
     """
     profile_mapping = get_automatic_profile_mapping(
         mock_vertica_conn.conn_id,
-        {"schema": "my_schema"},
+        {"schema": "my_database"},
     )
     assert isinstance(profile_mapping, VerticaUserPasswordProfileMapping)
 
@@ -145,8 +142,8 @@ def test_profile_args(
         "username": mock_vertica_conn.login,
         "password": "{{ env_var('COSMOS_CONN_VERTICA_PASSWORD') }}",
         "port": mock_vertica_conn.port,
+        "database": mock_vertica_conn.schema,
         "schema": "my_schema",
-        "database": mock_vertica_conn.extra_dejson.get("database"),
     }