sdlc-hub/tests/test_db.py

import json
import logging
import os
import sqlite3
import stat
import threading

import pytest

from sdlc_hub.core import api, backup, db, paths, projects, queue, settings
from sdlc_hub.modules.jira import history


def test_connect_creates_schema(open_db):
    conn = open_db()
    names = {r["name"] for r in conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'")}
    assert {"setting", "issue", "issue_link", "issue_comment", "issue_note", "batch", "command", "sync_run",
            "agent_status", "project", "issue_change"} <= names
    assert "created_at" in {r["name"] for r in conn.execute("PRAGMA table_info(issue_comment)")}
    assert "history_at" in {r["name"] for r in conn.execute("PRAGMA table_info(sync_run)")}
    assert "project" in {r["name"] for r in conn.execute("PRAGMA table_info(batch)")}
    assert {"jira_board", "jira_sprint", "issue_sprint"} <= names
    assert {"rank", "story_points"} <= {r["name"] for r in conn.execute("PRAGMA table_info(issue)")}
    assert conn.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone()["value"] == str(db.SCHEMA_VERSION)
    assert conn.execute("SELECT COUNT(*) AS n FROM agent_status").fetchone()["n"] == 2
    cols = {r["name"] for r in conn.execute("PRAGMA table_info(command)")}
    assert "deliveries" in cols


def _add_fact_status(conn):
    """Колонка issue_note.fact_status, как в базах до v8 (fact-status-removal)."""
    conn.execute("ALTER TABLE issue_note ADD COLUMN fact_status TEXT")


def _note_after_v8(conn, key):
    """После миграции до v8: колонки fact_status нет; work_kind заметки."""
    assert "fact_status" not in {r["name"] for r in conn.execute("PRAGMA table_info(issue_note)")}
    return conn.execute("SELECT work_kind FROM issue_note WHERE key = ?", (key,)).fetchone()["work_kind"]


def _drop_v6(conn):
    """Снять с базы текущей схемы то, что добавила v6 (вложения)."""
    conn.execute("DROP TABLE attachment_file")
    conn.execute("DROP TABLE issue_attachment")


def _drop_v5(conn):
    """Снять с базы текущей схемы то, что добавили v5 и v6: так строятся базы прежних версий."""
    _drop_v6(conn)
    conn.execute("DROP TABLE issue_change")
    conn.execute("ALTER TABLE issue_comment DROP COLUMN created_at")
    conn.execute("ALTER TABLE sync_run DROP COLUMN history_at")


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))
    _add_fact_status(conn)
    conn.execute("INSERT INTO issue_note(key, fact_status, work_kind) VALUES ('A-1', 'В работе', 'Аналитика')")
    _drop_v5(conn)
    conn.execute("ALTER TABLE command DROP COLUMN deliveries")
    conn.execute("ALTER TABLE batch DROP COLUMN project")
    conn.execute("DROP TABLE project")
    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"] == str(db.SCHEMA_VERSION)
    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 _note_after_v8(conn, "A-1") == "Аналитика"
    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"] == 2
    conn.close()


def test_transaction_rolls_back_on_error(tmp_path, open_db):
    conn = open_db(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, open_db):
    conn = open_db(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, open_db):
    conn = open_db(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, open_db):
    conn = open_db(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 sdlc_hub.core 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"] == str(db.SCHEMA_VERSION)
    conn.close()


def test_migrates_v2_with_backup(tmp_path):
    """База v2 (без issue_comment) с задачей и заметкой → v3: копия schema-v2, таблица комментариев пустая, данные на месте."""
    path = tmp_path / "hub.sqlite"
    conn = db.connect(path)
    conn.execute("INSERT INTO issue(key, summary, fetched_at) VALUES ('A-1', 'Т', '2026-10-15T00:00:00+00:00')")
    _add_fact_status(conn)
    conn.execute("INSERT INTO issue_note(key, fact_status, work_kind) VALUES ('A-1', 'В работе', 'Аналитика')")
    _drop_v5(conn)
    conn.execute("DROP TABLE issue_comment")
    conn.execute("ALTER TABLE batch DROP COLUMN project")
    conn.execute("DROP TABLE project")
    conn.execute("UPDATE setting SET value = '2' WHERE key = 'schema_version'")
    conn.close()
    conn = db.connect(path)
    assert conn.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone()["value"] == str(db.SCHEMA_VERSION)
    assert conn.execute("SELECT COUNT(*) AS n FROM issue_comment").fetchone()["n"] == 0
    assert conn.execute("SELECT summary FROM issue WHERE key = 'A-1'").fetchone()["summary"] == "Т"
    assert _note_after_v8(conn, "A-1") == "Аналитика"
    copies = list((tmp_path / "backups").glob("*-schema-v2.sqlite"))
    assert len(copies) == 1
    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, "jira")
    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


