70 lines
1.8 KiB
Python
70 lines
1.8 KiB
Python
"""Tests for migrations/001_initial_schema.sql via scripts/db_migrate.py."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
def test_migration_creates_documents_table(tmp_path):
|
|
db_path = str(tmp_path / "test.db")
|
|
schema = Path("migrations/001_initial_schema.sql").read_text()
|
|
|
|
conn = sqlite3.connect(db_path)
|
|
conn.executescript(schema)
|
|
conn.commit()
|
|
|
|
tables = {
|
|
r[0]
|
|
for r in conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table'"
|
|
).fetchall()
|
|
}
|
|
assert "documents" in tables
|
|
assert "page_chunks" in tables
|
|
|
|
|
|
def test_documents_table_has_required_columns(tmp_path):
|
|
db_path = str(tmp_path / "test.db")
|
|
schema = Path("migrations/001_initial_schema.sql").read_text()
|
|
|
|
conn = sqlite3.connect(db_path)
|
|
conn.executescript(schema)
|
|
|
|
cols = {r[1] for r in conn.execute("PRAGMA table_info(documents)").fetchall()}
|
|
assert {
|
|
"id",
|
|
"title",
|
|
"file_path",
|
|
"status",
|
|
"task_id",
|
|
"page_count",
|
|
"created_at",
|
|
} <= cols
|
|
|
|
|
|
def test_page_chunks_foreign_key_cascades(tmp_path):
|
|
db_path = str(tmp_path / "test.db")
|
|
schema = Path("migrations/001_initial_schema.sql").read_text()
|
|
|
|
conn = sqlite3.connect(db_path)
|
|
conn.executescript(schema)
|
|
conn.execute("PRAGMA foreign_keys = ON")
|
|
|
|
conn.execute(
|
|
"INSERT INTO documents(id, title, file_path, status) VALUES ('d1','Book','path.pdf','ready')"
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO page_chunks(doc_id, page_number, text, source, word_count) VALUES ('d1',1,'text','text_layer',1)"
|
|
)
|
|
conn.commit()
|
|
|
|
conn.execute("DELETE FROM documents WHERE id='d1'")
|
|
conn.commit()
|
|
|
|
count = conn.execute(
|
|
"SELECT COUNT(*) FROM page_chunks WHERE doc_id='d1'"
|
|
).fetchone()[0]
|
|
assert count == 0
|