test(alembic): регрессия на подзапросы в FROM без алиаса (PostgreSQL 15)

Co-Authored-By: Claude Haiku 5.5 <noreply@anthropic.com>

api/tests/unit/test_sql_migrations_pg15_compat.py

@@ -0,0 +1,126 @@
+"""Регрессия: подзапрос в FROM без алиаса падает на PostgreSQL 15.
+
+PostgreSQL 16 принимает `select count(*) from (select ...)` без алиаса,
+PostgreSQL 15 — нет («subquery in FROM must have an alias»). Тест сканирует
+миграции и требует алиас у каждого подзапроса в FROM.
+"""
+
+import re
+from pathlib import Path
+
+API_ROOT = Path(__file__).resolve().parents[2]
+VERSIONS_DIR = API_ROOT / "alembic" / "versions"
+
+FROM_SUBQUERY_RE = re.compile(r"\bFROM\s*\(", re.IGNORECASE)
+SUBQUERY_START_RE = re.compile(r"\s*(SELECT|WITH|VALUES)\b", re.IGNORECASE)
+# Слова, которые могут идти сразу после `)` и не являются алиасом.
+NON_ALIAS_KEYWORDS = {
+    "WHERE", "GROUP", "ORDER", "LIMIT", "UNION", "ON", "JOIN", "LEFT", "RIGHT",
+    "FULL", "INNER", "OUTER", "CROSS", "NATURAL", "HAVING", "WINDOW", "OFFSET",
+    "EXCEPT", "INTERSECT", "FETCH", "FOR", "USING", "LATERAL", "TABLESAMPLE",
+}
+ALIAS_RE = re.compile(r'\s*(?:AS\s+)?([A-Za-z_][A-Za-z0-9_$]*|"[^"]+")', re.IGNORECASE)
+AS_ONLY_RE = re.compile(r"\s*AS\b", re.IGNORECASE)
+
+
+def _matching_close_paren(text: str, open_idx: int) -> int:
+    """Индекс `)`, парной к `(` на open_idx; строки и комментарии пропускаются."""
+    depth = 0
+    i = open_idx
+    n = len(text)
+    while i < n:
+        ch = text[i]
+        if text.startswith("--", i):
+            nl = text.find("\n", i)
+            i = n if nl == -1 else nl
+            continue
+        if text.startswith("/*", i):
+            end = text.find("*/", i + 2)
+            i = n if end == -1 else end + 2
+            continue
+        if ch == "'":
+            i += 1
+            while i < n:
+                if text[i] == "'":
+                    if i + 1 < n and text[i + 1] == "'":
+                        i += 2
+                        continue
+                    break
+                i += 1
+            i += 1
+            continue
+        if ch == "(":
+            depth += 1
+        elif ch == ")":
+            depth -= 1
+            if depth == 0:
+                return i
+        i += 1
+    return -1
+
+
+def _has_alias_after(text: str, close_idx: int) -> bool:
+    rest = text[close_idx + 1 :]
+    if AS_ONLY_RE.match(rest):
+        # `) AS <имя>` — имя обязательно
+        m = ALIAS_RE.match(rest)
+        return bool(m)
+    m = ALIAS_RE.match(rest)
+    if not m:
+        return False
+    word = m.group(1)
+    if word.startswith('"'):
+        return True
+    return word.upper() not in NON_ALIAS_KEYWORDS
+
+
+def _find_violations() -> list[str]:
+    files = sorted((VERSIONS_DIR / "sql").glob("*.sql")) + sorted(VERSIONS_DIR.glob("*.py"))
+    violations: list[str] = []
+    for path in files:
+        text = path.read_text(encoding="utf-8")
+        for m in FROM_SUBQUERY_RE.finditer(text):
+            open_idx = m.end() - 1
+            if not SUBQUERY_START_RE.match(text, open_idx + 1):
+                continue  # `FROM (t1 JOIN t2 ...)` и подобное — не подзапрос
+            close_idx = _matching_close_paren(text, open_idx)
+            if close_idx == -1:
+                continue
+            if not _has_alias_after(text, close_idx):
+                line = text.count("\n", 0, m.start()) + 1
+                rel = path.relative_to(API_ROOT)
+                violations.append(f"{rel}:{line}")
+    return violations
+
+
+def test_subqueries_in_from_have_alias():
+    violations = _find_violations()
+    assert not violations, (
+        "Подзапрос в FROM без алиаса (PostgreSQL 15 такое не принимает, "
+        "только 16): " + ", ".join(violations) + ". Добавьте алиас, например `) AS sub`."
+    )
+
+
+def test_scanner_detects_missing_alias_and_accepts_alias():
+    """Самопроверка сканера на синтетическом тексте."""
+    bad = "select count(*) from (\n  select 1 from t\n) where x = 1"
+    good = "select count(*) from (\n  select 1 from t\n) AS s where x = 1"
+    good_no_as = "select count(*) from (select 1 from t) s"
+    values_good = "FROM (VALUES ('a'), ('b')) AS v(x)"
+    not_subquery = "a IS NOT DISTINCT FROM (v.data->>'x')"
+
+    def check(sql: str) -> bool:
+        m = FROM_SUBQUERY_RE.search(sql)
+        if not m:
+            return True
+        open_idx = m.end() - 1
+        if not SUBQUERY_START_RE.match(sql, open_idx + 1):
+            return True
+        close_idx = _matching_close_paren(sql, open_idx)
+        return _has_alias_after(sql, close_idx)
+
+    assert check(bad) is False
+    assert check(good) is True
+    assert check(good_no_as) is True
+    assert check(values_good) is True
+    assert check(not_subquery) is True