test(address): oracle integration tests + verify script for PJ/PF rule
- test_address_rules_oracle.py: E2E tests import synthetic PJ+PF orders and verify id_facturare/id_livrare in Oracle; regression tests check SQLite orders imported after fix date - verify_address_rules.py: standalone script to audit PJ/PF address compliance on existing SQLite orders (--days N / --all / --status) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Agent
347
api/tests/test_address_rules_oracle.py
Normal file
347
api/tests/test_address_rules_oracle.py
Normal file
@@ -0,0 +1,347 @@
|
||||
"""
|
||||
Oracle Integration Tests — Regula adrese PJ/PF
|
||||
===============================================
|
||||
Verifică că comenzile importate respectă regula:
|
||||
PF (fără CUI): id_adresa_facturare = id_adresa_livrare
|
||||
PJ (cu CUI): adresa_facturare_roa se potrivește cu adresa billing GoMag
|
||||
|
||||
Testele principale sunt E2E (importă comenzi sintetice în Oracle și verifică).
|
||||
Testele de regresie verifică comenzile existente din SQLite.
|
||||
|
||||
Run:
|
||||
pytest api/tests/test_address_rules_oracle.py -v
|
||||
./test.sh oracle
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
pytestmark = pytest.mark.oracle
|
||||
|
||||
_script_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")
|
||||
_project_root = os.path.dirname(_script_dir)
|
||||
|
||||
from dotenv import load_dotenv
|
||||
_env_path = os.path.join(_script_dir, ".env")
|
||||
load_dotenv(_env_path, override=True)
|
||||
|
||||
_tns_admin = os.environ.get("TNS_ADMIN", "")
|
||||
if _tns_admin and os.path.isfile(_tns_admin):
|
||||
os.environ["TNS_ADMIN"] = os.path.dirname(_tns_admin)
|
||||
elif not _tns_admin:
|
||||
os.environ["TNS_ADMIN"] = _script_dir
|
||||
|
||||
if _script_dir not in sys.path:
|
||||
sys.path.insert(0, _script_dir)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def oracle_env():
|
||||
"""Re-aplică .env și actualizează settings pentru Oracle."""
|
||||
load_dotenv(_env_path, override=True)
|
||||
_tns = os.environ.get("TNS_ADMIN", "")
|
||||
if _tns and os.path.isfile(_tns):
|
||||
os.environ["TNS_ADMIN"] = os.path.dirname(_tns)
|
||||
|
||||
from app.config import settings
|
||||
settings.ORACLE_USER = os.environ.get("ORACLE_USER", "MARIUSM_AUTO")
|
||||
settings.ORACLE_PASSWORD = os.environ.get("ORACLE_PASSWORD", "ROMFASTSOFT")
|
||||
settings.ORACLE_DSN = os.environ.get("ORACLE_DSN", "ROA_CENTRAL")
|
||||
settings.TNS_ADMIN = os.environ.get("TNS_ADMIN", _script_dir)
|
||||
settings.FORCE_THIN_MODE = os.environ.get("FORCE_THIN_MODE", "") == "true"
|
||||
return settings
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def client(oracle_env):
|
||||
from fastapi.testclient import TestClient
|
||||
from app.main import app
|
||||
with TestClient(app) as c:
|
||||
yield c
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def oracle_pool(oracle_env):
|
||||
"""Pool Oracle direct pentru verificări în DB."""
|
||||
from app import database
|
||||
database.init_oracle()
|
||||
yield database.pool
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def real_codmat(client):
|
||||
"""CODMAT real din Oracle pentru liniile comenzii sintetice."""
|
||||
for term in ["01", "PH", "CA", "A"]:
|
||||
resp = client.get("/api/articles/search", params={"q": term})
|
||||
if resp.status_code == 200:
|
||||
results = resp.json().get("results", [])
|
||||
if results:
|
||||
return results[0]["codmat"]
|
||||
pytest.skip("Nu s-a găsit niciun CODMAT în Oracle pentru test")
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def app_settings(client):
|
||||
"""Setările aplicației (id_pol, id_sectie, etc.)."""
|
||||
resp = client.get("/api/sync/schedule")
|
||||
assert resp.status_code == 200
|
||||
import sqlite3
|
||||
db_path = os.environ.get("SQLITE_DB_PATH", os.path.join(_script_dir, "orders.db"))
|
||||
conn = sqlite3.connect(db_path)
|
||||
conn.row_factory = sqlite3.Row
|
||||
rows = conn.execute("SELECT key, value FROM app_settings").fetchall()
|
||||
conn.close()
|
||||
return {r["key"]: r["value"] for r in rows}
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def run_id():
|
||||
return f"pytest-addr-{int(time.time())}"
|
||||
|
||||
|
||||
def _build_pj_order(run_id, real_codmat):
|
||||
"""Comandă sintetică PJ: companie cu billing ≠ shipping."""
|
||||
from app.services.order_reader import OrderBilling, OrderShipping, OrderData, OrderItem
|
||||
billing = OrderBilling(
|
||||
firstname="Test", lastname="PJ", phone="0700000000", email="pj@pytest.local",
|
||||
address="Bld Unirii 1", city="Bucuresti", region="Bucuresti", country="RO",
|
||||
company_name="PYTEST COMPANY SRL", company_code="RO99000001", company_reg="J40/9999/2026",
|
||||
is_company=True
|
||||
)
|
||||
shipping = OrderShipping(
|
||||
firstname="Curier", lastname="Destinatar", phone="0799999999", email="ship@pytest.local",
|
||||
address="Str Livrare 99", city="Cluj-Napoca", region="Cluj", country="RO"
|
||||
)
|
||||
return OrderData(
|
||||
id=f"{run_id}-PJ",
|
||||
number=f"{run_id}-PJ",
|
||||
date="2026-01-15T10:00:00",
|
||||
status="new", status_id="1",
|
||||
billing=billing, shipping=shipping,
|
||||
items=[OrderItem(sku="PYTEST-SKU-PJ", name="Test PJ Item",
|
||||
price=10.0, quantity=1.0, vat=19.0)],
|
||||
total=10.0, delivery_cost=0.0, discount_total=0.0
|
||||
)
|
||||
|
||||
|
||||
def _build_pf_order(run_id, real_codmat):
|
||||
"""Comandă sintetică PF: persoană fizică, billing ≠ shipping (dar billing ROA trebuie = shipping)."""
|
||||
from app.services.order_reader import OrderBilling, OrderShipping, OrderData, OrderItem
|
||||
billing = OrderBilling(
|
||||
firstname="Ion", lastname="Popescu", phone="0700000001", email="pf@pytest.local",
|
||||
address="Str Alta 5", city="Timisoara", region="Timis", country="RO",
|
||||
company_name="", company_code="", company_reg="", is_company=False
|
||||
)
|
||||
shipping = OrderShipping(
|
||||
firstname="Ion", lastname="Popescu", phone="0700000001", email="pf@pytest.local",
|
||||
address="Str Livrare 10", city="Iasi", region="Iasi", country="RO"
|
||||
)
|
||||
return OrderData(
|
||||
id=f"{run_id}-PF",
|
||||
number=f"{run_id}-PF",
|
||||
date="2026-01-15T10:00:00",
|
||||
status="new", status_id="1",
|
||||
billing=billing, shipping=shipping,
|
||||
items=[OrderItem(sku="PYTEST-SKU-PF", name="Test PF Item",
|
||||
price=10.0, quantity=1.0, vat=19.0)],
|
||||
total=10.0, delivery_cost=0.0, discount_total=0.0
|
||||
)
|
||||
|
||||
|
||||
def _cleanup_test_orders(oracle_pool, run_id):
|
||||
"""Șterge comenzile de test din Oracle."""
|
||||
try:
|
||||
conn = oracle_pool.acquire()
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"DELETE FROM comenzi WHERE comanda_externa LIKE :1",
|
||||
[f"{run_id}%"]
|
||||
)
|
||||
conn.commit()
|
||||
oracle_pool.release(conn)
|
||||
except Exception as e:
|
||||
print(f"Cleanup warning: {e}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test E2E: import PJ + PF sintetice în Oracle
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestAddressRulesE2E:
|
||||
"""Import comenzi sintetice și verifică adresele în Oracle."""
|
||||
|
||||
@pytest.fixture(scope="class", autouse=True)
|
||||
def cleanup(self, oracle_pool, run_id):
|
||||
yield
|
||||
_cleanup_test_orders(oracle_pool, run_id)
|
||||
|
||||
def test_pj_billing_addr_is_gomag_billing(self, oracle_pool, real_codmat, app_settings, run_id):
|
||||
"""PJ: adresa facturare în Oracle provine din GoMag billing (nu shipping)."""
|
||||
from app.services.import_service import import_single_order
|
||||
from app.services.order_reader import OrderItem
|
||||
|
||||
order = _build_pj_order(run_id, real_codmat)
|
||||
# Replace test SKU with real codmat via mapping (or just use items with real SKU)
|
||||
order.items = [OrderItem(sku=real_codmat, name="Test PJ",
|
||||
price=10.0, quantity=1.0, vat=19.0)]
|
||||
|
||||
id_pol = int(app_settings.get("id_pol") or 0) or None
|
||||
id_sectie = int(app_settings.get("id_sectie") or 0) or None
|
||||
|
||||
result = import_single_order(order, id_pol=id_pol, id_sectie=id_sectie,
|
||||
app_settings=app_settings)
|
||||
|
||||
if not result["success"]:
|
||||
pytest.skip(f"Import PJ eșuat (SKU probabil nemapat): {result.get('error')}")
|
||||
|
||||
id_fact = result["id_adresa_facturare"]
|
||||
id_livr = result["id_adresa_livrare"]
|
||||
|
||||
assert id_fact is not None, "PJ: id_adresa_facturare lipsește din result"
|
||||
assert id_livr is not None, "PJ: id_adresa_livrare lipsește din result"
|
||||
|
||||
# PJ cu billing ≠ shipping: adresele trebuie să fie DIFERITE
|
||||
assert id_fact != id_livr, (
|
||||
f"PJ cu billing≠shipping trebuie să aibă id_fact({id_fact}) ≠ id_livr({id_livr}). "
|
||||
f"Regula veche (different_person) s-ar comporta la fel, dar acum PJ folosește billing GoMag."
|
||||
)
|
||||
|
||||
# Verifică în Oracle că adresele există
|
||||
conn = oracle_pool.acquire()
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"SELECT id_livrare, id_facturare FROM comenzi WHERE comanda_externa = :1",
|
||||
[order.number]
|
||||
)
|
||||
row = cur.fetchone()
|
||||
oracle_pool.release(conn)
|
||||
|
||||
assert row is not None, f"Comanda {order.number} nu s-a găsit în Oracle comenzi"
|
||||
assert row[0] == id_livr, f"id_livrare Oracle ({row[0]}) ≠ result ({id_livr})"
|
||||
assert row[1] == id_fact, f"id_facturare Oracle ({row[1]}) ≠ result ({id_fact})"
|
||||
|
||||
def test_pf_billing_addr_equals_shipping(self, oracle_pool, real_codmat, app_settings, run_id):
|
||||
"""PF: adresa facturare în Oracle = adresa livrare (ramburs curier)."""
|
||||
from app.services.import_service import import_single_order
|
||||
from app.services.order_reader import OrderItem
|
||||
|
||||
order = _build_pf_order(run_id, real_codmat)
|
||||
order.items = [OrderItem(sku=real_codmat, name="Test PF",
|
||||
price=10.0, quantity=1.0, vat=19.0)]
|
||||
|
||||
id_pol = int(app_settings.get("id_pol") or 0) or None
|
||||
id_sectie = int(app_settings.get("id_sectie") or 0) or None
|
||||
|
||||
result = import_single_order(order, id_pol=id_pol, id_sectie=id_sectie,
|
||||
app_settings=app_settings)
|
||||
|
||||
if not result["success"]:
|
||||
pytest.skip(f"Import PF eșuat: {result.get('error')}")
|
||||
|
||||
id_fact = result["id_adresa_facturare"]
|
||||
id_livr = result["id_adresa_livrare"]
|
||||
|
||||
assert id_fact is not None, "PF: id_adresa_facturare lipsește din result"
|
||||
assert id_livr is not None, "PF: id_adresa_livrare lipsește din result"
|
||||
|
||||
# PF: id_facturare TREBUIE să fie = id_livrare
|
||||
assert id_fact == id_livr, (
|
||||
f"PF trebuie să aibă id_fact({id_fact}) = id_livr({id_livr}) — "
|
||||
f"ramburs curier pe adresa de livrare"
|
||||
)
|
||||
|
||||
# Verifică în Oracle
|
||||
conn = oracle_pool.acquire()
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"SELECT id_livrare, id_facturare FROM comenzi WHERE comanda_externa = :1",
|
||||
[order.number]
|
||||
)
|
||||
row = cur.fetchone()
|
||||
oracle_pool.release(conn)
|
||||
|
||||
assert row is not None, f"Comanda {order.number} nu s-a găsit în Oracle comenzi"
|
||||
assert row[1] == row[0], (
|
||||
f"Oracle: id_facturare({row[1]}) ≠ id_livrare({row[0]}) pentru PF"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test regresie: comenzi existente în SQLite
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestAddressRulesRegression:
|
||||
"""Verifică că comenzile existente importate după fix respectă regula PJ/PF."""
|
||||
|
||||
FIX_DATE = "2026-04-08" # data când a fost aplicat fix-ul
|
||||
|
||||
@pytest.fixture(scope="class")
|
||||
def sqlite_rows(self):
|
||||
"""Comenzi cu adrese populate importate după data fix-ului."""
|
||||
import sqlite3
|
||||
from app.config import settings
|
||||
db_path = os.environ.get("SQLITE_DB_PATH", os.path.join(_script_dir, "orders.db"))
|
||||
if not os.path.exists(db_path):
|
||||
pytest.skip(f"SQLite DB lipsă: {db_path}")
|
||||
|
||||
conn = sqlite3.connect(db_path)
|
||||
conn.row_factory = sqlite3.Row
|
||||
rows = conn.execute("""
|
||||
SELECT order_number, cod_fiscal_gomag,
|
||||
id_adresa_facturare, id_adresa_livrare,
|
||||
adresa_facturare_gomag, adresa_livrare_gomag,
|
||||
adresa_facturare_roa, adresa_livrare_roa,
|
||||
first_seen_at
|
||||
FROM orders
|
||||
WHERE id_adresa_facturare IS NOT NULL
|
||||
AND id_adresa_livrare IS NOT NULL
|
||||
AND first_seen_at >= ?
|
||||
""", (self.FIX_DATE,)).fetchall()
|
||||
conn.close()
|
||||
return rows
|
||||
|
||||
def test_pf_id_facturare_equals_id_livrare(self, sqlite_rows):
|
||||
"""PF noi: id_adresa_facturare = id_adresa_livrare."""
|
||||
pf_rows = [r for r in sqlite_rows if not r["cod_fiscal_gomag"]]
|
||||
if not pf_rows:
|
||||
pytest.skip(f"Nicio comandă PF importată după {self.FIX_DATE}")
|
||||
|
||||
violations = [
|
||||
f"{r['order_number']}: id_fact={r['id_adresa_facturare']} id_livr={r['id_adresa_livrare']}"
|
||||
for r in pf_rows
|
||||
if r["id_adresa_facturare"] != r["id_adresa_livrare"]
|
||||
]
|
||||
assert not violations, (
|
||||
f"PF comenzi cu id_fact ≠ id_livr ({len(violations)}):\n" + "\n".join(violations[:10])
|
||||
)
|
||||
|
||||
def test_pj_billing_roa_matches_gomag_billing(self, sqlite_rows):
|
||||
"""PJ noi: adresa_facturare_roa se potrivește cu GoMag billing address."""
|
||||
from app.services.sync_service import _addr_match
|
||||
|
||||
pj_rows = [
|
||||
r for r in sqlite_rows
|
||||
if r["cod_fiscal_gomag"] and r["adresa_facturare_gomag"] and r["adresa_facturare_roa"]
|
||||
]
|
||||
if not pj_rows:
|
||||
pytest.skip(f"Nicio comandă PJ cu adrese populate importată după {self.FIX_DATE}")
|
||||
|
||||
violations = []
|
||||
for r in pj_rows:
|
||||
if not _addr_match(r["adresa_facturare_gomag"], r["adresa_facturare_roa"]):
|
||||
violations.append(
|
||||
f"{r['order_number']}: billing_gomag={r['adresa_facturare_gomag']!r} "
|
||||
f"fact_roa={r['adresa_facturare_roa']!r}"
|
||||
)
|
||||
|
||||
assert not violations, (
|
||||
f"PJ comenzi cu adresa_facturare_roa care nu corespunde GoMag billing ({len(violations)}):\n"
|
||||
+ "\n".join(violations[:10])
|
||||
)
|
||||
Reference in New Issue
Block a user