feat(ui): #15 U5 — web upload import (HTMX) drop→mapare→preview→confirma
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>
This commit is contained in:
450
tests/test_import_ui.py
Normal file
450
tests/test_import_ui.py
Normal file
@@ -0,0 +1,450 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user