airbyte_cdk.test.standard_tests.pytest_hooks

Pytest hooks for Airbyte CDK tests.

These hooks are used to customize the behavior of pytest during test discovery and execution.

To use these hooks within a connector, add the following lines to the connector's conftest.py file, or to another file that is imported during test discovery:

  1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
  2"""Pytest hooks for Airbyte CDK tests.
  3
  4These hooks are used to customize the behavior of pytest during test discovery and execution.
  5
  6To use these hooks within a connector, add the following lines to the connector's `conftest.py`
  7file, or to another file that is imported during test discovery:
  8
  9```python
 10pytest_plugins = [
 11    "airbyte_cdk.test.standard_tests.pytest_hooks",
 12]
 13```
 14"""
 15
 16from typing import Literal, cast
 17
 18import pytest
 19
 20
 21@pytest.fixture
 22def connector_image_override(request: pytest.FixtureRequest) -> str | None:
 23    """Return the value of --connector-image, or None if not set."""
 24    return cast(str | None, request.config.getoption("--connector-image"))
 25
 26
 27@pytest.fixture
 28def read_from_streams(
 29    request: pytest.FixtureRequest,
 30) -> Literal["all", "none", "default"] | list[str]:
 31    """Specify if the test should read from streams.
 32
 33    The input can be one of the following:
 34    - [Omitted] - Default to False, meaning no streams will be read.
 35    - `--read-from-streams`: Read from all suggested streams.
 36    - `--read-from-streams=true`: Read from all suggested streams.
 37    - `--read-from-streams=suggested`: Read from all suggested streams.
 38    - `--read-from-streams=default`: Read from all suggested streams.
 39    - `--read-from-streams=all`: Read from all streams.
 40    - `--read-from-streams=stream1,stream2`: Read from the specified streams only.
 41    - `--read-from-streams=false`: Do not read from any streams.
 42    - `--read-from-streams=none`: Do not read from any streams.
 43    """
 44    input_val: str = request.config.getoption(
 45        "--read-from-streams",
 46        default="default",  # type: ignore
 47    )  # type: ignore
 48
 49    if isinstance(input_val, str):
 50        if input_val.lower() == "false":
 51            return "none"
 52        if input_val.lower() in ["true", "suggested", "default"]:
 53            # Default to 'default' (suggested) streams if the input is 'true', 'suggested', or
 54            # 'default'.
 55            # This is the default behavior if the option is not set.
 56            return "default"
 57        if input_val.lower() == "all":
 58            # This will sometimes fail if the account doesn't have permissions
 59            # to premium or restricted stream data.
 60            return "all"
 61
 62        # If the input is a comma-separated list, split it into a list.
 63        # This will return a one-element list if the input is a single stream name.
 64        return input_val.split(",")
 65
 66    # Else, probably a bool; return it as is.
 67    return input_val or "none"
 68
 69
 70@pytest.fixture
 71def read_scenarios(
 72    request: pytest.FixtureRequest,
 73) -> list[str] | Literal["all", "default"]:
 74    """Return the value of `--read-scenarios`.
 75
 76    This argument is ignored if `--read-from-streams` is False or not set.
 77
 78    The input can be one of the following:
 79    - [Omitted] - Default to 'config.json', meaning the default scenario will be read.
 80    - `--read-scenarios=all`: Read all scenarios.
 81    - `--read-scenarios=none`: Read no scenarios. (Overrides `--read-from-streams`, if set.)
 82    - `--read-scenarios=scenario1,scenario2`: Read the specified scenarios only.
 83
 84    """
 85    input_val = cast(
 86        str,
 87        request.config.getoption(
 88            "--read-scenarios",
 89            default="default",  # type: ignore
 90        ),
 91    )
 92
 93    if input_val.lower() == "default":
 94        # Default config scenario is always 'config.json'.
 95        return "default"
 96
 97    if input_val.lower() == "none":
 98        # Default config scenario is always 'config.json'.
 99        return []
