rar-autopass/tests/test_mapari_tabel.py

"""Teste US-005 (PRD 3.6): "De rezolvat" si "Operatii salvate" randate ca TABELE.

Doar marcaj de template — POST-urile /mapari, /mapari/salvate, /mapari/salvate/sterge
raman identice. H4: comutatorul auto-send din tabelul salvate reflecta valoarea STOCATA
per mapare (din _load_saved_op_mappings), nu un default hard.
"""

from __future__ import annotations

import json
import os
import re
import tempfile

import pytest
from starlette.testclient import TestClient


def _create_account_user(email: str, name: str = "Service", password: str = "parolasecreta10"):
    from app.accounts import create_account
    from app.users import create_user
    from app.db import get_connection

    conn = get_connection()
    try:
        acct_id = create_account(conn, name, active=True)
        create_user(conn, acct_id, email, password)
        return acct_id
    finally:
        conn.close()


def _login(client, email: str, password: str = "parolasecreta10") -> None:
    resp = client.get("/login")
    m = re.search(r'name="csrf_token"\s+value="([^"]+)"', resp.text) or \
        re.search(r'value="([^"]+)"\s+name="csrf_token"', resp.text)
    assert m
    resp = client.post("/login", data={"email": email, "parola": password, "csrf_token": m.group(1)})
    assert resp.status_code == 303


def _csrf(client) -> str:
    resp = client.get("/?tab=mapari")
    m = re.search(r'name="csrf_token"\s+value="([^"]+)"', resp.text) or \
        re.search(r'value="([^"]+)"\s+name="csrf_token"', resp.text)
    assert m, "csrf_token negasit"
    return m.group(1)


def _seed_nomenclator(cod: str, nume: str = "Test prestatie") -> None:
    from app.db import get_connection
    conn = get_connection()
    try:
        conn.execute(
            "INSERT OR REPLACE INTO nomenclator_rar (cod_prestatie, nume_prestatie) VALUES (?, ?)",
            (cod, nume),
        )
        conn.commit()
    finally:
        conn.close()


def _seed_op_mapping(acct: int, op: str, cod: str, auto_send: int = 1) -> None:
    from app.db import get_connection
    conn = get_connection()
    try:
        conn.execute(
            "INSERT INTO operations_mapping (account_id, cod_op_service, cod_prestatie, auto_send) "
            "VALUES (?, ?, ?, ?)",
            (acct, op, cod, auto_send),
        )
        conn.commit()
    finally:
        conn.close()


def _seed_needs_mapping(acct: int, op: str) -> int:
    """Submission needs_mapping (canal API, batch_id NULL) cu o operatie nemapata."""
    from app.db import get_connection
    conn = get_connection()
    try:
        cur = conn.execute(
            "INSERT INTO submissions (idempotency_key, account_id, status, payload_json) "
            "VALUES (?, ?, 'needs_mapping', ?)",
            (
                f"k-{op}-{os.urandom(4).hex()}",
                acct,
                json.dumps({
                    "vin": "WVWZZZ1JZXW000111",
                    "nr_inmatriculare": "B11AAA",
                    "data_prestatie": "2026-06-18",
                    "odometru_final": "12345",
                    "prestatii": [{"cod_op_service": op, "denumire": "ceva"}],
                }),
            ),
        )
        conn.commit()
        return cur.lastrowid
    finally:
        conn.close()


def _status_of(sid: int) -> str:
    from app.db import get_connection
    conn = get_connection()
    try:
        return conn.execute("SELECT status FROM submissions WHERE id=?", (sid,)).fetchone()["status"]
    finally:
        conn.close()


def _slice(text: str, start_marker: str, end_marker: str | None) -> str:
    i = text.index(start_marker)
    j = text.index(end_marker, i) if end_marker else len(text)
    return text[i:j]


@pytest.fixture()
def client(monkeypatch):
    tmp = tempfile.mkdtemp()
    monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "mapari_tabel.db"))
    monkeypatch.setenv("AUTOPASS_WEB_AUTH_REQUIRED", "true")
    from app.config import get_settings
    get_settings.cache_clear()
    from app.web import ratelimit
    ratelimit._hits.clear()
    from app.main import app
    with TestClient(app, follow_redirects=False) as c:
        yield c
    ratelimit._hits.clear()
    get_settings.cache_clear()


