sdlc-hub/tests/test_db.py

import logging
import os
import sqlite3
import stat
import threading

from jira_hub import backup, db, paths, queue, settings


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"]


def _mode(p):
    return stat.S_IMODE(p.stat().st_mode)


def test_data_dir_is_private(tmp_path):
    home = tmp_path / "data"
    home.mkdir()
    home.chmod(0o755)
    db.connect(home / "hub.sqlite").close()
    assert _mode(home) == 0o700


def test_existing_backups_dir_made_private_on_start(tmp_path):
    from jira_hub import api
    backups = tmp_path / "backups"
    backups.mkdir()
    backups.chmod(0o755)
    api.Hub(tmp_path / "hub.sqlite", backups_dir=backups)
    assert _mode(backups) == 0o700


def test_connect_without_directory_does_not_chmod_cwd(tmp_path, monkeypatch):
    monkeypatch.chdir(tmp_path)
    tmp_path.chmod(0o755)
    db.connect("hub.sqlite").close()
    assert _mode(tmp_path) == 0o755


def test_backups_dir_is_private(tmp_path):
    conn = db.connect(tmp_path / "hub.sqlite")
    backup.make_backup(conn, tmp_path / "backups", "batch-1")
    conn.close()
    assert _mode(tmp_path / "backups") == 0o700


def test_secure_dir_failure_only_warns(tmp_path, monkeypatch, caplog):
    def deny(*a, **k):
        raise PermissionError("нет прав")
    monkeypatch.setattr(os, "chmod", deny)
    with caplog.at_level(logging.WARNING):
        paths.secure_dir(tmp_path / "x")
    assert (tmp_path / "x").is_dir() and "0700" in caplog.text


def test_parallel_connect_makes_one_backup(tmp_path):
    path = tmp_path / "hub.sqlite"
    _v1_database(path)
    barrier, errors = threading.Barrier(4), []

    def go():
        try:
            barrier.wait()
            db.connect(path).close()
        except Exception as e:  # noqa: BLE001
            errors.append(e)

    threads = [threading.Thread(target=go) for _ in range(4)]
    for t in threads:
        t.start()
    for t in threads:
        t.join(30)
    assert errors == []
    assert len(list((tmp_path / "backups").glob("*-schema-v1.sqlite"))) == 1
    conn = db.connect(path)
    assert conn.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone()["value"] == "2"
    conn.close()


def test_migrated_claimed_redelivered_end_to_end(tmp_path):
    """База v1 с командой в claimed -> миграция (deliveries = 1) -> claim довыдаёт её с deliveries = 2 (M-10 #12)."""
    path = tmp_path / "hub.sqlite"
    _v1_database(path)
    conn = db.connect(path)
    settings.ensure_defaults(conn)
    cmd = queue.claim(conn)
    assert cmd["id"] == 1
    assert queue.get_command(conn, 1)["deliveries"] == 2
    conn.close()


def test_sqlite_without_unix_vfs_deadlock():
    # O-14 (#29): в SQLite 3.51.0–3.51.1 взаимная блокировка в unix-VFS (исправлена в 3.51.2) вешала хаб под
    # параллельными запросами — потоки ждали друг друга в sqlite3.connect/close. Python хаба — от uv (pyproject, tool.uv)
    assert sqlite3.sqlite_version_info[:2] != (3, 51) or sqlite3.sqlite_version_info >= (3, 51, 2), sqlite3.sqlite_version