airbyte_cdk.test.models

 28class ConnectorTestScenario(BaseModel):
 29    """Acceptance test scenario, as a Pydantic model.
 30
 31    This class represents an acceptance test scenario, which is a single test case
 32    that can be run against a connector. It is used to deserialize and validate the
 33    acceptance test configuration file.
 34    """
 35
 36    # Allows the class to be hashable, which PyTest will require
 37    # when we use to parameterize tests.
 38    model_config = ConfigDict(frozen=True)
 39
 40    class AcceptanceTestExpectRecords(BaseModel):
 41        path: Path
 42        exact_order: bool = False
 43
 44    class AcceptanceTestFileTypes(BaseModel):
 45        skip_test: bool
 46        bypass_reason: str
 47
 48    class AcceptanceTestEmptyStream(BaseModel):
 49        name: str
 50        bypass_reason: str | None = None
 51
 52        # bypass reason does not affect equality
 53        def __hash__(self) -> int:
 54            return hash(self.name)
 55
 56    config_path: Path | None = None
 57    config_dict: dict[str, Any] | None = None
 58
 59    _id: str | None = None  # Used to override the default ID generation
 60
 61    configured_catalog_path: Path | None = None
 62    empty_streams: list[AcceptanceTestEmptyStream] | None = None
 63    timeout_seconds: int | None = None
 64    expect_records: AcceptanceTestExpectRecords | None = None
 65    file_types: AcceptanceTestFileTypes | None = None
 66    status: Literal["succeed", "failed", "exception"] | None = None
 67
 68    def get_config_dict(
 69        self,
 70        *,
 71        connector_root: Path,
 72        empty_if_missing: bool,
 73    ) -> dict[str, Any]:
 74        """Return the config dictionary.
 75
 76        If a config dictionary has already been loaded, return it. Otherwise, load
 77        the config file and return the dictionary.
 78
 79        If `self.config_dict` and `self.config_path` are both `None`:
 80        - return an empty dictionary if `empty_if_missing` is True
 81        - raise a ValueError if `empty_if_missing` is False
 82        """
 83        if self.config_dict is not None:
 84            return self.config_dict
 85
 86        if self.config_path is not None:
 87            config_path = self.config_path
 88            if not config_path.is_absolute():
 89                # We usually receive a relative path here. Let's resolve it.
 90                config_path = (connector_root / self.config_path).resolve().absolute()
 91
 92            return cast(
 93                dict[str, Any],
 94                yaml.safe_load(config_path.read_text()),
 95            )
 96
 97        if empty_if_missing:
 98            return {}
 99
