feat(python-sources): add unit integration testing utilities for simplification (#43338)

Author: strosek

airbyte-cdk/python/airbyte_cdk/test/entrypoint_wrapper.py

@@ -16,6 +16,7 @@
 
 import json
 import logging
+import re
 import tempfile
 import traceback
 from io import StringIO
@@ -115,6 +116,14 @@ def _get_message_by_types(self, message_types: List[Type]) -> List[AirbyteMessag
     def _get_trace_message_by_trace_type(self, trace_type: TraceType) -> List[AirbyteMessage]:
         return [message for message in self._get_message_by_types([Type.TRACE]) if message.trace.type == trace_type]
 
+    def is_in_logs(self, pattern: str) -> bool:
+        """Check if any log message case-insensitive matches the pattern."""
+        return any(re.search(pattern, entry.log.message, flags=re.IGNORECASE) for entry in self.logs)
+
+    def is_not_in_logs(self, pattern: str) -> bool:
+        """Check if no log message matches the case-insensitive pattern."""
+        return not self.is_in_logs(pattern)
+
 
 def _run_command(source: Source, args: List[str], expecting_exception: bool = False) -> EntrypointOutput:
     log_capture_buffer = StringIO()

airbyte-cdk/python/airbyte_cdk/test/mock_http/response_builder.py

@@ -7,6 +7,7 @@
 from typing import Any, Dict, List, Optional, Union
 
 from airbyte_cdk.test.mock_http import HttpResponse
+from airbyte_cdk.test.utils.data import get_unit_test_folder
 
 
 def _extract(path: List[str], response_template: Dict[str, Any]) -> Any:
@@ -169,16 +170,12 @@ def build(self) -> HttpResponse:
 
 
 def _get_unit_test_folder(execution_folder: str) -> FilePath:
-    path = FilePath(execution_folder)
-    while path.name != "unit_tests":
-        if path.name == path.root or path.name == path.drive:
-            raise ValueError(f"Could not find `unit_tests` folder as a parent of {execution_folder}")
-        path = path.parent
-    return path
+    # FIXME: This function should be removed after the next CDK release to avoid breaking amazon-seller-partner test code.
+    return get_unit_test_folder(execution_folder)
 
 
 def find_template(resource: str, execution_folder: str) -> Dict[str, Any]:
-    response_template_filepath = str(_get_unit_test_folder(execution_folder) / "resource" / "http" / "response" / f"{resource}.json")
+    response_template_filepath = str(get_unit_test_folder(execution_folder) / "resource" / "http" / "response" / f"{resource}.json")
     with open(response_template_filepath, "r") as template_file:
         return json.load(template_file)  # type: ignore  # we assume the dev correctly set up the resource file
 

airbyte-cdk/python/airbyte_cdk/test/utils/__init__.py

@@ -0,0 +1 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.

airbyte-cdk/python/airbyte_cdk/test/utils/data.py

@@ -0,0 +1,20 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+
+from pydantic import FilePath
+
+
+def get_unit_test_folder(execution_folder: str) -> FilePath:
+    path = FilePath(execution_folder)
+    while path.name != "unit_tests":
+        if path.name == path.root or path.name == path.drive:
+            raise ValueError(f"Could not find `unit_tests` folder as a parent of {execution_folder}")
+        path = path.parent
+    return path
+
+
+def read_resource_file_contents(resource: str, test_location: str) -> str:
+    """Read the contents of a test data file from the test resource folder."""
+    file_path = str(get_unit_test_folder(test_location) / "resource" / "http" / "response" / f"{resource}")
+    with open(file_path) as f:
+        response = f.read()
+    return response

airbyte-cdk/python/airbyte_cdk/test/utils/http_mocking.py

@@ -0,0 +1,14 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+
+import re
+from typing import Any, Mapping
+
+from requests_mock import Mocker
+
+
+def register_mock_responses(mocker: Mocker, http_calls: list[Mapping[str, Mapping[str, Any]]]) -> None:
+    """Register a list of HTTP request-response pairs."""
+    for call in http_calls:
+        request, response = call["request"], call["response"]
+        matcher = re.compile(request["url"]) if request["is_regex"] else request["url"]
+        mocker.register_uri(request["method"], matcher, **response)

airbyte-cdk/python/airbyte_cdk/test/utils/reading.py

@@ -0,0 +1,26 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+
+from typing import Any, List, Mapping, Optional
+
+from airbyte_cdk import AbstractSource
+from airbyte_cdk.test.catalog_builder import CatalogBuilder
+from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput, read
+from airbyte_protocol.models import AirbyteStateMessage, ConfiguredAirbyteCatalog, SyncMode
+
+
+def catalog(stream_name: str, sync_mode: SyncMode) -> ConfiguredAirbyteCatalog:
+    """Create a catalog with a single stream."""
+    return CatalogBuilder().with_stream(stream_name, sync_mode).build()
+
+
+def read_records(
+    source: AbstractSource,
+    config: Mapping[str, Any],
+    stream_name: str,
+    sync_mode: SyncMode,
+    state: Optional[List[AirbyteStateMessage]] = None,
+    expecting_exception: bool = False,
+) -> EntrypointOutput:
+    """Read records from a stream."""
+    _catalog = catalog(stream_name, sync_mode)
+    return read(source, config, _catalog, state, expecting_exception)

airbyte-integrations/connectors/source-amazon-seller-partner/unit_tests/integration/utils.py

@@ -53,6 +53,7 @@ def get_stream_by_name(stream_name: str, config_: Mapping[str, Any]) -> Stream:
 
 def find_template(resource: str, execution_folder: str, template_format: Optional[str] = "csv") -> str:
     response_template_filepath = str(
+        # FIXME: the below function should be replaced with the public version after next CDK release
         _get_unit_test_folder(execution_folder) / "resource" / "http" / "response" / f"{resource}.{template_format}"
     )
     with open(response_template_filepath, "r") as template_file: