feat(T5/dashboard): import DBF idempotent + nomenclator browser + audit CSV + stare RAR
T5 (tools/import_dbf.py): citire prestatii_rar.DBF / mapare_prestatii.DBF cu dbfread, raport dry-run (randuri valide/duplicate/goale, mapari orfane = cod necunoscut in nomenclator), --commit cu upsert idempotent in tranzactie. Dashboard: browser nomenclator, indicator stare RAR (indisponibil? derivat din ultimul login < 30h, coada arata ultima stare locala), export audit CSV (/v1/audit/export?status=sent|all&date_from&date_to, b64Image exclus, coloana purge_after pentru retentia 90z). Verify: 11 teste noi (test_import_dbf 6, test_dashboard 5), suita 111 pass, dry-run real pe DBF-urile din repo + smoke live dashboard/CSV. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Agent
205
tests/test_import_dbf.py
Normal file
205
tests/test_import_dbf.py
Normal file
@@ -0,0 +1,205 @@
|
||||
"""Teste T5: import DBF -> SQLite (dry-run/raport + commit idempotent).
|
||||
|
||||
`mapare_prestatii.DBF` real e gol, deci pentru maparile reale scriem fixturi DBF
|
||||
dBASE III minimale. Nomenclatorul real (`prestatii_rar.DBF`, 20 randuri) e citit
|
||||
direct pentru un test de read pe date reale.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import struct
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from app.config import ROOT
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Helper: scriitor dBASE III minimal pentru fixturi #
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
def write_dbf(path: Path, fields: list[tuple[str, str, int]], records: list[dict]) -> None:
|
||||
"""Scrie un DBF dBASE III. fields = [(nume, tip C/L, lungime)]."""
|
||||
header = bytearray(32)
|
||||
header[0] = 0x03 # dBASE III, fara memo
|
||||
header[1:4] = bytes((25, 1, 1)) # data ultimei actualizari (fictiva)
|
||||
n_fields = len(fields)
|
||||
header_len = 32 + 32 * n_fields + 1
|
||||
record_len = 1 + sum(length for _, _, length in fields)
|
||||
struct.pack_into("<I", header, 4, len(records))
|
||||
struct.pack_into("<H", header, 8, header_len)
|
||||
struct.pack_into("<H", header, 10, record_len)
|
||||
|
||||
field_descs = bytearray()
|
||||
for name, ftype, length in fields:
|
||||
fd = bytearray(32)
|
||||
fd[0:11] = name.encode("ascii").ljust(11, b"\x00")[:11]
|
||||
fd[11] = ord(ftype)
|
||||
fd[16] = length
|
||||
field_descs += fd
|
||||
|
||||
body = bytearray()
|
||||
for rec in records:
|
||||
body += b"\x20" # flag stergere = spatiu (activ)
|
||||
for name, ftype, length in fields:
|
||||
val = rec.get(name)
|
||||
if ftype == "L":
|
||||
body += (b"T" if val else b"F")
|
||||
else:
|
||||
s = "" if val is None else str(val)
|
||||
body += s.encode("cp1252", "replace").ljust(length, b" ")[:length]
|
||||
|
||||
path.write_bytes(bytes(header) + bytes(field_descs) + b"\x0d" + bytes(body) + b"\x1a")
|
||||
|
||||
|
||||
MAPARE_FIELDS = [("COD_OP", "C", 10), ("DESCR_OP", "C", 40), ("COD_RAR", "C", 10), ("AUTO_SEND", "L", 1)]
|
||||
PREST_FIELDS = [("COD_PREST", "C", 10), ("NUME_PREST", "C", 60)]
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def env(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()
|
||||
yield Path(tmp)
|
||||
get_settings.cache_clear()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Read pe DBF #
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
def test_read_nomenclator_real_dbf():
|
||||
from tools.import_dbf import read_nomenclator
|
||||
|
||||
rep = read_nomenclator(ROOT / "prestatii_rar.DBF")
|
||||
assert len(rep["rows"]) == 20
|
||||
codes = {r["cod_prestatie"] for r in rep["rows"]}
|
||||
assert "OE-1" in codes and "R-ODO" in codes
|
||||
# codurile sunt normalizate upper
|
||||
assert all(r["cod_prestatie"] == r["cod_prestatie"].upper() for r in rep["rows"])
|
||||
|
||||
|
||||
def test_read_mapari_valid_blank_duplicate(tmp_path):
|
||||
from tools.import_dbf import read_mapari
|
||||
|
||||
p = tmp_path / "m.DBF"
|
||||
write_dbf(
|
||||
p,
|
||||
MAPARE_FIELDS,
|
||||
[
|
||||
{"COD_OP": "OP1", "DESCR_OP": "Reparatie motor", "COD_RAR": "oe-1", "AUTO_SEND": True},
|
||||
{"COD_OP": "OP2", "DESCR_OP": "Schimb ulei", "COD_RAR": "OE-2", "AUTO_SEND": False},
|
||||
{"COD_OP": "", "DESCR_OP": "fara cod op", "COD_RAR": "OE-3", "AUTO_SEND": True}, # blank
|
||||
{"COD_OP": "OP4", "DESCR_OP": "fara cod rar", "COD_RAR": "", "AUTO_SEND": True}, # blank
|
||||
{"COD_OP": "OP1", "DESCR_OP": "duplicat", "COD_RAR": "OE-9", "AUTO_SEND": True}, # duplicate
|
||||
],
|
||||
)
|
||||
rep = read_mapari(p)
|
||||
assert len(rep["rows"]) == 2
|
||||
assert rep["blanks"] == 2
|
||||
assert rep["duplicates"] == ["OP1"]
|
||||
# cod_rar normalizat upper, auto_send pastrat
|
||||
assert rep["rows"][0]["cod_prestatie"] == "OE-1"
|
||||
assert rep["rows"][0]["auto_send"] is True
|
||||
assert rep["rows"][1]["auto_send"] is False
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Orfane #
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
def test_find_orphans():
|
||||
from tools.import_dbf import find_orphans
|
||||
|
||||
mapari = [
|
||||
{"cod_op_service": "OP1", "cod_prestatie": "OE-1", "denumire": "x"},
|
||||
{"cod_op_service": "OP2", "cod_prestatie": "ZZZ", "denumire": "y"},
|
||||
]
|
||||
orphans = find_orphans(mapari, {"OE-1", "OE-2"})
|
||||
assert len(orphans) == 1
|
||||
assert orphans[0]["cod_op_service"] == "OP2"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# End-to-end dry-run + commit + idempotenta #
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
def _fixtures(dirpath: Path) -> tuple[Path, Path]:
|
||||
mp = dirpath / "mapare.DBF"
|
||||
np_ = dirpath / "prest.DBF"
|
||||
write_dbf(
|
||||
np_,
|
||||
PREST_FIELDS,
|
||||
[{"COD_PREST": "OE-1", "NUME_PREST": "REPARATIE"}, {"COD_PREST": "R-ODO", "NUME_PREST": "REPARATIE ODOMETRU"}],
|
||||
)
|
||||
write_dbf(
|
||||
mp,
|
||||
MAPARE_FIELDS,
|
||||
[
|
||||
{"COD_OP": "OP1", "DESCR_OP": "Reparatie", "COD_RAR": "OE-1", "AUTO_SEND": True},
|
||||
{"COD_OP": "OP2", "DESCR_OP": "Operatie necunoscuta", "COD_RAR": "XYZ", "AUTO_SEND": False}, # orfan
|
||||
],
|
||||
)
|
||||
return mp, np_
|
||||
|
||||
|
||||
def test_dry_run_does_not_write(env):
|
||||
from tools.import_dbf import run
|
||||
from app.db import get_connection
|
||||
|
||||
mp, np_ = _fixtures(env)
|
||||
res = run(commit=False, mapare_path=mp, prest_path=np_)
|
||||
assert res["written"] == {"nomenclator": 0, "mapari": 0}
|
||||
assert len(res["orphans"]) == 1
|
||||
assert res["orphans"][0]["cod_op_service"] == "OP2"
|
||||
|
||||
conn = get_connection()
|
||||
try:
|
||||
n = conn.execute("SELECT COUNT(*) AS n FROM operations_mapping").fetchone()["n"]
|
||||
assert n == 0
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_commit_writes_and_is_idempotent(env):
|
||||
from tools.import_dbf import run
|
||||
from app.db import get_connection
|
||||
|
||||
mp, np_ = _fixtures(env)
|
||||
res1 = run(commit=True, mapare_path=mp, prest_path=np_)
|
||||
assert res1["written"] == {"nomenclator": 2, "mapari": 2}
|
||||
|
||||
conn = get_connection()
|
||||
try:
|
||||
maps = conn.execute(
|
||||
"SELECT cod_op_service, cod_prestatie, auto_send FROM operations_mapping ORDER BY cod_op_service"
|
||||
).fetchall()
|
||||
assert [(m["cod_op_service"], m["cod_prestatie"], m["auto_send"]) for m in maps] == [
|
||||
("OP1", "OE-1", 1),
|
||||
("OP2", "XYZ", 0),
|
||||
]
|
||||
nom = conn.execute("SELECT COUNT(*) AS n FROM nomenclator_rar").fetchone()["n"]
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
# A doua rulare nu duplica (upsert pe cheile UNIQUE).
|
||||
run(commit=True, mapare_path=mp, prest_path=np_)
|
||||
conn = get_connection()
|
||||
try:
|
||||
assert conn.execute("SELECT COUNT(*) AS n FROM operations_mapping").fetchone()["n"] == 2
|
||||
assert conn.execute("SELECT COUNT(*) AS n FROM nomenclator_rar").fetchone()["n"] == nom
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_missing_dbf_raises(env):
|
||||
from tools.import_dbf import run
|
||||
|
||||
with pytest.raises(FileNotFoundError):
|
||||
run(commit=False, mapare_path=env / "nope.DBF", prest_path=env / "nope2.DBF")
|
||||
Reference in New Issue
Block a user