gomag-vending/api/tests/test_partner_cui_lookup.py

"""
Partner CUI Lookup — Oracle PL/SQL Strict Mode Regression
==========================================================
Tests for cauta_partener_dupa_cod_fiscal (PACK_IMPORT_PARTENERI).

Regression for FG COFFE #485065210: GoMag CUI "RO 34963277" (with space)
must find the existing ROA partner stored as "RO34963277" (no space) instead
of creating a duplicate.

Business rule in strict mode:
- Input with RO prefix (platitor TVA) → only match RO<bare> / RO <bare>
- Input without RO prefix (neplatitor) → only match <bare> (no cross-match)

Run:
    ./test.sh oracle
    pytest api/tests/test_partner_cui_lookup.py -v
"""

import os
import sys
import time

import pytest

pytestmark = pytest.mark.oracle

_script_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")

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)


@pytest.fixture(scope="module")
def oracle_pool():
    from app.config import settings
    from app import database
    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"
    database.init_oracle()
    yield database.pool


@pytest.fixture(scope="module")
def test_suffix():
    """Unique suffix per test run to avoid partner name collisions."""
    return f"PYT{int(time.time()) % 100000}"


def _unique_bare(pool, prefix: str) -> str:
    """Generate a CUI that doesn't exist in any form in nom_parteneri."""
    conn = pool.acquire()
    try:
        with conn.cursor() as cur:
            for i in range(100):
                candidate = f"{prefix}{int(time.time() * 1000) % 100000 + i:05d}"
                cur.execute("""
                    SELECT COUNT(*) FROM nom_parteneri
                    WHERE UPPER(TRIM(cod_fiscal)) IN (:1, 'RO' || :2, 'RO ' || :3)
                """, [candidate, candidate, candidate])
                if cur.fetchone()[0] == 0:
                    return candidate
            raise RuntimeError("Could not find unique CUI after 100 attempts")
    finally:
        pool.release(conn)


def _seed_partner(pool, cod_fiscal: str, denumire: str) -> int:
    """Insert a test partner row directly. Returns actual id_part (table trigger assigns ID)."""
    import oracledb
    conn = pool.acquire()
    try:
        with conn.cursor() as cur:
            id_out = cur.var(oracledb.DB_TYPE_NUMBER)
            cur.execute("""
                INSERT INTO nom_parteneri (id_part, denumire, cod_fiscal, sters, inactiv)
                VALUES (NVL((SELECT MAX(id_part)+1 FROM nom_parteneri), 1), :1, :2, 0, 0)
                RETURNING id_part INTO :3
            """, [denumire, cod_fiscal, id_out])
            conn.commit()
            return int(id_out.getvalue()[0])
    finally:
        pool.release(conn)


def _cleanup_partners(pool, id_list):
    if not id_list:
        return
    conn = pool.acquire()
    try:
        with conn.cursor() as cur:
            placeholders = ",".join(f":{i+1}" for i in range(len(id_list)))
            cur.execute(f"DELETE FROM nom_parteneri WHERE id_part IN ({placeholders})", id_list)
            conn.commit()
    except Exception as e:
        print(f"Cleanup warning: {e}")
    finally:
        pool.release(conn)


def _call_lookup(pool, cod_fiscal: str, strict: int | None):
    """Call PACK_IMPORT_PARTENERI.cauta_partener_dupa_cod_fiscal."""
    import oracledb
    conn = pool.acquire()
    try:
        with conn.cursor() as cur:
            return cur.callfunc(
                "PACK_IMPORT_PARTENERI.cauta_partener_dupa_cod_fiscal",
                oracledb.DB_TYPE_NUMBER,
                [cod_fiscal, strict],
            )
    finally:
        pool.release(conn)


# ===========================================================================
# Strict mode: RO prefix tolerance (FIX 2a regression)
# ===========================================================================

