Avoid breaking existing connections

airbyte-integrations/connectors/source-firebolt/source_firebolt/database.py

@@ -33,7 +33,9 @@ def parse_config(config: json, logger: AirbyteLogger) -> Dict[str, Any]:
 
     :return: dictionary of firebolt.db.Connection-compatible kwargs
     """
-    auth = _determine_auth(config["client_id"], config["client_secret"])
+    # We should use client_id/client_secret, this code supports username/password for legacy users
+    auth = _determine_auth(config.get("client_id", config.get("username")),
+                           config.get("client_secret", config.get("password")))
     connection_args = {
         "database": config["database"],
         "auth": auth,

airbyte-integrations/connectors/source-firebolt/unit_tests/test_firebolt_source.py

@@ -33,6 +33,16 @@ def config(request):
     }
     return args
 
+@fixture()
+def legacy_config(request):
+    args = {
+        "database": "my_database",
+        # @ is important here to determine the auth type
+        "username": "my@username",
+        "password": "my_password",
+        "engine": "my_engine",
+    }
+    return args
 
 @fixture()
 def config_no_engine():
@@ -99,6 +109,11 @@ def test_parse_config(config, logger):
     result = parse_config(config, logger)
     assert result["engine_url"] == "override_engine.api.firebolt.io"
 
+def test_parse_legacy_config(legacy_config, logger):
+    result = parse_config(legacy_config, logger)
+    assert result["database"] == "my_database"
+    assert result["auth"].username == "my@username"
+    assert result["auth"].password == "my_password"
 
 @patch("source_firebolt.database.connect")
 def test_connection(mock_connection, config, config_no_engine, logger):