rar-autopass/tests/test_auto_send_schema.py

"""Schema comutator Auto per cont + flag `held` pe submission.

Acopera: migrarea aditiva idempotenta (coloane + index partial pe DB vechi), helperii
`get_auto_send`/`set_auto_send`/`held_for_account` din app/accounts.py si default-ul 0.
"""

from __future__ import annotations

import os
import sqlite3
import tempfile

import pytest


@pytest.fixture()
def fresh_conn(monkeypatch):
    """DB nou cu schema curenta (init_db)."""
    tmp = tempfile.mkdtemp()
    monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "t.db"))
    from app.config import get_settings
    get_settings.cache_clear()
    from app.db import get_connection, init_db
    init_db()
    c = get_connection()
    yield c
    c.close()
    get_settings.cache_clear()


def _old_db(path: str) -> sqlite3.Connection:
    """DB in forma PRE-5.19: accounts fara auto_send_enabled, submissions fara held."""
    conn = sqlite3.connect(path, isolation_level=None)
    conn.row_factory = sqlite3.Row
    conn.execute(
        "CREATE TABLE accounts (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, "
        "cui TEXT)"
    )
    conn.execute(
        "CREATE TABLE submissions (id INTEGER PRIMARY KEY AUTOINCREMENT, "
        "idempotency_key TEXT NOT NULL UNIQUE, account_id INTEGER, status TEXT, "
        "payload_json TEXT NOT NULL)"
    )
    return conn


def _migrate_old(path: str, monkeypatch) -> sqlite3.Connection:
    from app.config import get_settings
    get_settings.cache_clear()
    conn = sqlite3.connect(path, isolation_level=None)
    conn.row_factory = sqlite3.Row
    from app.db import _migrate
    _migrate(conn)
    return conn


def test_migrare_adauga_coloane_si_index(tmp_path, monkeypatch):
    """Pe DB vechi (fara coloane) migrarea adauga held, auto_send_enabled si indexul partial."""
    path = str(tmp_path / "old.db")
    _old_db(path).close()
    conn = _migrate_old(path, monkeypatch)

    sub = {r["name"] for r in conn.execute("PRAGMA table_info(submissions)").fetchall()}
    assert "held" in sub
    acc = {r["name"] for r in conn.execute("PRAGMA table_info(accounts)").fetchall()}
    assert "auto_send_enabled" in acc

    idx = {r["name"] for r in conn.execute(
        "SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='submissions'"
    ).fetchall()}
    assert "idx_submissions_held" in idx
    conn.close()


def test_migrare_idempotenta_a_doua_rulare(tmp_path, monkeypatch):
    """A doua rulare _migrate pe DB deja migrat nu arunca si nu dubleaza nimic."""
    path = str(tmp_path / "old.db")
    _old_db(path).close()
    conn = _migrate_old(path, monkeypatch)
    from app.db import _migrate
    _migrate(conn)  # nu arunca
    acc = {r["name"] for r in conn.execute("PRAGMA table_info(accounts)").fetchall()}
    assert "auto_send_enabled" in acc
    conn.close()


def test_migrare_idempotenta_pe_db_fresh(fresh_conn):
    """_migrate re-rulat pe schema curenta (init_db) nu strica nimic."""
    from app.db import _migrate
    _migrate(fresh_conn)
    sub = {r["name"] for r in fresh_conn.execute("PRAGMA table_info(submissions)").fetchall()}
    assert "held" in sub


def test_held_for_account_auto_on(fresh_conn):
    from app.accounts import create_account, held_for_account, set_auto_send
    aid = create_account(fresh_conn, "Auto ON")
    set_auto_send(fresh_conn, aid, True)
    assert held_for_account(fresh_conn, aid) == 0


def test_held_for_account_auto_off(fresh_conn):
    from app.accounts import create_account, held_for_account
    aid = create_account(fresh_conn, "Auto OFF")
    assert held_for_account(fresh_conn, aid) == 1


def test_held_for_account_cont_inexistent(fresh_conn):
    from app.accounts import held_for_account
    assert held_for_account(fresh_conn, 99999) == 1


def test_get_set_auto_send_round_trip(fresh_conn):
    from app.accounts import create_account, get_auto_send, set_auto_send
    aid = create_account(fresh_conn, "Round trip")
    assert get_auto_send(fresh_conn, aid) is False
    set_auto_send(fresh_conn, aid, True)
    assert get_auto_send(fresh_conn, aid) is True
    # Idempotent: acelasi set nu arunca si pastreaza valoarea.
    set_auto_send(fresh_conn, aid, True)
    assert get_auto_send(fresh_conn, aid) is True
    set_auto_send(fresh_conn, aid, False)
    assert get_auto_send(fresh_conn, aid) is False


def test_get_auto_send_cont_inexistent(fresh_conn):
    from app.accounts import get_auto_send
    assert get_auto_send(fresh_conn, 99999) is False


@pytest.mark.no_auto_send_seed  # verifica default-ul BRUT id=1==0; nu wrapa init_db (conftest)
def test_default_auto_send_zero(fresh_conn):
    """Cont nou (create_account) si contul id=1 pornesc pe auto_send_enabled=0 (Auto OFF)."""
    from app.accounts import create_account
    aid = create_account(fresh_conn, "Nou")
    row = fresh_conn.execute(
        "SELECT auto_send_enabled FROM accounts WHERE id=?", (aid,)
    ).fetchone()
    assert row["auto_send_enabled"] == 0
    row1 = fresh_conn.execute(
        "SELECT auto_send_enabled FROM accounts WHERE id=1"
    ).fetchone()
    assert row1["auto_send_enabled"] == 0


def test_config_held_zile():
    from app.config import Settings
    s = Settings()
    assert s.held_warn_days == 7
    assert s.held_retention_days == 90