def _v3_database(path, base_jql="project = AURORA AND labels = AWPCA", board_id=4216):
    """База в состоянии схемы v3: без project и batch.project, base_jql/board_id в setting, очередь и журнал на старой JQL."""
    conn = db.connect(path)
    settings.ensure_defaults(conn)
    for key, value in (("base_jql", base_jql), ("board_id", board_id)):
        conn.execute("INSERT OR REPLACE INTO setting(key, value) VALUES (?, ?)", (key, json.dumps(value)))
    ts = "2026-10-13T00:00:00+00:00"
    for status, jql in (("done", base_jql), ("queued", base_jql), ("claimed", base_jql), ("queued", "key = A-1")):
        conn.execute("INSERT INTO batch(title, author, created_at) VALUES ('x', 'claude', ?)", (ts,))
        batch = conn.execute("SELECT MAX(id) AS id FROM batch").fetchone()["id"]
        conn.execute("INSERT INTO command(batch_id, seq, type, params, status) VALUES (?, 1, 'refresh', ?, ?)",
                     (batch, json.dumps({"jql": jql}), status))
        if status == "done" and jql == base_jql:
            cid = conn.execute("SELECT MAX(id) AS id FROM command").fetchone()["id"]
            conn.execute("INSERT INTO sync_run(started_at, finished_at, kind, query, command_id) VALUES (?, ?, 'jql', ?, ?)",
                         (ts, ts, base_jql, cid))
    conn.execute("INSERT INTO issue(key, summary, fetched_at) VALUES ('A-1', 'Т', ?)", (ts,))
    _add_fact_status(conn)
    conn.execute("INSERT INTO issue_note(key, fact_status, work_kind) VALUES ('A-1', 'В работе', 'Аналитика')")
    _drop_v5(conn)
    conn.execute("ALTER TABLE batch DROP COLUMN project")
    conn.execute("DROP TABLE project")
    conn.execute("UPDATE setting SET value = '3' WHERE key = 'schema_version'")
    conn.close()


def test_migrates_v3_creates_awpca(tmp_path):
    path = tmp_path / "hub.sqlite"
    _v3_database(path)
    conn = db.connect(path)
    new = 'project = AURORA AND labels in ("AWPCA")'
    assert {k: v for k, v in api.project_rows(conn)[0].items() if not k.startswith(("wiki_", "jira_"))} | {
        "created_at": None, "issues_count": None} == {
        "label": "AWPCA", "name": "АРМ КА", "board_id": 4216, "board_type": None, "created_at": None, "issues_count": None}
    assert settings.get(conn, "base_jql") == "project = AURORA AND labels = AWPCA"     # ключи на месте
    assert settings.get(conn, "board_id") == 4216
    assert conn.execute("SELECT query FROM sync_run").fetchone()["query"] == new
    jqls = {r["status"]: json.loads(r["params"]).get("jql") for r in conn.execute("SELECT status, params FROM command WHERE type = 'refresh'")}
    assert jqls["done"] == "project = AURORA AND labels = AWPCA"                        # завершённые не трогаем
    assert jqls["claimed"] == new
    queued = [json.loads(r["params"])["jql"] for r in conn.execute("SELECT params FROM command WHERE status = 'queued'")]
    assert sorted(queued) == sorted([new, "key = A-1"])
    assert conn.execute("SELECT project FROM batch").fetchall()[0]["project"] is None
    assert conn.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone()["value"] == str(db.SCHEMA_VERSION)
    assert len(list((tmp_path / "backups").glob("*-schema-v3.sqlite"))) == 1
    conn.close()


def test_migrates_v3_two_labels_board_to_nobody(tmp_path, caplog):
    _v3_database(tmp_path / "hub.sqlite", base_jql="project = AURORA AND labels in (AWPCA, STOCKS_DRRK)")
    conn = db.connect(tmp_path / "hub.sqlite")
    got = {p["label"]: (p["name"], p["board_id"]) for p in api.project_rows(conn)}
    assert got == {"AWPCA": ("АРМ КА", None), "STOCKS_DRRK": ("STOCKS_DRRK", None)}
    assert "4216" in caplog.text
    conn.close()


def test_migrates_v3_unparsed_jql_keeps_running(tmp_path, caplog):
    _v3_database(tmp_path / "hub.sqlite", base_jql="project = AURORA AND labels = AWPCA ORDER BY key")
    conn = db.connect(tmp_path / "hub.sqlite")
    assert projects.list_projects(conn) == []
    assert "ORDER BY key" in caplog.text
    assert conn.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone()["value"] == str(db.SCHEMA_VERSION)
    conn.close()


