113 lines
5.2 KiB
Python
113 lines
5.2 KiB
Python
import sqlite3
|
||
|
||
from jira_hub import db
|
||
|
||
|
||
def test_connect_creates_schema():
|
||
conn = db.connect(":memory:")
|
||
names = {r["name"] for r in conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'")}
|
||
assert {"setting", "issue", "issue_link", "issue_note", "batch", "command", "sync_run", "agent_status"} <= names
|
||
assert conn.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone()["value"] == "2"
|
||
assert conn.execute("SELECT COUNT(*) AS n FROM agent_status").fetchone()["n"] == 1
|
||
cols = {r["name"] for r in conn.execute("PRAGMA table_info(command)")}
|
||
assert "deliveries" in cols
|
||
|
||
|
||
def _v1_database(path):
|
||
"""База в состоянии схемы v1: без command.deliveries, schema_version = 1, с живыми данными."""
|
||
conn = db.connect(path)
|
||
conn.execute("INSERT INTO batch(title, author, created_at) VALUES ('x', 'claude', '2026-10-13T00:00:00+00:00')")
|
||
for status in ("claimed", "done", "queued"):
|
||
conn.execute("INSERT INTO command(batch_id, seq, type, params, status, claimed_at) VALUES (1, 1, 'refresh', '{}', ?, ?)",
|
||
(status, "2026-10-13T00:00:00+00:00" if status == "claimed" else None))
|
||
conn.execute("INSERT INTO issue_note(key, fact_status) VALUES ('A-1', 'В работе')")
|
||
conn.execute("ALTER TABLE command DROP COLUMN deliveries")
|
||
conn.execute("UPDATE setting SET value = '1' WHERE key = 'schema_version'")
|
||
conn.close()
|
||
|
||
|
||
def test_migrates_v1_with_backup(tmp_path):
|
||
path = tmp_path / "hub.sqlite"
|
||
_v1_database(path)
|
||
conn = db.connect(path)
|
||
assert conn.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone()["value"] == "2"
|
||
rows = conn.execute("SELECT status, deliveries FROM command ORDER BY id").fetchall()
|
||
assert [(r["status"], r["deliveries"]) for r in rows] == [("claimed", 1), ("done", 0), ("queued", 0)]
|
||
assert conn.execute("SELECT fact_status FROM issue_note WHERE key = 'A-1'").fetchone()["fact_status"] == "В работе"
|
||
copies = list((tmp_path / "backups").glob("*-schema-v1.sqlite"))
|
||
assert len(copies) == 1
|
||
old = sqlite3.connect(copies[0]) # не db.connect — он бы мигрировал копию
|
||
assert "deliveries" not in {r[1] for r in old.execute("PRAGMA table_info(command)")}
|
||
assert old.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone()[0] == "1"
|
||
old.close()
|
||
conn.close()
|
||
db.connect(path).close() # повторный connect не мигрирует и не копирует
|
||
assert len(list((tmp_path / "backups").glob("*.sqlite"))) == 1
|
||
|
||
|
||
def test_new_database_makes_no_backup(tmp_path):
|
||
db.connect(tmp_path / "hub.sqlite").close()
|
||
assert not (tmp_path / "backups").exists()
|
||
|
||
|
||
def test_connect_is_idempotent(tmp_path):
|
||
path = tmp_path / "hub.sqlite"
|
||
db.connect(path).close()
|
||
conn = db.connect(path)
|
||
assert conn.execute("SELECT COUNT(*) AS n FROM agent_status").fetchone()["n"] == 1
|
||
|
||
|
||
def test_transaction_rolls_back_on_error(tmp_path):
|
||
conn = db.connect(tmp_path / "hub.sqlite")
|
||
try:
|
||
with db.transaction(conn):
|
||
conn.execute("INSERT INTO batch(title, author, created_at) VALUES ('x', 'y', 'z')")
|
||
raise RuntimeError("boom")
|
||
except RuntimeError:
|
||
pass
|
||
assert conn.execute("SELECT COUNT(*) AS n FROM batch").fetchone()["n"] == 0
|
||
|
||
|
||
def test_nested_transaction_joins_outer(tmp_path):
|
||
conn = db.connect(tmp_path / "hub.sqlite")
|
||
try:
|
||
with db.transaction(conn):
|
||
with db.transaction(conn):
|
||
conn.execute("INSERT INTO batch(title, author, created_at) VALUES ('x', 'y', 'z')")
|
||
assert conn.in_transaction # вложенный блок не закоммитил сам
|
||
assert conn.execute("SELECT COUNT(*) AS n FROM batch").fetchone()["n"] == 1
|
||
raise RuntimeError("boom")
|
||
except RuntimeError:
|
||
pass
|
||
assert conn.execute("SELECT COUNT(*) AS n FROM batch").fetchone()["n"] == 0 # откат внешней откатил и вложенную
|
||
|
||
|
||
def test_nested_transaction_error_propagates_to_outer(tmp_path):
|
||
conn = db.connect(tmp_path / "hub.sqlite")
|
||
try:
|
||
with db.transaction(conn):
|
||
conn.execute("INSERT INTO batch(title, author, created_at) VALUES ('a', 'y', 'z')")
|
||
with db.transaction(conn):
|
||
raise RuntimeError("inner")
|
||
except RuntimeError:
|
||
pass
|
||
assert conn.execute("SELECT COUNT(*) AS n FROM batch").fetchone()["n"] == 0
|
||
|
||
|
||
def test_parse_and_utcnow():
|
||
assert db.parse("2020-01-01T00:00:00").tzinfo is not None
|
||
assert db.parse(db.now()) <= db.utcnow()
|
||
|
||
|
||
def test_nested_transaction_rollback_is_local(tmp_path):
|
||
conn = db.connect(tmp_path / "hub.sqlite")
|
||
with db.transaction(conn):
|
||
conn.execute("INSERT INTO batch(title, author, created_at) VALUES ('outer', 'y', 'z')")
|
||
try:
|
||
with db.transaction(conn):
|
||
conn.execute("INSERT INTO batch(title, author, created_at) VALUES ('inner', 'y', 'z')")
|
||
raise RuntimeError("inner")
|
||
except RuntimeError:
|
||
pass
|
||
assert [r["title"] for r in conn.execute("SELECT title FROM batch")] == ["outer"]
|