SAT: basic read on full catalog when test_strictness_level == high (#18937)

Author: alafanechere

airbyte-integrations/bases/source-acceptance-test/CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+## 0.2.16
+Run `basic_read` on the discovered catalog in `high` `test_strictness_level`. [#18937](https://github.com/airbytehq/airbyte/pull/18937).
+
 ## 0.2.15
 Make `expect_records` mandatory in `high` `test_strictness_level`. [#18497](https://github.com/airbytehq/airbyte/pull/18497/).
 

airbyte-integrations/bases/source-acceptance-test/Dockerfile

@@ -33,7 +33,7 @@ COPY pytest.ini setup.py ./
 COPY source_acceptance_test ./source_acceptance_test
 RUN pip install .
 
-LABEL io.airbyte.version=0.2.15
+LABEL io.airbyte.version=0.2.16
 LABEL io.airbyte.name=airbyte/source-acceptance-test
 
 ENTRYPOINT ["python", "-m", "pytest", "-p", "source_acceptance_test.plugin", "-r", "fEsx"]

airbyte-integrations/bases/source-acceptance-test/source_acceptance_test/conftest.py

@@ -13,20 +13,20 @@
 from typing import Any, List, MutableMapping, Optional, Set
 
 import pytest
-from airbyte_cdk.models import (
- AirbyteRecordMessage,
- AirbyteStream,
- ConfiguredAirbyteCatalog,
- ConfiguredAirbyteStream,
- ConnectorSpecification,
- DestinationSyncMode,
- Type,
-)
+from airbyte_cdk.models import AirbyteRecordMessage, AirbyteStream, ConfiguredAirbyteCatalog, ConnectorSpecification, Type
 from docker import errors
 from source_acceptance_test.base import BaseTest
-from source_acceptance_test.config import Config, EmptyStreamConfiguration
+from source_acceptance_test.config import Config, EmptyStreamConfiguration, ExpectedRecordsConfig
 from source_acceptance_test.tests import TestBasicRead
-from source_acceptance_test.utils import ConnectorRunner, SecretDict, filter_output, load_config, load_yaml_or_json_path
+from source_acceptance_test.utils import (
+ ConnectorRunner,
+ SecretDict,
+ build_configured_catalog_from_custom_catalog,
+ build_configured_catalog_from_discovered_catalog_and_empty_streams,
+ filter_output,
+ load_config,
+ load_yaml_or_json_path,
+)
 
 
 @pytest.fixture(name="acceptance_test_config", scope="session")
@@ -44,7 +44,7 @@ def base_path_fixture(pytestconfig, acceptance_test_config) -> Path:
 
 
 @pytest.fixture(name="test_strictness_level", scope="session")
-def test_strictness_level_fixture(acceptance_test_config: Config):
+def test_strictness_level_fixture(acceptance_test_config: Config) -> Config.TestStrictnessLevel:
  return acceptance_test_config.test_strictness_level
 
 
@@ -75,24 +75,17 @@ def configured_catalog_path_fixture(inputs, base_path) -> Optional[str]:
 
 
 @pytest.fixture(name="configured_catalog")
-def configured_catalog_fixture(configured_catalog_path, discovered_catalog) -> ConfiguredAirbyteCatalog:
- """Take ConfiguredAirbyteCatalog from discover command by default"""
+def configured_catalog_fixture(
+ configured_catalog_path: Optional[str],
+ discovered_catalog: MutableMapping[str, AirbyteStream],
+) -> ConfiguredAirbyteCatalog:
+ """Build a configured catalog.
+ If a configured catalog path is given we build a configured catalog from it, we build it from the discovered catalog otherwise.
+ """
  if configured_catalog_path:
- catalog = ConfiguredAirbyteCatalog.parse_file(configured_catalog_path)
- for configured_stream in catalog.streams:
- configured_stream.stream = discovered_catalog.get(configured_stream.stream.name, configured_stream.stream)
- return catalog
- streams = [
- ConfiguredAirbyteStream(
- stream=stream,
- sync_mode=stream.supported_sync_modes[0],
- destination_sync_mode=DestinationSyncMode.append,
- cursor_field=stream.default_cursor_field,
- primary_key=stream.source_defined_primary_key,
- )
- for _, stream in discovered_catalog.items()
- ]
- return ConfiguredAirbyteCatalog(streams=streams)
+ return build_configured_catalog_from_custom_catalog(configured_catalog_path, discovered_catalog)
+ else:
+ return build_configured_catalog_from_discovered_catalog_and_empty_streams(discovered_catalog, set())
 
 
 @pytest.fixture(name="image_tag")
@@ -178,12 +171,17 @@ def empty_streams_fixture(inputs, test_strictness_level) -> Set[EmptyStreamConfi
  return empty_streams
 
 
+@pytest.fixture(name="expect_records_config")
+def expect_records_config_fixture(inputs):
+ return inputs.expect_records
+
+
 @pytest.fixture(name="expected_records_by_stream")
 def expected_records_by_stream_fixture(
  test_strictness_level: Config.TestStrictnessLevel,
  configured_catalog: ConfiguredAirbyteCatalog,
  empty_streams: Set[EmptyStreamConfiguration],
- inputs,
+ expect_records_config: ExpectedRecordsConfig,
  base_path,
 ) -> MutableMapping[str, List[MutableMapping]]:
  def enforce_high_strictness_level_rules(expect_records_config, configured_catalog, empty_streams, records_by_stream) -> Optional[str]:
@@ -197,8 +195,9 @@ def enforce_high_strictness_level_rules(expect_records_config, configured_catalo
  error_prefix
  + f"{', '.join(not_seeded_streams)} streams are declared in the catalog but do not have expected records. Please add expected records to {expect_records_config.path} or declare these streams in empty_streams."
  )
-
- expect_records_config = inputs.expect_records
+ else:
+ if not getattr(expect_records_config, "bypass_reason", None):
+ pytest.fail(error_prefix / "A bypass reason must be filled if no path to expected records is provided.")
 
  expected_records_by_stream = {}
  if expect_records_config:
@@ -246,7 +245,6 @@ def discovered_catalog_fixture(connector_config, docker_runner: ConnectorRunner,
  catalogs = [message.catalog for message in output if message.type == Type.CATALOG]
  for stream in catalogs[-1].streams:
  cached_schemas[stream.name] = stream
-
  return cached_schemas
 
 

airbyte-integrations/bases/source-acceptance-test/source_acceptance_test/tests/test_core.py

@@ -8,7 +8,8 @@
 from collections import Counter, defaultdict
 from functools import reduce
 from logging import Logger
-from typing import Any, Dict, List, Mapping, MutableMapping, Set
+from typing import Any, Dict, List, Mapping, MutableMapping, Optional, Set
+from xmlrpc.client import Boolean
 
 import dpath.util
 import jsonschema
@@ -27,10 +28,23 @@
 from docker.errors import ContainerError
 from jsonschema._utils import flatten
 from source_acceptance_test.base import BaseTest
-from source_acceptance_test.config import BasicReadTestConfig, ConnectionTestConfig, DiscoveryTestConfig, SpecTestConfig
+from source_acceptance_test.config import (
+ BasicReadTestConfig,
+ Config,
+ ConnectionTestConfig,
+ DiscoveryTestConfig,
+ EmptyStreamConfiguration,
+ ExpectedRecordsConfig,
+ SpecTestConfig,
+)
 from source_acceptance_test.utils import ConnectorRunner, SecretDict, filter_output, make_hashable, verify_records_schema
 from source_acceptance_test.utils.backward_compatibility import CatalogDiffChecker, SpecDiffChecker, validate_previous_configs
-from source_acceptance_test.utils.common import find_all_values_for_key_in_schema, find_keyword_schema
+from source_acceptance_test.utils.common import (
+ build_configured_catalog_from_custom_catalog,
+ build_configured_catalog_from_discovered_catalog_and_empty_streams,
+ find_all_values_for_key_in_schema,
+ find_keyword_schema,
+)
 from source_acceptance_test.utils.json_schema_helper import JsonSchemaHelper, get_expected_schema_structure, get_object_structure
 
 
@@ -462,11 +476,58 @@ def _validate_expected_records(
  detailed_logger=detailed_logger,
  )
 
+ @pytest.fixture(name="should_validate_schema")
+ def should_validate_schema_fixture(self, inputs: BasicReadTestConfig, test_strictness_level: Config.TestStrictnessLevel):
+ if not inputs.validate_schema and test_strictness_level is Config.TestStrictnessLevel.high:
+ pytest.fail("High strictness level error: validate_schema must be set to true in the basic read test configuration.")
+ else:
+ return inputs.validate_schema
+
+ @pytest.fixture(name="should_validate_data_points")
+ def should_validate_data_points_fixture(self, inputs: BasicReadTestConfig) -> Boolean:
+ # TODO: we might want to enforce this when Config.TestStrictnessLevel.high
+ return inputs.validate_data_points
+
+ @pytest.fixture(name="configured_catalog")
+ def configured_catalog_fixture(
+ self,
+ test_strictness_level: Config.TestStrictnessLevel,
+ configured_catalog_path: Optional[str],
+ discovered_catalog: MutableMapping[str, AirbyteStream],
+ empty_streams: Set[EmptyStreamConfiguration],
+ ) -> ConfiguredAirbyteCatalog:
+ """Build a configured catalog for basic read only.
+ We discard the use of custom configured catalog if:
+ - No custom configured catalog is declared with configured_catalog_path.
+ - We are in high test strictness level.
+ When a custom configured catalog is discarded we use the discovered catalog from which we remove the declared empty streams.
+ We use a custom configured catalog if a configured_catalog_path is declared and we are not in high test strictness level.
+ Args:
+ test_strictness_level (Config.TestStrictnessLevel): The current test strictness level according to the global test configuration.
+ configured_catalog_path (Optional[str]): Path to a JSON file containing a custom configured catalog.
+ discovered_catalog (MutableMapping[str, AirbyteStream]): The discovered catalog.
+ empty_streams (Set[EmptyStreamConfiguration]): The empty streams declared in the test configuration.
+
+ Returns:
+ ConfiguredAirbyteCatalog: the configured Airbyte catalog.
+ """
+ if test_strictness_level is Config.TestStrictnessLevel.high or not configured_catalog_path:
+ if configured_catalog_path:
+ pytest.fail(
+ "High strictness level error: you can't set a custom configured catalog on the basic read test when strictness level is high."
+ )
+ return build_configured_catalog_from_discovered_catalog_and_empty_streams(discovered_catalog, empty_streams)
+ else:
+ return build_configured_catalog_from_custom_catalog(configured_catalog_path, discovered_catalog)
+
  def test_read(
  self,
  connector_config,
  configured_catalog,
- inputs: BasicReadTestConfig,
+ expect_records_config: ExpectedRecordsConfig,
+ should_validate_schema: Boolean,
+ should_validate_data_points: Boolean,
+ empty_streams: Set[EmptyStreamConfiguration],
  expected_records_by_stream: MutableMapping[str, List[MutableMapping]],
  docker_runner: ConnectorRunner,
  detailed_logger,
@@ -476,25 +537,25 @@ def test_read(
 
  assert records, "At least one record should be read using provided catalog"
 
- if inputs.validate_schema:
+ if should_validate_schema:
  self._validate_schema(records=records, configured_catalog=configured_catalog)
 
- self._validate_empty_streams(records=records, configured_catalog=configured_catalog, allowed_empty_streams=inputs.empty_streams)
+ self._validate_empty_streams(records=records, configured_catalog=configured_catalog, allowed_empty_streams=empty_streams)
  for pks, record in primary_keys_for_records(streams=configured_catalog.streams, records=records):
  for pk_path, pk_value in pks.items():
  assert (
  pk_value is not None
  ), f"Primary key subkeys {repr(pk_path)} have null values or not present in {record.stream} stream records."
 
  # TODO: remove this condition after https://github.com/airbytehq/airbyte/issues/8312 is done
- if inputs.validate_data_points:
+ if should_validate_data_points:
  self._validate_field_appears_at_least_once(records=records, configured_catalog=configured_catalog)
 
  if expected_records_by_stream:
  self._validate_expected_records(
  records=records,
  expected_records_by_stream=expected_records_by_stream,
- flags=inputs.expect_records,
+ flags=expect_records_config,
  detailed_logger=detailed_logger,
  )
 

airbyte-integrations/bases/source-acceptance-test/source_acceptance_test/utils/__init__.py

@@ -1,8 +1,17 @@
 #
-# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+# Copyright (c) 2022 Airbyte, Inc., all rights reserved.
 #
 from .asserts import verify_records_schema
-from .common import SecretDict, filter_output, full_refresh_only_catalog, incremental_only_catalog, load_config, load_yaml_or_json_path
+from .common import (
+ SecretDict,
+ build_configured_catalog_from_custom_catalog,
+ build_configured_catalog_from_discovered_catalog_and_empty_streams,
+ filter_output,
+ full_refresh_only_catalog,
+ incremental_only_catalog,
+ load_config,
+ load_yaml_or_json_path,
+)
 from .compare import diff_dicts, make_hashable
 from .connector_runner import ConnectorRunner
 from .json_schema_helper import JsonSchemaHelper
@@ -19,4 +28,6 @@
  "diff_dicts",
  "make_hashable",
  "verify_records_schema",
+ "build_configured_catalog_from_custom_catalog",
+ "build_configured_catalog_from_discovered_catalog_and_empty_streams",
 ]

airbyte-integrations/bases/source-acceptance-test/source_acceptance_test/utils/common.py

@@ -3,9 +3,10 @@
 #
 
 import json
+import logging
 from collections import UserDict
 from pathlib import Path
-from typing import Iterable, List, Union
+from typing import Iterable, List, MutableMapping, Set, Union
 
 import pytest
 from yaml import load
@@ -15,8 +16,15 @@
 except ImportError:
  from yaml import Loader
 
-from airbyte_cdk.models import AirbyteMessage, ConfiguredAirbyteCatalog, SyncMode
-from source_acceptance_test.config import Config
+from airbyte_cdk.models import (
+ AirbyteMessage,
+ AirbyteStream,
+ ConfiguredAirbyteCatalog,
+ ConfiguredAirbyteStream,
+ DestinationSyncMode,
+ SyncMode,
+)
+from source_acceptance_test.config import Config, EmptyStreamConfiguration
 
 
 def load_config(path: str) -> Config:
@@ -111,3 +119,58 @@ def find_all_values_for_key_in_schema(schema: dict, searched_key: str):
  yield value
  if isinstance(value, dict) or isinstance(value, list):
  yield from find_all_values_for_key_in_schema(value, searched_key)
+
+
+def build_configured_catalog_from_discovered_catalog_and_empty_streams(
+ discovered_catalog: MutableMapping[str, AirbyteStream], empty_streams: Set[EmptyStreamConfiguration]
+):
+ """Build a configured catalog from the discovered catalog with empty streams removed.
+
+ Args:
+ discovered_catalog (MutableMapping[str, AirbyteStream]): The discovered catalog.
+ empty_streams (Set[EmptyStreamConfiguration]): The set of empty streams declared in the test configuration.
+
+ Returns:
+ ConfiguredAirbyteCatalog: a configured Airbyte catalog.
+ """
+ empty_stream_names = [empty_stream.name for empty_stream in empty_streams]
+ streams = [
+ ConfiguredAirbyteStream(
+ stream=stream,
+ sync_mode=stream.supported_sync_modes[0],
+ destination_sync_mode=DestinationSyncMode.append,
+ cursor_field=stream.default_cursor_field,
+ primary_key=stream.source_defined_primary_key,
+ )
+ for _, stream in discovered_catalog.items()
+ if stream.name not in empty_stream_names
+ ]
+ if empty_stream_names:
+ logging.warning(
+ f"The configured catalog was built with the discovered catalog from which the following empty streams were removed: {', '.join(empty_stream_names)}."
+ )
+ else:
+ logging.info("The configured catalog is built from a fully discovered catalog.")
+ return ConfiguredAirbyteCatalog(streams=streams)
+
+
+def build_configured_catalog_from_custom_catalog(configured_catalog_path: str, discovered_catalog: MutableMapping[str, AirbyteStream]):
+ """Build a configured catalog from a local one stored in a JSON file.
+
+ Args:
+ configured_catalog_path (str): Local path to a custom configured catalog path
+ discovered_catalog (MutableMapping[str, AirbyteStream]): The discovered catalog
+
+ Returns:
+ ConfiguredAirbyteCatalog: a configured Airbyte catalog
+ """
+ catalog = ConfiguredAirbyteCatalog.parse_file(configured_catalog_path)
+ for configured_stream in catalog.streams:
+ try:
+ configured_stream.stream = discovered_catalog[configured_stream.stream.name]
+ except KeyError:
+ pytest.fail(
+ f"The {configured_stream.stream.name} stream you have set in {configured_catalog_path} is not part of the discovered_catalog"
+ )
+ logging.info("The configured catalog is built from a custom configured catalog.")
+ return catalog