@pytest.mark.parametrize("raw, want", [
    ("2026-10-22T13:05:11.000+0300", "2026-10-22T10:05:11+00:00"),
    ("2026-10-22T13:05:11+03:00", "2026-10-22T10:05:11+00:00"),
    ("2026-10-22T10:05:11Z", "2026-10-22T10:05:11+00:00"),
    ("2026-10-22T10:05:11.123Z", "2026-10-22T10:05:11+00:00"),
    ("2026-10-22T10:05:11.123", "2026-10-22T10:05:11+00:00"),
    ("2026-10-22T02:05:11-0500", "2026-10-22T07:05:11+00:00"),
    ("вчера", None), ("", None), (None, None), (42, None),
    ("9999-12-31T23:00:00-0300", None), ("0001-01-01T00:00:00+0300", None),      # M-1: выход за диапазон datetime
])
def test_jira_utc(raw, want):
    assert history.jira_utc(raw) == want


def _v4_database(path):
    """База v4: комментарий со временем Jira, полный прогон в журнале синхронизаций."""
    conn = db.connect(path)
    conn.execute("INSERT INTO issue_comment(key, comment_id, created, body) VALUES ('X', '1', '2026-10-22T13:05:11.000+0300', 'a')")
    conn.execute("INSERT INTO issue_comment(key, comment_id, created, body) VALUES ('X', '2', 'мусор', 'b')")
    conn.execute("INSERT INTO sync_run(started_at, finished_at, kind, query) VALUES "
                 "('2026-10-22T09:00:00+00:00', '2026-10-22T09:01:00+00:00', 'jql', 'q')")
    _drop_v5(conn)
    conn.execute("UPDATE setting SET value = '4' WHERE key = 'schema_version'")
    conn.close()


def test_migrates_v4(tmp_path):
    path = tmp_path / "hub.sqlite"
    _v4_database(path)
    conn = db.connect(path)
    assert conn.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone()["value"] == str(db.SCHEMA_VERSION)
    rows = {r["comment_id"]: (r["created"], r["created_at"]) for r in conn.execute("SELECT * FROM issue_comment")}
    assert rows == {"1": ("2026-10-22T13:05:11.000+0300", "2026-10-22T10:05:11+00:00"), "2": ("мусор", None)}
    assert conn.execute("SELECT history_at FROM sync_run").fetchone()["history_at"] is None
    assert conn.execute("SELECT COUNT(*) AS n FROM issue_change").fetchone()["n"] == 0
    assert len(list((tmp_path / "backups").glob("*-schema-v4.sqlite"))) == 1
    conn.close()


def test_migrate_v5_to_v6(tmp_path):
    path = tmp_path / "hub.sqlite"
    conn = db.connect(path)
    _drop_v6(conn)
    conn.execute("UPDATE setting SET value = '5' WHERE key = 'schema_version'")
    conn.close()
    conn = db.connect(path)
    assert conn.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone()["value"] == str(db.SCHEMA_VERSION)
    assert conn.execute("SELECT COUNT(*) AS n FROM issue_attachment").fetchone()["n"] == 0
    assert conn.execute("SELECT COUNT(*) AS n FROM attachment_file").fetchone()["n"] == 0
    assert len(list((tmp_path / "backups").glob("*-schema-v5.sqlite"))) == 1
    conn.close()


def _v7_database(path, settings_version=12, with_keys=True):
    """База v7: заметка со статусом по факту, ключи status_labels/transitions_map в setting (fact-status-removal)."""
    conn = db.connect(path)
    settings.ensure_defaults(conn)
    _add_fact_status(conn)
    conn.execute("INSERT INTO issue_note(key, fact_status, work_kind, updated_at, updated_by) "
                 "VALUES ('A-1', 'В работе', 'Аналитика', '2026-10-20T00:00:00+00:00', 'import')")
    rows = [("settings_version", settings_version)]
    if with_keys:
        rows += [("status_labels", {"В работе": "fact-in-progress"}), ("transitions_map", {"Снято": "Снято"})]
    for key, value in rows:
        conn.execute("INSERT OR REPLACE INTO setting(key, value) VALUES (?, ?)", (key, json.dumps(value, ensure_ascii=False)))
    conn.execute("UPDATE setting SET value = '7' WHERE key = 'schema_version'")
    conn.close()