100
101    return (
102        [
103            scenario_name.strip().lower().removesuffix(".json")
104            for scenario_name in input_val.split(",")
105        ]
106        if input_val
107        else []
108    )
109
110
111def pytest_addoption(parser: pytest.Parser) -> None:
112    """Add --connector-image to pytest's CLI."""
113    parser.addoption(
114        "--connector-image",
115        action="store",
116        default=None,
117        help="Use this pre-built connector Docker image instead of building one.",
118    )
119    parser.addoption(
120        "--read-from-streams",
121        action="store",
122        default=None,
123        help=read_from_streams.__doc__,
124    )
125    parser.addoption(
126        "--read-scenarios",
127        action="store",
128        default="default",
129        help=read_scenarios.__doc__,
130    )
131
132
133def pytest_generate_tests(metafunc: pytest.Metafunc) -> None:
134    """A helper for pytest_generate_tests hook.
135
136    If a test method (in a class subclassed from our base class)
137    declares an argument 'scenario', this function retrieves the
138    'scenarios' attribute from the test class and parametrizes that
139    test with the values from 'scenarios'.
140
141    ## Usage
142
143    ```python
144    from airbyte_cdk.test.standard_tests.connector_base import (
145        generate_tests,
146        ConnectorTestSuiteBase,
147    )
148
149    def pytest_generate_tests(metafunc):
150        generate_tests(metafunc)
151
152    class TestMyConnector(ConnectorTestSuiteBase):
153        ...
154
155    ```
156    """
157    # Check if the test function requires an 'scenario' argument
158    if "scenario" in metafunc.fixturenames:
159        # Retrieve the test class
160        test_class = metafunc.cls
161        if test_class is None:
162            return
163
164        # Check that the class is compatible with our test suite
165        scenarios_attr = getattr(test_class, "get_scenarios", None)
166        if scenarios_attr is None:
167            raise ValueError(
168                f"Test class {test_class} does not have a 'scenarios' attribute. "
169                "Please define the 'scenarios' attribute in the test class."
170            )
171
172        # Get the scenarios defined or discovered in the test class
173        scenarios = test_class.get_scenarios()
174
175        # Create pytest.param objects with special marks as needed
176        parametrized_scenarios = [
177            pytest.param(
178                scenario,
179                marks=[pytest.mark.requires_creds] if scenario.requires_creds else [],
180            )
181            for scenario in scenarios
182        ]
183
184        # Parametrize the 'scenario' argument with the scenarios
185        metafunc.parametrize(
186            "scenario",
187            parametrized_scenarios,
188            ids=[str(scenario) for scenario in scenarios],
189        )
@pytest.fixture
def connector_image_override(request: _pytest.fixtures.FixtureRequest) -> str | None:
22@pytest.fixture
23def connector_image_override(request: pytest.FixtureRequest) -> str | None:
24    """Return the value of --connector-image, or None if not set."""
25    return cast(str | None, request.config.getoption("--connector-image"))

Return the value of --connector-image, or None if not set.

@pytest.fixture
def read_from_streams( request: _pytest.fixtures.FixtureRequest) -> Union[Literal['all', 'none', 'default'], list[str]]:
28@pytest.fixture
29def read_from_streams(
30    request: pytest.FixtureRequest,
31) -> Literal["all", "none", "default"] | list[str]:
32    """Specify if the test should read from streams.
33
34    The input can be one of the following:
35    - [Omitted] - Default to False, meaning no streams will be read.
36    - `--read-from-streams`: Read from all suggested streams.
37    - `--read-from-streams=true`: Read from all suggested streams.
38    - `--read-from-streams=suggested`: Read from all suggested streams.
39    - `--read-from-streams=default`: Read from all suggested streams.
40    - `--read-from-streams=all`: Read from all streams.
41    - `--read-from-streams=stream1,stream2`: Read from the specified streams only.
42    - `--read-from-streams=false`: Do not read from any streams.
43    - `--read-from-streams=none`: Do not read from any streams.
44    """
45    input_val: str = request.config.getoption(
46        "--read-from-streams",
47        default="default",  # type: ignore
48    )  # type: ignore
49
50    if isinstance(input_val, str):
51        if input_val.lower() == "false":
52            return "none"
53        if input_val.lower() in ["true", "suggested", "default"]:
54            # Default to 'default' (suggested) streams if the input is 'true', 'suggested', or
55            # 'default'.
56            # This is the default behavior if the option is not set.
57            return "default"
58        if input_val.lower() == "all":
59            # This will sometimes fail if the account doesn't have permissions
60            # to premium or restricted stream data.
61            return "all"
62
63        # If the input is a comma-separated list, split it into a list.
64        # This will return a one-element list if the input is a single stream name.
65        return input_val.split(",")
66
67    # Else, probably a bool; return it as is.
68    return input_val or "none"

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.
@pytest.fixture
def read_scenarios( request: _pytest.fixtures.FixtureRequest) -> Union[list[str], Literal['all', 'default']]:
 71@pytest.fixture
 72def read_scenarios(
 73    request: pytest.FixtureRequest,
 74) -> list[str] | Literal["all", "default"]:
 75    """Return the value of `--read-scenarios`.
 76
 77    This argument is ignored if `--read-from-streams` is False or not set.
 78
 79    The input can be one of the following:
 80    - [Omitted] - Default to 'config.json', meaning the default scenario will be read.
 81    - `--read-scenarios=all`: Read all scenarios.
 82    - `--read-scenarios=none`: Read no scenarios. (Overrides `--read-from-streams`, if set.)
 83    - `--read-scenarios=scenario1,scenario2`: Read the specified scenarios only.
 84
 85    """
 86    input_val = cast(
 87        str,
 88        request.config.getoption(
 89            "--read-scenarios",
 90            default="default",  # type: ignore
 91        ),
 92    )
 93
 94    if input_val.lower() == "default":
 95        # Default config scenario is always 'config.json'.
 96        return "default"
 97
 98    if input_val.lower() == "none":
 99        # Default config scenario is always 'config.json'.
