File size: 5,303 Bytes
cc2ed2f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 | """Тесты на DbConnector и SqlExecutor.
Покрывают чтение схемы SQLite-баз, генерацию DDL и проверку того, что
SQLite-подключение действительно открывается в режиме read-only —
модифицирующие операции должны падать с sqlite3.OperationalError.
"""
import sqlite3
from pathlib import Path
import pytest
from src.db.connector import DbConnector, TableInfo
from src.db.executor import QueryResult, SqlExecutor
@pytest.fixture
def tiny_sqlite(tmp_path: Path) -> Path:
db = tmp_path / "tiny.sqlite"
conn = sqlite3.connect(db)
conn.execute(
"CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT NOT NULL, city TEXT)"
)
conn.executemany(
"INSERT INTO users (id, name, city) VALUES (?, ?, ?)",
[(1, "Иван", "Казань"), (2, "Анна", "Москва"), (3, "Олег", "Казань")],
)
conn.commit()
conn.close()
return db
# ──────────────────────────────────────────────────────────────────────
# DbConnector
# ──────────────────────────────────────────────────────────────────────
def test_connector_lists_tables(tiny_sqlite: Path):
c = DbConnector(str(tiny_sqlite))
assert c.list_tables() == ["users"]
def test_connector_reads_columns(tiny_sqlite: Path):
c = DbConnector(str(tiny_sqlite))
tables = c.get_schema(include_samples=False)
assert len(tables) == 1
table = tables[0]
assert isinstance(table, TableInfo)
names = [col.name for col in table.columns]
assert names == ["id", "name", "city"]
# id — primary key, name — NOT NULL
pk = next(col for col in table.columns if col.name == "id")
assert pk.primary_key is True
nn = next(col for col in table.columns if col.name == "name")
assert nn.nullable is False
def test_connector_renders_ddl(tiny_sqlite: Path):
c = DbConnector(str(tiny_sqlite))
schema_text = c.render_schema(include_samples=True)
assert "CREATE TABLE users" in schema_text
assert "PRIMARY KEY" in schema_text
# sample-строки прокинуты комментариями
assert "Иван" in schema_text or "Олег" in schema_text
def test_connector_accepts_sqlite_uri(tiny_sqlite: Path):
c = DbConnector(f"sqlite:///{tiny_sqlite}")
assert c.list_tables() == ["users"]
# ──────────────────────────────────────────────────────────────────────
# SqlExecutor
# ──────────────────────────────────────────────────────────────────────
def test_executor_runs_select(tiny_sqlite: Path):
ex = SqlExecutor(str(tiny_sqlite))
res = ex.run("SELECT id, name FROM users ORDER BY id")
assert isinstance(res, QueryResult)
assert res.success
assert res.columns == ["id", "name"]
assert res.row_count == 3
assert res.rows[0] == [1, "Иван"]
def test_executor_aggregation(tiny_sqlite: Path):
ex = SqlExecutor(str(tiny_sqlite))
res = ex.run("SELECT city, COUNT(*) AS cnt FROM users GROUP BY city ORDER BY cnt DESC")
assert res.success
assert res.rows[0] == ["Казань", 2]
def test_executor_returns_error_on_bad_sql(tiny_sqlite: Path):
ex = SqlExecutor(str(tiny_sqlite))
res = ex.run("SELEC nonsense FROM users")
assert not res.success
assert res.error is not None
def test_executor_blocks_modifications(tiny_sqlite: Path):
"""Ключевая проверка: SQLite-соединение открывается в read-only
режиме (URI mode=ro&immutable=1), модифицирующие операции должны
падать ошибкой, а не выполняться втихую."""
ex = SqlExecutor(str(tiny_sqlite))
res = ex.run("DELETE FROM users WHERE id = 1")
assert not res.success
assert res.error is not None
assert "read" in res.error.lower() or "readonly" in res.error.lower() \
or "только для чтения" in res.error.lower()
# Подтверждение, что данные не пострадали
check = ex.run("SELECT COUNT(*) FROM users")
assert check.success
assert check.rows == [[3]]
def test_executor_blocks_drop_table(tiny_sqlite: Path):
ex = SqlExecutor(str(tiny_sqlite))
res = ex.run("DROP TABLE users")
assert not res.success
# Подтверждение, что таблица на месте
check = ex.run("SELECT COUNT(*) FROM users")
assert check.success
def test_queryresult_to_markdown(tiny_sqlite: Path):
ex = SqlExecutor(str(tiny_sqlite))
res = ex.run("SELECT id, name FROM users WHERE id = 1")
md = res.to_markdown_table()
assert "id" in md and "name" in md
assert "Иван" in md
|