14e1c463f0
Catalog central pur app/errors.py ca sursa unica cod->{problema,fix},
consumat de API+UI+worker. Aditiv (field/message pastrate la octet) +
rar_error stocat superset. Scope: fluxul de declarare; login/signup/CSRF
neatinse. labels.parse_erori degradeaza gratios; UI progresiv AA light+dark.
631 teste.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
181 lines
7.2 KiB
Python
181 lines
7.2 KiB
Python
"""Teste US-007 (PRD 5.4): erori de import pe 3 niveluri in interfata web.
|
|
|
|
Verifica ca fragmentele HTML intoarse de rutele web /_import/* contin
|
|
textele structurate (problema + fix) din catalog pentru erorile de upload
|
|
si de mapare coloane.
|
|
|
|
Rutele testate sunt WEB (HTML), nu API JSON — raspunsul este HTML fragment.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import os
|
|
import tempfile
|
|
|
|
import openpyxl
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Fixture client cu WEB_AUTH_REQUIRED=false (dev mode, cont 1 implicit) #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
@pytest.fixture()
|
|
def client(monkeypatch):
|
|
"""Client FastAPI cu DB temporara izolata, auth web dezactivat (dev mode)."""
|
|
tmp = tempfile.mkdtemp()
|
|
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "erori3n.db"))
|
|
monkeypatch.setenv("AUTOPASS_WEB_AUTH_REQUIRED", "false")
|
|
from app.config import get_settings
|
|
get_settings.cache_clear()
|
|
from app.crypto import reset_cache
|
|
reset_cache()
|
|
from app.main import app
|
|
with TestClient(app) as c:
|
|
yield c
|
|
get_settings.cache_clear()
|
|
reset_cache()
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Helpere #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def _make_xlsx_prea_mare(n_randuri: int = 5001) -> bytes:
|
|
"""Xlsx cu mai mult de 5000 de randuri de date (declanseaza FileTooLarge)."""
|
|
wb = openpyxl.Workbook()
|
|
ws = wb.active
|
|
ws.append(["VIN", "Nr inmatriculare", "Data prestatie", "Odometru final", "Operatie"])
|
|
for i in range(n_randuri):
|
|
ws.append([
|
|
f"WVWZZZ1KZA{i:07d}",
|
|
f"B{i:04d}TST",
|
|
"2026-06-15",
|
|
str(100000 + i),
|
|
"Revizie",
|
|
])
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
return buf.getvalue()
|
|
|
|
|
|
def _upload_web(client: TestClient, data: bytes, filename: str = "test.xlsx") -> object:
|
|
"""POST pe ruta web de upload (intoarce HTML fragment)."""
|
|
return client.post(
|
|
"/_import/upload",
|
|
files={"file": (filename, io.BytesIO(data), "application/octet-stream")},
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 1. Upload fisier prea mare → fragment cu fix din IMPORT_FISIER_PREA_MARE #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_upload_fisier_prea_mare_3niveluri(client):
|
|
"""Upload xlsx >5000 randuri → fragment HTML contine textul fix din catalog.
|
|
|
|
Textul asteptat din CATALOG['IMPORT_FISIER_PREA_MARE']['fix']:
|
|
'Imparte fisierul in bucati de maxim 5000 de randuri si incarca-le pe rand.'
|
|
"""
|
|
from app.errors import CATALOG
|
|
fix_asteptat = CATALOG["IMPORT_FISIER_PREA_MARE"]["fix"]
|
|
|
|
data = _make_xlsx_prea_mare(5001)
|
|
r = _upload_web(client, data, "mare.xlsx")
|
|
|
|
assert r.status_code == 200, f"Asteptat 200, primit {r.status_code}: {r.text[:300]}"
|
|
html = r.text
|
|
assert fix_asteptat in html, (
|
|
f"Textul fix din IMPORT_FISIER_PREA_MARE nu apare in fragmentul HTML.\n"
|
|
f"Asteptat: {fix_asteptat!r}\n"
|
|
f"Fragment (primii 800 chars): {html[:800]}"
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 2. Upload fisier nerecunoscut → fragment cu fix din IMPORT_FISIER_NERECUNOSCUT
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_upload_fisier_nerecunoscut_3niveluri(client):
|
|
"""Upload bytes invalizi (nu e xlsx/csv valid) → fragment HTML cu fix din catalog.
|
|
|
|
Textul asteptat din CATALOG['IMPORT_FISIER_NERECUNOSCUT']['fix']:
|
|
'Incarca un fisier .xlsx sau .csv valid.'
|
|
"""
|
|
from app.errors import CATALOG
|
|
fix_asteptat = CATALOG["IMPORT_FISIER_NERECUNOSCUT"]["fix"]
|
|
problema_asteptata = CATALOG["IMPORT_FISIER_NERECUNOSCUT"]["problema"]
|
|
|
|
# Bytes invalizi: nu este un zip/xlsx, nu este text CSV
|
|
date_invalide = b"\x00\x01\x02\x03\x04\x05binar_junk_non_xlsx"
|
|
r = _upload_web(client, date_invalide, "test.xlsx")
|
|
|
|
assert r.status_code == 200, f"Asteptat 200, primit {r.status_code}: {r.text[:300]}"
|
|
html = r.text
|
|
assert fix_asteptat in html, (
|
|
f"Textul fix din IMPORT_FISIER_NERECUNOSCUT nu apare in fragmentul HTML.\n"
|
|
f"Asteptat: {fix_asteptat!r}\n"
|
|
f"Fragment (primii 800 chars): {html[:800]}"
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 3. Mapare coloane cu JSON invalid → fragment cu fix din COLOANE_FORMAT_JSON #
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def test_mapcoloane_format_json_3niveluri(client):
|
|
"""POST direct pe ruta de mapare coloane cu corp JSON invalid → fragment cu fix din catalog.
|
|
|
|
Textul asteptat din CATALOG['COLOANE_FORMAT_JSON']['fix']:
|
|
'Verifica sintaxa JSON a maparii de coloane (ghilimele duble, acolade inchise corect).'
|
|
|
|
Ruta /_import/{id}/mapare-coloane accepta form data. Codul de test simuleaza
|
|
un batch_id invalid (0) pentru a forta ramura de eroare, sau injecteaza direct
|
|
un batch_id valid cu JSON invalid in campul de mapare. Deoarece ruta nu primeste
|
|
JSON direct, testam ramura din routes.py unde se face json.dumps si se valideaza
|
|
manual, sau modificam testul sa plaseze un batch valid si sa trimita date malformate.
|
|
|
|
Strategia: upload un fisier valid, obtine import_id, trimite un form cu valoare
|
|
json invalida in campul coloane (via Content-Type: application/json intentionat gresit
|
|
sau via parametru form malformat). In implementarea din routes.py, eroarea
|
|
COLOANE_FORMAT_JSON este randata cand json.loads esueaza pe un camp special.
|
|
"""
|
|
from app.errors import CATALOG
|
|
fix_asteptat = CATALOG["COLOANE_FORMAT_JSON"]["fix"]
|
|
|
|
# Primul pas: upload un fisier valid pentru a obtine un import_id real
|
|
wb = openpyxl.Workbook()
|
|
ws = wb.active
|
|
ws.append(["VIN", "Nr", "Data", "KM", "Op"])
|
|
ws.append(["WVWZZZ1KZAW000123", "B001TST", "2026-06-15", "100000", "Revizie"])
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
xlsx_data = buf.getvalue()
|
|
|
|
r_upload = _upload_web(client, xlsx_data, "test.xlsx")
|
|
assert r_upload.status_code == 200, r_upload.text[:300]
|
|
|
|
# Extrage import_id din raspuns
|
|
import re
|
|
m = re.search(r"/_import/(\d+)/mapare-coloane", r_upload.text)
|
|
assert m, f"Nu s-a gasit import_id in raspuns: {r_upload.text[:500]}"
|
|
import_id = int(m.group(1))
|
|
|
|
# Trimite un request pe ruta de mapare coloane cu Content-Type: application/json
|
|
# si corp JSON invalid → ruta ar trebui sa intoarca eroarea COLOANE_FORMAT_JSON
|
|
r = client.post(
|
|
f"/_import/{import_id}/mapare-coloane",
|
|
content=b"{invalid json}",
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
|
|
assert r.status_code == 200, f"Asteptat 200, primit {r.status_code}: {r.text[:300]}"
|
|
html = r.text
|
|
assert fix_asteptat in html, (
|
|
f"Textul fix din COLOANE_FORMAT_JSON nu apare in fragmentul HTML.\n"
|
|
f"Asteptat: {fix_asteptat!r}\n"
|
|
f"Fragment (primii 800 chars): {html[:800]}"
|
|
)
|