diff --git a/api/tests/unit/test_sql_migrations_pg15_compat.py b/api/tests/unit/test_sql_migrations_pg15_compat.py new file mode 100644 index 0000000..3ba0c0f --- /dev/null +++ b/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