class TestStrictROPrefix:
    """Strict mode must cross-match 'RO123' and 'RO 123' (only space differs)."""

    def test_input_ro_space_finds_partner_ro_no_space(self, oracle_pool, test_suffix):
        """GoMag sends 'RO 34963277', ROA has 'RO34963277' → MUST find it (FG COFFE regression)."""
        cuf_bare = _unique_bare(oracle_pool, "9911")
        ro_no_space = f"RO{cuf_bare}"
        ids = []
        try:
            pid = _seed_partner(oracle_pool, ro_no_space, f"TEST_FG_COFFE_{test_suffix}")
            ids.append(pid)

            # GoMag input with space must still locate the partner stored without space
            found = _call_lookup(oracle_pool, f"RO {cuf_bare}", strict=1)
            assert found == pid, (
                f"Strict lookup for 'RO {cuf_bare}' must find partner stored as '{ro_no_space}'"
            )
        finally:
            _cleanup_partners(oracle_pool, ids)

    def test_input_ro_no_space_finds_partner_ro_space(self, oracle_pool, test_suffix):
        """Partner stored as 'RO 34963277' (with space) found via 'RO34963277' input."""
        cuf_bare = _unique_bare(oracle_pool, "9922")
        ro_space = f"RO {cuf_bare}"
        ids = []
        try:
            pid = _seed_partner(oracle_pool, ro_space, f"TEST_AUTOKLASS_{test_suffix}")
            ids.append(pid)

            found = _call_lookup(oracle_pool, f"RO{cuf_bare}", strict=1)
            assert found == pid
        finally:
            _cleanup_partners(oracle_pool, ids)

    def test_strict_bare_input_does_not_match_ro_form(self, oracle_pool, test_suffix):
        """Business rule: neplatitor TVA (bare '123') must NOT match platitor stored as 'RO123'."""
        cuf_bare = _unique_bare(oracle_pool, "9933")
        ro_form = f"RO{cuf_bare}"
        ids = []
        try:
            pid = _seed_partner(oracle_pool, ro_form, f"TEST_OLLYS_{test_suffix}")
            ids.append(pid)

            # Bare input + strict=1 → must NOT find the RO-form partner
            found = _call_lookup(oracle_pool, cuf_bare, strict=1)
            assert found is None, (
                f"Strict bare '{cuf_bare}' must not cross-match 'RO{cuf_bare}' "
                f"(different fiscal entities)"
            )
        finally:
            _cleanup_partners(oracle_pool, ids)

    def test_strict_ro_input_does_not_match_bare_form(self, oracle_pool, test_suffix):
        """Business rule: RO input (platitor) must NOT match bare stored form (neplatitor)."""
        cuf_bare = _unique_bare(oracle_pool, "9944")
        ids = []
        try:
            pid = _seed_partner(oracle_pool, cuf_bare, f"TEST_VENUS_{test_suffix}")
            ids.append(pid)

            found = _call_lookup(oracle_pool, f"RO{cuf_bare}", strict=1)
            assert found is None, (
                f"Strict 'RO{cuf_bare}' must not cross-match bare '{cuf_bare}'"
            )
        finally:
            _cleanup_partners(oracle_pool, ids)


# ===========================================================================
# Non-strict mode: backward compat — match any of 3 forms
# ===========================================================================

class TestNonStrict:
    """Non-strict (p_strict_search=NULL) matches all 3 forms (anti-dedup fallback)."""

    def test_non_strict_bare_finds_ro_form(self, oracle_pool, test_suffix):
        cuf_bare = _unique_bare(oracle_pool, "9955")
        ids = []
        try:
            pid = _seed_partner(oracle_pool, f"RO{cuf_bare}", f"TEST_CONVER_{test_suffix}")
            ids.append(pid)
            found = _call_lookup(oracle_pool, cuf_bare, strict=None)
            assert found == pid, "Non-strict must cross-match (anti-dedup fallback)"
        finally:
            _cleanup_partners(oracle_pool, ids)