446 lines
19 KiB
Python
446 lines
19 KiB
Python
"""
|
|
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
|
|
from app.config import settings as _s
|
|
db_path = _s.SQLITE_DB_PATH if os.path.isabs(_s.SQLITE_DB_PATH) else os.path.join(_script_dir, _s.SQLITE_DB_PATH)
|
|
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: parsare componente adresă (strada, numar, bloc, scara, apart, etaj)
|
|
# Apelează direct parseaza_adresa_semicolon din Oracle — fără import comandă.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestAddressComponentParsing:
|
|
"""Verifică extragerea componentelor adresei direct prin parseaza_adresa_semicolon."""
|
|
|
|
def _parse_address(self, oracle_pool, address, city="Bucuresti", region="Bucuresti"):
|
|
"""Call Oracle parseaza_adresa_semicolon and return parsed components."""
|
|
from app.services.import_service import format_address_for_oracle
|
|
formatted = format_address_for_oracle(address, city, region)
|
|
|
|
conn = oracle_pool.acquire()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
p_judet = cur.var(str, 200)
|
|
p_localitate = cur.var(str, 200)
|
|
p_strada = cur.var(str, 100)
|
|
p_numar = cur.var(str, 100)
|
|
p_sector = cur.var(str, 100)
|
|
p_bloc = cur.var(str, 30)
|
|
p_scara = cur.var(str, 10)
|
|
p_apart = cur.var(str, 10)
|
|
p_etaj = cur.var(str, 20)
|
|
|
|
cur.callproc("PACK_IMPORT_PARTENERI.parseaza_adresa_semicolon", [
|
|
formatted, p_judet, p_localitate, p_strada, p_numar,
|
|
p_sector, p_bloc, p_scara, p_apart, p_etaj
|
|
])
|
|
|
|
return {
|
|
"strada": p_strada.getvalue(),
|
|
"numar": p_numar.getvalue(),
|
|
"bloc": p_bloc.getvalue(),
|
|
"scara": p_scara.getvalue(),
|
|
"apart": p_apart.getvalue(),
|
|
"etaj": p_etaj.getvalue(),
|
|
"localitate": p_localitate.getvalue(),
|
|
"judet": p_judet.getvalue(),
|
|
}
|
|
finally:
|
|
oracle_pool.release(conn)
|
|
|
|
def test_full_address_all_components(self, oracle_pool):
|
|
"""Adresa completă cu nr, bl, sc, ap — toate componentele se extrag din strada."""
|
|
addr = self._parse_address(oracle_pool,
|
|
"Bd. 1 Decembrie 1918 nr. 26 bl. 6 sc. 2 ap. 36")
|
|
assert addr["numar"] == "26", f"numar={addr['numar']}"
|
|
assert addr["bloc"] == "6", f"bloc={addr['bloc']}"
|
|
assert addr["scara"] == "2", f"scara={addr['scara']}"
|
|
assert addr["apart"] == "36", f"apart={addr['apart']}"
|
|
assert "SC" not in (addr["strada"] or ""), f"SC ramas in strada: {addr['strada']}"
|
|
assert "AP" not in (addr["strada"] or ""), f"AP ramas in strada: {addr['strada']}"
|
|
|
|
def test_alphanumeric_bloc_and_letter_scara(self, oracle_pool):
|
|
"""Bloc alfanumeric (VN9) și scara literă (A) + etaj."""
|
|
addr = self._parse_address(oracle_pool,
|
|
"Strada Becatei nr 29 bl. VN9 sc. A et. 10 ap. 42")
|
|
assert addr["numar"] == "29", f"numar={addr['numar']}"
|
|
assert addr["bloc"] == "VN9", f"bloc={addr['bloc']}"
|
|
assert addr["scara"] == "A", f"scara={addr['scara']}"
|
|
assert addr["etaj"] == "10", f"etaj={addr['etaj']}"
|
|
assert addr["apart"] == "42", f"apart={addr['apart']}"
|
|
|
|
def test_address_without_commas_uppercase(self, oracle_pool):
|
|
"""Adresa uppercase fără virgule — keywords spațiu-separate."""
|
|
addr = self._parse_address(oracle_pool,
|
|
"STR DACIA NR 15 BLOC Z2 SC 1 AP 7 ET 3")
|
|
assert addr["numar"] == "15", f"numar={addr['numar']}"
|
|
assert addr["bloc"] == "Z2", f"bloc={addr['bloc']}"
|
|
assert addr["scara"] == "1", f"scara={addr['scara']}"
|
|
assert addr["apart"] == "7", f"apart={addr['apart']}"
|
|
assert addr["etaj"] == "3", f"etaj={addr['etaj']}"
|
|
|
|
def test_address_with_existing_commas(self, oracle_pool):
|
|
"""Adresa care deja are virgule — nu se strică parsarea."""
|
|
addr = self._parse_address(oracle_pool,
|
|
"Str Victoriei, nr. 10, bl. A1, sc. B, et. 2, ap. 15")
|
|
assert addr["numar"] == "10", f"numar={addr['numar']}"
|
|
assert addr["bloc"] == "A1", f"bloc={addr['bloc']}"
|
|
assert addr["scara"] == "B", f"scara={addr['scara']}"
|
|
assert addr["etaj"] == "2", f"etaj={addr['etaj']}"
|
|
assert addr["apart"] == "15", f"apart={addr['apart']}"
|
|
|
|
def test_no_keywords_street_unchanged(self, oracle_pool):
|
|
"""Adresa simplă fără keywords — strada rămâne intactă."""
|
|
addr = self._parse_address(oracle_pool, "Strada Victoriei 10")
|
|
assert "VICTORIEI" in (addr["strada"] or ""), f"strada={addr['strada']}"
|
|
|
|
def test_blocuri_neighborhood_not_extracted_as_bloc(self, oracle_pool):
|
|
"""'Blocuri' in street name must NOT be parsed as BLOC keyword."""
|
|
result = self._parse_address(oracle_pool, "Str Principala Modarzau Blocuri", "Zemes", "Bacau")
|
|
assert "MODARZAU BLOCURI" in (result.get("strada") or ""), f"strada should contain MODARZAU BLOCURI, got {result}"
|
|
assert result.get("bloc") is None, f"bloc should be NULL for neighborhood name, got {result.get('bloc')}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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])
|
|
)
|