From e25e8e84b30b20072a3c7d2b37e6eab4a690dcbc Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 19 Oct 2026 18:37:06 +0000 Subject: [PATCH 1/3] fix(mcp): port Connector Builder readiness behavior Co-Authored-By: AJ Steers --- airbyte/cloud/connector_builder/__init__.py | 6 ++ airbyte/cloud/connector_builder/_util.py | 55 +++++++++----- airbyte/cloud/connector_builder/models.py | 81 +++++++++++++++++---- airbyte/cloud/connector_builder/project.py | 60 ++++++++++----- airbyte/mcp/connector_builder.py | 51 +++++++++---- 5 files changed, 190 insertions(+), 63 deletions(-) diff --git a/airbyte/cloud/connector_builder/__init__.py b/airbyte/cloud/connector_builder/__init__.py index e5e60dda0..e88783b2c 100644 --- a/airbyte/cloud/connector_builder/__init__.py +++ b/airbyte/cloud/connector_builder/__init__.py @@ -22,6 +22,9 @@ ) from airbyte.cloud.connector_builder.project import ( CONNECTOR_BUILDER_TEST_DESTINATION_NAME, + TEST_READ_MAX_PAGES, + TEST_READ_MAX_RECORDS, + TEST_READ_MAX_SLICES, CloudConnectorBuilderProject, ) @@ -39,4 +42,7 @@ "ConnectorBuilderStreamReadResult", "ManifestTextEditMode", "ReadinessCheckResult", + "TEST_READ_MAX_PAGES", + "TEST_READ_MAX_RECORDS", + "TEST_READ_MAX_SLICES", ] diff --git a/airbyte/cloud/connector_builder/_util.py b/airbyte/cloud/connector_builder/_util.py index 1d552914b..64948f752 100644 --- a/airbyte/cloud/connector_builder/_util.py +++ b/airbyte/cloud/connector_builder/_util.py @@ -141,8 +141,12 @@ def readiness_next_steps( not_attempted_streams: list[str], ) -> list[str]: steps = [check.description for check in checks[:4] if check.result != "passed"] - failed = [stream.stream_name for stream in streams if not stream.passed] - capped = [stream.stream_name for stream in streams if stream.passed and not stream.read_to_end] + failed = [stream.stream_name for stream in streams if stream.failed] + capped = [ + stream.stream_name + for stream in streams + if stream.error is None and not stream.http_errors and not stream.read_to_end + ] if failed: steps.append(f"Fix the failed streams and test them again: {', '.join(failed)}") if pending_streams: @@ -156,16 +160,14 @@ def readiness_next_steps( "Run the report again with the not-attempted streams, skipping any already " f"validated or unchanged since: {', '.join(not_attempted_streams)}" ) - if not (failed or pending_streams or not_attempted_streams): - steps.append( - "All named streams passed. Once every stream is validated, publish the custom " - "source definition (if not yet) and run a full sync of every stream to a no-op " - "destination." - ) if capped: steps.append( - f"These streams hit a test-read cap; check them with the full sync: {', '.join(capped)}" + "These streams hit a test-read cap, so later pages weren't read; " + f"check them with a full sync: {', '.join(capped)}" ) + warning_streams = [stream.stream_name for stream in streams if stream.warnings] + if warning_streams: + steps.append(f"Report these streams' `warnings` to the user: {', '.join(warning_streams)}") return steps @@ -180,8 +182,15 @@ def readiness_checks( takes_no_inputs = isinstance(spec, dict) and not spec.get("properties") testing_values = project_data.get("testingValues") or {} source_definition_id = project_data["builderProject"].get("sourceDefinitionId") - failed = [stream.stream_name for stream in streams if not stream.passed] - not_read_to_end = [stream.stream_name for stream in streams if not stream.read_to_end] + failed = [stream.stream_name for stream in streams if stream.failed] + capped = [ + stream.stream_name + for stream in streams + if stream.error is None and not stream.http_errors and not stream.read_to_end + ] + not_read = [ + stream.stream_name for stream in streams if stream.error is not None or stream.http_errors + ] return [ ConnectorBuilderReadinessCheck( description="Save a draft manifest", @@ -201,7 +210,7 @@ def readiness_checks( detail=url, ), ConnectorBuilderReadinessCheck( - description="Pass a test read on the named streams", + description="Test-read the named streams without errors", result=_result(not failed and not pending_streams), detail="; ".join( part @@ -215,13 +224,21 @@ def readiness_checks( ), ConnectorBuilderReadinessCheck( description="Read the named streams to the end within the test-read caps", - result=_result(not not_read_to_end and not pending_streams), - detail=( - f"hit a cap: {', '.join(not_read_to_end)}. Check these with a full sync " - "after publishing." - if not_read_to_end - else None - ), + result=_result(not pending_streams and all(stream.read_to_end for stream in streams)), + detail="; ".join( + part + for part in ( + ( + f"hit a cap: {', '.join(capped)}. Check these with a full sync " + "after publishing." + ) + if capped + else "", + f"not read: {', '.join(not_read)} (test read failed)" if not_read else "", + ) + if part + ) + or None, ), ConnectorBuilderReadinessCheck( description="Publish the custom source definition", diff --git a/airbyte/cloud/connector_builder/models.py b/airbyte/cloud/connector_builder/models.py index 28e810a45..df7023165 100644 --- a/airbyte/cloud/connector_builder/models.py +++ b/airbyte/cloud/connector_builder/models.py @@ -33,6 +33,40 @@ class ConnectorBuilderPageSummary(BaseModel): ) +def _record_count_warnings( + record_count: int, + pages: list[ConnectorBuilderPageSummary], + *, + test_read_limit_reached: bool, +) -> list[str]: + """Return informational hints for record counts that may indicate a partial read.""" + if test_read_limit_reached: + return [] + if record_count == 0: + return ["Read 0 records. Check the request, record selector, and testing values."] + + warnings = [] + if record_count % 10 == 0: + warnings.append( + f"Read {record_count:,} records, a multiple of 10. A page size or API limit may " + "have cut the read short; check the paginator." + ) + + record_pages = [page for page in pages if page.record_count > 0] + if ( + record_pages + and record_pages[0].record_count > 1 + and pages[-1].record_count > 0 + and pages[-1].record_count == record_pages[0].record_count + ): + record_count_per_page = record_pages[0].record_count + warnings.append( + f"The last page returned a full page of {record_count_per_page:,} records, the " + "same as the first page, so more pages may exist; check the paginator." + ) + return warnings + + class ConnectorBuilderStreamReadResult(BaseModel): """Result of a Connector Builder test read of one stream.""" @@ -45,6 +79,13 @@ class ConnectorBuilderStreamReadResult(BaseModel): pages: list[ConnectorBuilderPageSummary] logs: list[dict[str, Any]] inferred_schema: dict[str, Any] | None = None + warnings: list[str] = Field( + default_factory=list, + description=( + "Hints that the read may be incomplete, such as round record counts. " + "They don't fail the read." + ), + ) @classmethod def from_api_response( @@ -77,12 +118,13 @@ def from_api_response( ) ) + test_read_limit_reached = bool(response.get("test_read_limit_reached")) return cls( stream_name=stream_name, record_count=len(records), slice_count=len(slices), page_count=len(pages), - test_read_limit_reached=bool(response.get("test_read_limit_reached")), + test_read_limit_reached=test_read_limit_reached, records=records, pages=pages, logs=[ @@ -90,6 +132,11 @@ def from_api_response( for log in response.get("logs") or [] ], inferred_schema=response.get("inferred_schema"), + warnings=_record_count_warnings( + len(records), + pages, + test_read_limit_reached=test_read_limit_reached, + ), ) @@ -129,19 +176,31 @@ class ConnectorBuilderStreamReadiness(BaseModel): """Readiness of one stream, from a Connector Builder test read.""" stream_name: str - passed: bool + failed: bool = Field( + description=( + "True when the test read raised an error, returned HTTP errors, or logged errors." + ), + ) record_count: int = 0 page_count: int = 0 read_to_end: bool = Field( default=False, description=( - "True when the stream ended within the test-read caps. False means the read hit a " - "record, page, or slice cap, so later pages were never read." + "True when the stream ended within the test-read caps without HTTP errors. False " + "means a record, page, or slice cap was hit, an HTTP error occurred, or the read " + "did not run (see `error`)." ), ) http_errors: list[ConnectorBuilderPageSummary] = Field(default_factory=list) problem_logs: list[dict[str, Any]] = Field(default_factory=list) inferred_schema: dict[str, Any] | None = None + warnings: list[str] = Field( + default_factory=list, + description=( + "Hints that the read may be incomplete, such as round record counts. " + "They don't fail the read." + ), + ) error: str | None = Field( default=None, description="The error raised by the test read itself, if it failed to run.", @@ -151,7 +210,7 @@ class ConnectorBuilderStreamReadiness(BaseModel): def from_stream_read( cls, result: ConnectorBuilderStreamReadResult ) -> ConnectorBuilderStreamReadiness: - """Judge a stream from its test read: records returned, no HTTP or error logs.""" + """Summarize errors and results from a stream test read.""" http_errors = [ page for page in result.pages @@ -163,13 +222,14 @@ def from_stream_read( ) return cls( stream_name=result.stream_name, - passed=result.record_count > 0 and not has_errors, + failed=has_errors, record_count=result.record_count, page_count=result.page_count, - read_to_end=not result.test_read_limit_reached, + read_to_end=not result.test_read_limit_reached and not http_errors, http_errors=http_errors, problem_logs=problem_logs, inferred_schema=result.inferred_schema, + warnings=result.warnings, ) @@ -200,13 +260,6 @@ class ConnectorBuilderFullSync(BaseModel): class ConnectorBuilderReadinessReport(BaseModel): """Readiness checks and per-stream test reads for a Connector Builder draft.""" - passed: bool = Field( - description=( - "True when every named stream passed and was read to the end. Covers only the " - "named streams; a full sync validates the whole connector." - ), - ) - passed_streams: list[str] = Field(default_factory=list) failed_streams: list[str] = Field(default_factory=list) pending_streams: list[str] = Field( default_factory=list, diff --git a/airbyte/cloud/connector_builder/project.py b/airbyte/cloud/connector_builder/project.py index bf409544c..e379530f0 100644 --- a/airbyte/cloud/connector_builder/project.py +++ b/airbyte/cloud/connector_builder/project.py @@ -51,6 +51,33 @@ """Shortest HTTP timeout for a test read started just before the readiness deadline.""" CONNECTOR_BUILDER_TEST_DESTINATION_NAME = "Connector Builder Test (No-op)" """Name of the no-op destination that full syncs find or create by default.""" +TEST_READ_MAX_RECORDS = 5_000 +"""Most records read per stream by `run_readiness_checks()`.""" +TEST_READ_MAX_PAGES = 20 +TEST_READ_MAX_SLICES = 20 + + +def _check_test_read_limits( + record_limit: int, + page_limit: int, + slice_limit: int, +) -> None: + """Reject test-read limits outside Airbyte Cloud's supported range.""" + limits = ( + ("record_limit", record_limit, TEST_READ_MAX_RECORDS), + ("page_limit", page_limit, TEST_READ_MAX_PAGES), + ("slice_limit", slice_limit, TEST_READ_MAX_SLICES), + ) + if any(not 1 <= value <= maximum for _, value, maximum in limits): + raise exc.AirbyteLibInputError( + message="Test-read limits are out of range.", + guidance=( + f"Airbyte Cloud accepts record_limit 1 to {TEST_READ_MAX_RECORDS:,}, " + f"page_limit 1 to {TEST_READ_MAX_PAGES}, and " + f"slice_limit 1 to {TEST_READ_MAX_SLICES}." + ), + context={name: value for name, value, _ in limits}, + ) class CloudConnectorBuilderProject: # noqa: PLR0904 # Builder project API facade. @@ -271,10 +298,11 @@ def test_read_stream( Args: stream_name: The stream to read. - record_limit: Maximum records to return (1 to 10,000). + record_limit: Maximum records to return (1 to 5,000). page_limit: Maximum pages to fetch per slice (1 to 20). slice_limit: Maximum slices to read (1 to 20). """ + _check_test_read_limits(record_limit, page_limit, slice_limit) return self._test_read_stream( self._require_draft_manifest(), stream_name, @@ -339,9 +367,9 @@ def run_readiness_checks( self, stream_names: list[str], *, - record_limit: int = 10_000, - page_limit: int = 20, - slice_limit: int = 20, + record_limit: int = TEST_READ_MAX_RECORDS, + page_limit: int = TEST_READ_MAX_PAGES, + slice_limit: int = TEST_READ_MAX_SLICES, timeout_seconds: float | None = None, run_full_sync: bool = False, test_source_id: str | None = None, @@ -349,26 +377,27 @@ def run_readiness_checks( ) -> ConnectorBuilderReadinessReport: """Check the draft's progress and test-read the given streams on Airbyte Cloud. - Reads run in parallel. A stream passes when its test read returns records with no - HTTP errors or error logs, and is `read_to_end` when it ended within the test-read - caps (which default to Cloud's maximums). The readiness checks are derived from the project, - so nothing is stored. `passed` covers only the named streams: each must pass and be - read to the end. A full sync after publishing validates the whole connector. + Reads run in parallel. A stream is `failed` when its read errored, returned HTTP + errors, or logged errors. The report lists failures and warnings without judging + readiness; `read_to_end` indicates whether the read ended within the test-read caps. + The readiness checks are derived from the project, so nothing is stored. A full sync + after publishing validates the whole connector. Args: stream_names: Streams to test-read (at least 1). For dynamic streams, get the generated names from `resolve_manifest()`. - record_limit: Maximum records to read per stream (1 to 10,000). + record_limit: Maximum records to read per stream (1 to 5,000). page_limit: Maximum pages to fetch per slice (1 to 20). slice_limit: Maximum slices to read (1 to 20). timeout_seconds: Return after this many seconds with the streams finished so far; the rest are listed in `pending_streams` and `not_attempted_streams`. No limit by default. - run_full_sync: If every named stream finished and passed (or none are named), + run_full_sync: If no named stream failed or is still pending (or none are named), publish the draft and start a full sync with `run_full_sync()`. test_source_id: The source for the full sync. See `run_full_sync()`. on_resource_created: Passed to `run_full_sync()`. """ + _check_test_read_limits(record_limit, page_limit, slice_limit) stream_names = list(dict.fromkeys(stream_names)) if not stream_names and not run_full_sync: raise exc.AirbyteLibInputError( @@ -414,12 +443,12 @@ def read(stream_name: str) -> ConnectorBuilderStreamReadiness: if ex.guidance: error += f" {ex.guidance}" return ConnectorBuilderStreamReadiness( - stream_name=stream_name, passed=False, error=error + stream_name=stream_name, failed=True, error=error ) except requests.RequestException as ex: return ConnectorBuilderStreamReadiness( stream_name=stream_name, - passed=False, + failed=True, error=f"Request to Airbyte Cloud failed: {type(ex).__name__}", ) return ConnectorBuilderStreamReadiness.from_stream_read(result) @@ -461,10 +490,7 @@ def _readiness_report( remaining = pending_streams + not_attempted_streams checks = readiness_checks(project_data, draft, streams, remaining, self.url) return ConnectorBuilderReadinessReport( - passed=not remaining - and all(stream.passed and stream.read_to_end for stream in streams), - passed_streams=[stream.stream_name for stream in streams if stream.passed], - failed_streams=[stream.stream_name for stream in streams if not stream.passed], + failed_streams=[stream.stream_name for stream in streams if stream.failed], pending_streams=pending_streams, not_attempted_streams=not_attempted_streams, checks=checks, diff --git a/airbyte/mcp/connector_builder.py b/airbyte/mcp/connector_builder.py index 7d07ed94c..632e82496 100644 --- a/airbyte/mcp/connector_builder.py +++ b/airbyte/mcp/connector_builder.py @@ -15,6 +15,9 @@ from pydantic import Field from airbyte.cloud.connector_builder import ( + TEST_READ_MAX_PAGES, + TEST_READ_MAX_RECORDS, + TEST_READ_MAX_SLICES, ManifestTextEditMode, ) from airbyte.exceptions import ( @@ -392,17 +395,32 @@ def test_connector_builder_stream( ], record_limit: Annotated[ int, - Field(description="Maximum records to read (1 to 10,000).", default=100, ge=1, le=10000), + Field( + description=f"Maximum records to read (1 to {TEST_READ_MAX_RECORDS:,}).", + default=100, + ge=1, + le=TEST_READ_MAX_RECORDS, + ), TraceArg.VALUE, ] = 100, page_limit: Annotated[ int, - Field(description="Maximum pages to fetch per slice (1 to 20).", default=5, ge=1, le=20), + Field( + description=f"Maximum pages to fetch per slice (1 to {TEST_READ_MAX_PAGES}).", + default=5, + ge=1, + le=TEST_READ_MAX_PAGES, + ), TraceArg.VALUE, ] = 5, slice_limit: Annotated[ int, - Field(description="Maximum slices to read (1 to 20).", default=5, ge=1, le=20), + Field( + description=f"Maximum slices to read (1 to {TEST_READ_MAX_SLICES}).", + default=5, + ge=1, + le=TEST_READ_MAX_SLICES, + ), TraceArg.VALUE, ] = 5, ) -> dict[str, Any]: @@ -411,7 +429,11 @@ def test_connector_builder_stream( Uses the project's saved testing values as the connector config. Returns counts, a per-page summary of HTTP requests (with response bodies for HTTP errors), connector logs, the inferred schema, and a preview of the first records (`records` holds at most - 10 records; `record_count` is the total read). + 10 records; `record_count` is the total read). `test_read_limit_reached` true means later + records weren't read. Cloud stops at page boundaries, so `record_count` can exceed + `record_limit`. + `warnings` flags reads that may be incomplete (round record counts, a full last page); + check them, but they don't fail the read. """ workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id) project = workspace.get_connector_builder_project( @@ -527,9 +549,9 @@ def run_connector_builder_readiness_checks( bool, Field( description=( - "Optional. If every named stream passed (or none are named), publish the draft " - "and start a full sync of every stream to a no-op destination. Returns the job " - "in `full_sync`; poll it with `get_cloud_sync_status`." + "Optional. If no named stream failed or is still pending (or none are named), " + "publish the draft and start a full sync of every stream to a no-op destination. " + "Returns the job in `full_sync`; poll it with `get_cloud_sync_status`." ), default=False, ), @@ -548,13 +570,16 @@ def run_connector_builder_readiness_checks( """Check a Connector Builder draft's progress and test-read its streams on Airbyte Cloud. Returns readiness checks for the steps to a finished custom source, with next steps, plus - per-stream results from parallel test reads using the project's testing values: pass/fail, + per-stream results from parallel test reads using the project's testing values: `failed`, `read_to_end`, record and page counts, HTTP errors, warning and error logs, and inferred - schemas. A stream passes when it returns records with no HTTP errors or error logs. - Cloud test reads stop at 20 pages per slice and 20 slices, so a stream that hits a cap - isn't `read_to_end` and the report doesn't pass. The verdict covers only the named - streams; a full sync validates the whole connector. With `run_full_sync`, the tool - publishes the draft and starts that sync once the named streams pass. + schemas. A stream is `failed` when its read errored, returned HTTP errors, or logged + errors. Cloud test reads stop at 5,000 records, 20 pages per slice, and 20 slices, so a + stream that hits a cap isn't `read_to_end`. The report calls out failures and warnings + without judging readiness; report every failure and warning to the user. A full sync + validates the whole connector. With `run_full_sync`, the tool publishes the draft and + starts that sync once no named stream failed or is pending. + `warnings` flags reads that may be incomplete (round record counts, a full last page); + check them, but they don't fail the read. """ workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id) project = workspace.get_connector_builder_project( From 2d3c4fe040242b42cb3a14caeddea6b77450242f Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 19 Oct 2026 18:37:09 +0000 Subject: [PATCH 2/3] test(mcp): cover Connector Builder readiness semantics Co-Authored-By: AJ Steers --- .../test_cloud_connector_builder.py | 333 ++++++++++++++++-- tests/unit_tests/test_mcp_cloud.py | 64 ++++ 2 files changed, 367 insertions(+), 30 deletions(-) diff --git a/tests/unit_tests/test_cloud_connector_builder.py b/tests/unit_tests/test_cloud_connector_builder.py index 1069b5679..14c61ad38 100644 --- a/tests/unit_tests/test_cloud_connector_builder.py +++ b/tests/unit_tests/test_cloud_connector_builder.py @@ -17,8 +17,10 @@ CloudConnectorBuilderProject, ConnectorBuilderPageSummary, ConnectorBuilderResolvedManifest, + ConnectorBuilderStreamReadiness, ConnectorBuilderStreamReadResult, ) +from airbyte.cloud.connector_builder.models import _record_count_warnings from airbyte.cloud.connector_builder._util import ( apply_manifest_text_edit, drop_masked_secrets, @@ -27,7 +29,11 @@ spec_has_secrets, ) from airbyte.cloud.workspaces import CloudWorkspace -from airbyte.exceptions import AirbyteDuplicateResourcesError, AirbyteLibInputError +from airbyte.exceptions import ( + AirbyteCloudError, + AirbyteDuplicateResourcesError, + AirbyteLibInputError, +) from airbyte.secrets.base import SecretString MANIFEST: dict[str, Any] = { @@ -140,6 +146,103 @@ def test_stream_read_result_from_api_response() -> None: assert result.inferred_schema == {"type": "object"} +@pytest.mark.parametrize( + ("record_count", "page_counts", "limit_reached", "expected"), + [ + pytest.param( + 100, + [100], + False, + [ + "Read 100 records, a multiple of 10. A page size or API limit may have cut " + "the read short; check the paginator.", + "The last page returned a full page of 100 records, the same as the first " + "page, so more pages may exist; check the paginator.", + ], + id="round-single-page", + ), + pytest.param(7, [5, 2], False, [], id="non-round-partial-last-page"), + pytest.param(5, [5, 0], False, [], id="empty-final-page"), + pytest.param( + 10, + [5, 5], + False, + [ + "Read 10 records, a multiple of 10. A page size or API limit may have cut " + "the read short; check the paginator.", + "The last page returned a full page of 5 records, the same as the first " + "page, so more pages may exist; check the paginator.", + ], + id="full-final-page", + ), + pytest.param( + 0, + [0], + False, + ["Read 0 records. Check the request, record selector, and testing values."], + id="zero-records", + ), + pytest.param(1, [1], False, [], id="one-record"), + pytest.param( + 150, + [50, 50, 50], + False, + [ + "Read 150 records, a multiple of 10. A page size or API limit may have cut " + "the read short; check the paginator.", + "The last page returned a full page of 50 records, the same as the first " + "page, so more pages may exist; check the paginator.", + ], + id="multiple-full-pages", + ), + pytest.param(100, [100], True, [], id="reached-test-read-limit"), + ], +) +def test_record_count_warnings( + record_count: int, + page_counts: list[int], + limit_reached: bool, + expected: list[str], +) -> None: + pages = [ConnectorBuilderPageSummary(record_count=count) for count in page_counts] + + assert ( + _record_count_warnings( + record_count, + pages, + test_read_limit_reached=limit_reached, + ) + == expected + ) + + +@pytest.mark.parametrize("limit_reached", [False, True]) +def test_stream_read_warnings_propagate_to_readiness_without_failing( + limit_reached: bool, +) -> None: + response = { + "slices": [ + { + "pages": [ + {"records": [{"id": index} for index in range(100)]}, + ] + } + ], + "test_read_limit_reached": limit_reached, + } + + result = ConnectorBuilderStreamReadResult.from_api_response("users", response) + readiness = ConnectorBuilderStreamReadiness.from_stream_read(result) + + if limit_reached: + assert result.warnings == [] + else: + assert len(result.warnings) == 2 + assert readiness.warnings == result.warnings + assert readiness.failed is False + assert readiness.read_to_end is not limit_reached + + def test_resolved_manifest_lists_stream_names() -> None: resolved = ConnectorBuilderResolvedManifest.from_manifest(MANIFEST) assert resolved.stream_names == ["users", "orders"] @@ -526,7 +629,7 @@ def fake_read( monkeypatch.setattr(project, "_test_read_stream", fake_read) -def test_run_readiness_checks_judges_each_stream( +def test_run_readiness_checks_summarizes_each_stream( monkeypatch: pytest.MonkeyPatch, ) -> None: names = ["ok", "empty", "http_error", "error_log", "warn_log", "broken", "dropped"] @@ -539,8 +642,11 @@ def test_run_readiness_checks_judges_each_stream( project, { "ok": _stream_read("ok", record_count=3), - "empty": _stream_read("empty", record_count=0), - "http_error": _stream_read("http_error", record_count=1, status=401), + "empty": ConnectorBuilderStreamReadResult.from_api_response( + "empty", + {"slices": [{"pages": [{"records": []}]}]}, + ), + "http_error": _stream_read("http_error", record_count=1, status=422), "error_log": _stream_read( "error_log", record_count=1, @@ -555,35 +661,43 @@ def test_run_readiness_checks_judges_each_stream( report = project.run_readiness_checks(names) assert [stream.stream_name for stream in report.streams] == names - assert {stream.stream_name: stream.passed for stream in report.streams} == { - "ok": True, + assert {stream.stream_name: stream.failed for stream in report.streams} == { + "ok": False, "empty": False, - "http_error": False, - "error_log": False, - "warn_log": True, - "broken": False, - "dropped": False, + "http_error": True, + "error_log": True, + "warn_log": False, + "broken": True, + "dropped": True, } - assert report.passed is False - assert report.passed_streams == ["ok", "warn_log"] assert report.failed_streams == [ - "empty", "http_error", "error_log", "broken", "dropped", ] by_name = {stream.stream_name: stream for stream in report.streams} - assert by_name["http_error"].http_errors[0].status == 401 + assert by_name["empty"].warnings == [ + "Read 0 records. Check the request, record selector, and testing values." + ] + assert "empty" not in report.failed_streams + assert by_name["http_error"].http_errors[0].status == 422 + assert by_name["http_error"].read_to_end is False assert by_name["warn_log"].problem_logs == [{"level": "WARN", "message": "hmm"}] assert by_name["broken"].error == "read failed Fix the stream." assert ( by_name["dropped"].error == "Request to Airbyte Cloud failed: ConnectionError" ) + assert report.checks[4].detail == "failed: http_error, error_log, broken, dropped" + assert report.checks[5].result == "failed" + assert "not read: http_error, broken, dropped (test read failed)" in ( + report.checks[5].detail or "" + ) assert ( - report.checks[4].detail - == "failed: empty, http_error, error_log, broken, dropped" + report.next_steps[-1] == "Report these streams' `warnings` to the user: empty" ) + assert not any("All named streams passed" in step for step in report.next_steps) + assert not any("ready" in step.lower() for step in report.next_steps) def test_run_readiness_checks_requires_stream_names() -> None: @@ -592,6 +706,127 @@ def test_run_readiness_checks_requires_stream_names() -> None: project.run_readiness_checks([]) +@pytest.mark.parametrize("method", ["test_read_stream", "run_readiness_checks"]) +@pytest.mark.parametrize( + ("limit_name", "limit_value"), + [ + ("record_limit", 5_001), + ("page_limit", 21), + ("slice_limit", 21), + ], +) +def test_test_read_limits_reject_out_of_range_without_api_calls( + monkeypatch: pytest.MonkeyPatch, + method: str, + limit_name: str, + limit_value: int, +) -> None: + def unexpected_api_call(*_: Any, **__: Any) -> Any: + pytest.fail("A test-read limit was rejected only after an API call.") + + monkeypatch.setattr(api_util, "get_connector_builder_project", unexpected_api_call) + monkeypatch.setattr( + api_util, "read_connector_builder_project_stream", unexpected_api_call + ) + project = CloudConnectorBuilderProject(_fake_workspace(), "bp") + expected: dict[str, int] = ( + {"record_limit": 100, "page_limit": 5, "slice_limit": 5} + if method == "test_read_stream" + else { + "record_limit": connector_builder.TEST_READ_MAX_RECORDS, + "page_limit": connector_builder.TEST_READ_MAX_PAGES, + "slice_limit": connector_builder.TEST_READ_MAX_SLICES, + } + ) + + with pytest.raises(AirbyteLibInputError) as error: + if method == "test_read_stream": + project.test_read_stream("users", **{limit_name: limit_value}) + else: + project.run_readiness_checks(["users"], **{limit_name: limit_value}) + + expected[limit_name] = limit_value + assert error.value.get_message() == "Test-read limits are out of range." + assert error.value.context == expected + + +def test_readiness_report_defaults_forward_cloud_test_read_limits( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _patch_project_store(monkeypatch, draft={"streams": [{"name": "users"}]}) + received: list[dict[str, Any]] = [] + + def fake_read(**kwargs: Any) -> dict[str, Any]: + received.append(kwargs) + return {"slices": [{"pages": [{"records": [{"id": 1}]}]}]} + + monkeypatch.setattr(api_util, "read_connector_builder_project_stream", fake_read) + project = CloudConnectorBuilderProject(_fake_workspace(), "bp") + + project.run_readiness_checks(["users"]) + + assert len(received) == 1 + assert received[0]["record_limit"] == connector_builder.TEST_READ_MAX_RECORDS + assert received[0]["page_limit"] == connector_builder.TEST_READ_MAX_PAGES + assert received[0]["slice_limit"] == connector_builder.TEST_READ_MAX_SLICES + + +def test_readiness_checklist_does_not_call_failed_reads_capped( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _patch_project_store(monkeypatch, draft={"streams": [{"name": "users"}]}) + project = CloudConnectorBuilderProject(_fake_workspace(), "bp") + + def fail_read(*_: Any, **__: Any) -> ConnectorBuilderStreamReadResult: + raise AirbyteCloudError(message="Cloud read failed.") + + monkeypatch.setattr(project, "_test_read_stream", fail_read) + + report = project.run_readiness_checks(["users"]) + + assert report.streams[0].read_to_end is False + assert report.checks[5].result == "failed" + assert report.checks[5].detail == "not read: users (test read failed)" + assert "hit a cap" not in (report.checks[5].detail or "") + + +def test_readiness_checklist_marks_failed_capped_stream_incomplete( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _patch_project_store(monkeypatch, draft={"streams": [{"name": "users"}]}) + project = CloudConnectorBuilderProject(_fake_workspace(), "bp") + _patch_reads( + monkeypatch, + project, + { + "users": _stream_read( + "users", + record_count=1, + logs=[{"level": "ERROR", "message": "read failed"}], + limit_reached=True, + ) + }, + ) + + report = project.run_readiness_checks(["users"]) + + assert report.streams[0].failed is True + assert report.streams[0].error is None + assert report.streams[0].http_errors == [] + assert report.failed_streams == ["users"] + assert report.checks[5].result == "failed" + assert report.checks[5].detail == ( + "hit a cap: users. Check these with a full sync after publishing." + ) + assert any( + "Fix the failed streams and test them again: users" in step + for step in report.next_steps + ) + assert any( + "hit a test-read cap" in step and "users" in step for step in report.next_steps + ) + + def test_run_readiness_checks_without_draft_fails( monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -602,7 +837,7 @@ def test_run_readiness_checks_without_draft_fails( ]) -def test_readiness_report_passes_only_when_every_stream_is_read_to_end( +def test_readiness_report_lists_caps_without_a_success_verdict( monkeypatch: pytest.MonkeyPatch, ) -> None: _patch_project_store(monkeypatch, draft=MANIFEST) @@ -618,15 +853,22 @@ def test_readiness_report_passes_only_when_every_stream_is_read_to_end( reads["orders"] = _stream_read("orders", record_count=1) full = project.run_readiness_checks(["users", "orders", "users"]) - assert partial.passed is True - assert capped.passed is False - assert capped.passed_streams == ["users", "orders"] - assert "hit a test-read cap" in capped.next_steps[-1] + assert partial.failed_streams == [] + assert capped.failed_streams == [] + assert "passed" not in partial.model_dump() + assert any( + "These streams hit a test-read cap, so later pages weren't read; " + "check them with a full sync: orders" == step + for step in capped.next_steps + ) + assert not any("Not ready yet" in step for step in capped.next_steps) + assert not any("All named streams passed" in step for step in capped.next_steps) + assert not any("ready" in step.lower() for step in capped.next_steps) assert [s.read_to_end for s in capped.streams] == [True, False] assert capped.checks[5].result == "failed" - assert full.passed is True - assert full.passed_streams == ["users", "orders"] - assert full.next_steps[0].startswith("All named streams passed.") + assert full.failed_streams == [] + assert not any("All named streams passed" in step for step in full.next_steps) + assert not any("ready" in step.lower() for step in full.next_steps) def test_readiness_checks_require_testing_values_when_spec_has_inputs( @@ -653,7 +895,7 @@ def test_readiness_checks_require_testing_values_when_spec_has_inputs( "passed", "unknown", ] - assert report.passed is True + assert report.failed_streams == [] def test_publish_spec_uses_api_field_names() -> None: @@ -719,10 +961,9 @@ def fake_read( report = project.run_readiness_checks(names, timeout_seconds=0.5) finally: release.set() - assert report.passed_streams == ["fast"] + assert report.failed_streams == [] assert report.pending_streams == sorted(slow) assert report.not_attempted_streams == ["queued"] - assert report.passed is False assert any( "not-attempted streams" in step and "queued" in step for step in report.next_steps @@ -730,6 +971,35 @@ def fake_read( assert any("still reading at the timeout" in step for step in report.next_steps) +def test_readiness_checklist_is_incomplete_when_a_stream_is_pending( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _patch_project_store(monkeypatch, draft=MANIFEST) + project = CloudConnectorBuilderProject(_fake_workspace(), "bp") + release = threading.Event() + + def fake_read( + _manifest: Any, stream_name: str, **_: Any + ) -> ConnectorBuilderStreamReadResult: + if stream_name == "pending": + release.wait(5) + return _stream_read(stream_name, record_count=1) + + monkeypatch.setattr(project, "_test_read_stream", fake_read) + try: + report = project.run_readiness_checks( + ["ready", "pending"], + timeout_seconds=0.5, + ) + finally: + release.set() + + assert report.pending_streams == ["pending"] + assert [stream.stream_name for stream in report.streams] == ["ready"] + assert all(stream.read_to_end for stream in report.streams) + assert report.checks[5].result == "failed" + + def test_readiness_bounds_read_requests_by_timeout( monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -1003,8 +1273,8 @@ def test_run_full_sync_without_test_source_fails( ("stream_names", "record_count", "expect_sync"), [ pytest.param([], 0, True, id="no-streams"), - pytest.param(["users"], 1, True, id="all-passed"), - pytest.param(["users"], 0, False, id="failed-skips-sync"), + pytest.param(["users"], 1, True, id="no-failed-streams"), + pytest.param(["users"], 0, True, id="zero-records-no-error"), ], ) def test_readiness_full_sync_gating( @@ -1170,6 +1440,9 @@ def test_run_readiness_checks_full_sync_only_without_draft( report = project.run_readiness_checks([], run_full_sync=True, test_source_id="src") assert report.full_sync is not None + report_data = report.model_dump() + assert "passed" not in report_data + assert report_data["full_sync"] is not None assert report.full_sync.destination_created is False project.publish.assert_not_called() workspace.deploy_connection.assert_called_once() diff --git a/tests/unit_tests/test_mcp_cloud.py b/tests/unit_tests/test_mcp_cloud.py index 91c47d104..50e7298ac 100644 --- a/tests/unit_tests/test_mcp_cloud.py +++ b/tests/unit_tests/test_mcp_cloud.py @@ -3573,6 +3573,70 @@ def test_request_cloud_search_index_backfill_is_a_write_tool() -> None: assert policy.pipeline_change is False +def test_connector_builder_tools_return_read_warnings( + monkeypatch: pytest.MonkeyPatch, +) -> None: + warnings = ["Read 100 records, a multiple of 10. Check the paginator."] + read_result = SimpleNamespace( + model_dump=lambda: { + "stream_name": "users", + "warnings": warnings, + "pages": [], + "http_errors": [], + "logs": [], + "problem_logs": [], + "error": None, + }, + records=[], + ) + readiness_report = SimpleNamespace( + full_sync=None, + model_dump=lambda: { + "full_sync": None, + "streams": [ + { + "stream_name": "users", + "warnings": warnings, + "pages": [], + "http_errors": [], + "problem_logs": [], + "error": None, + } + ], + }, + ) + project = SimpleNamespace( + builder_project_id="builder-project", + url="https://cloud.airbyte.com/workspaces/ws/connector-builder/edit/bp", + test_read_stream=lambda *_args, **_kwargs: read_result, + run_readiness_checks=lambda *_args, **_kwargs: readiness_report, + ) + workspace = SimpleNamespace( + get_connector_builder_project=lambda *_args, **_kwargs: project + ) + monkeypatch.setattr( + connector_builder_mcp, "_get_cloud_workspace", lambda *_args: workspace + ) + ctx = cast(Context, object()) + + stream_result = connector_builder_mcp.test_connector_builder_stream( + ctx, + "users", + builder_project_id="builder-project", + workspace_id="ws", + ) + readiness_result = connector_builder_mcp.run_connector_builder_readiness_checks( + ctx, + ["users"], + builder_project_id="builder-project", + workspace_id="ws", + ) + + assert stream_result["warnings"] == warnings + assert readiness_result["streams"][0]["warnings"] == warnings + assert "passed" not in readiness_result + + def test_redact_stream_diagnostics() -> None: leaky = "GET https://api.example.com/data?api_key=s3cret failed: Bearer s3cret" data = connector_builder_mcp._redact_stream_diagnostics({ From b22d6083dfe353756b6535afd25e15fe1dbc08d3 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 19 Oct 2026 18:43:37 +0000 Subject: [PATCH 3/3] test(mcp): cover HTTP error full-sync gating Co-Authored-By: AJ Steers --- .../test_cloud_connector_builder.py | 21 ++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) diff --git a/tests/unit_tests/test_cloud_connector_builder.py b/tests/unit_tests/test_cloud_connector_builder.py index 14c61ad38..a4af43a07 100644 --- a/tests/unit_tests/test_cloud_connector_builder.py +++ b/tests/unit_tests/test_cloud_connector_builder.py @@ -1270,17 +1270,19 @@ def test_run_full_sync_without_test_source_fails( @pytest.mark.parametrize( - ("stream_names", "record_count", "expect_sync"), + ("stream_names", "record_count", "status", "expect_sync"), [ - pytest.param([], 0, True, id="no-streams"), - pytest.param(["users"], 1, True, id="no-failed-streams"), - pytest.param(["users"], 0, True, id="zero-records-no-error"), + pytest.param([], 0, 200, True, id="no-streams"), + pytest.param(["users"], 1, 200, True, id="no-failed-streams"), + pytest.param(["users"], 0, 200, True, id="zero-records-no-error"), + pytest.param(["users"], 1, 422, False, id="http-error-skips-sync"), ], ) def test_readiness_full_sync_gating( monkeypatch: pytest.MonkeyPatch, stream_names: list[str], record_count: int, + status: int, expect_sync: bool, ) -> None: _patch_project_store(monkeypatch, draft=MANIFEST) @@ -1296,10 +1298,14 @@ def test_readiness_full_sync_gating( ) run_full_sync = MagicMock(return_value=full_sync) monkeypatch.setattr(project, "run_full_sync", run_full_sync) + publish = MagicMock() + monkeypatch.setattr(project, "publish", publish) monkeypatch.setattr( project, "_test_read_stream", - lambda _m, name, **_: _stream_read(name, record_count=record_count), + lambda _m, name, **_: _stream_read( + name, record_count=record_count, status=status + ), ) report = project.run_readiness_checks( @@ -1311,6 +1317,11 @@ def test_readiness_full_sync_gating( run_full_sync.assert_called_once_with( test_source_id="src", on_resource_created=None ) + else: + assert report.full_sync is None + assert report.failed_streams == ["users"] + run_full_sync.assert_not_called() + publish.assert_not_called() def test_readiness_requires_streams_without_full_sync(