You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
85 lines
3.0 KiB
85 lines
3.0 KiB
import os
|
|
import re
|
|
import pathlib
|
|
import pytest
|
|
# small migration filename/header tests; keep imports minimal
|
|
|
|
|
|
MIGRATION_FILENAME = "2026-03-22-add-similarity-cache.sql"
|
|
MIGRATION_PATH = pathlib.Path("migrations") / MIGRATION_FILENAME
|
|
|
|
|
|
def _strip_sql_comments(sql: str) -> str:
|
|
"""Remove SQL single-line (-- ...) and C-style (/* ... */) comments.
|
|
|
|
This is a best-effort stripper sufficient for the test's purpose.
|
|
"""
|
|
# remove block comments
|
|
sql = re.sub(r"/\*.*?\*/", "", sql, flags=re.S)
|
|
# remove line comments
|
|
sql = re.sub(r"--.*?$", "", sql, flags=re.M)
|
|
return sql.strip()
|
|
|
|
|
|
def test_migration_file_exists_and_header():
|
|
# file must exist
|
|
assert MIGRATION_PATH.exists(), f"Migration file {MIGRATION_PATH} not found"
|
|
|
|
text = MIGRATION_PATH.read_text(encoding="utf8")
|
|
|
|
# header should reference the filename and purpose
|
|
assert MIGRATION_FILENAME in text.splitlines()[0], (
|
|
"First line should include the filename"
|
|
)
|
|
assert "similarity" in text.lower(), "Header should mention similarity"
|
|
|
|
|
|
def test_optional_apply_migration_safe():
|
|
# If TEST_DB_URL is set, try to apply the SQL only if it contains non-comment statements.
|
|
db_url = os.environ.get("TEST_DB_URL")
|
|
sql = MIGRATION_PATH.read_text(encoding="utf8")
|
|
stripped = _strip_sql_comments(sql)
|
|
|
|
# If there is no DB url, consider this a filename/header validation test only.
|
|
if not db_url:
|
|
pytest.skip("TEST_DB_URL not set; skipping DB apply step")
|
|
|
|
# If the SQL is empty (only comments), nothing to apply — test passes.
|
|
if not stripped:
|
|
pytest.skip("Migration contains no executable SQL; nothing to apply")
|
|
|
|
# Otherwise attempt to execute the SQL. Be conservative: if drivers are missing or
|
|
# connection fails, skip the test rather than failing CI. Only unexpected errors
|
|
# during execution should fail the test.
|
|
try:
|
|
if db_url.startswith("sqlite:"):
|
|
import sqlite3
|
|
|
|
# sqlite URL might be sqlite:///path or sqlite:///:memory:
|
|
path = db_url.split("sqlite:", 1)[1]
|
|
# normalize prefixes like ///
|
|
path = path.lstrip("/") or ":memory:"
|
|
conn = sqlite3.connect(path)
|
|
try:
|
|
conn.executescript(sql)
|
|
finally:
|
|
conn.close()
|
|
elif db_url.startswith("postgresql:") or db_url.startswith("postgres:"):
|
|
try:
|
|
import psycopg2
|
|
except Exception as e: # pragma: no cover - driver may be absent in CI
|
|
pytest.skip(f"psycopg2 not available: {e}")
|
|
|
|
# psycopg2 accepts a DSN; rely on that here.
|
|
conn = psycopg2.connect(db_url)
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute(sql)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
else:
|
|
pytest.skip(f"DB URL scheme not supported by this test: {db_url}")
|
|
except Exception as exc:
|
|
# Unexpected error while applying SQL should fail the test.
|
|
raise
|
|
|