Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions airbyte_cdk/test/standard_tests/connector_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -114,6 +114,6 @@ def test_check(
connector_root=self.get_connector_root_dir(),
)
assert len(result.connection_status_messages) == 1, (
f"Expected exactly one CONNECTION_STATUS message. "
"Got: {result.connection_status_messages!s}"
"Expected exactly one CONNECTION_STATUS message. "
f"Got: {result.connection_status_messages!s}"
)
4 changes: 2 additions & 2 deletions airbyte_cdk/test/standard_tests/source_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -96,8 +96,8 @@ def test_spec(self) -> None:
# If an error occurs, it will be raised above.

assert len(result.spec_messages) == 1, (
"Expected exactly 1 spec message but got {len(result.spec_messages)}",
result.errors,
f"Expected exactly 1 spec message but got {len(result.spec_messages)}. "
f"Errors: {result.errors!s}"
)

def test_basic_read(
Expand Down
70 changes: 70 additions & 0 deletions unit_tests/test/test_assertion_messages.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
"""Guards against assertion messages that never render as intended.

Two failure modes are covered:

- A string message containing a `{placeholder}` but missing the `f` prefix, which
prints the literal braces instead of the interpolated value.
- A tuple message, which is always truthy and prints as a tuple repr.
"""

from __future__ import annotations

import ast
import re
from pathlib import Path

import pytest

import airbyte_cdk.test.standard_tests as standard_tests

_PLACEHOLDER_PATTERN = re.compile(r"\{[A-Za-z_][A-Za-z0-9_.\[\]()!:'\"\s]*\}")
_STANDARD_TESTS_DIR = Path(standard_tests.__file__).parent


def _uninterpolated_placeholders(node: ast.expr) -> list[str]:
"""Return placeholder-looking substrings that will print literally."""
if isinstance(node, ast.Constant) and isinstance(node.value, str):
return _PLACEHOLDER_PATTERN.findall(node.value)

if isinstance(node, ast.JoinedStr):
# Implicit concatenation of f-string and plain-string fragments collapses
# into a single JoinedStr; plain fragments survive as Constant values.
found: list[str] = []
for value in node.values:
found.extend(_uninterpolated_placeholders(value))
return found

if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add):
# Messages built with `+` concatenation.
return _uninterpolated_placeholders(node.left) + _uninterpolated_placeholders(node.right)

return []


@pytest.mark.parametrize(
"source_file",
sorted(_STANDARD_TESTS_DIR.rglob("*.py")),
ids=lambda path: path.name,
)
def test_assert_messages_are_renderable(source_file: Path) -> None:
tree = ast.parse(source_file.read_text(), filename=str(source_file))
problems: list[str] = []
for node in ast.walk(tree):
if not isinstance(node, ast.Assert) or node.msg is None:
continue

if isinstance(node.msg, ast.Tuple):
problems.append(
f"{source_file.name}:{node.lineno}: assert message is a tuple, "
"which is always truthy and prints as a tuple repr"
)
Comment thread
devin-ai-integration[bot] marked this conversation as resolved.
continue

for placeholder in _uninterpolated_placeholders(node.msg):
problems.append(
f"{source_file.name}:{node.lineno}: assert message contains "
f"{placeholder} but the fragment is missing the `f` prefix"
)

assert not problems, "\n".join(problems)
Loading