def test_migrates_v7_drops_fact_status(tmp_path):
    path = tmp_path / "hub.sqlite"
    _v7_database(path)
    conn = db.connect(path)
    assert db._schema_version(conn) == db.SCHEMA_VERSION
    assert _note_after_v8(conn, "A-1") == "Аналитика"
    row = conn.execute("SELECT updated_at, updated_by FROM issue_note WHERE key = 'A-1'").fetchone()
    assert (row["updated_at"], row["updated_by"]) == ("2026-10-20T00:00:00+00:00", "import")
    assert settings.get(conn, "status_labels") is None and settings.get(conn, "transitions_map") is None
    assert settings.get(conn, "settings_version") == 13
    assert len(list((tmp_path / "backups").glob("*-schema-v7.sqlite"))) == 1
    conn.close()


def test_migrates_v7_without_keys_keeps_settings_version(tmp_path):
    path = tmp_path / "hub.sqlite"
    _v7_database(path, with_keys=False)
    conn = db.connect(path)
    assert db._schema_version(conn) == db.SCHEMA_VERSION and settings.get(conn, "settings_version") == 12
    conn.close()


def test_migrates_v7_without_settings_version(tmp_path):
    path = tmp_path / "hub.sqlite"
    _v7_database(path)
    raw = sqlite3.connect(path)
    raw.execute("DELETE FROM setting WHERE key = 'settings_version'")
    raw.commit()
    raw.close()
    conn = db.connect(path)
    assert db._schema_version(conn) == db.SCHEMA_VERSION
    assert conn.execute("SELECT COUNT(*) AS n FROM setting WHERE key = 'settings_version'").fetchone()["n"] == 0
    conn.close()


def test_v8_reconnect_is_stable(tmp_path):
    path = tmp_path / "hub.sqlite"
    _v7_database(path)
    db.connect(path).close()
    conn = db.connect(path)
    settings.ensure_defaults(conn)
    assert settings.get(conn, "settings_version") == 13
    assert "status_labels" not in settings.get_all(conn) and "transitions_map" not in settings.get_all(conn)
    conn.close()


def test_obsolete_keys_removed_by_ensure_defaults(tmp_path):
    """I-1: процесс со старым кодом (MCP-сессия до reconnect) после миграции снова вставляет status_labels/transitions_map
    своим ensure_defaults; новый ensure_defaults (старт хаба, новый MCP) их убирает."""
    path = tmp_path / "hub.sqlite"
    _v7_database(path)
    conn = db.connect(path)
    for key in ("status_labels", "transitions_map"):              # как INSERT OR IGNORE старых DEFAULTS
        conn.execute("INSERT OR IGNORE INTO setting(key, value) VALUES (?, '{}')", (key,))
    version = settings.get(conn, "settings_version")
    settings.ensure_defaults(conn)
    assert {"status_labels", "transitions_map"}.isdisjoint(settings.get_all(conn))
    assert settings.get(conn, "settings_version") == version
    conn.close()


def _v8_database(path):
    """База v8: задача X со спринтом «Спринт 2», без колонок rank/story_points и таблиц досок (sprints-backlog)."""
    conn = db.connect(path)
    settings.ensure_defaults(conn)
    conn.execute("INSERT INTO issue(key, summary, sprint, labels, fetched_at) VALUES ('X', 'т', 'Спринт 2', '[]', ?)", (db.now(),))
    for col in ("rank", "story_points"):
        conn.execute(f"ALTER TABLE issue DROP COLUMN {col}")
    for t in ("jira_board", "jira_sprint", "issue_sprint"):
        conn.execute(f"DROP TABLE {t}")
    conn.execute("UPDATE setting SET value = '8' WHERE key = 'schema_version'")
    conn.close()


def test_migrates_v8_adds_agile_columns(tmp_path):            # сценарий «Миграция базы v8»
    path = tmp_path / "hub.sqlite"; _v8_database(path)
    conn = db.connect(path)
    assert db._schema_version(conn) == 9
    row = conn.execute("SELECT rank, story_points, sprint FROM issue WHERE key = 'X'").fetchone()
    assert (row["rank"], row["story_points"], row["sprint"]) == (None, None, "Спринт 2")
    types = {r["name"]: r["type"] for r in conn.execute("PRAGMA table_info(issue)")}
    assert types["rank"] == "TEXT" and types["story_points"] == "REAL"
    for t in ("jira_board", "jira_sprint", "issue_sprint"):
        assert conn.execute(f"SELECT COUNT(*) AS n FROM {t}").fetchone()["n"] == 0
    assert len(list((tmp_path / "backups").glob("*-schema-v8.sqlite"))) == 1
    conn.close()


def test_v9_reconnect_is_stable(tmp_path):                    # повторный connect не мигрирует и не копирует
    path = tmp_path / "hub.sqlite"; _v8_database(path)
    db.connect(path).close(); db.connect(path).close()
    assert len(list((tmp_path / "backups").glob("*-schema-v*.sqlite"))) == 1