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

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

@@ -0,0 +1,23 @@
+import re
+
+from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput
+
+
+def is_in_logs(pattern: str, output: EntrypointOutput) -> bool:
+    return any(re.search(pattern, entry.log.message, flags=re.IGNORECASE) for entry in output.logs)
+
+
+def is_not_in_logs(pattern: str, output: EntrypointOutput) -> bool:
+    return not is_in_logs(pattern, output)
+
+
+def assert_good_read(output: EntrypointOutput, expected_records: int) -> None:
+    # checks there are no errors in output, plus the amount of records is correct.
+    assert len(output.records) == expected_records
+    assert is_not_in_logs("error|exception", output)
+
+
+def assert_bad_read(output: EntrypointOutput, expected_records: int) -> None:
+    # checks there are errors in output, plus the amount of records expected amount of records.
+    assert len(output.records) == expected_records
+    assert is_in_logs("error|exception", output)

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

@@ -0,0 +1,17 @@
+from pathlib import Path as 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 get_json_contents(resource: str, test_location: FilePath) -> str:
+    json_path = str(_get_unit_test_folder(test_location) / "resource" / "http" / "response" / f"{resource}")
+    with open(json_path) as f:
+        response = f.read()
+    return response

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

@@ -0,0 +1,22 @@
+from typing import Optional, List, Mapping, Any
+
+from airbyte_cdk import AbstractSource
+from airbyte_cdk.test.catalog_builder import CatalogBuilder
+from airbyte_protocol.models import AirbyteStateMessage, ConfiguredAirbyteCatalog, SyncMode
+from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput, read
+
+
+def catalog(stream_name: str, sync_mode: SyncMode) -> ConfiguredAirbyteCatalog:
+    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: Optional[bool] = False,
+) -> EntrypointOutput:
+    _catalog = catalog(stream_name, sync_mode)
+    return read(source, config, _catalog, state, expecting_exception)

airbyte-integrations/connectors/source-woocommerce/unit_tests/integration/common.py

@@ -0,0 +1,35 @@
+from typing import MutableMapping, Any
+
+import source_woocommerce
+from airbyte_cdk.connector_builder.connector_builder_handler import resolve_manifest
+from airbyte_cdk import AbstractSource
+
+
+def config() -> MutableMapping[str, Any]:
+    return {
+        "api_key": "test_api_key",
+        "api_secret": "test_api_secret",
+        "shop": "airbyte.store",
+        "start_date": "2017-01-01",
+    }
+
+
+def source() -> AbstractSource:
+    return source_woocommerce.SourceWoocommerce()
+
+
+def common_params():
+    return "orderby=id&order=asc&dates_are_gmt=true&per_page=100"
+
+
+def endpoint_url(stream_name: str, path_parameter: str = None) -> str:
+    if path_parameter:
+        return f"{url_base()}/{path_parameter}/{stream_name}"
+
+    return f"{url_base()}/{stream_name}"
+
+
+def url_base() -> str:
+    url = resolve_manifest(source()).record.data["manifest"]["definitions"]["requester"]["url_base"]
+    url = url.replace("{{ config['shop'] }}", config()["shop"])
+    return url

airbyte-integrations/connectors/source-woocommerce/unit_tests/integration/test_basic_read.py

@@ -0,0 +1,57 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+
+import pytest
+import requests_mock
+
+from airbyte_cdk.test.utils.assertions import assert_good_read
+from airbyte_cdk.test.utils.data import get_json_contents
+from airbyte_protocol.models import SyncMode
+from freezegun import freeze_time
+
+
+from airbyte_cdk.test.utils.reading import read_records
+from .common import common_params, config, source, endpoint_url
+
+
+def modified_after() -> str:
+    return "2017-01-01T00:00:00"
+
+
+def modified_before() -> str:
+    return "2017-01-30T23:59:59"
+
+
+def date_range() -> str:
+    # Create and encode date range
+    return f"modified_after={modified_after()}&modified_before={modified_before()}".replace(":", "%3A")
+
+
+def build_url(stream_name: str) -> str:
+    return f"{endpoint_url(stream_name)}?{common_params()}&{date_range()}"
+
+
+@freeze_time(modified_before())
+@requests_mock.Mocker(kw="mock")
+@pytest.mark.parametrize(
+    "stream_name, json_response_file, num_records",
+    [
+        ("coupons", "coupons.json", 2),
+        ("orders", "orders.json", 2),
+        # ("customers", "customers.json", 2),
+        ("products", "products.json", 2),
+    ],
+)
+def test_read_simple_endpoints_successfully(stream_name, json_response_file, num_records, **kwargs) -> None:
+    """Test basic read for  all streams that use date ranges and don't have parent streams."""
+
+    # Register mock response
+    kwargs["mock"].get(
+        build_url(stream_name),
+        text=get_json_contents(json_response_file, __file__)
+    )
+
+    # Read records
+    output = read_records(source(), config(), stream_name, SyncMode.full_refresh)
+
+    # Check read was successful
+    assert_good_read(output, num_records)

airbyte-integrations/connectors/source-woocommerce/unit_tests/integration/utils.py

@@ -1,7 +1,7 @@
 # Copyright (c) 2024 Airbyte, Inc., all rights reserved.
 
 from pathlib import Path as FilePath
-from typing import List, Optional
+from typing import List, Optional, MutableMapping, Any
 
 from airbyte_cdk.test.catalog_builder import CatalogBuilder
 from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput, read
@@ -16,12 +16,13 @@ def catalog(stream_name: str, sync_mode: SyncMode) -> ConfiguredAirbyteCatalog:
     return CatalogBuilder().with_stream(stream_name, sync_mode).build()
 
 
-def config() -> ConfigBuilder:
-    return ConfigBuilder()
-
-
-def source() -> SourceWoocommerce:
-    return SourceWoocommerce()
+def config() -> MutableMapping[str, Any]:
+    return {
+        "api_key": "test_api_key",
+        "api_secret": "test_api_secret",
+        "shop": "airbyte.store",
+        "start_date": "2017-01-01",
+    }
 
 
 def read_output(
@@ -33,20 +34,4 @@ def read_output(
 ) -> EntrypointOutput:
     _catalog = catalog(stream_name, sync_mode)
     _config = config_builder.build()
-    return read(source(), _config, _catalog, state, expecting_exception)
-
-
-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 get_json_http_response(resource: str, status_code: int) -> HttpResponse:
-    json_path = str(_get_unit_test_folder(__file__) / "resource" / "http" / "response" / f"{resource}")
-    with open(json_path) as f:
-        response = f.read()
-    return HttpResponse(response, status_code)
+    return read(SourceWoocommerce(), _config, _catalog, state, expecting_exception)