chore(source-gcs): check AirbyteTracedException raised unit tests for file-based sources (#44770)

Author: strosek

airbyte-integrations/connectors/source-gcs/unit_tests/resource/catalog/catalog_bad_encoding.json

@@ -0,0 +1,22 @@
+{
+  "streams": [
+    {
+      "stream": {
+        "name": "stream_1",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh", "incremental"]
+      },
+      "sync_mode": "incremental",
+      "destination_sync_mode": "overwrite"
+    },
+    {
+      "stream": {
+        "name": "stream_2",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh", "incremental"]
+      },
+      "sync_mode": "incremental",
+      "destination_sync_mode": "overwrite"
+    }
+  ]
+}

airbyte-integrations/connectors/source-gcs/unit_tests/resource/config/config_bad_encoding.json

@@ -0,0 +1,22 @@
+{
+  "bucket": "some-bucket",
+  "service_account": "{\"type\": \"service_account\"}",
+  "streams": [
+    {
+      "name": "stream_1",
+      "validation_policy": "Emit Record",
+      "format": {
+        "filetype": "csv",
+        "encoding": "utf16"
+      }
+    },
+    {
+      "name": "stream_2",
+      "validation_policy": "Emit Record",
+      "format": {
+        "filetype": "csv",
+        "encoding": "utf16"
+      }
+    }
+  ]
+}

airbyte-integrations/connectors/source-gcs/unit_tests/resource/config/config_bad_encoding_single_stream.json

@@ -0,0 +1,14 @@
+{
+  "bucket": "some-bucket",
+  "service_account": "{\"type\": \"service_account\"}",
+  "streams": [
+    {
+      "name": "stream_1",
+      "validation_policy": "Emit Record",
+      "format": {
+        "filetype": "csv",
+        "encoding": "utf16"
+      }
+    }
+  ]
+}

airbyte-integrations/connectors/source-gcs/unit_tests/test_source.py

@@ -0,0 +1,57 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+
+import re
+from pathlib import Path
+from unittest.mock import Mock
+
+import pytest
+from airbyte_cdk import AirbyteTracedException
+from airbyte_cdk.sources.file_based.availability_strategy import DefaultFileBasedAvailabilityStrategy
+from source_gcs import Config, Cursor, SourceGCS, SourceGCSStreamReader
+
+
+def _source_gcs(catalog, config):
+    return SourceGCS(
+        SourceGCSStreamReader(),
+        Config,
+        catalog,
+        config,
+        None,
+        cursor_cls=Cursor,
+    )
+
+
+def _catalog_path():
+    return Path(__file__).resolve().parent.joinpath("resource/catalog/")
+
+
+def _config_path():
+    return Path(__file__).resolve().parent.joinpath("resource/config/")
+
+
+def _mock_encoding_error():
+    DefaultFileBasedAvailabilityStrategy.check_availability_and_parsability = Mock(side_effect=AirbyteTracedException(message="encoding error"))
+
+
+def test_check_connection_with_airbyte_traced_exception(logger):
+    config = SourceGCS.read_config(f"{_config_path()}/config_bad_encoding_single_stream.json")
+    catalog = SourceGCS.read_catalog(f"{_catalog_path()}/catalog_bad_encoding.json")
+    source = _source_gcs(catalog, config)
+    _mock_encoding_error()
+
+    with pytest.raises(AirbyteTracedException) as ate:
+        source.check_connection(logger, config)
+
+    assert re.search(r"^Unable to connect.+encoding error", ate.value.message)
+
+
+def test_check_connection_with_airbyte_traced_exception_multiple_failures(logger):
+    config = SourceGCS.read_config(f"{_config_path()}/config_bad_encoding.json")
+    catalog = SourceGCS.read_catalog(f"{_catalog_path()}/catalog_bad_encoding.json")
+    source = _source_gcs(catalog, config)
+    _mock_encoding_error()
+
+    with pytest.raises(AirbyteTracedException) as ate:
+        source.check_connection(logger, config)
+
+    assert re.search(r"2 streams with errors.+encoding error", ate.value.message)