Marius Mutu
52454a5925
- Complete manual testing of all Oracle PL/SQL packages - Document 75% success rate (3/4 components passing) - Move all test scripts from api/ to api/tests/ subdirectory - Update P1-004 story with comprehensive test results - Identify external dependency blocking full order import - Mark Phase 1 as 95% complete, ready for Phase 2 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
301 lines
12 KiB
Python
301 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script pentru package-urile Oracle IMPORT_PARTENERI și IMPORT_COMENZI
|
|
Rulează testele manuale conform P1-004
|
|
"""
|
|
|
|
import oracledb
|
|
import os
|
|
import json
|
|
from datetime import datetime
|
|
from dotenv import load_dotenv
|
|
|
|
# Load environment
|
|
load_dotenv('api/.env')
|
|
|
|
# Oracle configuration
|
|
user = os.environ['ORACLE_USER']
|
|
password = os.environ['ORACLE_PASSWORD']
|
|
dsn = os.environ['ORACLE_DSN']
|
|
|
|
# Initialize Oracle client (thick mode)
|
|
try:
|
|
instantclient_path = os.environ.get('INSTANTCLIENTPATH', '/opt/oracle/instantclient_23_9')
|
|
oracledb.init_oracle_client(lib_dir=instantclient_path)
|
|
print(f"✅ Oracle thick mode initialized: {instantclient_path}")
|
|
except Exception as e:
|
|
print(f"⚠️ Oracle thick mode failed, using thin mode: {e}")
|
|
|
|
def test_connection():
|
|
"""Test basic Oracle connection"""
|
|
print("\n🔍 Testing Oracle Connection...")
|
|
try:
|
|
with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute("SELECT SYSDATE FROM DUAL")
|
|
db_time = cur.fetchone()[0]
|
|
print(f"✅ Connected to Oracle: {db_time}")
|
|
return True
|
|
except Exception as e:
|
|
print(f"❌ Connection failed: {e}")
|
|
return False
|
|
|
|
def test_import_parteneri():
|
|
"""Test IMPORT_PARTENERI package functions"""
|
|
print("\n🧪 Testing IMPORT_PARTENERI Package...")
|
|
|
|
try:
|
|
with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
|
|
with conn.cursor() as cur:
|
|
|
|
# Test 1: Creare partener nou (persoană fizică)
|
|
print("\n📝 Test 1: Creare partener nou (persoană fizică)")
|
|
cur.execute("""
|
|
SELECT PACK_IMPORT_PARTENERI.cauta_sau_creeaza_partener(
|
|
NULL,
|
|
'Ion Popescu Test',
|
|
'JUD:Bucuresti;BUCURESTI;Str. Testului;10',
|
|
'0721234567',
|
|
'ion.test@email.com'
|
|
) FROM DUAL
|
|
""")
|
|
partner_id1 = cur.fetchone()[0]
|
|
print(f"✅ Partener nou creat: ID = {partner_id1}")
|
|
|
|
# Test 2: Căutare partener existent după denumire
|
|
print("\n🔍 Test 2: Căutare partener existent după denumire")
|
|
cur.execute("""
|
|
SELECT PACK_IMPORT_PARTENERI.cauta_sau_creeaza_partener(
|
|
NULL,
|
|
'Ion Popescu Test',
|
|
'JUD:Bucuresti;BUCURESTI;Str. Testului;10',
|
|
'0721234567',
|
|
'ion.test@email.com'
|
|
) FROM DUAL
|
|
""")
|
|
partner_id2 = cur.fetchone()[0]
|
|
print(f"✅ Partener existent găsit: ID = {partner_id2}")
|
|
|
|
if partner_id1 == partner_id2:
|
|
print("✅ PASS: Același ID returnat pentru același partener")
|
|
else:
|
|
print("❌ FAIL: IDs diferite pentru același partener")
|
|
|
|
# Test 3: Creare partener companie cu CUI
|
|
print("\n🏢 Test 3: Creare partener companie cu CUI")
|
|
cur.execute("""
|
|
SELECT PACK_IMPORT_PARTENERI.cauta_sau_creeaza_partener(
|
|
'RO12345678',
|
|
'Test Company SRL',
|
|
'JUD:Cluj;CLUJ-NAPOCA;Str. Companiei;25',
|
|
'0264123456',
|
|
'office@testcompany.ro'
|
|
) FROM DUAL
|
|
""")
|
|
partner_id3 = cur.fetchone()[0]
|
|
print(f"✅ Companie nouă creată: ID = {partner_id3}")
|
|
|
|
# Test 4: Căutare companie după CUI
|
|
print("\n🔍 Test 4: Căutare companie după CUI")
|
|
cur.execute("""
|
|
SELECT PACK_IMPORT_PARTENERI.cauta_sau_creeaza_partener(
|
|
'RO12345678',
|
|
'Test Company SRL Modified',
|
|
'JUD:Bucuresti;BUCURESTI;Str. Alta;1',
|
|
'0722999888',
|
|
'new@testcompany.ro'
|
|
) FROM DUAL
|
|
""")
|
|
partner_id4 = cur.fetchone()[0]
|
|
print(f"✅ Companie existentă găsită: ID = {partner_id4}")
|
|
|
|
if partner_id3 == partner_id4:
|
|
print("✅ PASS: Căutare după CUI funcționează corect")
|
|
else:
|
|
print("❌ FAIL: CUI-ul nu a fost găsit corect")
|
|
|
|
conn.commit()
|
|
|
|
except Exception as e:
|
|
print(f"❌ IMPORT_PARTENERI test failed: {e}")
|
|
return False
|
|
|
|
return True
|
|
|
|
def test_import_comenzi():
|
|
"""Test IMPORT_COMENZI package functions"""
|
|
print("\n🧪 Testing IMPORT_COMENZI Package...")
|
|
|
|
try:
|
|
with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
|
|
with conn.cursor() as cur:
|
|
|
|
# Test 1: gaseste_articol_roa - SKU simplu
|
|
print("\n📝 Test 1: gaseste_articol_roa - SKU simplu (nu există în ARTICOLE_TERTI)")
|
|
cur.execute("""
|
|
SELECT * FROM TABLE(PACK_IMPORT_COMENZI.gaseste_articol_roa('SIMPLE_SKU', 5, 100.0))
|
|
""")
|
|
results = cur.fetchall()
|
|
if results:
|
|
print(f"✅ SKU simplu găsit: {results}")
|
|
else:
|
|
print("⚠️ SKU simplu nu a fost găsit (normal dacă nu există în nom_articole)")
|
|
|
|
# Test 2: gaseste_articol_roa - Reîmpachetare CAFE100
|
|
print("\n☕ Test 2: gaseste_articol_roa - Reîmpachetare CAFE100")
|
|
cur.execute("""
|
|
SELECT * FROM TABLE(PACK_IMPORT_COMENZI.gaseste_articol_roa('CAFE100', 2, 50.0))
|
|
""")
|
|
results = cur.fetchall()
|
|
if results:
|
|
for row in results:
|
|
print(f"✅ CAFE100 mapare: CODMAT={row[0]}, Cant={row[1]}, Pret={row[2]}")
|
|
else:
|
|
print("❌ CAFE100 nu a fost găsit")
|
|
|
|
# Test 3: gaseste_articol_roa - Set compus SET01
|
|
print("\n🎁 Test 3: gaseste_articol_roa - Set compus SET01")
|
|
cur.execute("""
|
|
SELECT * FROM TABLE(PACK_IMPORT_COMENZI.gaseste_articol_roa('SET01', 1, 200.0))
|
|
""")
|
|
results = cur.fetchall()
|
|
if results:
|
|
total_percent = 0
|
|
for row in results:
|
|
print(f"✅ SET01 component: CODMAT={row[0]}, Cant={row[1]}, Pret={row[2]}")
|
|
# Calculate percentage based on price
|
|
total_percent += (row[2] / 200.0 * 100)
|
|
print(f"Total percentage: {total_percent}%")
|
|
else:
|
|
print("❌ SET01 nu a fost găsit")
|
|
|
|
except Exception as e:
|
|
print(f"❌ IMPORT_COMENZI gaseste_articol_roa test failed: {e}")
|
|
return False
|
|
|
|
return True
|
|
|
|
def test_full_order_import():
|
|
"""Test complete order import workflow"""
|
|
print("\n🧪 Testing Complete Order Import...")
|
|
|
|
try:
|
|
with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
|
|
with conn.cursor() as cur:
|
|
|
|
# First create a partner for the order
|
|
print("\n👤 Creating test partner for order...")
|
|
cur.execute("""
|
|
SELECT PACK_IMPORT_PARTENERI.cauta_sau_creeaza_partener(
|
|
NULL,
|
|
'Maria Testescu',
|
|
'JUD:Bucuresti;BUCURESTI;Calea Victoriei;100',
|
|
'0723456789',
|
|
'maria.test@email.com'
|
|
) FROM DUAL
|
|
""")
|
|
partner_id = cur.fetchone()[0]
|
|
print(f"✅ Partner created/found: ID = {partner_id}")
|
|
|
|
# Test complete order with multiple article types
|
|
print("\n📦 Test: Complete order import with mixed articles")
|
|
|
|
# JSON with mixed article types: simple, repack, and set
|
|
articles_json = json.dumps([
|
|
{"sku": "CAFE100", "cantitate": 2, "pret": 45.0}, # Repack: 2x10=20 units CAF01
|
|
{"sku": "SET01", "cantitate": 1, "pret": 150.0} # Set: 2x CAF01 + 1x FILTRU01
|
|
])
|
|
|
|
print(f"Articles JSON: {articles_json}")
|
|
|
|
order_number = 'TEST-ORDER-' + datetime.now().strftime('%Y%m%d-%H%M%S')
|
|
cur.execute("""
|
|
SELECT PACK_IMPORT_COMENZI.importa_comanda(
|
|
:order_num,
|
|
SYSDATE,
|
|
:partner_id,
|
|
:articles_json,
|
|
NULL,
|
|
NULL,
|
|
'Test order - P1-004 validation'
|
|
) FROM DUAL
|
|
""", {
|
|
'order_num': order_number,
|
|
'partner_id': partner_id,
|
|
'articles_json': articles_json
|
|
})
|
|
|
|
order_id = cur.fetchone()[0]
|
|
|
|
if order_id and order_id > 0:
|
|
print(f"✅ Order imported successfully: ID = {order_id}")
|
|
|
|
# Verify order details
|
|
cur.execute("""
|
|
SELECT c.id_comanda, c.numar, c.data_comanda, p.denumire
|
|
FROM comenzi c
|
|
JOIN parteneri p ON c.id_partener = p.id_partener
|
|
WHERE c.id_comanda = ?
|
|
""", [order_id])
|
|
|
|
order_info = cur.fetchone()
|
|
if order_info:
|
|
print(f"Order details: ID={order_info[0]}, Numar={order_info[1]}, Data={order_info[2]}, Partner={order_info[3]}")
|
|
|
|
# Verify order items
|
|
cur.execute("""
|
|
SELECT a.codmat, dc.cantitate, dc.pret_unitar
|
|
FROM det_comenzi dc
|
|
JOIN articole a ON dc.id_articol = a.id_articol
|
|
WHERE dc.id_comanda = ?
|
|
ORDER BY dc.id_det_comanda
|
|
""", [order_id])
|
|
|
|
items = cur.fetchall()
|
|
print(f"\nOrder items ({len(items)} total):")
|
|
for item in items:
|
|
print(f" - CODMAT: {item[0]}, Quantity: {item[1]}, Price: {item[2]}")
|
|
|
|
else:
|
|
print(f"❌ Order import failed: returned ID = {order_id}")
|
|
return False
|
|
|
|
conn.commit()
|
|
|
|
except Exception as e:
|
|
print(f"❌ Full order import test failed: {e}")
|
|
return False
|
|
|
|
return True
|
|
|
|
def main():
|
|
"""Run all tests"""
|
|
print("🚀 Starting P1-004: Manual Package Testing")
|
|
print("=" * 60)
|
|
|
|
# Test connection first
|
|
if not test_connection():
|
|
return False
|
|
|
|
# Test IMPORT_PARTENERI
|
|
if not test_import_parteneri():
|
|
return False
|
|
|
|
# Test IMPORT_COMENZI
|
|
if not test_import_comenzi():
|
|
return False
|
|
|
|
# Test full workflow
|
|
if not test_full_order_import():
|
|
return False
|
|
|
|
print("\n" + "=" * 60)
|
|
print("🎉 All P1-004 tests completed successfully!")
|
|
print("✅ IMPORT_PARTENERI package: PASS")
|
|
print("✅ IMPORT_COMENZI package: PASS")
|
|
print("✅ End-to-end workflow: PASS")
|
|
return True
|
|
|
|
if __name__ == "__main__":
|
|
success = main()
|
|
exit(0 if success else 1) |