100        return []
101
102    return (
103        [
104            scenario_name.strip().lower().removesuffix(".json")
105            for scenario_name in input_val.split(",")
106        ]
107        if input_val
108        else []
109    )

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.
def pytest_addoption(parser: _pytest.config.argparsing.Parser) -> None:
112def pytest_addoption(parser: pytest.Parser) -> None:
113    """Add --connector-image to pytest's CLI."""
114    parser.addoption(
115        "--connector-image",
116        action="store",
117        default=None,
118        help="Use this pre-built connector Docker image instead of building one.",
119    )
120    parser.addoption(
121        "--read-from-streams",
122        action="store",
123        default=None,
124        help=read_from_streams.__doc__,
125    )
126    parser.addoption(
127        "--read-scenarios",
128        action="store",
129        default="default",
130        help=read_scenarios.__doc__,
131    )

Add --connector-image to pytest's CLI.

def pytest_generate_tests(metafunc: _pytest.python.Metafunc) -> None:
134def pytest_generate_tests(metafunc: pytest.Metafunc) -> None:
135    """A helper for pytest_generate_tests hook.
136
137    If a test method (in a class subclassed from our base class)
138    declares an argument 'scenario', this function retrieves the
139    'scenarios' attribute from the test class and parametrizes that
140    test with the values from 'scenarios'.
141
142    ## Usage
143
144    ```python
145    from airbyte_cdk.test.standard_tests.connector_base import (
146        generate_tests,
147        ConnectorTestSuiteBase,
148    )
149
150    def pytest_generate_tests(metafunc):
151        generate_tests(metafunc)
152
153    class TestMyConnector(ConnectorTestSuiteBase):
154        ...
155
156    ```
157    """
158    # Check if the test function requires an 'scenario' argument
159    if "scenario" in metafunc.fixturenames:
160        # Retrieve the test class
161        test_class = metafunc.cls
162        if test_class is None:
163            return
164
165        # Check that the class is compatible with our test suite
166        scenarios_attr = getattr(test_class, "get_scenarios", None)
167        if scenarios_attr is None:
168            raise ValueError(
169                f"Test class {test_class} does not have a 'scenarios' attribute. "
170                "Please define the 'scenarios' attribute in the test class."
171            )
172
173        # Get the scenarios defined or discovered in the test class
174        scenarios = test_class.get_scenarios()
175
176        # Create pytest.param objects with special marks as needed
177        parametrized_scenarios = [
178            pytest.param(
179                scenario,
180                marks=[pytest.mark.requires_creds] if scenario.requires_creds else [],
181            )
182            for scenario in scenarios
183        ]
184
185        # Parametrize the 'scenario' argument with the scenarios
186        metafunc.parametrize(
187            "scenario",
188            parametrized_scenarios,
189            ids=[str(scenario) for scenario in scenarios],
190        )

A helper for pytest_generate_tests hook.

If a test method (in a class subclassed from our base class) declares an argument 'scenario', this function retrieves the 'scenarios' attribute from the test class and parametrizes that test with the values from 'scenarios'.

Usage

from airbyte_cdk.test.standard_tests.connector_base import (
    generate_tests,
    ConnectorTestSuiteBase,
)

def pytest_generate_tests(metafunc):
    generate_tests(metafunc)

class TestMyConnector(ConnectorTestSuiteBase):
    ...