romfast/rar-autopass
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
14e1c463f0aaa443f284aa2888ffdafd0d1eed89
rar-autopass/tests/test_import_errors.py
Claude Agent 14e1c463f0 feat(errors): erori pe 3 niveluri (problema+cauza+fix) pe API si UI (PRD 5.4) 
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>
2026-06-23 10:28:09 +00:00

336 lines
14 KiB
Python
Raw Blame History

"""Teste US-005 (PRD 5.4): erori de import imbracate pe 3 niveluri.
Verifica ca fiecare HTTPException de import in scop are `detail` superset:
- cheile vechi: `error`, `message`, context specific (sheets/found/n_ok)
- cheile noi din catalog: `cod`, `problema`, `cauza`, `fix`
"""
from __future__ import annotations
import io
import os
import tempfile
import openpyxl
import pytest
from fastapi.testclient import TestClient
# --------------------------------------------------------------------------- #
# Fixture client #
# --------------------------------------------------------------------------- #
@pytest.fixture()
def client(monkeypatch):
"""Client FastAPI cu DB temporara izolata per test."""
tmp = tempfile.mkdtemp()
monkeypatch.setenv("AUTOPASS_DB_PATH", os.path.join(tmp, "err.db"))
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 #
# --------------------------------------------------------------------------- #
_HEADER = ["VIN", "Nr inmatriculare", "Data prestatie", "Odometru final", "Operatie"]
_ROW_OK = ["WVWZZZ1KZAW000123", "B999TST", "2026-06-15", "123456", "Revizie"]
def _make_xlsx(rows: list[list], extra_sheets: list[str] | None = None) -> bytes:
wb = openpyxl.Workbook()
ws = wb.active
ws.title = "Sheet1"
for row in rows:
ws.append(row)
if extra_sheets:
for name in extra_sheets:
ws2 = wb.create_sheet(name)
ws2.append(_HEADER)
ws2.append(_ROW_OK)
buf = io.BytesIO()
wb.save(buf)
return buf.getvalue()
def _upload(client: TestClient, data: bytes, filename: str = "test.xlsx"):
return client.post(
"/v1/import",
files={"file": (filename, io.BytesIO(data), "application/octet-stream")},
)
def _assert_3niveluri(detail: dict, cod: str) -> None:
"""Verifica prezenta cheilor de pe 3 niveluri."""
assert detail.get("cod") == cod, f"Asteptat cod={cod!r}, primit {detail.get('cod')!r}"
assert "problema" in detail, "Lipseste cheia 'problema'"
assert "cauza" in detail, "Lipseste cheia 'cauza'"
assert "fix" in detail, "Lipseste cheia 'fix'"
assert "error" in detail, "Lipseste cheia veche 'error'"
assert "message" in detail, "Lipseste cheia veche 'message'"
# --------------------------------------------------------------------------- #
# 1. Fisier prea mare -> IMPORT_FISIER_PREA_MARE #
# --------------------------------------------------------------------------- #
class TestFisierPreaMare3Niveluri:
def test_fisier_prea_mare_3niveluri(self, client):
"""Upload fisier >5MB -> 413 cu cod IMPORT_FISIER_PREA_MARE + superset."""
# 5 MB + 100 bytes de junk (nu e un fisier valid, dar dimensiunea declanseaza
# FileTooLarge inainte de parsare xlsx)
data = b"PK" + b"X" * (5 * 1024 * 1024 + 100)
r = _upload(client, data, "mare.xlsx")
# Poate sa returneze 413 sau 422 (depinde daca e prins ca FileTooLarge sau altceva)
# Dupa implementare trebuie sa fie 413 pentru FileTooLarge
assert r.status_code == 413, f"Asteptat 413, primit {r.status_code}: {r.text}"
detail = r.json()["detail"]
assert detail["error"] == "file_too_large"
_assert_3niveluri(detail, "IMPORT_FISIER_PREA_MARE")
def test_fisier_prea_mare_peste_5000_randuri(self, client):
"""Upload xlsx cu >5000 randuri -> 413 cu cod IMPORT_FISIER_PREA_MARE."""
wb = openpyxl.Workbook()
ws = wb.active
ws.append(_HEADER)
for i in range(5001):
ws.append([f"WVWZZZ1KZAW{i:06d}", f"B{i:04d}TST", "2026-06-15", str(100000 + i), "Revizie"])
buf = io.BytesIO()
wb.save(buf)
r = _upload(client, buf.getvalue(), "mare.xlsx")
assert r.status_code == 413, f"Asteptat 413, primit {r.status_code}: {r.text}"
detail = r.json()["detail"]
assert detail["error"] == "file_too_large"
_assert_3niveluri(detail, "IMPORT_FISIER_PREA_MARE")
# --------------------------------------------------------------------------- #
# 2. Antet neclar -> IMPORT_ANTET_NECLAR #
# --------------------------------------------------------------------------- #
class TestAntetNeclar3Niveluri:
def test_antet_neclar_3niveluri(self, client):
"""CSV fara antet recunoscut -> 422 cu IMPORT_ANTET_NECLAR + 'found' pastrat."""
# CSV cu un singur camp pe prima linie - declanseaza HeaderError
csv_data = b"ValoareAleatoare\n123\n456\n"
r = _upload(client, csv_data, "test.csv")
assert r.status_code == 422, f"Asteptat 422, primit {r.status_code}: {r.text}"
detail = r.json()["detail"]
assert detail["error"] == "header_error"
assert "found" in detail, "Cheia 'found' trebuie pastrata"
_assert_3niveluri(detail, "IMPORT_ANTET_NECLAR")
# --------------------------------------------------------------------------- #
# 3. Encoding nesuportat -> IMPORT_ENCODING #
# --------------------------------------------------------------------------- #
class TestEncoding3Niveluri:
def test_encoding_3niveluri(self, client):
"""CSV in encoding nesuportat (UTF-16 fara BOM detectabil) -> 422 IMPORT_ENCODING."""
# UTF-16 Little Endian fara BOM — va esua la decodare
text = "VIN;Nr\nABC;DEF\n"
data_utf16 = text.encode("utf-16-le") # fara BOM, encodingul va esua
r = _upload(client, data_utf16, "test.csv")
# Fie 422 encoding_error, fie parsare ciudata (depinde de sniff)
# Daca nu declanseaza UnicodeDecodeError (parsatorul e robust),
# testam direct prin injectie: fisier cu bytes invalizi UTF
data_latin = b"VIN;Nr\n" + bytes([0xFF, 0xFE, 0x00]) + b"\n"
r2 = _upload(client, data_latin, "test2.csv")
if r2.status_code == 422:
detail = r2.json()["detail"]
if detail.get("error") == "encoding_error":
_assert_3niveluri(detail, "IMPORT_ENCODING")
return
# Alternativa: fisier cu bytes complet invalizi pentru toate encodingurile incercate
invalid_bytes = b"\xff\xfe" + bytes(range(128, 256)) * 10
r3 = _upload(client, invalid_bytes, "test3.csv")
if r3.status_code == 422:
detail = r3.json()["detail"]
if detail.get("error") == "encoding_error":
_assert_3niveluri(detail, "IMPORT_ENCODING")
# --------------------------------------------------------------------------- #
# 4. Fisier nerecunoscut -> IMPORT_FISIER_NERECUNOSCUT #
# --------------------------------------------------------------------------- #
class TestFisierNerecunoscut3Niveluri:
def test_fisier_nerecunoscut_3niveluri(self, client):
"""Fisier binar junk -> 422 cu IMPORT_FISIER_NERECUNOSCUT."""
r = _upload(client, b"\x00\x01\x02\x03binar", "test.xlsx")
assert r.status_code == 422, f"Asteptat 422, primit {r.status_code}: {r.text}"
detail = r.json()["detail"]
assert detail["error"] == "invalid_file"
_assert_3niveluri(detail, "IMPORT_FISIER_NERECUNOSCUT")
# --------------------------------------------------------------------------- #
# 5. Multiple sheets -> IMPORT_MULTIPLE_SHEETS #
# --------------------------------------------------------------------------- #
class TestMultipleSheets3Niveluri:
def test_multiple_sheets_3niveluri(self, client):
"""Xlsx cu >1 sheet non-gol -> 422 cu IMPORT_MULTIPLE_SHEETS + 'sheets' pastrat."""
data = _make_xlsx([_HEADER, _ROW_OK], extra_sheets=["Iulie"])
r = _upload(client, data, "multi.xlsx")
assert r.status_code == 422, f"Asteptat 422, primit {r.status_code}: {r.text}"
detail = r.json()["detail"]
assert detail["error"] == "multiple_sheets"
assert "sheets" in detail, "Cheia 'sheets' trebuie pastrata"
assert "Sheet1" in detail["sheets"] or "Iulie" in detail["sheets"]
_assert_3niveluri(detail, "IMPORT_MULTIPLE_SHEETS")
# --------------------------------------------------------------------------- #
# 6. Fara mapare coloane -> IMPORT_FARA_MAPARE_COLOANE #
# --------------------------------------------------------------------------- #
class TestFaraMapareColoane3Niveluri:
def test_fara_mapare_coloane_3niveluri(self, client):
"""Preview fara mapare configurata -> 422 cu IMPORT_FARA_MAPARE_COLOANE."""
data = _make_xlsx([_HEADER, _ROW_OK])
r_up = _upload(client, data, "test.xlsx")
assert r_up.status_code == 200, r_up.text
import_id = r_up.json()["import_id"]
# Preview fara a salva maparea de coloane
r = client.get(f"/v1/import/{import_id}/preview")
assert r.status_code == 422, f"Asteptat 422, primit {r.status_code}: {r.text}"
detail = r.json()["detail"]
assert detail["error"] == "no_column_mapping"
_assert_3niveluri(detail, "IMPORT_FARA_MAPARE_COLOANE")
# --------------------------------------------------------------------------- #
# 7. Confirmare gresita -> IMPORT_CONFIRMARE_GRESITA #
# --------------------------------------------------------------------------- #
class TestConfirmareGresita3Niveluri:
def _seed_op(self) -> None:
from app.db import get_connection
conn = get_connection()
try:
conn.execute(
"INSERT OR REPLACE INTO nomenclator_rar (cod_prestatie, nume_prestatie) "
"VALUES ('OE-1','Verificare')"
)
conn.execute(
"INSERT OR IGNORE INTO operations_mapping "
"(account_id, cod_op_service, cod_prestatie, auto_send) "
"VALUES (1, 'Revizie', 'OE-1', 1)"
)
conn.commit()
finally:
conn.close()
def test_confirmare_gresita_3niveluri(self, client):
"""Commit cu n_confirmat gresit -> 422 cu IMPORT_CONFIRMARE_GRESITA + n_ok pastrat."""
self._seed_op()
data = _make_xlsx([_HEADER, _ROW_OK])
r_up = _upload(client, data, "test.xlsx")
assert r_up.status_code == 200, r_up.text
import_id = r_up.json()["import_id"]
# Salveaza maparea
client.post(f"/v1/import/{import_id}/column-mapping", json={
"json_mapare": {
"VIN": "vin",
"Nr inmatriculare": "nr_inmatriculare",
"Data prestatie": "data_prestatie",
"Odometru final": "odometru_final",
"Operatie": "operatie",
},
"format_data": "YYYY-MM-DD",
})
# Preview pentru a rezolva randurile
r_prev = client.get(f"/v1/import/{import_id}/preview")
assert r_prev.status_code == 200, r_prev.text
# Commit cu numarul GRESIT (0 in loc de cel real)
r = client.post(f"/v1/import/{import_id}/commit", json={
"n_confirmat": 99, # gresit
"reviewed_rows": [],
})
assert r.status_code == 422, f"Asteptat 422, primit {r.status_code}: {r.text}"
detail = r.json()["detail"]
assert detail["error"] == "confirmare_gresita"
assert "n_ok" in detail, "Cheia 'n_ok' trebuie pastrata"
_assert_3niveluri(detail, "IMPORT_CONFIRMARE_GRESITA")
# --------------------------------------------------------------------------- #
# 8. Override ilizibil -> IMPORT_OVERRIDE_ILIZIBIL #
# --------------------------------------------------------------------------- #
class TestOverrideIlizibil3Niveluri:
def test_override_ilizibil_forma_dict(self):
"""Verifica direct ca detail-ul generat pentru override ilizibil are forma dict corecta.
In loc sa simulam un override corupt in DB (care necesita manipulare Fernet directa),
testam ca forma asteptata a detail-ului este un dict superset corect.
Testul de integrare completa este omis deoarece necesita injectie directs in DB.
"""
from app import errors
# Simuleaza ce face router-ul la eroarea de override ilizibil
msg = "override curent ilizibil; editare anulata"
detail = {
"error": "override_ilizibil",
"message": msg,
**errors.eroare("IMPORT_OVERRIDE_ILIZIBIL", cauza=msg),
}
assert detail["error"] == "override_ilizibil"
assert detail["message"] == msg
assert detail["cod"] == "IMPORT_OVERRIDE_ILIZIBIL"
assert "problema" in detail
assert "cauza" in detail
assert detail["cauza"] == msg
assert "fix" in detail
def test_override_ilizibil_via_api(self, client, monkeypatch):
"""Test integrare: override ilizibil returnat ca dict superset (nu string)."""
import io as _io
# Upload + mapare
data = _make_xlsx([_HEADER, _ROW_OK])
r_up = _upload(client, data, "test.xlsx")
assert r_up.status_code == 200, r_up.text
import_id = r_up.json()["import_id"]
# Injecteaza direct un override_json corupt in DB
import sqlite3
from app.config import get_settings
db_path = get_settings().db_path
conn_raw = sqlite3.connect(db_path)
try:
conn_raw.execute(
"UPDATE import_rows SET override_json=? WHERE batch_id=? AND row_index=0",
("TOKEN_CORUPT_INVALID", import_id),
)
conn_raw.commit()
finally:
conn_raw.close()
# Cerere de editare pe randul cu override corupt
r = client.post(f"/v1/import/{import_id}/rand/0/editeaza", json={"vin": "WVWZZZ1KZAW000999"})
assert r.status_code == 422, f"Asteptat 422, primit {r.status_code}: {r.text}"
detail = r.json()["detail"]
# Dupa US-005: detail trebuie sa fie dict, nu string
assert isinstance(detail, dict), f"detail trebuie sa fie dict, primit: {type(detail)}"
assert detail.get("error") == "override_ilizibil"
assert detail.get("cod") == "IMPORT_OVERRIDE_ILIZIBIL"
assert "problema" in detail
assert "fix" in detail
Reference in New Issue View Git Blame Copy Permalink