diff --git a/.github/workflows/connector-tests.yml b/.github/workflows/connector-tests.yml index 3eff71380d..81bcf36bc7 100644 --- a/.github/workflows/connector-tests.yml +++ b/.github/workflows/connector-tests.yml @@ -189,8 +189,11 @@ jobs: env: GCP_GSM_CREDENTIALS: ${{ secrets.GCP_GSM_CREDENTIALS }} POETRY_DYNAMIC_VERSIONING_BYPASS: "0.0.0" + # The Docker read runs in the "Container Build and Test" step below; skip it here so it + # does not run twice. run: | - poetry run airbyte-cdk connector test + poetry run airbyte-cdk connector test \ + --pytest-arg=--read-from-streams=false # Manifest-only connector images are built `FROM` the published `source-declarative-manifest` # base image declared in metadata.yaml, so by default the container tests exercise the @@ -212,6 +215,7 @@ jobs: # branch-built SDM base image here too. Otherwise those tests build the connector image # `FROM` the published metadata base image (running the released CDK instead of this # branch's), and the later `image test` step can't reuse its layers. + # The Docker read runs in the later `image test` step, so it is skipped here. - name: Run FAST Standard Tests [Manifest-Only Connectors] if: env.CONNECTOR_LANGUAGE == 'manifest-only' working-directory: airbyte-python-cdk @@ -221,7 +225,8 @@ jobs: run: | poetry run airbyte-cdk connector test ${{ matrix.connector }} \ --pytest-arg=--connector-base-image \ - --pytest-arg=airbyte/source-declarative-manifest:dev + --pytest-arg=airbyte/source-declarative-manifest:dev \ + --pytest-arg=--read-from-streams=false - name: Container Build and Test [All Connectors] if: steps.no_changes.outputs.status != 'cancelled' diff --git a/airbyte_cdk/test/standard_tests/docker_base.py b/airbyte_cdk/test/standard_tests/docker_base.py index b66b11f270..2a279fd47e 100644 --- a/airbyte_cdk/test/standard_tests/docker_base.py +++ b/airbyte_cdk/test/standard_tests/docker_base.py @@ -39,6 +39,10 @@ assert_check_outcome, assert_config_matches_spec, ) +from airbyte_cdk.test.standard_tests._read_assertions import ( + assert_read_records, + assert_schema_validation_opt_out_allowed, +) from airbyte_cdk.test.standard_tests._spec_compatibility import ( DEPLOYMENT_MODE_ENV_VARS, REGISTRY_UNAVAILABLE_ERRORS, @@ -151,6 +155,71 @@ def get_read_streams( return streams +DEFAULT_READ_SCENARIO_IDS: tuple[str, ...] = ("config", "valid_config", "default") +"""The config ids (file names without `.json`) that `--read-scenarios=default` reads.""" + + +def get_default_read_scenario_ids(scenarios: list[ConnectorTestScenario]) -> set[str]: + """Return the ids of the configs that the Docker read test reads under `--read-scenarios=default`. + + These are the read configs named in `DEFAULT_READ_SCENARIO_IDS`. A connector that names none of + its read configs that way reads its first read config instead, so the default never skips + every Docker read of a connector. Configs expected to fail are never picked. + """ + read_ids = [ + scenario.id + for scenario in scenarios + if scenario.runs("read") and not scenario.expected_outcome.expect_exception() + ] + preferred = { + scenario_id for scenario_id in read_ids if scenario_id in DEFAULT_READ_SCENARIO_IDS + } + return preferred or set(read_ids[:1]) + + +def select_docker_read_streams( + read_streams: list[AirbyteStream], + *, + read_from_streams: Literal["all", "default"] | list[str], + suggested_streams: list[str] | None, +) -> list[AirbyteStream]: + """Narrow a config's read streams by `--read-from-streams` (see the `read_from_streams` fixture). + + `default` keeps the connector's suggested streams, unless it declares none or none of them is + read for this config: then every stream is kept, as `all` does. A stream list keeps the listed + streams, which may leave none. + """ + if isinstance(read_from_streams, list): + wanted = set(read_from_streams) + return [stream for stream in read_streams if stream.name in wanted] + + if read_from_streams == "default" and suggested_streams: + suggested = set(suggested_streams) + narrowed = [stream for stream in read_streams if stream.name in suggested] + if narrowed: + return narrowed + + return read_streams + + +CONNECTOR_WORKDIR = "/source" +"""The connector's working directory in the Docker tests: the platform's, for a source.""" + +CONNECTOR_WORKDIR_ARGS: tuple[str, ...] = ( + "--tmpfs", + f"{CONNECTOR_WORKDIR}:rw,exec,mode=1777", + "--workdir", + CONNECTOR_WORKDIR, +) +"""`docker run` arguments that give the connector a writable working directory, as the platform does. + +The platform runs a connector with its working directory on a writable volume. An image's own +`WORKDIR` is usually not writable by the image's non-root user, so a connector that writes a +temporary file to its working directory (a bulk export download, for example) would fail only in +the tests. +""" + + _SPEC_FAILURE_OUTPUT_LINES = 50 """How many trailing lines of the connector's output a failed `spec` run reports.""" @@ -735,6 +804,7 @@ def test_docker_image_build_and_check( "docker", "run", "--rm", + *CONNECTOR_WORKDIR_ARGS, "-v", f"{temp_config_file}:{container_config_path}:rw", connector_image, @@ -850,6 +920,7 @@ def test_docker_image_build_and_discover( "docker", "run", "--rm", + *CONNECTOR_WORKDIR_ARGS, "-v", f"{temp_config_file}:{container_config_path}:rw", connector_image, @@ -875,7 +946,7 @@ def test_docker_image_build_and_read( read_from_streams: Literal["all", "none", "default"] | list[str], read_scenarios: Literal["all", "none", "default"] | list[str], ) -> None: - """Read from the connector's Docker image. + """Read from the connector's Docker image, and check the records as `test_basic_read` does. This test builds the connector image and runs a full-refresh `read` command inside the container, for configs listed under `basic_read` or `full_refresh` in @@ -883,6 +954,18 @@ def test_docker_image_build_and_read( `configured_catalog_path`, or every discovered stream when it declares none, minus its `empty_streams` (see `get_read_streams`), narrowed further by `--read-from-streams`. + The records get the same checks as in `test_basic_read`: a config listed under + `basic_read` must return at least one record from every stream it reads, and every record + must match its stream's JSON schema (opt out with `validate_schema: false`, which a + connector at `test_strictness_level: high` cannot). A config listed only under + `full_refresh` must return some records. + + The test runs by default. `--read-scenarios` picks which configs are read (by default + `config`, `valid_config` and `default`, or the first read config when a connector has + none of those), and `--read-from-streams` which of their streams (by default the + connector's suggested streams); `--read-from-streams=false` (or `none`) skips the test. See the + `read_from_streams` and `read_scenarios` fixtures. + Note: - If the --connector-image arg is provided, it will be used instead of building the image. """ @@ -895,25 +978,31 @@ def test_docker_image_build_and_read( pytest.skip("Skipping (expected to fail).") if read_from_streams == "none": - pytest.skip("Skipping read test (`--read-from-streams=false`).") + pytest.skip( + "Skipping the Docker read test: it was disabled with `--read-from-streams`." + ) if read_scenarios == "none": - pytest.skip("Skipping (`--read-scenarios=none`).") + pytest.skip("Skipping the Docker read test: it was disabled with `--read-scenarios`.") - default_scenario_ids = ["config", "valid_config", "default"] - if read_scenarios == "all": - pass - elif read_scenarios == "default": - if scenario.id not in default_scenario_ids: + if read_scenarios == "default": + default_ids = get_default_read_scenario_ids(self.get_scenarios()) + if scenario.id not in default_ids: pytest.skip( - f"Skipping read test for scenario '{scenario.id}' " - f"(not in default scenarios list '{default_scenario_ids}')." + f"Skipping the Docker read of config '{scenario.id}': " + f"`--read-scenarios=default` reads {', '.join(sorted(map(repr, default_ids)))}. " + "Pass `--read-scenarios=all` to read every config." ) - elif scenario.id not in read_scenarios: - # pytest.skip( - raise ValueError( - f"Skipping read test for scenario '{scenario.id}' " - f"(not in --read-scenarios={read_scenarios})." + elif read_scenarios != "all" and scenario.id.lower() not in read_scenarios: + pytest.skip( + f"Skipping the Docker read of config '{scenario.id}': not in " + f"`--read-scenarios={','.join(read_scenarios)}`." + ) + + if scenario.is_basic_read_config: + assert_schema_validation_opt_out_allowed( + validate_schema=scenario.validate_schema, + test_strictness_level=scenario.test_strictness_level, ) connector_root = self.get_connector_root_dir() @@ -941,6 +1030,7 @@ def test_docker_image_build_and_read( "docker", "run", "--rm", + *CONNECTOR_WORKDIR_ARGS, "-v", f"{temp_config_file}:{container_config_path}:rw", connector_image, @@ -951,19 +1041,22 @@ def test_docker_image_build_and_read( raise_if_errors=True, ) - read_streams = get_read_streams( - scenario, - get_discovered_streams(discover_result, connector_name), - connector_root, + suggested_streams = metadata.data.suggestedStreams + read_streams = select_docker_read_streams( + get_read_streams( + scenario, + get_discovered_streams(discover_result, connector_name), + connector_root, + ), + read_from_streams=read_from_streams, + suggested_streams=suggested_streams.streams if suggested_streams else None, ) - if read_from_streams == "default" and metadata.data.suggestedStreams: - # Read only the suggested streams. - suggested_streams = set(metadata.data.suggestedStreams.streams) - read_streams = [s for s in read_streams if s.name in suggested_streams] - - if isinstance(read_from_streams, list): - # If `read_from_streams` is a list, read only those streams. - read_streams = [s for s in read_streams if s.name in read_from_streams] + if not read_streams and isinstance(read_from_streams, list): + pytest.skip( + f"Skipping the Docker read of config '{scenario.id}': none of the streams in " + f"`--read-from-streams={','.join(read_from_streams)}` is " + "read for it." + ) configured_catalog: ConfiguredAirbyteCatalog = ConfiguredAirbyteCatalog( streams=[ @@ -984,6 +1077,7 @@ def test_docker_image_build_and_read( "docker", "run", "--rm", + *CONNECTOR_WORKDIR_ARGS, "-v", f"{temp_config_file}:{container_config_path}:rw", "-v", @@ -997,3 +1091,14 @@ def test_docker_image_build_and_read( ], raise_if_errors=True, ) + + if scenario.is_basic_read_config: + # Also asserts that the read returned records, in the same pass over them. + assert_read_records( + records=read_result.records_iterator, + configured_catalog=configured_catalog, + require_records_per_stream=True, + validate_schema=scenario.validate_schema, + ) + elif next(read_result.records_iterator, None) is None: + raise AssertionError("Expected records but got none.") diff --git a/airbyte_cdk/test/standard_tests/pytest_hooks.py b/airbyte_cdk/test/standard_tests/pytest_hooks.py index fa905e7157..90088288a6 100644 --- a/airbyte_cdk/test/standard_tests/pytest_hooks.py +++ b/airbyte_cdk/test/standard_tests/pytest_hooks.py @@ -38,92 +38,102 @@ def connector_base_image_override(request: pytest.FixtureRequest) -> str | None: return cast(str | None, request.config.getoption("--connector-base-image")) +READ_FROM_STREAMS_DEFAULT = "default" +"""The `--read-from-streams` value used when the option is omitted: the Docker read test runs.""" + +READ_SCENARIOS_DEFAULT = "default" +"""The `--read-scenarios` value used when the option is omitted.""" + + @pytest.fixture def read_from_streams( request: pytest.FixtureRequest, ) -> Literal["all", "none", "default"] | list[str]: - """Specify if the test should read from streams. - - The input can be one of the following: - - [Omitted] - Default to False, meaning no streams will be read. - - `--read-from-streams`: Read from all suggested streams. - - `--read-from-streams=true`: Read from all suggested streams. - - `--read-from-streams=suggested`: Read from all suggested streams. - - `--read-from-streams=default`: Read from all suggested streams. - - `--read-from-streams=all`: Read from all streams. - - `--read-from-streams=stream1,stream2`: Read from the specified streams only. - - `--read-from-streams=false`: Do not read from any streams. - - `--read-from-streams=none`: Do not read from any streams. + """Return the value of `--read-from-streams`: which streams the Docker read test reads. + + The Docker read test (`test_docker_image_build_and_read`) runs a full-refresh `read` inside the + connector image for each config listed under `basic_read` or `full_refresh` in + `acceptance-test-config.yml`, of the streams of its `configured_catalog_path` (or every + discovered stream) minus its `empty_streams`. This option narrows those streams further: + + - [Omitted], `--read-from-streams=default`, `=true` or `=suggested`: Read the connector's + `suggestedStreams` from `metadata.yaml`. A connector that declares none, or whose suggested + streams are all excluded for the config, reads all of the config's streams. + - `--read-from-streams=all`: Read all of the config's streams. + - `--read-from-streams=stream1,stream2`: Read only these of the config's streams. A config + that reads none of them is skipped. + - `--read-from-streams=false` or `=none`: Skip the Docker read test (the explicit opt-out). + Prefer `false` in CI files: CDK 7.36.0 and earlier read `none` as a stream name. + + Which configs are read is set by `--read-scenarios`. The two options combine: the test reads + the streams selected here for each config selected there, so the default of both reads at + least one config of every connector that lists a config under a read section. """ - input_val: str = request.config.getoption( + input_val = request.config.getoption( "--read-from-streams", - default="default", # type: ignore - ) # type: ignore - - if isinstance(input_val, str): - if input_val.lower() == "false": - return "none" - if input_val.lower() in ["true", "suggested", "default"]: - # Default to 'default' (suggested) streams if the input is 'true', 'suggested', or - # 'default'. - # This is the default behavior if the option is not set. - return "default" - if input_val.lower() == "all": - # This will sometimes fail if the account doesn't have permissions - # to premium or restricted stream data. - return "all" - - # If the input is a comma-separated list, split it into a list. - # This will return a one-element list if the input is a single stream name. - return input_val.split(",") - - # Else, probably a bool; return it as is. - return input_val or "none" + default=READ_FROM_STREAMS_DEFAULT, + ) + if input_val is None: + input_val = READ_FROM_STREAMS_DEFAULT + if not isinstance(input_val, str): + # Only a bool can get here, from a caller that sets the option programmatically. + return "default" if input_val else "none" + + normalized = input_val.strip().lower() + if normalized in ("false", "none"): + return "none" + if normalized in ("true", "suggested", "default"): + return "default" + if normalized == "all": + # This will sometimes fail if the account doesn't have permissions + # to premium or restricted stream data. + return "all" + + # A comma-separated list of stream names; a single name gives a one-element list. + return [name.strip() for name in input_val.split(",") if name.strip()] @pytest.fixture def read_scenarios( request: pytest.FixtureRequest, -) -> list[str] | Literal["all", "default"]: - """Return the value of `--read-scenarios`. - - This argument is ignored if `--read-from-streams` is False or not set. - - The input can be one of the following: - - [Omitted] - Default to 'config.json', meaning the default scenario will be read. - - `--read-scenarios=all`: Read all scenarios. - - `--read-scenarios=none`: Read no scenarios. (Overrides `--read-from-streams`, if set.) - - `--read-scenarios=scenario1,scenario2`: Read the specified scenarios only. - +) -> list[str] | Literal["all", "none", "default"]: + """Return the value of `--read-scenarios`: which configs the Docker read test reads. + + Only configs listed under `basic_read` or `full_refresh` in `acceptance-test-config.yml` are + ever read. Of those: + + - [Omitted] or `--read-scenarios=default`: Read the configs whose file name (without `.json`) + is `config`, `valid_config` or `default`. A connector with none of those reads its first + read config instead, so the default never skips every Docker read of a connector. + - `--read-scenarios=all`: Read every read config. + - `--read-scenarios=scenario1,scenario2`: Read only these configs (by file name, with or + without `.json`); the others are skipped. + - `--read-scenarios=none`: Skip the Docker read test, like `--read-from-streams=false`. + + `--read-from-streams` then picks the streams read for each selected config. The in-process + read test (`test_basic_read`) ignores both options and reads every read config. """ - input_val = cast( - str, - request.config.getoption( - "--read-scenarios", - default="default", # type: ignore - ), + input_val = request.config.getoption( + "--read-scenarios", + default=READ_SCENARIOS_DEFAULT, ) + if not input_val: + input_val = READ_SCENARIOS_DEFAULT + input_val = cast(str, input_val) - if input_val.lower() == "default": - # Default config scenario is always 'config.json'. - return "default" + normalized = input_val.strip().lower() + if normalized in ("default", "all", "none"): + return cast(Literal["all", "none", "default"], normalized) - if input_val.lower() == "none": - # Default config scenario is always 'config.json'. - return [] - - return ( - [ - scenario_name.strip().lower().removesuffix(".json") - for scenario_name in input_val.split(",") - ] - if input_val - else [] - ) + return [ + scenario_name.strip().lower().removesuffix(".json") + for scenario_name in input_val.split(",") + if scenario_name.strip() + ] def pytest_addoption(parser: pytest.Parser) -> None: - """Add --connector-image to pytest's CLI.""" + """Add the standard tests' options to pytest's CLI.""" parser.addoption( "--connector-image", action="store", @@ -145,13 +155,13 @@ def pytest_addoption(parser: pytest.Parser) -> None: parser.addoption( "--read-from-streams", action="store", - default=None, + default=READ_FROM_STREAMS_DEFAULT, help=read_from_streams.__doc__, ) parser.addoption( "--read-scenarios", action="store", - default="default", + default=READ_SCENARIOS_DEFAULT, help=read_scenarios.__doc__, ) diff --git a/cdk-migrations.md b/cdk-migrations.md index 31426600c4..28b2a4a530 100644 --- a/cdk-migrations.md +++ b/cdk-migrations.md @@ -2,6 +2,35 @@ ## Upgrading to the next minor release (unreleased) +### The Docker read test runs by default + +`test_docker_image_build_and_read`, the `read` inside the connector image, was skipped for every connector: `--read-from-streams` was registered without a default, and the test treated the missing value as `none`. Neither the CDK's connector CI nor the connector monorepo passed the option, so the test never ran. It now runs by default, in `airbyte-cdk image test` (what the connector monorepo runs for every Python and manifest-only connector) and wherever else the Docker suite is collected. For most Python connectors, whose in-process tests do not run in the monorepo, it is the first `read` that connector CI runs (only `source-s3` and `source-hardcoded-records` already read, through their own integration tests). + +The read records now get the checks of the in-process `test_basic_read`, through the same `assert_read_records`: + +- A config listed under `basic_read` must return at least one record from every stream it reads, and every record must match its stream's JSON schema. The exceptions are declared the same way: `empty_streams` (with a `bypass_reason`) for streams without data, and `validate_schema: false` to skip schema validation, which a connector at `test_strictness_level: high` cannot set. +- A config listed only under `full_refresh` must return some records. + +Previously the test only failed on a trace error or a non-zero exit. + +The Docker `check`, `discover` and `read` tests now run the connector with a writable working directory, `/source` (an in-memory `tmpfs`), as the platform does. They used to run in the image's `WORKDIR`, which the image's `airbyte` user cannot write, so a connector that writes a temporary file to its working directory (for example `source-shopify` and `source-salesforce`, for bulk results) failed in the test although it works in a sync. + +Two options pick what is read. The defaults read one config of every connector that lists a config under `basic_read` or `full_refresh`: + +| Option | Default | Other values | +|---|---|---| +| `--read-scenarios` (which configs) | `default`: the configs named `config`, `valid_config` or `default`. A connector with none of those reads its first read config instead. | `all`; a comma-separated list of config names (with or without `.json`); `none` skips the test. | +| `--read-from-streams` (which streams of each config) | `default` (also `true`, `suggested`): the connector's `suggestedStreams` from `metadata.yaml`, or every stream of the config when it declares none, or when none of them is read for the config. | `all`; a comma-separated list of stream names (a config that reads none of them is skipped); `false` or `none` skips the test (`false` on every CDK version; CDK 7.36.0 and earlier read `none` as a stream name and run the read). | + +The streams of a config are the ones the in-process read covers: those of its `configured_catalog_path`, or every discovered stream, minus `empty_streams`. The two options only narrow that set, and either one set to `none` skips the test, so they cannot cancel each other silently: the `--read-scenarios` default used to skip every Docker read of a connector whose read configs had other names, and now reads its first one. `--read-scenarios=none` used to raise a `ValueError` for every config, and a `--read-scenarios` list raised one for each config it left out; both now skip. The in-process `test_basic_read` ignores both options and reads every read config. + +Migration steps: + +- A connector whose Docker read fails gets the same fixes as a failing `test_basic_read`: refresh an expired secret, declare a stream without data in `empty_streams` with a `bypass_reason`, fix the schema or the records, or set `validate_schema: false` (with a reason) where that is allowed. +- To skip the Docker read for one run, pass `--read-from-streams=false` (to `airbyte-cdk connector test` as `--pytest-arg=--read-from-streams=false`), or set `PYTEST_ADDOPTS=--read-from-streams=false`, which `airbyte-cdk image test` also honours. Prefer `false` over `none` in CI files and poe tasks: `false` skips the read on every CDK version, while CDK 7.36.0 and earlier read `none` as a stream name, so they run the Docker `discover` and a `read` of an empty catalog. + +## Upgrading to 7.39.0 + ### `acceptance-test-config.yml` keys the Standard Tests do not read fail collection `ConnectorTestScenario` declared `expect_records` (with `exact_order`), `timeout_seconds` and `file_types`, and no test read them: they parsed, and the check they configured in the retired Connector Acceptance Tests (CAT) never ran. They are removed. Any other key that the model did not declare was dropped without a word, so a misspelled key (`empty_stream` for `empty_streams`) was ignored as well. diff --git a/unit_tests/test/test_docker_read.py b/unit_tests/test/test_docker_read.py new file mode 100644 index 0000000000..f197852cdb --- /dev/null +++ b/unit_tests/test/test_docker_read.py @@ -0,0 +1,358 @@ +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +"""The Docker read test runs by default and checks its records as `test_basic_read` does. + +- `--read-from-streams` and `--read-scenarios` default to reading; `false` or `none` is the explicit opt-out. +- `--read-scenarios=default` reads the `config`, `valid_config` and `default` configs, or the first + read config of a connector that has none of those, so it never skips every read. +- `--read-from-streams=default` reads the suggested streams, or every stream when none applies. +- A `basic_read` config must return records from every stream it reads, and the records must + match their schemas; a `full_refresh`-only config must return some records. +""" + +from __future__ import annotations + +import json +import re +from pathlib import Path +from typing import Any + +import pytest +import yaml + +from airbyte_cdk.test.models.scenario import ConnectorTestScenario +from airbyte_cdk.test.standard_tests.docker_base import ( + DockerConnectorTestSuite, + get_default_read_scenario_ids, +) +from unit_tests.test.test_scenario_sections import ( + _catalog_message, + _entry, + _record_message, + _scenario, + _stream_json, + _stub_docker, + _suite_class, +) + +pytest_plugins = ["pytester"] + +DEFAULTS: dict[str, Any] = { + "connector_image_override": "airbyte/source-test:dev", # skips the image build + "connector_base_image_override": None, + "read_from_streams": "default", + "read_scenarios": "default", +} + +DISCOVER = [ + _catalog_message( + _stream_json("users", ["full_refresh", "incremental"], ["updated_at"]), + _stream_json("events", ["full_refresh"]), + ) +] + + +def _read_suite( + tmp_path: Path, + acceptance_tests: dict[str, Any] | None = None, + *, + suggested_streams: list[str] | None = None, +) -> DockerConnectorTestSuite: + suite_class = _suite_class( + tmp_path, + acceptance_tests + if acceptance_tests is not None + else {"basic_read": {"tests": [_entry("secrets/config.json")]}}, + base=DockerConnectorTestSuite, + ) + if suggested_streams is not None: + metadata_path = tmp_path / "source-test" / "metadata.yaml" + metadata = yaml.safe_load(metadata_path.read_text()) + metadata["data"]["suggestedStreams"] = {"streams": suggested_streams} + metadata_path.write_text(yaml.safe_dump(metadata)) + return suite_class() + + +def _read(suite: DockerConnectorTestSuite, scenario: ConnectorTestScenario, **options: Any) -> None: + suite.test_docker_image_build_and_read(scenario, **(DEFAULTS | options)) + + +def _read_commands(calls: list[list[str]]) -> list[list[str]]: + return [cmd for cmd in calls if "read" in cmd] + + +# --- Options ----------------------------------------------------------------------------------- + +OPTIONS_TEST = """ +import json + +def test_options(read_from_streams, read_scenarios): + print("OPTIONS=" + json.dumps([read_from_streams, read_scenarios])) +""" + + +@pytest.mark.parametrize( + "args, expected", + [ + pytest.param([], ["default", "default"], id="omitted-reads-by-default"), + pytest.param(["--read-from-streams=none"], ["none", "default"], id="none"), + pytest.param(["--read-from-streams=False"], ["none", "default"], id="false"), + pytest.param(["--read-from-streams=true"], ["default", "default"], id="true"), + pytest.param(["--read-from-streams=suggested"], ["default", "default"], id="suggested"), + pytest.param(["--read-from-streams=all"], ["all", "default"], id="all"), + pytest.param(["--read-from-streams=users, events"], [["users", "events"], "default"]), + pytest.param(["--read-scenarios=none"], ["default", "none"], id="scenarios-none"), + pytest.param(["--read-scenarios=ALL"], ["default", "all"], id="scenarios-all"), + pytest.param( + ["--read-scenarios=Config.json,config_oauth"], + ["default", ["config", "config_oauth"]], + id="scenarios-list", + ), + ], +) +def test_read_options(pytester: pytest.Pytester, args: list[str], expected: list[Any]) -> None: + """The `addoption` defaults, the fixtures and their docstrings agree: omitted means read.""" + pytester.makepyfile(test_options=OPTIONS_TEST) + result = pytester.runpytest_inprocess( + "-p", "airbyte_cdk.test.standard_tests.pytest_hooks", "-s", *args + ) + result.assert_outcomes(passed=1) + match = re.search(r"OPTIONS=(.*)$", result.stdout.str(), re.MULTILINE) + assert match is not None, result.stdout.str() + assert json.loads(match.group(1)) == expected + + +# --- Opt-outs and config selection ------------------------------------------------------------- + + +def test_docker_read_runs_by_default(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + calls = _stub_docker( + monkeypatch, + {"discover": DISCOVER, "read": [_record_message("users"), _record_message("events")]}, + ) + _read(_read_suite(tmp_path), _scenario("basic_read")) + assert len(_read_commands(calls)) == 1 + + +@pytest.mark.parametrize( + "options, reason", + [ + pytest.param({"read_from_streams": "none"}, "--read-from-streams", id="read-from-streams"), + pytest.param({"read_scenarios": "none"}, "--read-scenarios", id="read-scenarios"), + ], +) +def test_docker_read_opt_outs_skip( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, options: dict[str, Any], reason: str +) -> None: + calls = _stub_docker(monkeypatch, {}) + with pytest.raises(pytest.skip.Exception, match=f"disabled with `{reason}`"): + _read(_read_suite(tmp_path), _scenario("basic_read"), **options) + assert calls == [] + + +def test_docker_read_scenario_list_skips_other_configs( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A config missing from `--read-scenarios` is skipped; it used to raise a `ValueError`.""" + calls = _stub_docker( + monkeypatch, + {"discover": DISCOVER, "read": [_record_message("users"), _record_message("events")]}, + ) + suite = _read_suite(tmp_path) + with pytest.raises(pytest.skip.Exception, match=r"not in `--read-scenarios=config_oauth`"): + _read(suite, _scenario("basic_read"), read_scenarios=["config_oauth"]) + assert calls == [] + + _read(suite, _scenario("basic_read", name="Config_OAuth"), read_scenarios=["config_oauth"]) + assert len(_read_commands(calls)) == 1 + + +@pytest.mark.parametrize( + "configs, expected", + [ + pytest.param( + ["secrets/config_oauth.json", "secrets/config.json", "secrets/valid_config.json"], + {"config", "valid_config"}, + id="named-defaults", + ), + pytest.param( + ["secrets/config_oauth.json", "secrets/config_token.json"], + {"config_oauth"}, + id="first-read-config-when-none-is-named-default", + ), + ], +) +def test_default_read_scenarios(tmp_path: Path, configs: list[str], expected: set[str]) -> None: + suite = _read_suite( + tmp_path, + { + # A `connection`-only config is never read, so it is never the fallback. + "connection": {"tests": [_entry("secrets/check_only.json")]}, + "basic_read": {"tests": [_entry(config) for config in configs]}, + }, + ) + assert get_default_read_scenario_ids(suite.get_scenarios()) == expected + + +def test_default_read_scenarios_never_pick_an_expected_failure() -> None: + scenarios = [ + _scenario("basic_read", name="bad", status="failed"), + _scenario("basic_read", name="config_oauth"), + ] + assert get_default_read_scenario_ids(scenarios) == {"config_oauth"} + assert get_default_read_scenario_ids([_scenario("connection")]) == set() + + +def test_docker_read_defaults_read_a_connector_without_a_default_named_config( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """`--read-scenarios=default` used to skip every read of such a connector.""" + calls = _stub_docker( + monkeypatch, + {"discover": DISCOVER, "read": [_record_message("users"), _record_message("events")]}, + ) + suite = _read_suite( + tmp_path, + { + "basic_read": { + "tests": [_entry("secrets/config_oauth.json"), _entry("secrets/config_token.json")] + } + }, + ) + _read(suite, _scenario("basic_read", name="config_oauth")) + with pytest.raises( + pytest.skip.Exception, match=r"`--read-scenarios=default` reads 'config_oauth'" + ): + _read(suite, _scenario("basic_read", name="config_token")) + assert len(_read_commands(calls)) == 1 + + +# --- Streams ----------------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "suggested, read_from_streams, expected", + [ + pytest.param(None, "default", ["users", "events"], id="no-suggested-streams"), + pytest.param(["events"], "default", ["events"], id="suggested-streams"), + pytest.param(["events"], "all", ["users", "events"], id="all"), + pytest.param(["other"], "default", ["users", "events"], id="no-suggested-stream-read"), + pytest.param(["events"], ["users"], ["users"], id="stream-list"), + ], +) +def test_docker_read_streams( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + suggested: list[str] | None, + read_from_streams: Any, + expected: list[str], +) -> None: + catalogs: list[dict[str, Any]] = [] + _stub_docker( + monkeypatch, + {"discover": DISCOVER, "read": [_record_message("users"), _record_message("events")]}, + read_catalogs=catalogs, + ) + suite = _read_suite(tmp_path, suggested_streams=suggested) + _read(suite, _scenario("basic_read"), read_from_streams=read_from_streams) + assert [stream["stream"]["name"] for stream in catalogs[-1]["streams"]] == expected + + +def test_docker_read_skips_when_the_stream_list_matches_nothing( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + calls = _stub_docker(monkeypatch, {"discover": DISCOVER}) + with pytest.raises(pytest.skip.Exception, match=r"`--read-from-streams=other,more`"): + _read(_read_suite(tmp_path), _scenario("basic_read"), read_from_streams=["other", "more"]) + assert _read_commands(calls) == [] + + +# --- Record checks ----------------------------------------------------------------------------- + + +def test_docker_read_requires_records_from_every_stream( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _stub_docker(monkeypatch, {"discover": DISCOVER, "read": [_record_message("users")]}) + suite = _read_suite(tmp_path) + + with pytest.raises(AssertionError, match=r"these returned none: 'events'"): + _read(suite, _scenario("basic_read")) + + # A stream declared in `empty_streams` is not read, so it needs no records. + _read( + suite, + _scenario("basic_read", empty_streams=[{"name": "events", "bypass_reason": "no data"}]), + ) + + +def test_docker_read_of_a_full_refresh_only_config_needs_some_records( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + responses = {"discover": DISCOVER, "read": [_record_message("users")]} + _stub_docker(monkeypatch, responses) + suite = _read_suite(tmp_path, {"full_refresh": {"tests": [_entry("secrets/config.json")]}}) + + _read(suite, _scenario("full_refresh")) + + responses["read"] = [json.dumps({"type": "LOG", "log": {"level": "INFO", "message": "hi"}})] + with pytest.raises(AssertionError, match="Expected records but got none."): + _read(suite, _scenario("full_refresh")) + + +def _record_with(stream: str, data: dict[str, Any]) -> str: + return json.dumps( + {"type": "RECORD", "record": {"stream": stream, "data": data, "emitted_at": 0}} + ) + + +STRICT_DISCOVER = [ + _catalog_message( + { + "name": "users", + "json_schema": {"type": "object", "properties": {"id": {"type": "string"}}}, + "supported_sync_modes": ["full_refresh"], + } + ) +] + + +def test_docker_read_validates_records_against_their_schema( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + _stub_docker( + monkeypatch, + {"discover": STRICT_DISCOVER, "read": [_record_with("users", {"id": 31337})]}, + ) + suite = _read_suite(tmp_path) + + with pytest.raises(AssertionError, match=r"do not match the stream's JSON schema") as exc_info: + _read(suite, _scenario("basic_read")) + message = str(exc_info.value) + assert "Stream 'users'" in message and "`$.id`" in message + assert "31337" not in message, "record values must never reach the failure message" + + # The opt-out is honoured... + _read(suite, _scenario("basic_read", validate_schema=False)) + + # ...but not at high strictness, where it fails before anything runs in Docker. + calls = _stub_docker(monkeypatch, {}) + with pytest.raises(AssertionError, match=r"`validate_schema: false` is not allowed"): + _read( + suite, + _scenario("basic_read", validate_schema=False, test_strictness_level="high"), + ) + assert calls == [] + + +def test_docker_read_runs_in_a_writable_working_directory( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The platform gives a connector a writable working directory; the image's own is not.""" + calls = _stub_docker( + monkeypatch, + {"discover": DISCOVER, "read": [_record_message("users"), _record_message("events")]}, + ) + _read(_read_suite(tmp_path), _scenario("basic_read")) + assert [cmd[3:7] for cmd in calls] == [ + ["--tmpfs", "/source:rw,exec,mode=1777", "--workdir", "/source"] + ] * 2, "both `discover` and `read`"