100        raise ValueError("No config dictionary or path provided.")
101
102    @property
103    def expected_outcome(self) -> ExpectedOutcome:
104        """Whether the test scenario expects an exception to be raised.
105
106        Returns True if the scenario expects an exception, False if it does not,
107        and None if there is no set expectation.
108        """
109        return ExpectedOutcome.from_status_str(self.status)
110
111    @property
112    def id(self) -> str:
113        """Return a unique identifier for the test scenario.
114
115        This is used by PyTest to identify the test scenario.
116        """
117        if self._id:
118            return self._id
119
120        if self.config_path:
121            return self.config_path.stem
122
123        return str(hash(self))
124
125    def __str__(self) -> str:
126        return f"'{self.id}' Test Scenario"
127
128    @contextmanager
129    def with_temp_config_file(
130        self,
131        connector_root: Path,
132    ) -> Generator[Path, None, None]:
133        """Yield a temporary JSON file path containing the config dict and delete it on exit."""
134        config = self.get_config_dict(
135            empty_if_missing=True,
136            connector_root=connector_root,
137        )
138        with tempfile.NamedTemporaryFile(
139            prefix="config-",
140            suffix=".json",
141            mode="w",
142            delete=False,  # Don't fail if cannot delete the file on exit
143            encoding="utf-8",
144        ) as temp_file:
145            temp_file.write(json.dumps(config))
146            temp_file.flush()
147            # Allow the file to be read by other processes
148            temp_path = Path(temp_file.name)
149            temp_path.chmod(temp_path.stat().st_mode | 0o444)
150            yield temp_path
151
152        # attempt cleanup, ignore errors
153        with suppress(OSError):
154            temp_path.unlink()
155
156    def without_expected_outcome(self) -> ConnectorTestScenario:
157        """Return a copy of the scenario that does not expect failure or success.
158
159        This is useful when running multiple steps, to defer the expectations to a later step.
160        """
161        return ConnectorTestScenario(
162            **self.model_dump(exclude={"status"}),
163        )
164
165    def with_expecting_failure(self) -> ConnectorTestScenario:
166        """Return a copy of the scenario that expects failure.
167
168        This is useful when deriving new scenarios from existing ones.
169        """
170        if self.status == "failed":
171            return self
172
173        return ConnectorTestScenario(
174            **self.model_dump(exclude={"status"}),
175            status="failed",
176        )
177
178    def with_expecting_success(self) -> ConnectorTestScenario:
179        """Return a copy of the scenario that expects success.
180
181        This is useful when deriving new scenarios from existing ones.
182        """
183        if self.status == "succeed":
184            return self
185
186        return ConnectorTestScenario(
187            **self.model_dump(exclude={"status"}),
188            status="succeed",
189        )
190
191    @property
192    def requires_creds(self) -> bool:
193        """Return True if the scenario requires credentials to run."""
194        return bool(self.config_path and "secrets" in self.config_path.parts)

40    class AcceptanceTestExpectRecords(BaseModel):
41        path: Path
42        exact_order: bool = False

44    class AcceptanceTestFileTypes(BaseModel):
45        skip_test: bool
46        bypass_reason: str

48    class AcceptanceTestEmptyStream(BaseModel):
49        name: str
50        bypass_reason: str | None = None
51
52        # bypass reason does not affect equality
53        def __hash__(self) -> int:
54            return hash(self.name)

16class ExpectedOutcome(Enum):
17    """Enum to represent the expected outcome of a test scenario.
18
19    Class supports comparisons to a boolean or None.
20    """
21
22    EXPECT_EXCEPTION = auto()
23    EXPECT_SUCCESS = auto()
24    ALLOW_ANY = auto()
25
26    @classmethod
27    def from_status_str(cls, status: str | None) -> ExpectedOutcome:
28        """Convert a status string to an ExpectedOutcome."""
29        if status is None:
30            return ExpectedOutcome.ALLOW_ANY
31
32        try:
33            return {
34                "succeed": ExpectedOutcome.EXPECT_SUCCESS,
35                "failed": ExpectedOutcome.EXPECT_EXCEPTION,
36                "exception": ExpectedOutcome.EXPECT_EXCEPTION,  # same as 'failed'
37            }[status]
38        except KeyError as ex:
39            raise ValueError(
40                f"Invalid status '{status}'. Expected 'succeed', 'failed', or 'exception'.",
41            ) from ex
42
43    @classmethod
44    def from_expecting_exception_bool(cls, expecting_exception: bool | None) -> ExpectedOutcome:
45        """Convert a boolean indicating whether an exception is expected to an ExpectedOutcome."""
46        if expecting_exception is None:
47            # Align with legacy behavior where default would be 'False' (no exception expected)
48            return ExpectedOutcome.EXPECT_SUCCESS
49
50        return (
51            ExpectedOutcome.EXPECT_EXCEPTION
52            if expecting_exception
53            else ExpectedOutcome.EXPECT_SUCCESS
54        )
55
56    def expect_exception(self) -> bool:
57        """Return whether the expectation is that an exception should be raised."""
58        return self == ExpectedOutcome.EXPECT_EXCEPTION
59
60    def expect_success(self) -> bool:
61        """Return whether the expectation is that the test should succeed without exceptions."""
62        return self == ExpectedOutcome.EXPECT_SUCCESS