68b73439a6
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014Typ5sXWNMFirr2va2uL8o
72 lines
2.9 KiB
Python
72 lines
2.9 KiB
Python
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"] == "1"
|
|
assert conn.execute("SELECT COUNT(*) AS n FROM agent_status").fetchone()["n"] == 1
|
|
|
|
|
|
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"]
|