def test_mapari_de_rezolvat_in_tabel(client):
    """Sectiunea "De rezolvat" randata ca tabel (.tablewrap + <table> + un rand per operatie)."""
    acct = _create_account_user("rez@test.com")
    _seed_nomenclator("R-FRANE", "Reparatie frane")
    _seed_needs_mapping(acct, "OP-REZ1")
    _login(client, "rez@test.com")

    resp = client.get("/?tab=mapari")
    assert resp.status_code == 200
    sec = _slice(resp.text, "De rezolvat", "Mapari operatii salvate")

    assert "tablewrap" in sec, "tabelul De rezolvat trebuie sa foloseasca .tablewrap"
    assert "<table" in sec and "<th" in sec, "De rezolvat trebuie randat ca tabel cu antet"
    assert "OP-REZ1" in sec, "operatia nemapata trebuie sa apara ca rand"
    # POST neschimbat: forma trimite tot la /mapari
    assert 'hx-post="/mapari"' in sec


def test_mapari_salvate_in_tabel(client):
    """Sectiunea "Operatii salvate" randata ca tabel; H4: auto-send reflecta valoarea STOCATA."""
    acct = _create_account_user("salv@test.com")
    _seed_nomenclator("R-FRANE", "Reparatie frane")
    _seed_nomenclator("R-MOTOR", "Reparatie motor")
    _seed_op_mapping(acct, "OP-AUTO", "R-FRANE", auto_send=1)
    _seed_op_mapping(acct, "OP-MANUAL", "R-MOTOR", auto_send=0)
    _login(client, "salv@test.com")

    resp = client.get("/?tab=mapari")
    assert resp.status_code == 200
    sec = _slice(resp.text, "Mapari operatii salvate", "Formate de coloane salvate")

    assert "tablewrap" in sec, "tabelul Operatii salvate trebuie sa foloseasca .tablewrap"
    assert "<table" in sec and "<th" in sec, "Operatii salvate trebuie randat ca tabel cu antet"
    assert "OP-AUTO" in sec and "OP-MANUAL" in sec
    assert 'name="auto_send"' in sec
    # H4: exact maparile cu auto_send STOCAT True sunt bifate (aici: o singura)
    assert sec.count("checked") == 1, "comutatorul auto-send trebuie sa reflecte valoarea stocata, nu un default"
    # POST-urile neschimbate
    assert 'hx-post="/mapari/salvate"' in sec
    assert 'hx-post="/mapari/salvate/sterge"' in sec


def test_mapari_salvare_si_stergere_inca_functioneaza(client):
    """POST /mapari (salveaza), /mapari/salvate (edit), /mapari/salvate/sterge (delete) neschimbate."""
    acct = _create_account_user("func@test.com")
    _seed_nomenclator("R-FRANE", "Reparatie frane")
    _seed_nomenclator("R-MOTOR", "Reparatie motor")
    sid = _seed_needs_mapping(acct, "OP-300")
    assert _status_of(sid) == "needs_mapping"
    _login(client, "func@test.com")
    csrf = _csrf(client)

    # Salveaza o mapare noua pentru operatia nemapata -> se deblocheaza
    resp = client.post("/mapari", data={
        "cod_op_service": "OP-300", "cod_prestatie": "R-FRANE", "auto_send": "true", "csrf_token": csrf,
    })
    assert resp.status_code == 200
    assert _status_of(sid) != "needs_mapping"

    from app.db import get_connection
    conn = get_connection()
    try:
        row = conn.execute(
            "SELECT cod_prestatie FROM operations_mapping WHERE account_id=? AND cod_op_service=?",
            (acct, "OP-300"),
        ).fetchone()
    finally:
        conn.close()
    assert row and row["cod_prestatie"] == "R-FRANE"

    # Editeaza maparea salvata
    resp = client.post("/mapari/salvate", data={
        "cod_op_service": "OP-300", "cod_prestatie": "R-MOTOR", "auto_send": "true", "csrf_token": csrf,
    })
    assert resp.status_code == 200
    conn = get_connection()
    try:
        row = conn.execute(
            "SELECT cod_prestatie FROM operations_mapping WHERE account_id=? AND cod_op_service=?",
            (acct, "OP-300"),
        ).fetchone()
    finally:
        conn.close()
    assert row["cod_prestatie"] == "R-MOTOR"

    # Sterge maparea salvata
    resp = client.post("/mapari/salvate/sterge", data={
        "cod_op_service": "OP-300", "csrf_token": csrf,
    })
    assert resp.status_code == 200
    conn = get_connection()
    try:
        row = conn.execute(
            "SELECT 1 FROM operations_mapping WHERE account_id=? AND cod_op_service=?",
            (acct, "OP-300"),
        ).fetchone()
    finally:
        conn.close()
    assert row is None