""" 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]) )