From c4e6ddbd38187ca70aff9039e6410cf3430a403f Mon Sep 17 00:00:00 2001
From: Aaron Steers <aj@airbyte.io>
Date: Fri, 6 Jun 2025 11:07:26 -0700
Subject: [PATCH 01/14] test: remove expected-to-fail tests from fast standard
 tests (resolves false-positive failures)

---
 .../test/standard_tests/connector_base.py     | 25 +++++++++++++------
 1 file changed, 18 insertions(+), 7 deletions(-)

diff --git a/airbyte_cdk/test/standard_tests/connector_base.py b/airbyte_cdk/test/standard_tests/connector_base.py
index 394028247..d78dbbf80 100644
--- a/airbyte_cdk/test/standard_tests/connector_base.py
+++ b/airbyte_cdk/test/standard_tests/connector_base.py
@@ -163,13 +163,24 @@ def get_scenarios(
             ):
                 continue
 
-            test_scenarios.extend(
-                [
-                    ConnectorTestScenario.model_validate(test)
-                    for test in all_tests_config["acceptance_tests"][category]["tests"]
-                    if "config_path" in test and "iam_role" not in test["config_path"]
-                ]
-            )
+            for test in all_tests_config["acceptance_tests"][category]["tests"]:
+                scenario = ConnectorTestScenario.model_validate(test)
+
+                if "config_path" in test and "iam_role" in test["config_path"]:
+                    # We skip iam_role tests for now, as they are not supported in the test suite.
+                    continue
+
+                if scenario.expect_exception:
+                    # For now, we skip tests that are expected to fail.
+                    # This is because they create false-positives in the test suite
+                    # if they fail later than expected.
+                    continue
+
+                if scenario.config_path in [s.config_path for s in test_scenarios]:
+                    # Skip duplicate scenarios based on config_path
+                    continue
+
+                test_scenarios.append(scenario)
 
         connector_root = cls.get_connector_root_dir().absolute()
         for test in test_scenarios:

From c9323151a02d6388b986c9440ba251d4d53cd8ff Mon Sep 17 00:00:00 2001
From: "Aaron (\"AJ\") Steers" <aj@airbyte.io>
Date: Fri, 6 Jun 2025 11:09:04 -0700
Subject: [PATCH 02/14] Update
 airbyte_cdk/test/standard_tests/connector_base.py

---
 airbyte_cdk/test/standard_tests/connector_base.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/airbyte_cdk/test/standard_tests/connector_base.py b/airbyte_cdk/test/standard_tests/connector_base.py
