61ae58ef25
Kit discount: v_disc_amt is per-kit, not per-unit — remove division by v_cantitate_web so discount lines compute correctly (e.g. -2 x 5 = -10). Price sync: stop auto-inserting missing articles into price policies (was inserting with wrong proc_tvav from GoMag). Log warning instead. Kit detection: extend to single-component repackagings (cantitate_roa > 1) in both PL/SQL package and price sync/validation services. Add repackaging kit pricing test for separate_line and distributed modes. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
551 lines
23 KiB
Python
551 lines
23 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Complete end-to-end test for order import functionality
|
|
Tests: Partner creation, Article mapping, Order import with full workflow
|
|
"""
|
|
|
|
import oracledb
|
|
import os
|
|
from dotenv import load_dotenv
|
|
import random
|
|
from datetime import datetime
|
|
|
|
load_dotenv('.env')
|
|
|
|
user = os.environ['ORACLE_USER']
|
|
password = os.environ['ORACLE_PASSWORD']
|
|
dsn = os.environ['ORACLE_DSN']
|
|
|
|
try:
|
|
instantclient_path = os.environ.get('INSTANTCLIENTPATH', '/opt/oracle/instantclient_23_9')
|
|
oracledb.init_oracle_client(lib_dir=instantclient_path)
|
|
except Exception as e:
|
|
pass
|
|
|
|
def setup_test_data(cur):
|
|
"""Setup test data by running SQL script"""
|
|
print("🔧 Setting up test data...")
|
|
|
|
# Read and execute setup script
|
|
with open('/app/tests/setup_test_data.sql', 'r') as f:
|
|
setup_sql = f.read()
|
|
|
|
# Split by statements and execute each
|
|
statements = [stmt.strip() for stmt in setup_sql.split(';') if stmt.strip() and not stmt.strip().startswith('--')]
|
|
|
|
for stmt in statements:
|
|
if stmt.upper().startswith(('INSERT', 'DELETE', 'COMMIT')):
|
|
try:
|
|
cur.execute(stmt)
|
|
if stmt.upper().startswith('COMMIT'):
|
|
print(" ✅ Test data setup committed")
|
|
except Exception as e:
|
|
if "unique constraint" not in str(e).lower():
|
|
print(f" ⚠️ Setup warning: {e}")
|
|
|
|
def teardown_test_data(cur):
|
|
"""Cleanup test data by running teardown script"""
|
|
print("🧹 Cleaning up test data...")
|
|
|
|
try:
|
|
# Read and execute teardown script
|
|
with open('/app/tests/teardown_test_data.sql', 'r') as f:
|
|
teardown_sql = f.read()
|
|
|
|
# Split by statements and execute each
|
|
statements = [stmt.strip() for stmt in teardown_sql.split(';') if stmt.strip() and not stmt.strip().startswith('--')]
|
|
|
|
for stmt in statements:
|
|
if stmt.upper().startswith(('DELETE', 'COMMIT')):
|
|
try:
|
|
cur.execute(stmt)
|
|
if stmt.upper().startswith('COMMIT'):
|
|
print(" ✅ Test data cleanup committed")
|
|
except Exception as e:
|
|
print(f" ⚠️ Cleanup warning: {e}")
|
|
|
|
except Exception as e:
|
|
print(f" ❌ Teardown error: {e}")
|
|
|
|
def test_complete_import():
|
|
"""
|
|
Complete test of order import workflow:
|
|
1. Setup test data
|
|
2. Test individual components
|
|
3. Create partner if doesn't exist
|
|
4. Import complete order with articles
|
|
5. Verify results
|
|
6. Cleanup test data
|
|
"""
|
|
print("🎯 COMPLETE ORDER IMPORT TEST")
|
|
print("=" * 60)
|
|
|
|
success_count = 0
|
|
total_tests = 0
|
|
|
|
try:
|
|
with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
|
|
with conn.cursor() as cur:
|
|
unique_suffix = random.randint(1000, 9999)
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
|
# ========================================
|
|
# SETUP: Initialize test data
|
|
# ========================================
|
|
setup_test_data(cur)
|
|
|
|
# ========================================
|
|
# TEST 1: Component Validation
|
|
# ========================================
|
|
print("\n📋 TEST 1: Individual Component Validation")
|
|
print("-" * 40)
|
|
|
|
# Test article mapping
|
|
total_tests += 1
|
|
print("1.1 Testing article mapping...")
|
|
cur.execute("SELECT * FROM TABLE(PACK_IMPORT_COMENZI.gaseste_articol_roa('CAFE100'))")
|
|
article_results = cur.fetchall()
|
|
if len(article_results) > 0:
|
|
print(f" ✅ Article mapping: Found {len(article_results)} mappings for CAFE100")
|
|
success_count += 1
|
|
else:
|
|
print(" ❌ Article mapping: No results for CAFE100")
|
|
|
|
# Test JSON parsing
|
|
total_tests += 1
|
|
print("1.2 Testing JSON parsing...")
|
|
test_json = '[{"sku": "CAFE100", "cantitate": 1, "pret": 25.0}]'
|
|
cur.execute("SELECT * FROM TABLE(PACK_JSON.parse_array(:json))", {'json': test_json})
|
|
json_results = cur.fetchall()
|
|
if len(json_results) > 0:
|
|
print(f" ✅ JSON parsing: Successfully parsed {len(json_results)} items")
|
|
success_count += 1
|
|
else:
|
|
print(" ❌ JSON parsing: Failed to parse JSON")
|
|
|
|
# ========================================
|
|
# TEST 2: Partner Management
|
|
# ========================================
|
|
print("\n👥 TEST 2: Partner Creation/Retrieval")
|
|
print("-" * 40)
|
|
|
|
total_tests += 1
|
|
partner_name = f'Test Client {timestamp}-{unique_suffix}'
|
|
partner_address = 'JUD:Bucuresti;BUCURESTI;Str. Test;12'
|
|
partner_phone = f'072{unique_suffix:04d}000'
|
|
partner_email = f'test{unique_suffix}@example.com'
|
|
|
|
print(f"2.1 Creating/finding partner: {partner_name}")
|
|
|
|
partner_var = cur.var(oracledb.NUMBER)
|
|
cur.execute("""
|
|
DECLARE
|
|
v_partner_id NUMBER;
|
|
BEGIN
|
|
v_partner_id := PACK_IMPORT_PARTENERI.cauta_sau_creeaza_partener(
|
|
NULL, -- cod_fiscal (NULL for individuals)
|
|
:partner_name,
|
|
:partner_address,
|
|
:partner_phone,
|
|
:partner_email
|
|
);
|
|
:result := v_partner_id;
|
|
END;
|
|
""", {
|
|
'partner_name': partner_name,
|
|
'partner_address': partner_address,
|
|
'partner_phone': partner_phone,
|
|
'partner_email': partner_email,
|
|
'result': partner_var
|
|
})
|
|
|
|
partner_id = partner_var.getvalue()
|
|
if partner_id and partner_id > 0:
|
|
print(f" ✅ Partner management: ID {partner_id}")
|
|
success_count += 1
|
|
else:
|
|
print(" ❌ Partner management: Failed to create/find partner")
|
|
return False
|
|
|
|
# ========================================
|
|
# TEST 3: Complete Order Import
|
|
# ========================================
|
|
print("\n📦 TEST 3: Complete Order Import")
|
|
print("-" * 40)
|
|
|
|
total_tests += 1
|
|
order_number = f'COMPLETE-{timestamp}-{unique_suffix}'
|
|
|
|
# Test with multiple articles including real GoMag SKU
|
|
test_articles = [
|
|
{"sku": "CAFE100", "cantitate": 2, "pret": 25.0}, # Mapped article
|
|
{"sku": "8000070028685", "cantitate": 1, "pret": 69.79} # Real GoMag SKU
|
|
]
|
|
articles_json = str(test_articles).replace("'", '"')
|
|
|
|
print(f"3.1 Importing order: {order_number}")
|
|
print(f" Articles: {articles_json}")
|
|
|
|
result_var = cur.var(oracledb.NUMBER)
|
|
cur.execute("""
|
|
DECLARE
|
|
v_order_id NUMBER;
|
|
BEGIN
|
|
v_order_id := PACK_IMPORT_COMENZI.importa_comanda(
|
|
:order_number,
|
|
SYSDATE,
|
|
:partner_id,
|
|
:articles_json,
|
|
NULL, -- id_adresa_livrare
|
|
NULL, -- id_adresa_facturare
|
|
'Complete end-to-end test order'
|
|
);
|
|
:result := v_order_id;
|
|
END;
|
|
""", {
|
|
'order_number': order_number,
|
|
'partner_id': partner_id,
|
|
'articles_json': articles_json,
|
|
'result': result_var
|
|
})
|
|
|
|
order_id = result_var.getvalue()
|
|
|
|
# Get detailed error information
|
|
cur.execute("SELECT PACK_IMPORT_COMENZI.get_last_error FROM DUAL")
|
|
error_msg = cur.fetchone()[0]
|
|
|
|
if order_id and order_id > 0:
|
|
print(f" ✅ Order import: SUCCESS! ID {order_id}")
|
|
success_count += 1
|
|
|
|
# ========================================
|
|
# TEST 4: Result Verification
|
|
# ========================================
|
|
print("\n🔍 TEST 4: Result Verification")
|
|
print("-" * 40)
|
|
|
|
total_tests += 1
|
|
# Verify order details
|
|
cur.execute("""
|
|
SELECT
|
|
c.NR_COMANDA,
|
|
c.DATA_COMANDA,
|
|
c.INTERNA,
|
|
c.ID_PART,
|
|
c.ID_GESTIUNE,
|
|
c.ID_SECTIE,
|
|
np.DENUMIRE as PARTNER_NAME
|
|
FROM COMENZI c
|
|
LEFT JOIN NOM_PARTENERI np ON c.ID_PART = np.ID_PART
|
|
WHERE c.ID_COMANDA = :order_id
|
|
""", {'order_id': order_id})
|
|
|
|
order_details = cur.fetchone()
|
|
if order_details:
|
|
print(f"4.1 Order verification:")
|
|
print(f" Number: {order_details[0]}")
|
|
print(f" Date: {order_details[1]}")
|
|
print(f" Type (INTERNA): {order_details[2]}")
|
|
print(f" Partner: {order_details[6]} (ID: {order_details[3]})")
|
|
print(f" Gestiune: {order_details[4]}")
|
|
print(f" Sectie: {order_details[5]}")
|
|
|
|
# Verify articles in order
|
|
cur.execute("""
|
|
SELECT
|
|
ce.CANTITATE,
|
|
ce.PRET,
|
|
na.CODMAT,
|
|
na.DENUMIRE
|
|
FROM COMENZI_ELEMENTE ce
|
|
JOIN NOM_ARTICOLE na ON ce.ID_ARTICOL = na.ID_ARTICOL
|
|
WHERE ce.ID_COMANDA = :order_id
|
|
ORDER BY na.CODMAT
|
|
""", {'order_id': order_id})
|
|
|
|
order_articles = cur.fetchall()
|
|
if order_articles:
|
|
print(f"4.2 Articles in order ({len(order_articles)} items):")
|
|
for art in order_articles:
|
|
print(f" - Qty: {art[0]:>3}, Price: {art[1]:>8.2f}, Code: {art[2]:>10} - {art[3]}")
|
|
success_count += 1
|
|
|
|
# Calculate totals
|
|
total_qty = sum(art[0] for art in order_articles)
|
|
total_value = sum(art[0] * art[1] for art in order_articles)
|
|
print(f" TOTAL: Qty={total_qty}, Value={total_value:.2f} RON")
|
|
|
|
else:
|
|
print(" ❌ No articles found in order")
|
|
else:
|
|
print(" ❌ Order verification failed")
|
|
|
|
else:
|
|
print(f" ❌ Order import: FAILED")
|
|
if error_msg:
|
|
print(f" Error: {error_msg}")
|
|
else:
|
|
print(f" No specific error message, ID returned: {order_id}")
|
|
|
|
conn.commit()
|
|
|
|
# ========================================
|
|
# FINAL RESULTS
|
|
# ========================================
|
|
print("\n" + "=" * 60)
|
|
print(f"📊 FINAL RESULTS: {success_count}/{total_tests} tests passed")
|
|
print("=" * 60)
|
|
|
|
# ========================================
|
|
# TEARDOWN: Cleanup test data
|
|
# ========================================
|
|
teardown_test_data(cur)
|
|
conn.commit()
|
|
|
|
if success_count == total_tests:
|
|
print("🎉 ALL TESTS PASSED! Order import system is fully functional.")
|
|
return True
|
|
elif success_count >= total_tests - 1:
|
|
print("⚠️ MOSTLY SUCCESSFUL: Core components working, minor issues remain.")
|
|
return True
|
|
else:
|
|
print("❌ SIGNIFICANT ISSUES: Multiple components need attention.")
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"❌ CRITICAL ERROR: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
# Attempt cleanup even on error
|
|
try:
|
|
with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
|
|
with conn.cursor() as cur:
|
|
print("\n🧹 Attempting cleanup after error...")
|
|
teardown_test_data(cur)
|
|
conn.commit()
|
|
except:
|
|
print(" ⚠️ Cleanup after error also failed")
|
|
|
|
return False
|
|
|
|
def test_repackaging_kit_pricing():
|
|
"""
|
|
Test single-component repackaging with kit pricing.
|
|
CAFE100 -> CAF01 with cantitate_roa=10 (1 web package = 10 ROA units).
|
|
Verifies that kit pricing applies: list price per unit + discount line.
|
|
"""
|
|
print("\n" + "=" * 60)
|
|
print("🎯 REPACKAGING KIT PRICING TEST")
|
|
print("=" * 60)
|
|
|
|
success_count = 0
|
|
total_tests = 0
|
|
|
|
try:
|
|
with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
|
|
with conn.cursor() as cur:
|
|
unique_suffix = random.randint(1000, 9999)
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
|
|
setup_test_data(cur)
|
|
|
|
# Create a test partner
|
|
partner_var = cur.var(oracledb.NUMBER)
|
|
partner_name = f'Test Repack {timestamp}-{unique_suffix}'
|
|
cur.execute("""
|
|
DECLARE v_id NUMBER;
|
|
BEGIN
|
|
v_id := PACK_IMPORT_PARTENERI.cauta_sau_creeaza_partener(
|
|
NULL, :name, 'JUD:Bucuresti;BUCURESTI;Str Test;1',
|
|
'0720000000', 'repack@test.com');
|
|
:result := v_id;
|
|
END;
|
|
""", {'name': partner_name, 'result': partner_var})
|
|
partner_id = partner_var.getvalue()
|
|
if not partner_id or partner_id <= 0:
|
|
print(" SKIP: Could not create test partner")
|
|
return False
|
|
|
|
# ---- Test separate_line mode ----
|
|
total_tests += 1
|
|
order_number = f'TEST-REPACK-SEP-{timestamp}-{unique_suffix}'
|
|
# Web price: 2 packages * 10 units * some_price = total
|
|
# With list price 51.50/unit, 2 packs of 10 = 20 units
|
|
# Web price per package = 450 lei => total web = 900
|
|
# Expected: 20 units @ 51.50 = 1030, discount = 130
|
|
web_price_per_pack = 450.0
|
|
articles_json = f'[{{"sku": "CAFE100", "cantitate": 2, "pret": {web_price_per_pack}}}]'
|
|
|
|
print(f"\n1. Testing separate_line mode: {order_number}")
|
|
print(f" CAFE100 x2 @ {web_price_per_pack} lei/pack, cantitate_roa=10")
|
|
|
|
result_var = cur.var(oracledb.NUMBER)
|
|
cur.execute("""
|
|
DECLARE v_id NUMBER;
|
|
BEGIN
|
|
PACK_IMPORT_COMENZI.importa_comanda(
|
|
:order_number, SYSDATE, :partner_id,
|
|
:articles_json,
|
|
NULL, NULL,
|
|
1, -- id_pol (default price policy)
|
|
NULL, NULL,
|
|
'separate_line', -- kit_mode
|
|
NULL, NULL, NULL,
|
|
v_id);
|
|
:result := v_id;
|
|
END;
|
|
""", {
|
|
'order_number': order_number,
|
|
'partner_id': partner_id,
|
|
'articles_json': articles_json,
|
|
'result': result_var
|
|
})
|
|
|
|
order_id = result_var.getvalue()
|
|
if order_id and order_id > 0:
|
|
print(f" Order created: ID {order_id}")
|
|
|
|
cur.execute("""
|
|
SELECT ce.CANTITATE, ce.PRET, na.CODMAT, na.DENUMIRE
|
|
FROM COMENZI_ELEMENTE ce
|
|
JOIN NOM_ARTICOLE na ON ce.ID_ARTICOL = na.ID_ARTICOL
|
|
WHERE ce.ID_COMANDA = :oid
|
|
ORDER BY ce.CANTITATE DESC
|
|
""", {'oid': order_id})
|
|
rows = cur.fetchall()
|
|
|
|
if len(rows) >= 2:
|
|
# Should have article line + discount line
|
|
art_line = [r for r in rows if r[0] > 0]
|
|
disc_line = [r for r in rows if r[0] < 0]
|
|
|
|
if art_line and disc_line:
|
|
print(f" Article: qty={art_line[0][0]}, price={art_line[0][1]:.2f} ({art_line[0][2]})")
|
|
print(f" Discount: qty={disc_line[0][0]}, price={disc_line[0][1]:.2f}")
|
|
total = sum(r[0] * r[1] for r in rows)
|
|
expected_total = web_price_per_pack * 2
|
|
print(f" Total: {total:.2f} (expected: {expected_total:.2f})")
|
|
if abs(total - expected_total) < 0.02:
|
|
print(" PASS: Total matches web price")
|
|
success_count += 1
|
|
else:
|
|
print(" FAIL: Total mismatch")
|
|
else:
|
|
print(f" FAIL: Expected article + discount lines, got {len(art_line)} art / {len(disc_line)} disc")
|
|
elif len(rows) == 1:
|
|
print(f" FAIL: Only 1 line (no discount). qty={rows[0][0]}, price={rows[0][1]:.2f}")
|
|
print(" Kit pricing did NOT activate for single-component repackaging")
|
|
else:
|
|
print(" FAIL: No order lines found")
|
|
else:
|
|
cur.execute("SELECT PACK_IMPORT_COMENZI.get_last_error FROM DUAL")
|
|
err = cur.fetchone()[0]
|
|
print(f" FAIL: Order import failed: {err}")
|
|
|
|
conn.commit()
|
|
|
|
# ---- Test distributed mode ----
|
|
total_tests += 1
|
|
order_number2 = f'TEST-REPACK-DIST-{timestamp}-{unique_suffix}'
|
|
print(f"\n2. Testing distributed mode: {order_number2}")
|
|
|
|
result_var2 = cur.var(oracledb.NUMBER)
|
|
cur.execute("""
|
|
DECLARE v_id NUMBER;
|
|
BEGIN
|
|
PACK_IMPORT_COMENZI.importa_comanda(
|
|
:order_number, SYSDATE, :partner_id,
|
|
:articles_json,
|
|
NULL, NULL,
|
|
1, NULL, NULL,
|
|
'distributed',
|
|
NULL, NULL, NULL,
|
|
v_id);
|
|
:result := v_id;
|
|
END;
|
|
""", {
|
|
'order_number': order_number2,
|
|
'partner_id': partner_id,
|
|
'articles_json': articles_json,
|
|
'result': result_var2
|
|
})
|
|
|
|
order_id2 = result_var2.getvalue()
|
|
if order_id2 and order_id2 > 0:
|
|
print(f" Order created: ID {order_id2}")
|
|
|
|
cur.execute("""
|
|
SELECT ce.CANTITATE, ce.PRET, na.CODMAT
|
|
FROM COMENZI_ELEMENTE ce
|
|
JOIN NOM_ARTICOLE na ON ce.ID_ARTICOL = na.ID_ARTICOL
|
|
WHERE ce.ID_COMANDA = :oid
|
|
""", {'oid': order_id2})
|
|
rows2 = cur.fetchall()
|
|
|
|
if len(rows2) == 1:
|
|
# Distributed: single line with adjusted price
|
|
total = rows2[0][0] * rows2[0][1]
|
|
expected_total = web_price_per_pack * 2
|
|
print(f" Line: qty={rows2[0][0]}, price={rows2[0][1]:.2f}, total={total:.2f}")
|
|
if abs(total - expected_total) < 0.02:
|
|
print(" PASS: Distributed price correct")
|
|
success_count += 1
|
|
else:
|
|
print(f" FAIL: Total {total:.2f} != expected {expected_total:.2f}")
|
|
else:
|
|
print(f" INFO: Got {len(rows2)} lines (expected 1 for distributed)")
|
|
for r in rows2:
|
|
print(f" qty={r[0]}, price={r[1]:.2f}, codmat={r[2]}")
|
|
else:
|
|
cur.execute("SELECT PACK_IMPORT_COMENZI.get_last_error FROM DUAL")
|
|
err = cur.fetchone()[0]
|
|
print(f" FAIL: Order import failed: {err}")
|
|
|
|
conn.commit()
|
|
|
|
# Cleanup
|
|
teardown_test_data(cur)
|
|
conn.commit()
|
|
|
|
print(f"\n{'=' * 60}")
|
|
print(f"RESULTS: {success_count}/{total_tests} tests passed")
|
|
print('=' * 60)
|
|
return success_count == total_tests
|
|
|
|
except Exception as e:
|
|
print(f"CRITICAL ERROR: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
try:
|
|
with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
|
|
with conn.cursor() as cur:
|
|
teardown_test_data(cur)
|
|
conn.commit()
|
|
except:
|
|
pass
|
|
return False
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("Starting complete order import test...")
|
|
print(f"Timestamp: {datetime.now()}")
|
|
|
|
success = test_complete_import()
|
|
|
|
print(f"\nTest completed at: {datetime.now()}")
|
|
if success:
|
|
print("🎯 PHASE 1 VALIDATION: SUCCESSFUL")
|
|
else:
|
|
print("🔧 PHASE 1 VALIDATION: NEEDS ATTENTION")
|
|
|
|
# Run repackaging kit pricing test
|
|
print("\n")
|
|
repack_success = test_repackaging_kit_pricing()
|
|
if repack_success:
|
|
print("🎯 REPACKAGING KIT PRICING: SUCCESSFUL")
|
|
else:
|
|
print("🔧 REPACKAGING KIT PRICING: NEEDS ATTENTION")
|
|
|
|
exit(0 if success else 1) |