""" PL/SQL package tests using direct Oracle connection. Verifies that key Oracle packages are VALID and that order import procedures work end-to-end with cleanup. """ import json import time import logging import pytest pytestmark = pytest.mark.oracle logger = logging.getLogger(__name__) PACKAGES_TO_CHECK = [ "PACK_IMPORT_COMENZI", "PACK_IMPORT_PARTENERI", "PACK_COMENZI", "PACK_FACTURARE", ] _STATUS_SQL = """ SELECT status FROM user_objects WHERE object_name = :name AND object_type = 'PACKAGE BODY' """ # --------------------------------------------------------------------------- # Module-scoped fixture for sharing test order ID between tests # --------------------------------------------------------------------------- @pytest.fixture(scope="module") def test_order_id(oracle_connection): """ Create a test order via PACK_IMPORT_COMENZI.importa_comanda and yield its ID. Cleans up (DELETE) after all module tests finish. """ import oracledb conn = oracle_connection order_id = None # Find a minimal valid partner ID try: with conn.cursor() as cur: cur.execute( "SELECT MIN(id_partener) FROM parteneri WHERE id_partener > 0" ) row = cur.fetchone() if not row or row[0] is None: pytest.skip("No partners found in Oracle — cannot create test order") partner_id = int(row[0]) except Exception as exc: pytest.skip(f"Cannot query parteneri table: {exc}") # Build minimal JSON articles — use a SKU known from NOM_ARTICOLE if possible with conn.cursor() as cur: cur.execute( "SELECT codmat FROM nom_articole WHERE rownum = 1" ) row = cur.fetchone() test_sku = row[0] if row else "CAFE100" nr_comanda_ext = f"PYTEST-{int(time.time())}" articles = json.dumps([{ "sku": test_sku, "cantitate": 1, "pret": 50.0, "denumire": "Test article (pytest)", "tva": 19, "discount": 0, }]) try: with conn.cursor() as cur: clob_var = cur.var(oracledb.DB_TYPE_CLOB) clob_var.setvalue(0, articles) id_comanda_var = cur.var(oracledb.DB_TYPE_NUMBER) cur.callproc("PACK_IMPORT_COMENZI.importa_comanda", [ nr_comanda_ext, # p_nr_comanda_ext None, # p_data_comanda (NULL = SYSDATE in pkg) partner_id, # p_id_partener clob_var, # p_json_articole None, # p_id_adresa_livrare None, # p_id_adresa_facturare None, # p_id_pol None, # p_id_sectie None, # p_id_gestiune None, # p_kit_mode None, # p_id_pol_productie None, # p_kit_discount_codmat None, # p_kit_discount_id_pol id_comanda_var, # v_id_comanda (OUT) ]) raw = id_comanda_var.getvalue() order_id = int(raw) if raw is not None else None if order_id and order_id > 0: conn.commit() logger.info(f"Test order created: ID={order_id}, NR={nr_comanda_ext}") else: conn.rollback() order_id = None except Exception as exc: try: conn.rollback() except Exception: pass logger.warning(f"Could not create test order: {exc}") order_id = None yield order_id # Cleanup — runs even if tests fail if order_id: try: with conn.cursor() as cur: cur.execute( "DELETE FROM comenzi_articole WHERE id_comanda = :id", {"id": order_id} ) cur.execute( "DELETE FROM com_antet WHERE id_comanda = :id", {"id": order_id} ) conn.commit() logger.info(f"Test order {order_id} cleaned up") except Exception as exc: logger.error(f"Cleanup failed for order {order_id}: {exc}") # --------------------------------------------------------------------------- # Package validity tests # --------------------------------------------------------------------------- def test_pack_import_comenzi_valid(oracle_connection): """PACK_IMPORT_COMENZI package body must be VALID.""" with oracle_connection.cursor() as cur: cur.execute(_STATUS_SQL, {"name": "PACK_IMPORT_COMENZI"}) row = cur.fetchone() assert row is not None, "PACK_IMPORT_COMENZI package body not found in user_objects" assert row[0] == "VALID", f"PACK_IMPORT_COMENZI is {row[0]}" def test_pack_import_parteneri_valid(oracle_connection): """PACK_IMPORT_PARTENERI package body must be VALID.""" with oracle_connection.cursor() as cur: cur.execute(_STATUS_SQL, {"name": "PACK_IMPORT_PARTENERI"}) row = cur.fetchone() assert row is not None, "PACK_IMPORT_PARTENERI package body not found in user_objects" assert row[0] == "VALID", f"PACK_IMPORT_PARTENERI is {row[0]}" def test_pack_comenzi_valid(oracle_connection): """PACK_COMENZI package body must be VALID.""" with oracle_connection.cursor() as cur: cur.execute(_STATUS_SQL, {"name": "PACK_COMENZI"}) row = cur.fetchone() assert row is not None, "PACK_COMENZI package body not found in user_objects" assert row[0] == "VALID", f"PACK_COMENZI is {row[0]}" def test_pack_facturare_valid(oracle_connection): """PACK_FACTURARE package body must be VALID.""" with oracle_connection.cursor() as cur: cur.execute(_STATUS_SQL, {"name": "PACK_FACTURARE"}) row = cur.fetchone() assert row is not None, "PACK_FACTURARE package body not found in user_objects" assert row[0] == "VALID", f"PACK_FACTURARE is {row[0]}" # --------------------------------------------------------------------------- # Order import tests # --------------------------------------------------------------------------- def test_import_order_with_articles(test_order_id): """PACK_IMPORT_COMENZI.importa_comanda must return a valid order ID > 0.""" if test_order_id is None: pytest.skip("Test order creation failed — see test_order_id fixture logs") assert test_order_id > 0, f"importa_comanda returned invalid ID: {test_order_id}" def test_cleanup_test_order(oracle_connection, test_order_id): """Verify the test order rows exist and can be queried (cleanup runs via fixture).""" if test_order_id is None: pytest.skip("No test order to verify") with oracle_connection.cursor() as cur: cur.execute( "SELECT COUNT(*) FROM com_antet WHERE id_comanda = :id", {"id": test_order_id} ) row = cur.fetchone() # At this point the order should still exist (fixture cleanup runs after module) assert row is not None assert row[0] >= 0 # may be 0 if already cleaned, just confirm query works