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:
pytest_plugins = [
"airbyte_cdk.test.standard_tests.pytest_hooks",
]
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 )
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.
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.
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.
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.
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):
...