854db66abc
Implementare completa U5 din plan-treapta2.md (sectiunea 13): - _upload.html: drop zone + buton accesibil (a11y: drag nu e la tastatura), drag-and-drop JS, mesaj 'NU se trimite nimic pana confirmi', selector foi pt multi-sheet xlsx, stari eroare/mesaj - _mapcoloane.html: formular mapare coloane cu .maprow/.mapcol.grow, sugestii fuzzy pre-selectate, etiichete <label> vizibile, sample values, format data configurabil - _preview_import.html: tabel 6 stari, pills rezumat, filtre pe stare, .chk per-rand pe needs_review (D11), banner declarant .banner.warn direct deasupra input-ului N (D12), bara confirmare sticky, text 'dubla cu randul N' pe duplicate_in_file (D10 daltonism), link export CSV randuri esuate - base.html: .s-needs_review (warn), .s-already_sent/.s-duplicate_in_file (muted), .drop-zone, .banner.warn, .sticky-bar, .htmx-indicator - routes.py: rute /_import/upload/mapare-coloane/preview/reset/confirma; helper _web_compute_preview refoloseste _resolve_row_for_preview, _already_sent_lookup, _signature din import_router (fara a-l edita); commit ON CONFLICT DO NOTHING (TOCTOU); log atestare - tests/test_import_ui.py: 15 teste (dashboard, upload, mapare, preview, confirmare N corect/gresit, reset, erori, multi-sheet, a11y D10/D11/D12) 279 teste total, 0 esecuri. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
451 lines
16 KiB
Python
451 lines
16 KiB
Python
"""Teste UI web import (U5) — upload → mapare coloane → preview → confirmare.
|
|
|
|
Verifica:
|
|
- Dashboard randeaza sectiunea de upload
|
|
- Upload xlsx → mapare noua → fragment _mapcoloane returnat
|
|
- Upload xlsx cu mapare existenta → preview direct
|
|
- Salvare mapare coloane → preview randat
|
|
- Preview afiseaza rezumat stari si randul tabelului
|
|
- Confirmare cu N corect → succes (in coada)
|
|
- Confirmare cu N gresit → eroare explicita
|
|
- Reset → drop zone gol
|
|
- Erori upload (fisier invalid, prea mare, header neclar)
|
|
- Sheet selector la multi-sheet xlsx
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import os
|
|
import tempfile
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(monkeypatch):
|
|
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.main import app
|
|
from fastapi.testclient import TestClient
|
|
|
|
with TestClient(app) as c:
|
|
yield c
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def _make_xlsx_bytes(rows: list[dict]) -> bytes:
|
|
"""Construieste un xlsx minimal cu openpyxl pentru fixture teste."""
|
|
openpyxl = pytest.importorskip("openpyxl")
|
|
wb = openpyxl.Workbook()
|
|
ws = wb.active
|
|
if not rows:
|
|
return b""
|
|
headers = list(rows[0].keys())
|
|
ws.append(headers)
|
|
for row in rows:
|
|
ws.append([row.get(h) for h in headers])
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
return buf.getvalue()
|
|
|
|
|
|
def _make_csv_bytes(rows: list[dict], sep: str = ";") -> bytes:
|
|
"""Construieste un CSV minimal pentru fixture teste."""
|
|
import csv
|
|
import io
|
|
|
|
buf = io.StringIO()
|
|
if not rows:
|
|
return b""
|
|
writer = csv.DictWriter(buf, fieldnames=list(rows[0].keys()), delimiter=sep)
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
return buf.getvalue().encode("utf-8")
|
|
|
|
|
|
_SAMPLE_ROWS = [
|
|
{
|
|
"VIN": "WVWZZZ1KZAW000123",
|
|
"Nr inmatriculare": "B001TST",
|
|
"Data prestatie": "15.06.2026",
|
|
"Odometru final": "123456",
|
|
"Operatie": "Revizie",
|
|
},
|
|
{
|
|
"VIN": "WVWZZZ1KZAW000456",
|
|
"Nr inmatriculare": "B002TST",
|
|
"Data prestatie": "16.06.2026",
|
|
"Odometru final": "200000",
|
|
"Operatie": "Revizie",
|
|
},
|
|
]
|
|
|
|
|
|
def _seed_op_mapping(client, cod_op: str = "Revizie", cod_prest: str = "OE-1") -> None:
|
|
"""Seeda o mapare de operatii cod_op → cod_prestatie via API."""
|
|
client.post("/v1/mapari", json={
|
|
"cod_op_service": cod_op,
|
|
"cod_prestatie": cod_prest,
|
|
"auto_send": True,
|
|
})
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Dashboard #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_dashboard_contine_drop_zone(client):
|
|
"""Dashboard-ul randeaza sectiunea de upload cu drop zone si mesaj warmth."""
|
|
r = client.get("/")
|
|
assert r.status_code == 200
|
|
assert "Primul fisier" in r.text
|
|
assert "drop-zone" in r.text
|
|
assert "NU se trimite nimic" in r.text
|
|
assert "import-section" in r.text
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Upload xlsx — mapare noua #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_upload_xlsx_fara_mapare_arata_formular_mapare(client):
|
|
"""Upload xlsx fara mapare salvata → fragment mapare coloane."""
|
|
xlsx = _make_xlsx_bytes(_SAMPLE_ROWS)
|
|
r = client.post(
|
|
"/_import/upload",
|
|
files={"file": ("test.xlsx", xlsx, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")},
|
|
)
|
|
assert r.status_code == 200
|
|
# Formular de mapare coloane
|
|
assert "Mapare coloane" in r.text
|
|
assert "mapare-coloane" in r.text # URL in form action
|
|
# Coloanele din fisier apar in formular
|
|
assert "VIN" in r.text
|
|
assert "Data prestatie" in r.text
|
|
# Sugestii fuzzy pentru VIN
|
|
assert "vin" in r.text.lower()
|
|
|
|
|
|
def test_upload_csv_fara_mapare_arata_formular_mapare(client):
|
|
"""Upload CSV cu separator ; → formular mapare coloane."""
|
|
csv_bytes = _make_csv_bytes(_SAMPLE_ROWS)
|
|
r = client.post(
|
|
"/_import/upload",
|
|
files={"file": ("test.csv", csv_bytes, "text/csv")},
|
|
)
|
|
assert r.status_code == 200
|
|
assert "Mapare coloane" in r.text
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Salvare mapare coloane → preview #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def _upload_and_get_import_id(client, rows=None) -> int:
|
|
"""Helper: incarca fisier si extrage import_id din raspuns."""
|
|
xlsx = _make_xlsx_bytes(rows or _SAMPLE_ROWS)
|
|
r = client.post(
|
|
"/_import/upload",
|
|
files={"file": ("test.xlsx", xlsx, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")},
|
|
)
|
|
assert r.status_code == 200
|
|
# Extrage import_id din URL-ul form action din raspuns
|
|
text = r.text
|
|
# Form action contine /_import/{id}/mapare-coloane
|
|
import re
|
|
m = re.search(r"/_import/(\d+)/mapare-coloane", text)
|
|
assert m, f"Nu s-a gasit import_id in raspuns: {text[:500]}"
|
|
return int(m.group(1))
|
|
|
|
|
|
def test_salvare_mapare_coloane_arata_preview(client):
|
|
"""Dupa salvarea maparii de coloane, raspunsul contine preview-ul."""
|
|
# Asigura ca nomenclatorul are OE-1 (seeding automat la init_db)
|
|
import_id = _upload_and_get_import_id(client)
|
|
|
|
r = client.post(
|
|
f"/_import/{import_id}/mapare-coloane",
|
|
data={
|
|
"colname": ["VIN", "Nr inmatriculare", "Data prestatie", "Odometru final", "Operatie"],
|
|
"canon": ["vin", "nr_inmatriculare", "data_prestatie", "odometru_final", "operatie"],
|
|
"format_data": "DD.MM.YYYY",
|
|
},
|
|
)
|
|
assert r.status_code == 200
|
|
# Preview trebuie sa contina elementele cheie
|
|
assert "Preview" in r.text
|
|
assert "confirm-form" in r.text
|
|
assert "n-confirmat" in r.text
|
|
# Rezumat stari
|
|
assert "gata de trimis" in r.text or "ok" in r.text
|
|
|
|
|
|
def test_preview_arata_randul_vin(client):
|
|
"""Preview contine VIN-ul din fisier."""
|
|
import_id = _upload_and_get_import_id(client)
|
|
client.post(
|
|
f"/_import/{import_id}/mapare-coloane",
|
|
data={
|
|
"colname": ["VIN", "Nr inmatriculare", "Data prestatie", "Odometru final", "Operatie"],
|
|
"canon": ["vin", "nr_inmatriculare", "data_prestatie", "odometru_final", "operatie"],
|
|
"format_data": "DD.MM.YYYY",
|
|
},
|
|
)
|
|
r = client.get(f"/_import/{import_id}/preview")
|
|
assert r.status_code == 200
|
|
assert "WVWZZZ1KZAW000123" in r.text
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Upload cu mapare existenta → preview direct #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_upload_cu_mapare_existenta_sare_direct_la_preview(client):
|
|
"""Al doilea upload cu acelasi antet → preview imediat (mapare retinuta)."""
|
|
# Primul upload + salvare mapare
|
|
import_id1 = _upload_and_get_import_id(client)
|
|
client.post(
|
|
f"/_import/{import_id1}/mapare-coloane",
|
|
data={
|
|
"colname": ["VIN", "Nr inmatriculare", "Data prestatie", "Odometru final", "Operatie"],
|
|
"canon": ["vin", "nr_inmatriculare", "data_prestatie", "odometru_final", "operatie"],
|
|
"format_data": "DD.MM.YYYY",
|
|
},
|
|
)
|
|
|
|
# Al doilea upload cu acelasi antet
|
|
xlsx = _make_xlsx_bytes(_SAMPLE_ROWS)
|
|
r = client.post(
|
|
"/_import/upload",
|
|
files={"file": ("test2.xlsx", xlsx, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")},
|
|
)
|
|
assert r.status_code == 200
|
|
# Trebuie sa ajunga direct la preview, nu la mapare
|
|
assert "Preview" in r.text
|
|
assert "confirm-form" in r.text
|
|
assert "Mapare retinuta aplicata automat" in r.text
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Confirmare (gate HARD) #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def _setup_preview(client) -> int:
|
|
"""Upload + mapare + seeda operatii + intoarce import_id gata de confirmare."""
|
|
_seed_op_mapping(client) # "Revizie" → "OE-1"
|
|
import_id = _upload_and_get_import_id(client)
|
|
client.post(
|
|
f"/_import/{import_id}/mapare-coloane",
|
|
data={
|
|
"colname": ["VIN", "Nr inmatriculare", "Data prestatie", "Odometru final", "Operatie"],
|
|
"canon": ["vin", "nr_inmatriculare", "data_prestatie", "odometru_final", "operatie"],
|
|
"format_data": "DD.MM.YYYY",
|
|
},
|
|
)
|
|
return import_id
|
|
|
|
|
|
def test_confirmare_n_corect_pune_in_coada(client):
|
|
"""Confirmare cu N corect → randurile ok ajung in coada."""
|
|
import_id = _setup_preview(client)
|
|
|
|
# Compute preview pentru a afla n_ok
|
|
r_prev = client.get(f"/_import/{import_id}/preview")
|
|
assert r_prev.status_code == 200
|
|
|
|
# Citeste summary din DB
|
|
from app.db import get_connection
|
|
|
|
conn = get_connection()
|
|
try:
|
|
batch = conn.execute(
|
|
"SELECT ok FROM import_batches WHERE id=?", (import_id,)
|
|
).fetchone()
|
|
n_ok = batch["ok"]
|
|
finally:
|
|
conn.close()
|
|
|
|
assert n_ok > 0, "Asteptat cel putin un rand ok dupa seeding corect"
|
|
|
|
r = client.post(
|
|
f"/_import/{import_id}/confirma",
|
|
data={"n_confirmat": str(n_ok), "confirmed_by": "test@test.ro"},
|
|
)
|
|
assert r.status_code == 200
|
|
# Succes → drop zone cu mesaj
|
|
assert "S-au pus in coada" in r.text or "prezentari" in r.text
|
|
# Sectiunea se reseteaza la drop zone
|
|
assert "drop-zone" in r.text
|
|
|
|
# Verifica ca submissions au fost create
|
|
conn2 = get_connection()
|
|
try:
|
|
n = conn2.execute(
|
|
"SELECT COUNT(*) FROM submissions WHERE batch_id=?", (import_id,)
|
|
).fetchone()[0]
|
|
assert n == n_ok, f"Asteptat {n_ok} submissions, gasit {n}"
|
|
finally:
|
|
conn2.close()
|
|
|
|
|
|
def test_confirmare_n_gresit_arata_eroare(client):
|
|
"""Confirmare cu N gresit → eroare clara, nu enqueue."""
|
|
import_id = _setup_preview(client)
|
|
client.get(f"/_import/{import_id}/preview") # calculeaza si stocheaza starea
|
|
|
|
r = client.post(
|
|
f"/_import/{import_id}/confirma",
|
|
data={"n_confirmat": "99", "confirmed_by": ""},
|
|
)
|
|
assert r.status_code == 200
|
|
# Trebuie sa arate eroare de confirmare sau preview cu eroare
|
|
assert (
|
|
"difera" in r.text
|
|
or "Numarul confirmat" in r.text
|
|
or "Niciun rand ok" in r.text
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Reset #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_reset_arata_drop_zone_gol(client):
|
|
"""GET /_import/reset → drop zone gol fara mesaje."""
|
|
r = client.get("/_import/reset")
|
|
assert r.status_code == 200
|
|
assert "drop-zone" in r.text
|
|
assert "Primul fisier" in r.text
|
|
assert "import-section" in r.text
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Erori upload #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_upload_fisier_invalid_arata_eroare(client):
|
|
"""Upload fisier invalid → mesaj de eroare in drop zone."""
|
|
r = client.post(
|
|
"/_import/upload",
|
|
files={"file": ("test.xlsx", b"not a real xlsx file", "application/octet-stream")},
|
|
)
|
|
assert r.status_code == 200
|
|
# Trebuie sa arate drop zone cu eroare
|
|
assert "drop-zone" in r.text or "import-section" in r.text
|
|
# Eroare vizibila
|
|
assert "nerecunoscut" in r.text.lower() or "invalid" in r.text.lower() or "eroare" in r.text.lower()
|
|
|
|
|
|
def test_upload_fisier_csv_antet_o_coloana_arata_eroare(client):
|
|
"""CSV cu o singura coloana reala → header acceptat sau eroare gestionata."""
|
|
bad = b"date_fara_header\nval1\nval2\n"
|
|
r = client.post(
|
|
"/_import/upload",
|
|
files={"file": ("test.csv", bad, "text/csv")},
|
|
)
|
|
# Fie detecteaza header OK fie arata eroare
|
|
assert r.status_code == 200
|
|
assert "import-section" in r.text
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Multi-sheet xlsx #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_upload_multi_sheet_arata_selector(client):
|
|
"""xlsx cu mai multe foi → selector de foaie in drop zone."""
|
|
openpyxl = pytest.importorskip("openpyxl")
|
|
wb = openpyxl.Workbook()
|
|
ws1 = wb.active
|
|
ws1.title = "Date"
|
|
ws1.append(["VIN", "Data"])
|
|
ws1.append(["WVW001", "15.06.2026"])
|
|
ws2 = wb.create_sheet("Raport")
|
|
ws2.append(["Total"])
|
|
ws2.append(["1"])
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
xlsx = buf.getvalue()
|
|
|
|
r = client.post(
|
|
"/_import/upload",
|
|
files={"file": ("multi.xlsx", xlsx, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")},
|
|
)
|
|
assert r.status_code == 200
|
|
# Trebuie sa arate selector de foi
|
|
assert "foi" in r.text.lower() or "sheet" in r.text.lower() or "foaie" in r.text.lower()
|
|
# Foile trebuie sa apara ca optiuni
|
|
assert "Date" in r.text or "Raport" in r.text
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Elemente a11y (D10/D11/D12) #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_preview_contine_banner_declarant(client):
|
|
"""Preview contine bannerul declarant (D12) cu text despre ireversibil."""
|
|
import_id = _setup_preview(client)
|
|
r = client.get(f"/_import/{import_id}/preview")
|
|
assert r.status_code == 200
|
|
assert "declarantul" in r.text
|
|
assert "ireversibil" in r.text
|
|
assert "banner" in r.text
|
|
|
|
|
|
def test_preview_contine_checkboxuri_needs_review(client):
|
|
"""Randurile needs_review au checkbox 'verificat' (D11)."""
|
|
# Cream un rand cu VIN numeric → needs_review
|
|
rows_with_review = [
|
|
{
|
|
"VIN": "1234567890", # VIN numeric → coercion flag → needs_review
|
|
"Nr inmatriculare": "B001TST",
|
|
"Data prestatie": "15.06.2026",
|
|
"Odometru final": "123456",
|
|
"Operatie": "OE-1",
|
|
}
|
|
]
|
|
import_id = _upload_and_get_import_id(client, rows=rows_with_review)
|
|
client.post(
|
|
f"/_import/{import_id}/mapare-coloane",
|
|
data={
|
|
"colname": ["VIN", "Nr inmatriculare", "Data prestatie", "Odometru final", "Operatie"],
|
|
"canon": ["vin", "nr_inmatriculare", "data_prestatie", "odometru_final", "operatie"],
|
|
"format_data": "DD.MM.YYYY",
|
|
},
|
|
)
|
|
r = client.get(f"/_import/{import_id}/preview")
|
|
assert r.status_code == 200
|
|
# Checkbox reviewed_rows prezent pentru randul needs_review
|
|
assert "reviewed_rows" in r.text
|
|
assert "needs_review" in r.text
|
|
|
|
|
|
def test_preview_duplicate_in_file_are_text(client):
|
|
"""Randurile duplicate_in_file arata text 'dubla cu randul N' (D10 — nu doar culoare)."""
|
|
_seed_op_mapping(client, "Revizie", "OE-1")
|
|
# Doua randuri identice → duplicate_in_file
|
|
dup_rows = [
|
|
{"VIN": "WVWZZZ1KZAW000123", "Nr inmatriculare": "B001TST",
|
|
"Data prestatie": "15.06.2026", "Odometru final": "123456", "Operatie": "Revizie"},
|
|
{"VIN": "WVWZZZ1KZAW000123", "Nr inmatriculare": "B001TST",
|
|
"Data prestatie": "15.06.2026", "Odometru final": "123456", "Operatie": "Revizie"},
|
|
]
|
|
import_id = _upload_and_get_import_id(client, rows=dup_rows)
|
|
client.post(
|
|
f"/_import/{import_id}/mapare-coloane",
|
|
data={
|
|
"colname": ["VIN", "Nr inmatriculare", "Data prestatie", "Odometru final", "Operatie"],
|
|
"canon": ["vin", "nr_inmatriculare", "data_prestatie", "odometru_final", "operatie"],
|
|
"format_data": "DD.MM.YYYY",
|
|
},
|
|
)
|
|
r = client.get(f"/_import/{import_id}/preview")
|
|
assert r.status_code == 200
|
|
# Text explicit pentru duplicate_in_file (nu doar culoare — cerinta daltonism D10)
|
|
assert "dubla cu randul" in r.text
|
|
assert "duplicate_in_file" in r.text
|