index d78dbbf80..34079b78c 100644
--- a/airbyte_cdk/test/standard_tests/connector_base.py
+++ b/airbyte_cdk/test/standard_tests/connector_base.py
@@ -176,7 +176,7 @@ def get_scenarios(
                     # if they fail later than expected.
                     continue
 
-                if scenario.config_path in [s.config_path for s in test_scenarios]:
+                if scenario.config_path and scenario.config_path in [s.config_path for s in test_scenarios]:
                     # Skip duplicate scenarios based on config_path
                     continue
 

From 0ef7a334bf61d928d5352f8445f1ab21af33a158 Mon Sep 17 00:00:00 2001
From: Aaron Steers <aj@airbyte.io>
Date: Fri, 6 Jun 2025 15:09:59 -0700
Subject: [PATCH 03/14] fix conditions for failure, add "with_()" and
 "without_()" morph methods

---
 .../test/standard_tests/_job_runner.py        |  6 ++-
 .../test/standard_tests/connector_base.py     | 16 ++++----
 .../standard_tests/declarative_sources.py     |  7 +++-
 .../test/standard_tests/models/scenario.py    | 41 ++++++++++++++++++-
 .../test/standard_tests/source_base.py        | 20 +++++----
 5 files changed, 69 insertions(+), 21 deletions(-)

diff --git a/airbyte_cdk/test/standard_tests/_job_runner.py b/airbyte_cdk/test/standard_tests/_job_runner.py
index ad8316d78..5bec9ba9f 100644
--- a/airbyte_cdk/test/standard_tests/_job_runner.py
+++ b/airbyte_cdk/test/standard_tests/_job_runner.py
@@ -58,6 +58,7 @@ def run_test_job(
     connector: IConnector | type[IConnector] | Callable[[], IConnector],
     verb: Literal["spec", "read", "check", "discover"],
     *,
+    connector_root: Path,
     test_scenario: ConnectorTestScenario | None = None,
     catalog: ConfiguredAirbyteCatalog | dict[str, Any] | None = None,
 ) -> entrypoint_wrapper.EntrypointOutput:
@@ -84,7 +85,10 @@ def run_test_job(
         )
 
     args: list[str] = [verb]
-    config_dict = test_scenario.get_config_dict(empty_if_missing=True)
+    config_dict = test_scenario.get_config_dict(
+        empty_if_missing=True,
+        connector_root=connector_root,
+    )
     if config_dict and verb != "spec":
         # Write the config to a temp json file and pass the path to the file as an argument.
         config_path = (
diff --git a/airbyte_cdk/test/standard_tests/connector_base.py b/airbyte_cdk/test/standard_tests/connector_base.py
index 34079b78c..b7b891c63 100644
--- a/airbyte_cdk/test/standard_tests/connector_base.py
+++ b/airbyte_cdk/test/standard_tests/connector_base.py
@@ -116,6 +116,7 @@ def test_check(
             self.create_connector(scenario),
             "check",
             test_scenario=scenario,
+            connector_root=self.get_connector_root_dir(),
         )
         conn_status_messages: list[AirbyteMessage] = [
             msg for msg in result._messages if msg.type == Type.CONNECTION_STATUS
@@ -164,12 +165,16 @@ def get_scenarios(
                 continue
 
             for test in all_tests_config["acceptance_tests"][category]["tests"]:
-                scenario = ConnectorTestScenario.model_validate(test)
+                if "config_path" not in test:
+                    # Skip tests without a config_path
+                    continue
 
-                if "config_path" in test and "iam_role" in test["config_path"]:
+                if "iam_role" in test["config_path"]:
                     # We skip iam_role tests for now, as they are not supported in the test suite.
                     continue
 
+                scenario = ConnectorTestScenario.model_validate(test)
+
                 if scenario.expect_exception:
                     # For now, we skip tests that are expected to fail.
                     # This is because they create false-positives in the test suite
@@ -182,11 +187,4 @@ def get_scenarios(
 
                 test_scenarios.append(scenario)
 
-        connector_root = cls.get_connector_root_dir().absolute()
-        for test in test_scenarios:
-            if test.config_path:
-                test.config_path = connector_root / test.config_path
-            if test.configured_catalog_path:
-                test.configured_catalog_path = connector_root / test.configured_catalog_path
-
         return test_scenarios
diff --git a/airbyte_cdk/test/standard_tests/declarative_sources.py b/airbyte_cdk/test/standard_tests/declarative_sources.py
index f454a267f..53e002f12 100644
--- a/airbyte_cdk/test/standard_tests/declarative_sources.py
+++ b/airbyte_cdk/test/standard_tests/declarative_sources.py
@@ -78,7 +78,12 @@ def create_connector(
         config = {
             "__injected_manifest": manifest_dict,
         }
-        config.update(scenario.get_config_dict(empty_if_missing=True))
+        config.update(
+            scenario.get_config_dict(
+                empty_if_missing=True,
+                connector_root=cls.get_connector_root_dir(),
+            ),
+        )
 
         if cls.components_py_path and cls.components_py_path.exists():
             os.environ["AIRBYTE_ENABLE_UNSAFE_CODE"] = "true"
diff --git a/airbyte_cdk/test/standard_tests/models/scenario.py b/airbyte_cdk/test/standard_tests/models/scenario.py
index 0ace85d33..a18625a53 100644
--- a/airbyte_cdk/test/standard_tests/models/scenario.py
+++ b/airbyte_cdk/test/standard_tests/models/scenario.py
@@ -13,7 +13,7 @@
 from typing import Any, Literal, cast
 
 import yaml
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict
 
 
 class ConnectorTestScenario(BaseModel):
@@ -24,6 +24,8 @@ class ConnectorTestScenario(BaseModel):
     acceptance test configuration file.
     """
 
+    model_config = ConfigDict(frozen=True)
+
     class AcceptanceTestExpectRecords(BaseModel):
         path: Path
         exact_order: bool = False
@@ -46,6 +48,7 @@ class AcceptanceTestFileTypes(BaseModel):
     def get_config_dict(
         self,
         *,
+        connector_root: Path,
         empty_if_missing: bool,
     ) -> dict[str, Any]:
         """Return the config dictionary.
@@ -61,7 +64,15 @@ def get_config_dict(
             return self.config_dict
 
         if self.config_path is not None:
-            return cast(dict[str, Any], yaml.safe_load(self.config_path.read_text()))
+            config_path = self.config_path
+            if not config_path.is_absolute():
+                # We usually receive a relative path here. Let's resolve it.
+                config_path = (connector_root / self.config_path).resolve().absolute()
+
+            return cast(
+                dict[str, Any],
+                yaml.safe_load(config_path.read_text()),
+            )
 
         if empty_if_missing:
             return {}
@@ -83,3 +94,29 @@ def __str__(self) -> str:
             return f"'{self.config_path.name}' Test Scenario"
 
         return f"'{hash(self)}' Test Scenario"
+
+    def without_expecting_failure(self) -> ConnectorTestScenario:
+        """Return a copy of the scenario that does not expect failure.
+
+        This is useful when you need to run multiple steps and you
+        want to defer failure expectation for one or more steps.
+        """
+        if self.status != "failed":
+            return self
+
+        return ConnectorTestScenario(
+            **self.model_dump(exclude={"status"}),
+        )
+
+    def with_expecting_failure(self) -> ConnectorTestScenario:
+        """Return a copy of the scenario that expects failure.
+
+        This is useful when deriving new scenarios from existing ones.
+        """
+        if self.status == "failed":
+            return self
+
+        return ConnectorTestScenario(
+            **self.model_dump(exclude={"status"}),
+            status="failed",
+        )
diff --git a/airbyte_cdk/test/standard_tests/source_base.py b/airbyte_cdk/test/standard_tests/source_base.py
index a256fa04c..a0ecb2ce2 100644
--- a/airbyte_cdk/test/standard_tests/source_base.py
+++ b/airbyte_cdk/test/standard_tests/source_base.py
@@ -43,6 +43,7 @@ def test_check(
             self.create_connector(scenario),
             "check",
             test_scenario=scenario,
+            connector_root=self.get_connector_root_dir(),
         )
         conn_status_messages: list[AirbyteMessage] = [
             msg for msg in result._messages if msg.type == Type.CONNECTION_STATUS
@@ -61,6 +62,7 @@ def test_discover(
         run_test_job(
             self.create_connector(scenario),
             "discover",
+            connector_root=self.get_connector_root_dir(),
             test_scenario=scenario,
         )
 
@@ -80,6 +82,7 @@ def test_spec(self) -> None:
             verb="spec",
             test_scenario=None,
             connector=self.create_connector(scenario=None),
+            connector_root=self.get_connector_root_dir(),
         )
         # If an error occurs, it will be raised above.
 
@@ -102,10 +105,11 @@ def test_basic_read(
         discover_result = run_test_job(
             self.create_connector(scenario),
             "discover",
-            test_scenario=scenario,
+            connector_root=self.get_connector_root_dir(),
+            test_scenario=scenario.without_expecting_failure(),
         )
-        if scenario.expect_exception:
-            assert discover_result.errors, "Expected exception but got none."
+        if scenario.expect_exception and discover_result.errors:
+            # Failed as expected; we're done.
             return
 
         configured_catalog = ConfiguredAirbyteCatalog(
@@ -122,6 +126,7 @@ def test_basic_read(
             self.create_connector(scenario),
             "read",
             test_scenario=scenario,
+            connector_root=self.get_connector_root_dir(),
             catalog=configured_catalog,
         )
 
@@ -149,15 +154,14 @@ def test_fail_read_with_bad_catalog(
                     ),
                     sync_mode="INVALID",  # type: ignore [reportArgumentType]
                     destination_sync_mode="INVALID",  # type: ignore [reportArgumentType]
-                )
-            ]
+                ),
+            ],
         )
-        # Set expected status to "failed" to ensure the test fails if the connector.
-        scenario.status = "failed"
         result: entrypoint_wrapper.EntrypointOutput = run_test_job(
             self.create_connector(scenario),
             "read",
-            test_scenario=scenario,
+            connector_root=self.get_connector_root_dir(),
+            test_scenario=scenario.with_expecting_failure(),  # Expect failure due to bad catalog
             catalog=asdict(invalid_configured_catalog),
         )
         assert result.errors, "Expected errors but got none."

From c754895a3b51513093b0d6020dcb3494236c7ddd Mon Sep 17 00:00:00 2001
From: "Aaron (\"AJ\") Steers" <aj@airbyte.io>
Date: Fri, 6 Jun 2025 15:22:23 -0700
Subject: [PATCH 04/14] Update
 airbyte_cdk/test/standard_tests/models/scenario.py

---
 airbyte_cdk/test/standard_tests/models/scenario.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/airbyte_cdk/test/standard_tests/models/scenario.py b/airbyte_cdk/test/standard_tests/models/scenario.py
index a18625a53..75c005fc0 100644
--- a/airbyte_cdk/test/standard_tests/models/scenario.py
+++ b/airbyte_cdk/test/standard_tests/models/scenario.py
@@ -24,6 +24,8 @@ class ConnectorTestScenario(BaseModel):
     acceptance test configuration file.
     """
 
+    # Allows the class to be hashable, which PyTest will require
+    # when we use to parameterize tests.
     model_config = ConfigDict(frozen=True)
 
     class AcceptanceTestExpectRecords(BaseModel):

From 41b0db1ccf334832aaf35b028960bce963d1d38b Mon Sep 17 00:00:00 2001
From: Aaron Steers <aj@airbyte.io>
Date: Fri, 6 Jun 2025 16:11:59 -0700
Subject: [PATCH 05/14] don't skip failure conditions

---
 airbyte_cdk/test/standard_tests/connector_base.py | 6 ------
 1 file changed, 6 deletions(-)

diff --git a/airbyte_cdk/test/standard_tests/connector_base.py b/airbyte_cdk/test/standard_tests/connector_base.py
index b7b891c63..981168489 100644
--- a/airbyte_cdk/test/standard_tests/connector_base.py
+++ b/airbyte_cdk/test/standard_tests/connector_base.py
@@ -175,12 +175,6 @@ def get_scenarios(
 
                 scenario = ConnectorTestScenario.model_validate(test)
 
-                if scenario.expect_exception:
-                    # For now, we skip tests that are expected to fail.
-                    # This is because they create false-positives in the test suite
-                    # if they fail later than expected.
-                    continue
-
                 if scenario.config_path and scenario.config_path in [s.config_path for s in test_scenarios]:
                     # Skip duplicate scenarios based on config_path
                     continue

From 5eefef7e5464bf29881149a5a672733c8c661b43 Mon Sep 17 00:00:00 2001
From: Aaron Steers <aj@airbyte.io>
Date: Fri, 6 Jun 2025 16:12:51 -0700
Subject: [PATCH 06/14] fix: classmethod override

---
 airbyte_cdk/test/standard_tests/util.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/airbyte_cdk/test/standard_tests/util.py b/airbyte_cdk/test/standard_tests/util.py
index 58ae19d85..f391d99ca 100644
--- a/airbyte_cdk/test/standard_tests/util.py
+++ b/airbyte_cdk/test/standard_tests/util.py
@@ -67,7 +67,7 @@ def create_connector_test_suite(
         )
 
     subclass_overrides: dict[str, Any] = {
-        "get_connector_root_dir": lambda: connector_directory,
+        "get_connector_root_dir": classmethod(lambda cls: connector_directory),
     }
 
     TestSuiteAuto = type(

From 76aa0e7b395e74afcf9a1a8b3856601cd295b46f Mon Sep 17 00:00:00 2001
From: Aaron Steers <aj@airbyte.io>
Date: Fri, 6 Jun 2025 16:17:13 -0700
Subject: [PATCH 07/14] format-fix

---
 airbyte_cdk/test/standard_tests/connector_base.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/airbyte_cdk/test/standard_tests/connector_base.py b/airbyte_cdk/test/standard_tests/connector_base.py
index 981168489..ef69180e9 100644
--- a/airbyte_cdk/test/standard_tests/connector_base.py
+++ b/airbyte_cdk/test/standard_tests/connector_base.py
@@ -175,7 +175,9 @@ def get_scenarios(
 
                 scenario = ConnectorTestScenario.model_validate(test)
 
-                if scenario.config_path and scenario.config_path in [s.config_path for s in test_scenarios]:
+                if scenario.config_path and scenario.config_path in [
+                    s.config_path for s in test_scenarios
+                ]:
                     # Skip duplicate scenarios based on config_path
                     continue
 

From 344a495f99f4a2e3c201a01aaddd51611f83903a Mon Sep 17 00:00:00 2001
From: Aaron Steers <aj@airbyte.io>
Date: Fri, 6 Jun 2025 18:05:01 -0700
Subject: [PATCH 08/14] tri-state ExpectedOutcome

---
 airbyte_cdk/test/entrypoint_wrapper.py        | 31 ++++++---
 .../test/standard_tests/_job_runner.py        | 15 ++--
 .../test/standard_tests/models/scenario.py    | 68 ++++++++++++++++---
 .../test/standard_tests/source_base.py        |  4 +-
 airbyte_cdk/test/utils/reading.py             |  5 +-
 .../declarative/file/test_file_stream.py      | 15 ++--
 .../test_resumable_full_refresh.py            |  3 +-
 unit_tests/test/test_entrypoint_wrapper.py    | 11 ++-
 8 files changed, 113 insertions(+), 39 deletions(-)

diff --git a/airbyte_cdk/test/entrypoint_wrapper.py b/airbyte_cdk/test/entrypoint_wrapper.py
index 79c328203..4af286be5 100644
--- a/airbyte_cdk/test/entrypoint_wrapper.py
+++ b/airbyte_cdk/test/entrypoint_wrapper.py
@@ -24,6 +24,7 @@
 from typing import Any, List, Mapping, Optional, Union
 
 import orjson
+from langsmith import expect
 from pydantic import ValidationError as V2ValidationError
 from serpyco_rs import SchemaValidationError
 
@@ -44,6 +45,7 @@
     Type,
 )
 from airbyte_cdk.sources import Source
+from airbyte_cdk.test.standard_tests.models.scenario import ExpectedOutcome
 
 
 class EntrypointOutput:
@@ -157,7 +159,9 @@ def is_not_in_logs(self, pattern: str) -> bool:
 
 
 def _run_command(
-    source: Source, args: List[str], expecting_exception: bool = False
+    source: Source,
+    args: list[str],
+    expected_outcome: ExpectedOutcome,
 ) -> EntrypointOutput:
     log_capture_buffer = StringIO()
     stream_handler = logging.StreamHandler(log_capture_buffer)
@@ -175,27 +179,29 @@ def _run_command(
         for message in source_entrypoint.run(parsed_args):
             messages.append(message)
     except Exception as exception:
-        if not expecting_exception:
+        if expected_outcome.expect_success():
             print("Printing unexpected error from entrypoint_wrapper")
             print("".join(traceback.format_exception(None, exception, exception.__traceback__)))
+
         uncaught_exception = exception
 
     captured_logs = log_capture_buffer.getvalue().split("\n")[:-1]
 
     parent_logger.removeHandler(stream_handler)
 
-    return EntrypointOutput(messages + captured_logs, uncaught_exception)
+    return EntrypointOutput(messages + captured_logs, uncaught_exception=uncaught_exception)
 
 
 def discover(
     source: Source,
     config: Mapping[str, Any],
-    expecting_exception: bool = False,
+    *,
+    expected_outcome: ExpectedOutcome = ExpectedOutcome.EXPECT_SUCCESS,
 ) -> EntrypointOutput:
     """
     config must be json serializable
-    :param expecting_exception: By default if there is an uncaught exception, the exception will be printed out. If this is expected, please
-        provide expecting_exception=True so that the test output logs are cleaner
+    :param expected_outcome: By default if there is an uncaught exception, the exception will be printed out. If this is expected, please
+        provide `expected_outcome=ExpectedOutcome.EXPECT_FAILURE` so that the test output logs are cleaner
     """
 
     with tempfile.TemporaryDirectory() as tmp_directory:
@@ -203,7 +209,9 @@ def discover(
         config_file = make_file(tmp_directory_path / "config.json", config)
 
         return _run_command(
-            source, ["discover", "--config", config_file, "--debug"], expecting_exception
+            source,
+            ["discover", "--config", config_file, "--debug"],
+            expected_outcome=expected_outcome,
         )
 
 
@@ -212,13 +220,14 @@ def read(
     config: Mapping[str, Any],
     catalog: ConfiguredAirbyteCatalog,
     state: Optional[List[AirbyteStateMessage]] = None,
-    expecting_exception: bool = False,
+    *,
+    expected_outcome: ExpectedOutcome = ExpectedOutcome.EXPECT_SUCCESS,
 ) -> EntrypointOutput:
     """
     config and state must be json serializable
 
-    :param expecting_exception: By default if there is an uncaught exception, the exception will be printed out. If this is expected, please
-        provide expecting_exception=True so that the test output logs are cleaner
+    :param expected_outcome: By default if there is an uncaught exception, the exception will be printed out. If this is expected, please
+        provide `expected_outcome=ExpectedOutcome.EXPECT_FAILURE` so that the test output logs are cleaner.
     """
     with tempfile.TemporaryDirectory() as tmp_directory:
         tmp_directory_path = Path(tmp_directory)
@@ -245,7 +254,7 @@ def read(
                 ]
             )
 
-        return _run_command(source, args, expecting_exception)
+        return _run_command(source, args, expected_outcome=expected_outcome)
 
 
 def make_file(
diff --git a/airbyte_cdk/test/standard_tests/_job_runner.py b/airbyte_cdk/test/standard_tests/_job_runner.py
index 5bec9ba9f..c6b951eb4 100644
--- a/airbyte_cdk/test/standard_tests/_job_runner.py
+++ b/airbyte_cdk/test/standard_tests/_job_runner.py
@@ -122,9 +122,9 @@ def run_test_job(
     result: entrypoint_wrapper.EntrypointOutput = entrypoint_wrapper._run_command(  # noqa: SLF001  # Non-public API
         source=connector_obj,  # type: ignore [arg-type]
         args=args,
-        expecting_exception=test_scenario.expect_exception,
+        expected_outcome=test_scenario.expected_outcome,
     )
-    if result.errors and not test_scenario.expect_exception:
+    if result.errors and test_scenario.expected_outcome.expect_success():
         raise AssertionError(
             f"Expected no errors but got {len(result.errors)}: \n" + _errors_to_str(result)
         )
@@ -139,7 +139,7 @@ def run_test_job(
             + "\n".join([str(msg) for msg in result.connection_status_messages])
             + _errors_to_str(result)
         )
-        if test_scenario.expect_exception:
+        if test_scenario.expected_outcome.expect_exception():
             conn_status = result.connection_status_messages[0].connectionStatus
             assert conn_status, (
                 "Expected CONNECTION_STATUS message to be present. Got: \n"
@@ -153,14 +153,15 @@ def run_test_job(
         return result
 
     # For all other verbs, we assert check that an exception is raised (or not).
-    if test_scenario.expect_exception:
+    if test_scenario.expected_outcome.expect_exception():
         if not result.errors:
             raise AssertionError("Expected exception but got none.")
 
         return result
 
-    assert not result.errors, (
-        f"Expected no errors but got {len(result.errors)}: \n" + _errors_to_str(result)
-    )
+    if test_scenario.expected_outcome.expect_success():
+        assert not result.errors, (
+            f"Expected no errors but got {len(result.errors)}: \n" + _errors_to_str(result)
+        )
 
     return result
diff --git a/airbyte_cdk/test/standard_tests/models/scenario.py b/airbyte_cdk/test/standard_tests/models/scenario.py
index 75c005fc0..088a24f42 100644
--- a/airbyte_cdk/test/standard_tests/models/scenario.py
+++ b/airbyte_cdk/test/standard_tests/models/scenario.py
@@ -9,13 +9,47 @@
 
 from __future__ import annotations
 
-from pathlib import Path
+from enum import Enum, auto
+from pathlib import Path  # noqa: TC003  # Pydantic needs this (don't move to 'if typing' block)
 from typing import Any, Literal, cast
 
 import yaml
 from pydantic import BaseModel, ConfigDict
 
 
+class ExpectedOutcome(Enum):
+    """Enum to represent the expected outcome of a test scenario.
+
+    Class supports comparisons to a boolean or None.
+    """
+
+    EXPECT_EXCEPTION = auto()
+    EXPECT_SUCCESS = auto()
+    ALLOW_ANY = auto()
+
+    @classmethod
+    def from_status_str(cls, status: str | None) -> ExpectedOutcome:
+        """Convert a status string to an ExpectedOutcome."""
+        if status is None:
+            return ExpectedOutcome.ALLOW_ANY
+
+        try:
+            return {
+                "succeed": ExpectedOutcome.EXPECT_SUCCESS,
+                "failed": ExpectedOutcome.EXPECT_EXCEPTION,
+            }[status]
+        except KeyError as ex:
+            raise ValueError(f"Invalid status '{status}'. Expected 'succeed' or 'failed'.") from ex
+
+    def expect_exception(self) -> bool:
+        """Return whether the expectation is that an exception should be raised."""
+        return self == ExpectedOutcome.EXPECT_EXCEPTION
+
+    def expect_success(self) -> bool:
+        """Return whether the expectation is that the test should succeed without exceptions."""
+        return self == ExpectedOutcome.EXPECT_SUCCESS
+
+
 class ConnectorTestScenario(BaseModel):
     """Acceptance test scenario, as a Pydantic model.
 
@@ -82,8 +116,13 @@ def get_config_dict(
         raise ValueError("No config dictionary or path provided.")
 
     @property
-    def expect_exception(self) -> bool:
-        return self.status and self.status == "failed" or False
+    def expected_outcome(self) -> ExpectedOutcome:
+        """Whether the test scenario expects an exception to be raised.
+
+        Returns True if the scenario expects an exception, False if it does not,
+        and None if there is no set expectation.
+        """
+        return ExpectedOutcome.from_status_str(self.status)
 
     @property
     def instance_name(self) -> str:
@@ -97,15 +136,11 @@ def __str__(self) -> str:
 
         return f"'{hash(self)}' Test Scenario"
 
-    def without_expecting_failure(self) -> ConnectorTestScenario:
-        """Return a copy of the scenario that does not expect failure.
+    def without_expected_outcome(self) -> ConnectorTestScenario:
+        """Return a copy of the scenario that does not expect failure or success.
 
-        This is useful when you need to run multiple steps and you
-        want to defer failure expectation for one or more steps.
+        This is useful when running multiple steps, to defer the expectations to a later step.
         """
-        if self.status != "failed":
-            return self
-
         return ConnectorTestScenario(
             **self.model_dump(exclude={"status"}),
         )
@@ -122,3 +157,16 @@ def with_expecting_failure(self) -> ConnectorTestScenario:
             **self.model_dump(exclude={"status"}),
             status="failed",
         )
+
+    def with_expecting_success(self) -> ConnectorTestScenario:
+        """Return a copy of the scenario that expects success.
+
+        This is useful when deriving new scenarios from existing ones.
+        """
+        if self.status == "succeed":
+            return self
+
+        return ConnectorTestScenario(
+            **self.model_dump(exclude={"status"}),
+            status="succeed",
+        )
diff --git a/airbyte_cdk/test/standard_tests/source_base.py b/airbyte_cdk/test/standard_tests/source_base.py
index a0ecb2ce2..754565095 100644
--- a/airbyte_cdk/test/standard_tests/source_base.py
+++ b/airbyte_cdk/test/standard_tests/source_base.py
@@ -106,9 +106,9 @@ def test_basic_read(
             self.create_connector(scenario),
             "discover",
             connector_root=self.get_connector_root_dir(),
-            test_scenario=scenario.without_expecting_failure(),
+            test_scenario=scenario.without_expected_outcome(),
         )
-        if scenario.expect_exception and discover_result.errors:
+        if scenario.expected_outcome.expect_exception() and discover_result.errors:
             # Failed as expected; we're done.
             return
 
diff --git a/airbyte_cdk/test/utils/reading.py b/airbyte_cdk/test/utils/reading.py
index 2d89cb870..b99903abd 100644
--- a/airbyte_cdk/test/utils/reading.py
+++ b/airbyte_cdk/test/utils/reading.py
@@ -6,6 +6,7 @@
 from airbyte_cdk.models import AirbyteStateMessage, ConfiguredAirbyteCatalog, SyncMode
 from airbyte_cdk.test.catalog_builder import CatalogBuilder
 from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput, read
+from airbyte_cdk.test.standard_tests.models.scenario import ExpectedOutcome
 
 
 def catalog(stream_name: str, sync_mode: SyncMode) -> ConfiguredAirbyteCatalog:
@@ -19,8 +20,8 @@ def read_records(
     stream_name: str,
     sync_mode: SyncMode,
     state: Optional[List[AirbyteStateMessage]] = None,
-    expecting_exception: bool = False,
+    expected_outcome: ExpectedOutcome = ExpectedOutcome.EXPECT_SUCCESS,
 ) -> EntrypointOutput:
     """Read records from a stream."""
     _catalog = catalog(stream_name, sync_mode)
-    return read(source, config, _catalog, state, expecting_exception)
+    return read(source, config, _catalog, state, expected_outcome=expected_outcome)
diff --git a/unit_tests/sources/declarative/file/test_file_stream.py b/unit_tests/sources/declarative/file/test_file_stream.py
index 7b645f540..0ec5d86b1 100644
--- a/unit_tests/sources/declarative/file/test_file_stream.py
+++ b/unit_tests/sources/declarative/file/test_file_stream.py
@@ -17,6 +17,7 @@
 from airbyte_cdk.test.entrypoint_wrapper import read as entrypoint_read
 from airbyte_cdk.test.mock_http import HttpMocker, HttpRequest, HttpResponse
 from airbyte_cdk.test.mock_http.response_builder import find_binary_response, find_template
+from airbyte_cdk.test.standard_tests.models.scenario import ExpectedOutcome
 from airbyte_cdk.test.state_builder import StateBuilder
 
 
@@ -53,8 +54,9 @@ def read(
     config_builder: ConfigBuilder,
     catalog: ConfiguredAirbyteCatalog,
     state_builder: Optional[StateBuilder] = None,
-    expecting_exception: bool = False,
     yaml_file: Optional[str] = None,
+    *,
+    expected_outcome: ExpectedOutcome = ExpectedOutcome.EXPECT_SUCCESS,
 ) -> EntrypointOutput:
     config = config_builder.build()
     state = state_builder.build() if state_builder else StateBuilder().build()
@@ -63,14 +65,19 @@ def read(
         config,
         catalog,
         state,
-        expecting_exception,
+        expected_outcome=expected_outcome,
     )
 
 
-def discover(config_builder: ConfigBuilder, expecting_exception: bool = False) -> EntrypointOutput:
+def discover(
+    config_builder: ConfigBuilder,
+    expected_outcome: ExpectedOutcome = ExpectedOutcome.EXPECT_SUCCESS,
+) -> EntrypointOutput:
     config = config_builder.build()
     return entrypoint_discover(
-        _source(CatalogBuilder().build(), config), config, expecting_exception
+        _source(CatalogBuilder().build(), config),
+        config,
+        expected_outcome=expected_outcome,
     )
 
 
diff --git a/unit_tests/sources/mock_server_tests/test_resumable_full_refresh.py b/unit_tests/sources/mock_server_tests/test_resumable_full_refresh.py
index 5ba58e384..3f624501d 100644
--- a/unit_tests/sources/mock_server_tests/test_resumable_full_refresh.py
+++ b/unit_tests/sources/mock_server_tests/test_resumable_full_refresh.py
@@ -27,6 +27,7 @@
     create_record_builder,
     create_response_builder,
 )
+from airbyte_cdk.test.standard_tests.models.scenario import ExpectedOutcome
 from airbyte_cdk.test.state_builder import StateBuilder
 from unit_tests.sources.mock_server_tests.mock_source_fixture import SourceFixture
 from unit_tests.sources.mock_server_tests.test_helpers import (
@@ -344,7 +345,7 @@ def test_resumable_full_refresh_failure(self, http_mocker):
             source,
             config=config,
             catalog=_create_catalog([("justice_songs", SyncMode.full_refresh, {})]),
-            expecting_exception=True,
+            expected_outcome=ExpectedOutcome.EXPECT_EXCEPTION,
         )
 
         status_messages = actual_messages.get_stream_statuses("justice_songs")
diff --git a/unit_tests/test/test_entrypoint_wrapper.py b/unit_tests/test/test_entrypoint_wrapper.py
index a8d02fca9..bbb0b3e55 100644
--- a/unit_tests/test/test_entrypoint_wrapper.py
+++ b/unit_tests/test/test_entrypoint_wrapper.py
@@ -32,6 +32,7 @@
 )
 from airbyte_cdk.sources.abstract_source import AbstractSource
 from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput, discover, read
+from airbyte_cdk.test.standard_tests.models.scenario import ExpectedOutcome
 from airbyte_cdk.test.state_builder import StateBuilder
 
 
@@ -229,7 +230,7 @@ def test_given_unexpected_exception_when_discover_then_print(self, entrypoint, p
     @patch("airbyte_cdk.test.entrypoint_wrapper.AirbyteEntrypoint")
     def test_given_expected_exception_when_discover_then_do_not_print(self, entrypoint, print_mock):
         entrypoint.return_value.run.side_effect = ValueError("This error should not be printed")
-        discover(self._a_source, _A_CONFIG, expecting_exception=True)
+        discover(self._a_source, _A_CONFIG, expected_outcome=ExpectedOutcome.EXPECT_EXCEPTION)
         assert print_mock.call_count == 0
 
     @patch("airbyte_cdk.test.entrypoint_wrapper.AirbyteEntrypoint")
@@ -380,7 +381,13 @@ def test_given_unexpected_exception_when_read_then_print(self, entrypoint, print
     @patch("airbyte_cdk.test.entrypoint_wrapper.AirbyteEntrypoint")
     def test_given_expected_exception_when_read_then_do_not_print(self, entrypoint, print_mock):
         entrypoint.return_value.run.side_effect = ValueError("This error should not be printed")
-        read(self._a_source, _A_CONFIG, _A_CATALOG, _A_STATE, expecting_exception=True)
+        read(
+            self._a_source,
+            _A_CONFIG,
+            _A_CATALOG,
+            _A_STATE,
+            expected_outcome=ExpectedOutcome.EXPECT_EXCEPTION,
+        )
         assert print_mock.call_count == 0
 
     @patch("airbyte_cdk.test.entrypoint_wrapper.AirbyteEntrypoint")

From 1ac503a47bae966add9371697e039b135dbb4356 Mon Sep 17 00:00:00 2001
From: "Aaron (\"AJ\") Steers" <aj@airbyte.io>
Date: Fri, 6 Jun 2025 18:12:36 -0700
Subject: [PATCH 09/14] Update airbyte_cdk/test/entrypoint_wrapper.py

---
 airbyte_cdk/test/entrypoint_wrapper.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/airbyte_cdk/test/entrypoint_wrapper.py b/airbyte_cdk/test/entrypoint_wrapper.py
index 4af286be5..dfdc0559b 100644
--- a/airbyte_cdk/test/entrypoint_wrapper.py
+++ b/airbyte_cdk/test/entrypoint_wrapper.py
@@ -24,7 +24,6 @@
 from typing import Any, List, Mapping, Optional, Union
 
 import orjson
-from langsmith import expect
 from pydantic import ValidationError as V2ValidationError
 from serpyco_rs import SchemaValidationError
 

From 845b8006d3a93b7b73a67e56e7b1982dd50d9184 Mon Sep 17 00:00:00 2001
From: Aaron Steers <aj@airbyte.io>
Date: Fri, 6 Jun 2025 18:52:25 -0700
Subject: [PATCH 10/14] fix circular import issue

---
 airbyte_cdk/test/entrypoint_wrapper.py        |  2 +-
 airbyte_cdk/test/models/__init__.py           | 10 +++++
 airbyte_cdk/test/models/outcome.py            | 45 +++++++++++++++++++
 .../{standard_tests => }/models/scenario.py   | 34 +-------------
 .../test/standard_tests/_job_runner.py        |  2 +-
 .../test/standard_tests/connector_base.py     |  4 +-
 .../standard_tests/declarative_sources.py     |  2 +-
 .../test/standard_tests/models/__init__.py    |  7 ---
 .../test/standard_tests/source_base.py        | 11 +++--
 airbyte_cdk/test/utils/reading.py             |  2 +-
 .../declarative/file/test_file_stream.py      |  2 +-
 .../test_resumable_full_refresh.py            |  2 +-
 unit_tests/test/test_entrypoint_wrapper.py    |  2 +-
 13 files changed, 72 insertions(+), 53 deletions(-)
 create mode 100644 airbyte_cdk/test/models/__init__.py
 create mode 100644 airbyte_cdk/test/models/outcome.py
 rename airbyte_cdk/test/{standard_tests => }/models/scenario.py (79%)
 delete mode 100644 airbyte_cdk/test/standard_tests/models/__init__.py

diff --git a/airbyte_cdk/test/entrypoint_wrapper.py b/airbyte_cdk/test/entrypoint_wrapper.py
index dfdc0559b..a5fc18985 100644
--- a/airbyte_cdk/test/entrypoint_wrapper.py
+++ b/airbyte_cdk/test/entrypoint_wrapper.py
@@ -44,7 +44,7 @@
     Type,
 )
 from airbyte_cdk.sources import Source
-from airbyte_cdk.test.standard_tests.models.scenario import ExpectedOutcome
+from airbyte_cdk.test.models.scenario import ExpectedOutcome
 
 
 class EntrypointOutput:
diff --git a/airbyte_cdk/test/models/__init__.py b/airbyte_cdk/test/models/__init__.py
new file mode 100644
index 000000000..70e6a3600
--- /dev/null
+++ b/airbyte_cdk/test/models/__init__.py
@@ -0,0 +1,10 @@
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+"""Models used for standard tests."""
+
+from airbyte_cdk.test.models.outcome import ExpectedOutcome
+from airbyte_cdk.test.models.scenario import ConnectorTestScenario
+
+__all__ = [
+    "ConnectorTestScenario",
+    "ExpectedOutcome",
+]
diff --git a/airbyte_cdk/test/models/outcome.py b/airbyte_cdk/test/models/outcome.py
new file mode 100644
index 000000000..bb1e81c8a
--- /dev/null
+++ b/airbyte_cdk/test/models/outcome.py
@@ -0,0 +1,45 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+"""Run acceptance tests in PyTest.
+
+These tests leverage the same `acceptance-test-config.yml` configuration files as the
+acceptance tests in CAT, but they run in PyTest instead of CAT. This allows us to run
+the acceptance tests in the same local environment as we are developing in, speeding
+up iteration cycles.
+"""
+
+from __future__ import annotations
+
+from enum import Enum, auto
+
+
+class ExpectedOutcome(Enum):
+    """Enum to represent the expected outcome of a test scenario.
+
+    Class supports comparisons to a boolean or None.
+    """
+
+    EXPECT_EXCEPTION = auto()
+    EXPECT_SUCCESS = auto()
+    ALLOW_ANY = auto()
+
+    @classmethod
+    def from_status_str(cls, status: str | None) -> ExpectedOutcome:
+        """Convert a status string to an ExpectedOutcome."""
+        if status is None:
+            return ExpectedOutcome.ALLOW_ANY
+
+        try:
+            return {
+                "succeed": ExpectedOutcome.EXPECT_SUCCESS,
+                "failed": ExpectedOutcome.EXPECT_EXCEPTION,
+            }[status]
+        except KeyError as ex:
+            raise ValueError(f"Invalid status '{status}'. Expected 'succeed' or 'failed'.") from ex
+
+    def expect_exception(self) -> bool:
+        """Return whether the expectation is that an exception should be raised."""
+        return self == ExpectedOutcome.EXPECT_EXCEPTION
+
+    def expect_success(self) -> bool:
+        """Return whether the expectation is that the test should succeed without exceptions."""
+        return self == ExpectedOutcome.EXPECT_SUCCESS
diff --git a/airbyte_cdk/test/standard_tests/models/scenario.py b/airbyte_cdk/test/models/scenario.py
similarity index 79%
rename from airbyte_cdk/test/standard_tests/models/scenario.py
rename to airbyte_cdk/test/models/scenario.py
index 088a24f42..0ac59498f 100644
--- a/airbyte_cdk/test/standard_tests/models/scenario.py
+++ b/airbyte_cdk/test/models/scenario.py
@@ -9,45 +9,13 @@
 
 from __future__ import annotations
 
-from enum import Enum, auto
 from pathlib import Path  # noqa: TC003  # Pydantic needs this (don't move to 'if typing' block)
 from typing import Any, Literal, cast
 
 import yaml
 from pydantic import BaseModel, ConfigDict
 
-
-class ExpectedOutcome(Enum):
-    """Enum to represent the expected outcome of a test scenario.
-
-    Class supports comparisons to a boolean or None.
-    """
-
-    EXPECT_EXCEPTION = auto()
-    EXPECT_SUCCESS = auto()
-    ALLOW_ANY = auto()
-
-    @classmethod
-    def from_status_str(cls, status: str | None) -> ExpectedOutcome:
-        """Convert a status string to an ExpectedOutcome."""
-        if status is None:
-            return ExpectedOutcome.ALLOW_ANY
-
-        try:
-            return {
-                "succeed": ExpectedOutcome.EXPECT_SUCCESS,
-                "failed": ExpectedOutcome.EXPECT_EXCEPTION,
-            }[status]
-        except KeyError as ex:
-            raise ValueError(f"Invalid status '{status}'. Expected 'succeed' or 'failed'.") from ex
-
-    def expect_exception(self) -> bool:
-        """Return whether the expectation is that an exception should be raised."""
-        return self == ExpectedOutcome.EXPECT_EXCEPTION
-
-    def expect_success(self) -> bool:
-        """Return whether the expectation is that the test should succeed without exceptions."""
-        return self == ExpectedOutcome.EXPECT_SUCCESS
+from airbyte_cdk.test.models.outcome import ExpectedOutcome
 
 
 class ConnectorTestScenario(BaseModel):
diff --git a/airbyte_cdk/test/standard_tests/_job_runner.py b/airbyte_cdk/test/standard_tests/_job_runner.py
index c6b951eb4..8f4174b1a 100644
--- a/airbyte_cdk/test/standard_tests/_job_runner.py
+++ b/airbyte_cdk/test/standard_tests/_job_runner.py
@@ -16,7 +16,7 @@
     Status,
 )
 from airbyte_cdk.test import entrypoint_wrapper
-from airbyte_cdk.test.standard_tests.models import (
+from airbyte_cdk.test.models import (
     ConnectorTestScenario,
 )
 
diff --git a/airbyte_cdk/test/standard_tests/connector_base.py b/airbyte_cdk/test/standard_tests/connector_base.py
index ef69180e9..073db0b26 100644
--- a/airbyte_cdk/test/standard_tests/connector_base.py
+++ b/airbyte_cdk/test/standard_tests/connector_base.py
@@ -20,10 +20,10 @@
     Type,
 )
 from airbyte_cdk.test import entrypoint_wrapper
-from airbyte_cdk.test.standard_tests._job_runner import IConnector, run_test_job
-from airbyte_cdk.test.standard_tests.models import (
+from airbyte_cdk.test.models import (
     ConnectorTestScenario,
 )
+from airbyte_cdk.test.standard_tests._job_runner import IConnector, run_test_job
 from airbyte_cdk.utils.connector_paths import (
     ACCEPTANCE_TEST_CONFIG,
     find_connector_root,
diff --git a/airbyte_cdk/test/standard_tests/declarative_sources.py b/airbyte_cdk/test/standard_tests/declarative_sources.py
index 53e002f12..18a2a5910 100644
--- a/airbyte_cdk/test/standard_tests/declarative_sources.py
+++ b/airbyte_cdk/test/standard_tests/declarative_sources.py
@@ -9,8 +9,8 @@
 from airbyte_cdk.sources.declarative.concurrent_declarative_source import (
     ConcurrentDeclarativeSource,
 )
+from airbyte_cdk.test.models import ConnectorTestScenario
 from airbyte_cdk.test.standard_tests._job_runner import IConnector
-from airbyte_cdk.test.standard_tests.models import ConnectorTestScenario
 from airbyte_cdk.test.standard_tests.source_base import SourceTestSuiteBase
 from airbyte_cdk.utils.connector_paths import MANIFEST_YAML
 
diff --git a/airbyte_cdk/test/standard_tests/models/__init__.py b/airbyte_cdk/test/standard_tests/models/__init__.py
deleted file mode 100644
index 13d67e16a..000000000
--- a/airbyte_cdk/test/standard_tests/models/__init__.py
+++ /dev/null
@@ -1,7 +0,0 @@
-from airbyte_cdk.test.standard_tests.models.scenario import (
-    ConnectorTestScenario,
-)
-
-__all__ = [
-    "ConnectorTestScenario",
-]
diff --git a/airbyte_cdk/test/standard_tests/source_base.py b/airbyte_cdk/test/standard_tests/source_base.py
index 754565095..f3cc1a50c 100644
--- a/airbyte_cdk/test/standard_tests/source_base.py
+++ b/airbyte_cdk/test/standard_tests/source_base.py
@@ -2,6 +2,7 @@
 """Base class for source test suites."""
 
 from dataclasses import asdict
+from typing import TYPE_CHECKING
 
 from airbyte_cdk.models import (
     AirbyteMessage,
@@ -12,14 +13,16 @@
     SyncMode,
     Type,
 )
-from airbyte_cdk.test import entrypoint_wrapper
+from airbyte_cdk.test.models import (
+    ConnectorTestScenario,
+)
 from airbyte_cdk.test.standard_tests._job_runner import run_test_job
 from airbyte_cdk.test.standard_tests.connector_base import (
     ConnectorTestSuiteBase,
 )
-from airbyte_cdk.test.standard_tests.models import (
-    ConnectorTestScenario,
-)
+
+if TYPE_CHECKING:
+    from airbyte_cdk.test import entrypoint_wrapper
 
 
 class SourceTestSuiteBase(ConnectorTestSuiteBase):
diff --git a/airbyte_cdk/test/utils/reading.py b/airbyte_cdk/test/utils/reading.py
index b99903abd..a0450af3e 100644
--- a/airbyte_cdk/test/utils/reading.py
+++ b/airbyte_cdk/test/utils/reading.py
@@ -6,7 +6,7 @@
 from airbyte_cdk.models import AirbyteStateMessage, ConfiguredAirbyteCatalog, SyncMode
 from airbyte_cdk.test.catalog_builder import CatalogBuilder
 from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput, read
-from airbyte_cdk.test.standard_tests.models.scenario import ExpectedOutcome
+from airbyte_cdk.test.models.outcome import ExpectedOutcome
 
 
 def catalog(stream_name: str, sync_mode: SyncMode) -> ConfiguredAirbyteCatalog:
diff --git a/unit_tests/sources/declarative/file/test_file_stream.py b/unit_tests/sources/declarative/file/test_file_stream.py
index 0ec5d86b1..410339029 100644
--- a/unit_tests/sources/declarative/file/test_file_stream.py
+++ b/unit_tests/sources/declarative/file/test_file_stream.py
@@ -17,7 +17,7 @@
 from airbyte_cdk.test.entrypoint_wrapper import read as entrypoint_read
 from airbyte_cdk.test.mock_http import HttpMocker, HttpRequest, HttpResponse
 from airbyte_cdk.test.mock_http.response_builder import find_binary_response, find_template
-from airbyte_cdk.test.standard_tests.models.scenario import ExpectedOutcome
+from airbyte_cdk.test.models.scenario import ExpectedOutcome
 from airbyte_cdk.test.state_builder import StateBuilder
 
 
diff --git a/unit_tests/sources/mock_server_tests/test_resumable_full_refresh.py b/unit_tests/sources/mock_server_tests/test_resumable_full_refresh.py
index 3f624501d..04e232083 100644
--- a/unit_tests/sources/mock_server_tests/test_resumable_full_refresh.py
+++ b/unit_tests/sources/mock_server_tests/test_resumable_full_refresh.py
@@ -27,7 +27,7 @@
     create_record_builder,
     create_response_builder,
 )
-from airbyte_cdk.test.standard_tests.models.scenario import ExpectedOutcome
+from airbyte_cdk.test.models.scenario import ExpectedOutcome
 from airbyte_cdk.test.state_builder import StateBuilder
 from unit_tests.sources.mock_server_tests.mock_source_fixture import SourceFixture
 from unit_tests.sources.mock_server_tests.test_helpers import (
diff --git a/unit_tests/test/test_entrypoint_wrapper.py b/unit_tests/test/test_entrypoint_wrapper.py
index bbb0b3e55..60e190c11 100644
--- a/unit_tests/test/test_entrypoint_wrapper.py
+++ b/unit_tests/test/test_entrypoint_wrapper.py
@@ -32,7 +32,7 @@
 )
 from airbyte_cdk.sources.abstract_source import AbstractSource
 from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput, discover, read
-from airbyte_cdk.test.standard_tests.models.scenario import ExpectedOutcome
+from airbyte_cdk.test.models.scenario import ExpectedOutcome
 from airbyte_cdk.test.state_builder import StateBuilder
 
 

From 0fc90b40f99b524f30da27f5840092549d676fcc Mon Sep 17 00:00:00 2001
From: Aaron Steers <aj@airbyte.io>
Date: Fri, 6 Jun 2025 20:57:51 -0700
Subject: [PATCH 11/14] restore older arg as deprecated

---
 airbyte_cdk/test/models/outcome.py | 12 ++++++++++++
 airbyte_cdk/test/utils/reading.py  | 10 +++++++++-
 2 files changed, 21 insertions(+), 1 deletion(-)

diff --git a/airbyte_cdk/test/models/outcome.py b/airbyte_cdk/test/models/outcome.py
index bb1e81c8a..36471e2d4 100644
--- a/airbyte_cdk/test/models/outcome.py
+++ b/airbyte_cdk/test/models/outcome.py
@@ -36,6 +36,18 @@ def from_status_str(cls, status: str | None) -> ExpectedOutcome:
         except KeyError as ex:
             raise ValueError(f"Invalid status '{status}'. Expected 'succeed' or 'failed'.") from ex
 
+    @classmethod
+    def from_expecting_exception_bool(cls, expecting_exception: bool | None) -> ExpectedOutcome:
+        """Convert a boolean indicating whether an exception is expected to an ExpectedOutcome."""
+        if expecting_exception is None:
+            return ExpectedOutcome.ALLOW_ANY
+
+        return (
+            ExpectedOutcome.EXPECT_EXCEPTION
+            if expecting_exception
+            else ExpectedOutcome.EXPECT_SUCCESS
+        )
+
     def expect_exception(self) -> bool:
         """Return whether the expectation is that an exception should be raised."""
         return self == ExpectedOutcome.EXPECT_EXCEPTION
diff --git a/airbyte_cdk/test/utils/reading.py b/airbyte_cdk/test/utils/reading.py
index a0450af3e..17b4ef556 100644
--- a/airbyte_cdk/test/utils/reading.py
+++ b/airbyte_cdk/test/utils/reading.py
@@ -20,8 +20,16 @@ def read_records(
     stream_name: str,
     sync_mode: SyncMode,
     state: Optional[List[AirbyteStateMessage]] = None,
-    expected_outcome: ExpectedOutcome = ExpectedOutcome.EXPECT_SUCCESS,
+    *,
+    expecting_exception: bool | None = None,  # Deprecated, use expected_outcome instead.
+    expected_outcome: ExpectedOutcome | None = None,
 ) -> EntrypointOutput:
     """Read records from a stream."""
+    if expecting_exception is not None and expected_outcome is not None:
+        raise ValueError("Cannot set both expecting_exception and expected_outcome.")
+
+    if expected_outcome is None:
+        expected_outcome = ExpectedOutcome.from_expecting_exception_bool(expecting_exception)
+
     _catalog = catalog(stream_name, sync_mode)
     return read(source, config, _catalog, state, expected_outcome=expected_outcome)

From 6bc4e333d3d2c85824c2704de7e227776969fd48 Mon Sep 17 00:00:00 2001
From: Aaron Steers <aj@airbyte.io>
Date: Wed, 11 Jun 2025 13:56:05 -0700
Subject: [PATCH 12/14] commit prev wip

---
 airbyte_cdk/test/entrypoint_wrapper.py | 28 +++++++++++++++++++++-----
 airbyte_cdk/test/utils/reading.py      | 17 ++++++++--------
 2 files changed, 32 insertions(+), 13 deletions(-)

diff --git a/airbyte_cdk/test/entrypoint_wrapper.py b/airbyte_cdk/test/entrypoint_wrapper.py
index a5fc18985..0a9a5ee0c 100644
--- a/airbyte_cdk/test/entrypoint_wrapper.py
+++ b/airbyte_cdk/test/entrypoint_wrapper.py
@@ -159,9 +159,19 @@ def is_not_in_logs(self, pattern: str) -> bool:
 
 def _run_command(
     source: Source,
-    args: list[str],
-    expected_outcome: ExpectedOutcome,
+    args: List[str],
+    expecting_exception: bool | None = None,  # Deprecated, use `expected_outcome` instead.
+    *,
+    expected_outcome: ExpectedOutcome | None = None
 ) -> EntrypointOutput:
+    """Internal function to run a command with the AirbyteEntrypoint.
+
+    Note: Even though this function is private, some connectors do call it directly.
+
+    Note: The `expecting_exception` arg is now deprecated in favor of the tri-state
+    `expected_outcome` arg. The old argument is supported (for now) for backwards compatibility.
+    """
+    expected_outcome = expected_outcome or ExpectedOutcome.from_expecting_exception_bool(expecting_exception)
     log_capture_buffer = StringIO()
     stream_handler = logging.StreamHandler(log_capture_buffer)
     stream_handler.setLevel(logging.INFO)
@@ -194,8 +204,9 @@ def _run_command(
 def discover(
     source: Source,
     config: Mapping[str, Any],
+    expecting_exception: bool | None = None,  # Deprecated, use `expected_outcome` instead.
     *,
-    expected_outcome: ExpectedOutcome = ExpectedOutcome.EXPECT_SUCCESS,
+    expected_outcome: ExpectedOutcome | None = None,
 ) -> EntrypointOutput:
     """
     config must be json serializable
@@ -210,6 +221,7 @@ def discover(
         return _run_command(
             source,
             ["discover", "--config", config_file, "--debug"],
+            expecting_exception=expecting_exception,  # Deprecated, but still supported.
             expected_outcome=expected_outcome,
         )
 
@@ -219,8 +231,9 @@ def read(
     config: Mapping[str, Any],
     catalog: ConfiguredAirbyteCatalog,
     state: Optional[List[AirbyteStateMessage]] = None,
+    expecting_exception: bool | None = None,  # Deprecated, use `expected_outcome` instead.
     *,
-    expected_outcome: ExpectedOutcome = ExpectedOutcome.EXPECT_SUCCESS,
+    expected_outcome: ExpectedOutcome | None = None,
 ) -> EntrypointOutput:
     """
     config and state must be json serializable
@@ -253,7 +266,12 @@ def read(
                 ]
             )
 
-        return _run_command(source, args, expected_outcome=expected_outcome)
+        return _run_command(
+            source,
+            args,
+            expecting_exception=expecting_exception,  # Deprecated, but still supported.
+            expected_outcome=expected_outcome,
+        )
 
 
 def make_file(
diff --git a/airbyte_cdk/test/utils/reading.py b/airbyte_cdk/test/utils/reading.py
index 17b4ef556..0d8a092cb 100644
--- a/airbyte_cdk/test/utils/reading.py
+++ b/airbyte_cdk/test/utils/reading.py
@@ -20,16 +20,17 @@ def read_records(
     stream_name: str,
     sync_mode: SyncMode,
     state: Optional[List[AirbyteStateMessage]] = None,
-    *,
     expecting_exception: bool | None = None,  # Deprecated, use expected_outcome instead.
+    *,
     expected_outcome: ExpectedOutcome | None = None,
 ) -> EntrypointOutput:
     """Read records from a stream."""
-    if expecting_exception is not None and expected_outcome is not None:
-        raise ValueError("Cannot set both expecting_exception and expected_outcome.")
-
-    if expected_outcome is None:
-        expected_outcome = ExpectedOutcome.from_expecting_exception_bool(expecting_exception)
-
     _catalog = catalog(stream_name, sync_mode)
-    return read(source, config, _catalog, state, expected_outcome=expected_outcome)
+    return read(
+        source,
+        config,
+        _catalog,
+        state,
+        expecting_exception=expecting_exception,  # Deprecated, for backward compatibility.
+        expected_outcome=expected_outcome,
+    )

From 4191d100ee08673e5a55d59045de45a3cd2ac523 Mon Sep 17 00:00:00 2001
From: Aaron Steers <aj@airbyte.io>
Date: Wed, 11 Jun 2025 14:08:30 -0700
Subject: [PATCH 13/14] format fix

---
 airbyte_cdk/test/entrypoint_wrapper.py | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/airbyte_cdk/test/entrypoint_wrapper.py b/airbyte_cdk/test/entrypoint_wrapper.py
index 0a9a5ee0c..276293744 100644
--- a/airbyte_cdk/test/entrypoint_wrapper.py
+++ b/airbyte_cdk/test/entrypoint_wrapper.py
@@ -162,7 +162,7 @@ def _run_command(
     args: List[str],
     expecting_exception: bool | None = None,  # Deprecated, use `expected_outcome` instead.
     *,
-    expected_outcome: ExpectedOutcome | None = None
+    expected_outcome: ExpectedOutcome | None = None,
 ) -> EntrypointOutput:
     """Internal function to run a command with the AirbyteEntrypoint.
 
@@ -171,7 +171,9 @@ def _run_command(
     Note: The `expecting_exception` arg is now deprecated in favor of the tri-state
     `expected_outcome` arg. The old argument is supported (for now) for backwards compatibility.
     """
-    expected_outcome = expected_outcome or ExpectedOutcome.from_expecting_exception_bool(expecting_exception)
+    expected_outcome = expected_outcome or ExpectedOutcome.from_expecting_exception_bool(
+        expecting_exception
+    )
     log_capture_buffer = StringIO()
     stream_handler = logging.StreamHandler(log_capture_buffer)
     stream_handler.setLevel(logging.INFO)

From 2434501ae4dc984d8bf68a7026cc06b5cf6592b7 Mon Sep 17 00:00:00 2001
From: Aaron Steers <aj@airbyte.io>
Date: Wed, 11 Jun 2025 14:30:59 -0700
Subject: [PATCH 14/14] fix: legacy default expectation behavior

---
 airbyte_cdk/test/entrypoint_wrapper.py | 2 +-
 airbyte_cdk/test/models/outcome.py     | 3 ++-
 2 files changed, 3 insertions(+), 2 deletions(-)

diff --git a/airbyte_cdk/test/entrypoint_wrapper.py b/airbyte_cdk/test/entrypoint_wrapper.py
index 276293744..b46e7f86a 100644
--- a/airbyte_cdk/test/entrypoint_wrapper.py
+++ b/airbyte_cdk/test/entrypoint_wrapper.py
@@ -172,7 +172,7 @@ def _run_command(
     `expected_outcome` arg. The old argument is supported (for now) for backwards compatibility.
     """
     expected_outcome = expected_outcome or ExpectedOutcome.from_expecting_exception_bool(
-        expecting_exception
+        expecting_exception,
     )
     log_capture_buffer = StringIO()
     stream_handler = logging.StreamHandler(log_capture_buffer)
diff --git a/airbyte_cdk/test/models/outcome.py b/airbyte_cdk/test/models/outcome.py
index 36471e2d4..2bc74730c 100644
--- a/airbyte_cdk/test/models/outcome.py
+++ b/airbyte_cdk/test/models/outcome.py
@@ -40,7 +40,8 @@ def from_status_str(cls, status: str | None) -> ExpectedOutcome:
     def from_expecting_exception_bool(cls, expecting_exception: bool | None) -> ExpectedOutcome:
         """Convert a boolean indicating whether an exception is expected to an ExpectedOutcome."""
         if expecting_exception is None:
-            return ExpectedOutcome.ALLOW_ANY
+            # Align with legacy behavior where default would be 'False' (no exception expected)
+            return ExpectedOutcome.EXPECT_SUCCESS
 
         return (
             ExpectedOutcome.EXPECT_EXCEPTION