fix: keep empty-fixture lint exemption after non-ASCII text

src/sqlbuild/lint/_helpers/native_format.py

@@ -68,7 +68,9 @@ def format_native_sql_bodies(
             body: str = contents[body_start:body_end]
             trailing: str = body[len(body.rstrip()) :]
             core: str = body[: len(body) - len(trailing)] if trailing else body
-            neutralized, sites = neutralize_interpolation(body=core, dialect=config.dialect)
+            neutralized, sites = neutralize_interpolation(
+                body=core, dialect=config.dialect, for_formatting=True
+            )
             cache_key: tuple[str, str] = (neutralized, config.dialect)
             requests_by_key.setdefault(
                 cache_key,

src/sqlbuild/lint/_helpers/sqlbuild_tokens.py

@@ -30,39 +30,49 @@
 from sqlbuild.lint.exceptions import InterpolationRestorationError
 from sqlbuild.lint.models import InterpolationSite
 
-_SQLBUILD_FUNCTION_NAMES: tuple[str, ...] = (
-    "__cursor_start",
-    "__cursor_end",
-    "__empty_fixture",
+_SQLBUILD_REFERENCE_FUNCTION_NAMES: tuple[str, ...] = (
     "__dbt_ref",
     "__table_fn",
     "__source",
     "__seed",
     "__udf",
     "__ref",
 )
+_SQLBUILD_FUNCTION_NAMES: tuple[str, ...] = (
+    "__cursor_start",
+    "__cursor_end",
+    "__empty_fixture",
+    *_SQLBUILD_REFERENCE_FUNCTION_NAMES,
+)
 _AUDIT_PARAMETER_SENTINEL_TEMPLATE: str = "__sqlbuild_audit_parameter_{index}__"
 _CONTEXT_SENTINEL_TEMPLATE: str = "__sqlbuild_context_parameter_{index}__"
 _LINE_COMMENT_START: str = "--"
 _BLOCK_COMMENT_START: str = "/*"
 _BLOCK_COMMENT_END: str = "*/"
-_SQLBUILD_FUNCTION_PATTERN: str = "|".join(
-    re.escape(name) for name in sorted(_SQLBUILD_FUNCTION_NAMES)
-)
 _NON_CODE_SCAN_PATTERN: str = (
     r"--[^\n]*(?:\n|\Z)"
     r"|/\*[\s\S]*?(?:\*/|\Z)"
     r"|'(?:\\.|''|[^'\\])*(?:'|\Z)"
     r'|"(?:\\.|""|[^"\\])*(?:"|\Z)'
 )
 _BACKTICK_SCAN_PATTERN: str = r"|`(?:``|[^`])*(?:`|\Z)"
-_SITE_SCAN_PATTERN: str = rf"|(?P<site>@@|\$\{{|@|(?:{_SQLBUILD_FUNCTION_PATTERN})\s*\()"
-_INTERPOLATION_SCAN_PATTERN: re.Pattern[str] = re.compile(
-    _NON_CODE_SCAN_PATTERN + _SITE_SCAN_PATTERN, re.IGNORECASE
-)
-_BACKTICK_INTERPOLATION_SCAN_PATTERN: re.Pattern[str] = re.compile(
-    _NON_CODE_SCAN_PATTERN + _BACKTICK_SCAN_PATTERN + _SITE_SCAN_PATTERN, re.IGNORECASE
-)
+
+
+@cache
+def _interpolation_scan_pattern(
+    *, backtick_identifiers: bool, for_formatting: bool
+) -> re.Pattern[str]:
+    """Keep lint fallback lexical policy identical to the native ASCII fast path."""
+
+    names: tuple[str, ...] = (
+        _SQLBUILD_FUNCTION_NAMES if for_formatting else _SQLBUILD_REFERENCE_FUNCTION_NAMES
+    )
+    functions: str = "|".join(re.escape(name) for name in sorted(names))
+    whitespace: str = r"\s*" if for_formatting else ""
+    sites: str = rf"|(?P<site>@@|\$\{{|@|(?:{functions}){whitespace}\()"
+    backticks: str = _BACKTICK_SCAN_PATTERN if backtick_identifiers else ""
+    flags: re.RegexFlag = re.IGNORECASE if for_formatting else re.NOFLAG
+    return re.compile(_NON_CODE_SCAN_PATTERN + backticks + sites, flags)
 
 
 @cache
@@ -142,17 +152,15 @@ def neutralize_generic_audit_parameters(*, body: str) -> tuple[str, tuple[Interp
 
 
 def neutralize_interpolation(
-    *, body: str, dialect: str
+    *, body: str, dialect: str, for_formatting: bool = False
 ) -> tuple[str, tuple[InterpolationSite, ...]]:
-    """Replace every interpolation site with a unique sentinel identifier."""
+    """Neutralize references for lint, or protect every intrinsic spelling for formatting."""
 
     if not _contains_interpolation_candidate(body=body):
         return body, ()
     backtick_identifiers: bool = _backtick_identifiers(dialect)
-    scan_pattern: re.Pattern[str] = (
-        _BACKTICK_INTERPOLATION_SCAN_PATTERN
-        if backtick_identifiers
-        else _INTERPOLATION_SCAN_PATTERN
+    scan_pattern: re.Pattern[str] = _interpolation_scan_pattern(
+        backtick_identifiers=backtick_identifiers, for_formatting=for_formatting
     )
     sites: list[InterpolationSite] = []
     pieces: list[str] = []

tests/e2e/src/sqlbuild/cli/commands/main/playground/_test_types.py

@@ -21,3 +21,10 @@ class PythonNodesPlaygroundLifecycleTestCase:
     expected_plan_fragments: tuple[str, ...]
     expected_build_fragments: tuple[str, ...]
     expected_check_fragments: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class UnicodeEmptyFixtureTestCase:
+    description: str
+    status: str
+    expected_added_findings: int = 0

tests/e2e/src/sqlbuild/cli/commands/main/playground/test_unicode_empty_fixture.py

@@ -0,0 +1,81 @@
+"""Unicode fixture exemptions through the generated playground and real CLI."""
+
+import json
+import re
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from tests.e2e.src.sqlbuild.cli.commands.main.playground._test_types import (
+    UnicodeEmptyFixtureTestCase,
+)
+from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import run_sqb
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        UnicodeEmptyFixtureTestCase(text, text)
+        for text in ("complété", "total €", "order 😀", "complété € 😀")
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_unicode_playground_fixture_when_running_rules_and_format_then_empty_input_is_allowed(
+    test_case: UnicodeEmptyFixtureTestCase, tmp_path: Path
+) -> None:
+    project: Path = tmp_path / "orders_playground"
+    scaffold: subprocess.CompletedProcess[str] = run_sqb(
+        command=("playground", str(project)), project_dir=tmp_path
+    )
+    assert scaffold.returncode == 0, scaffold.stdout + scaffold.stderr
+    fixture: Path = project / "tests" / "unit" / "test_fact_orders.sql"
+    sql: str = fixture.read_text(encoding="utf-8")
+    sql = re.sub(
+        r"__ref__stg_payments AS \(.*?\n\)",
+        "__ref__stg_payments AS (\n  SELECT * FROM __EMPTY_FIXTURE()\n)",
+        sql,
+        flags=re.DOTALL,
+    )
+    fixture.write_text(sql, encoding="utf-8")
+    baseline: subprocess.CompletedProcess[str] = run_sqb(
+        command=("rules", "--json", "run", "SQBRSQL021"), project_dir=project
+    )
+    baseline_count: int = json.loads(baseline.stdout)["finding_count"]
+    fixture.write_text(
+        sql.replace("'completed' AS status", f"'{test_case.status}' AS status"), encoding="utf-8"
+    )
+    before: subprocess.CompletedProcess[str] = run_sqb(
+        command=("rules", "--json", "run", "SQBRSQL021"), project_dir=project
+    )
+    assert before.returncode == baseline.returncode, before.stdout + before.stderr
+    assert (
+        json.loads(before.stdout)["finding_count"] - baseline_count
+        == test_case.expected_added_findings
+    )
+    assert json.loads(before.stdout)["findings"] == json.loads(baseline.stdout)["findings"]
+    formatted_result: subprocess.CompletedProcess[str] = run_sqb(
+        command=("format", "--json", str(fixture)), project_dir=project
+    )
+    assert formatted_result.returncode == 0, formatted_result.stdout + formatted_result.stderr
+    formatted: str = fixture.read_text(encoding="utf-8")
+    assert test_case.status in formatted
+    assert "__EMPTY_FIXTURE()" in formatted
+    after: subprocess.CompletedProcess[str] = run_sqb(
+        command=("rules", "--json", "run", "SQBRSQL021"), project_dir=project
+    )
+    assert after.returncode == baseline.returncode, after.stdout + after.stderr
+    assert (
+        json.loads(after.stdout)["finding_count"] - baseline_count
+        == test_case.expected_added_findings
+    )
+    checked: subprocess.CompletedProcess[str] = run_sqb(
+        command=("format", "--check", "--json", str(fixture)),
+        project_dir=project,
+    )
+    assert checked.returncode == 0, checked.stdout + checked.stderr
+    assert fixture.read_text(encoding="utf-8") == formatted
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-vv"])

tests/integration/src/sqlbuild/cli/commands/main/_test_types.py

@@ -697,3 +697,13 @@ class FormatterSyntaxTestCase:
     sql: str
     expected_fragments: tuple[str, ...]
     macro_source: str = ""
+
+
+@dataclass(frozen=True)
+class UnicodeRuleLocationTestCase:
+    """SQL rule diagnostics expressed in authored Unicode character coordinates."""
+
+    description: str
+    rule_code: str
+    sql: str
+    expected_anchors: tuple[str, ...]

tests/integration/src/sqlbuild/cli/commands/main/test_format_preservation.py

@@ -239,5 +239,62 @@ def test_given_cte_macros_and_comments_when_formatting_then_layout_and_rules_sta
     assert payload["findings"] == []
 
 
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        FormatterSyntaxTestCase(
+            f"Unicode restoration after {prefix}",
+            f"WITH orders AS (SELECT '{prefix}' AS label, order_id, payload FROM __ref(\"raw_orders\")),\n"
+            "values_input AS (SELECT column1 AS order_flag FROM VALUES (1)),\n"
+            f'-- Stage {prefix} orders.\n@copy_orders("staged_orders", "orders"),\n'
+            "final AS (SELECT label, "
+            f"CAST(order_id AS NUMERIC(/* type {prefix} */ 10, 2)) AS order_id, "
+            "payload:customer.name::VARCHAR AS customer_name, __cursor_start() AS start_at, "
+            "__cursor_end() AS end_at, FIRST_VALUE(order_id IGNORE NULLS) OVER (ORDER BY order_id) AS first_id, "
+            f"POSITION('{prefix}' IN label) AS label_position FROM staged_orders s CROSS JOIN values_input) SELECT * FROM final",
+            (
+                f"'{prefix}'",
+                f"-- Stage {prefix} orders.",
+                f"/* type {prefix} */",
+                "NUMERIC(",
+                "payload:customer.name::VARCHAR",
+                '__ref("raw_orders")',
+                "__cursor_start()",
+                "__cursor_end()",
+                "FROM VALUES (1)",
+                "FIRST_VALUE(order_id IGNORE NULLS)",
+                f"POSITION('{prefix}' IN label)",
+                '\n@copy_orders("staged_orders", "orders"),\n',
+                "\nfinal AS (",
+            ),
+        )
+        for prefix in ("é", "€", "😀", "é € 😀")
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_multibyte_text_before_restorations_when_formatting_then_output_is_preserved(
+    test_case: FormatterSyntaxTestCase, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    (tmp_path / "sqlbuild_project.toml").write_text('name = "orders"\nadapter = "snowflake"\n')
+    macros: Path = tmp_path / "models" / "_macros"
+    macros.mkdir(parents=True)
+    (macros / "orders.py").write_text(
+        "def copy_orders(name: str, source: str) -> str:\n"
+        '    return f"{name} AS (SELECT * FROM {source})"\n'
+    )
+    model: Path = tmp_path / "models" / "order_summary.sql"
+    model.write_text(
+        'MODEL (description "Orders é € 😀");\n' + test_case.sql + "\n", encoding="utf-8"
+    )
+    assert main(["--project-dir", str(tmp_path), "format", "--json"]) == 0
+    payload: dict[str, object] = json.loads(capsys.readouterr().out)
+    assert payload["faults"] == 0
+    formatted: str = model.read_text(encoding="utf-8")
+    assert all(fragment in formatted for fragment in test_case.expected_fragments)
+    assert 'description "Orders é € 😀"' in formatted
+    assert main(["--project-dir", str(tmp_path), "format", "--check", "--json"]) == 0
+    assert model.read_text(encoding="utf-8") == formatted
+
+
 if __name__ == "__main__":
     pytest.main([__file__, "-vv"])