275 lines
10 KiB
Python
275 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
P0 CRITICAL FIX Verification: Deep Merge for JSONB fields.
|
|
|
|
This test verifies that the deep_merge_dict() function in admin_packages.py
|
|
correctly preserves nested JSONB keys (pricing_zones.HU, lifecycle.available_until)
|
|
when a PATCH request only sends a subset of the data.
|
|
|
|
Test scenarios:
|
|
1. deep_merge_dict unit test — verify the function works correctly in isolation
|
|
2. API PATCH test — send a PATCH with only DEFAULT pricing zone, verify HU/US zones survive
|
|
3. API PATCH test — send a PATCH with lifecycle.is_public, verify available_until survives
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import asyncio
|
|
from copy import deepcopy
|
|
|
|
# Add backend to path
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '../../backend'))
|
|
|
|
# ── Import the deep_merge_dict function directly ──
|
|
from app.api.v1.endpoints.admin_packages import deep_merge_dict
|
|
|
|
|
|
def test_deep_merge_dict_unit():
|
|
"""Unit test: verify deep_merge_dict preserves nested keys."""
|
|
print("=" * 70)
|
|
print("🧪 TEST 1: deep_merge_dict() unit test")
|
|
print("=" * 70)
|
|
|
|
# ── Scenario A: Multi-zone pricing preservation ──
|
|
base = {
|
|
"type": "private",
|
|
"pricing_zones": {
|
|
"HU": {"monthly_price": 990.0, "yearly_price": 9990.0, "currency": "HUF", "credit_price": 25000},
|
|
"US": {"monthly_price": 3.5, "yearly_price": 35.9, "currency": "USD", "credit_price": 25000},
|
|
"DEFAULT": {"monthly_price": 2.99, "yearly_price": 29.9, "currency": "EUR", "credit_price": 25000},
|
|
},
|
|
"lifecycle": {"is_public": True, "available_until": "2026-12-31T23:59:59Z"},
|
|
"allowances": {"max_vehicles": 3, "max_garages": 1, "monthly_free_credits": 50},
|
|
}
|
|
|
|
# Frontend sends ONLY DEFAULT zone update (simulating the bug scenario)
|
|
override = {
|
|
"type": "private",
|
|
"pricing_zones": {
|
|
"DEFAULT": {"monthly_price": 5.99, "yearly_price": 59.9, "currency": "EUR"},
|
|
},
|
|
"lifecycle": {"is_public": True},
|
|
}
|
|
|
|
merged = deep_merge_dict(base, override)
|
|
|
|
# Verify HU zone is PRESERVED
|
|
assert "HU" in merged["pricing_zones"], "❌ FAIL: HU pricing zone was LOST!"
|
|
assert merged["pricing_zones"]["HU"]["monthly_price"] == 990.0, "❌ FAIL: HU price changed!"
|
|
assert merged["pricing_zones"]["HU"]["currency"] == "HUF", "❌ FAIL: HU currency changed!"
|
|
|
|
# Verify US zone is PRESERVED
|
|
assert "US" in merged["pricing_zones"], "❌ FAIL: US pricing zone was LOST!"
|
|
assert merged["pricing_zones"]["US"]["monthly_price"] == 3.5, "❌ FAIL: US price changed!"
|
|
|
|
# Verify DEFAULT zone is UPDATED
|
|
assert merged["pricing_zones"]["DEFAULT"]["monthly_price"] == 5.99, "❌ FAIL: DEFAULT price not updated!"
|
|
assert merged["pricing_zones"]["DEFAULT"]["yearly_price"] == 59.9, "❌ FAIL: DEFAULT yearly price not updated!"
|
|
|
|
# Verify lifecycle.available_until is PRESERVED
|
|
assert merged["lifecycle"]["available_until"] == "2026-12-31T23:59:59Z", \
|
|
"❌ FAIL: lifecycle.available_until was LOST!"
|
|
assert merged["lifecycle"]["is_public"] is True, "❌ FAIL: lifecycle.is_public changed!"
|
|
|
|
# Verify allowances are PRESERVED
|
|
assert merged["allowances"]["max_vehicles"] == 3, "❌ FAIL: allowances.max_vehicles changed!"
|
|
|
|
print(" ✅ PASS: All pricing zones preserved (HU, US, DEFAULT)")
|
|
print(" ✅ PASS: lifecycle.available_until preserved")
|
|
print(" ✅ PASS: Unrelated fields (allowances) preserved")
|
|
print(" ✅ PASS: Updated fields correctly applied (DEFAULT price changed)")
|
|
print()
|
|
|
|
# ── Scenario B: None values are skipped ──
|
|
base_b = {"pricing_zones": {"DEFAULT": {"monthly_price": 10.0}}}
|
|
override_b = {"pricing_zones": {"DEFAULT": {"monthly_price": None}}}
|
|
merged_b = deep_merge_dict(base_b, override_b)
|
|
assert merged_b["pricing_zones"]["DEFAULT"]["monthly_price"] == 10.0, \
|
|
"❌ FAIL: None override should preserve existing value!"
|
|
print(" ✅ PASS: None values correctly preserve existing data")
|
|
print()
|
|
|
|
# ── Scenario C: Empty override doesn't destroy data ──
|
|
base_c = {"pricing_zones": {"HU": {"monthly_price": 990.0}}}
|
|
override_c = {}
|
|
merged_c = deep_merge_dict(base_c, override_c)
|
|
assert merged_c["pricing_zones"]["HU"]["monthly_price"] == 990.0, \
|
|
"❌ FAIL: Empty override should preserve everything!"
|
|
print(" ✅ PASS: Empty override preserves all existing data")
|
|
print()
|
|
|
|
print("🎉 All unit tests PASSED!")
|
|
print()
|
|
return True
|
|
|
|
|
|
async def test_api_patch_preserves_zones():
|
|
"""
|
|
Test the actual API endpoint by sending a PATCH request.
|
|
Uses the admin auth token and tests against private_pro_v1 (id=14).
|
|
"""
|
|
print("=" * 70)
|
|
print("🧪 TEST 2: API PATCH — pricing zone preservation")
|
|
print("=" * 70)
|
|
|
|
import httpx
|
|
|
|
# Get admin token
|
|
BASE_URL = os.environ.get("API_BASE_URL", "http://localhost:8000")
|
|
ADMIN_EMAIL = os.environ.get("ADMIN_EMAIL", "admin@profibot.hu")
|
|
ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD", "Admin123!")
|
|
|
|
async with httpx.AsyncClient(base_url=BASE_URL, timeout=30) as client:
|
|
# ── Login (OAuth2 form) ──
|
|
login_resp = await client.post("/api/v1/auth/login", data={
|
|
"username": ADMIN_EMAIL,
|
|
"password": ADMIN_PASSWORD,
|
|
})
|
|
if login_resp.status_code != 200:
|
|
print(f" ⚠️ Login failed ({login_resp.status_code}), skipping API test.")
|
|
print(f" Response: {login_resp.text[:200]}")
|
|
return False
|
|
|
|
token = login_resp.json().get("access_token")
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
# ── Fetch private_pro_v1 (id=14) ──
|
|
list_resp = await client.get("/api/v1/admin/packages", headers=headers)
|
|
if list_resp.status_code != 200:
|
|
print(f" ⚠️ List failed ({list_resp.status_code}), skipping API test.")
|
|
return False
|
|
|
|
tiers = list_resp.json().get("tiers", [])
|
|
target = next((t for t in tiers if t["name"] == "private_pro_v1"), None)
|
|
if not target:
|
|
print(" ⚠️ private_pro_v1 not found, trying first package with pricing_zones...")
|
|
target = next((t for t in tiers if t.get("rules", {}).get("pricing_zones")), None)
|
|
|
|
if not target:
|
|
print(" ⚠️ No package with pricing_zones found, skipping API test.")
|
|
return False
|
|
|
|
tier_id = target["id"]
|
|
print(f" Target package: {target['name']} (id={tier_id})")
|
|
|
|
# Record zones BEFORE patch
|
|
before_zones = dict(target["rules"].get("pricing_zones", {}))
|
|
before_lifecycle = dict(target["rules"].get("lifecycle", {}))
|
|
print(f" Zones before: {list(before_zones.keys())}")
|
|
print(f" Lifecycle before: {before_lifecycle}")
|
|
|
|
# ── Send PATCH with ONLY DEFAULT zone ──
|
|
patch_payload = {
|
|
"rules": {
|
|
"type": target["rules"]["type"],
|
|
"pricing_zones": {
|
|
"DEFAULT": {
|
|
"monthly_price": 999.99,
|
|
"yearly_price": 9999.99,
|
|
"currency": "EUR",
|
|
}
|
|
},
|
|
"lifecycle": {"is_public": True},
|
|
}
|
|
}
|
|
|
|
patch_resp = await client.patch(
|
|
f"/api/v1/admin/packages/{tier_id}",
|
|
headers=headers,
|
|
json=patch_payload,
|
|
)
|
|
|
|
if patch_resp.status_code != 200:
|
|
print(f" ❌ PATCH failed ({patch_resp.status_code}): {patch_resp.text[:300]}")
|
|
return False
|
|
|
|
updated = patch_resp.json()
|
|
after_zones = updated["rules"].get("pricing_zones", {})
|
|
after_lifecycle = updated["rules"].get("lifecycle", {})
|
|
|
|
print(f" Zones after: {list(after_zones.keys())}")
|
|
print(f" Lifecycle after: {after_lifecycle}")
|
|
|
|
# ── VERIFY: All original zones still exist ──
|
|
all_preserved = True
|
|
for zone_key in before_zones:
|
|
if zone_key not in after_zones:
|
|
print(f" ❌ FAIL: Zone '{zone_key}' was LOST!")
|
|
all_preserved = False
|
|
else:
|
|
print(f" ✅ Zone '{zone_key}' preserved")
|
|
|
|
# ── VERIFY: DEFAULT price was updated ──
|
|
if "DEFAULT" in after_zones:
|
|
if after_zones["DEFAULT"]["monthly_price"] == 999.99:
|
|
print(f" ✅ DEFAULT price updated correctly")
|
|
else:
|
|
print(f" ⚠️ DEFAULT price mismatch: {after_zones['DEFAULT']['monthly_price']}")
|
|
|
|
# ── VERIFY: lifecycle.available_until preserved ──
|
|
if before_lifecycle.get("available_until"):
|
|
if after_lifecycle.get("available_until") == before_lifecycle["available_until"]:
|
|
print(f" ✅ lifecycle.available_until preserved")
|
|
else:
|
|
print(f" ❌ FAIL: lifecycle.available_until was LOST or changed!")
|
|
print(f" Before: {before_lifecycle.get('available_until')}")
|
|
print(f" After: {after_lifecycle.get('available_until')}")
|
|
all_preserved = False
|
|
|
|
if all_preserved:
|
|
print("\n🎉 API test PASSED! Deep merge is working correctly.")
|
|
else:
|
|
print("\n❌ API test FAILED — some data was lost!")
|
|
|
|
# ── Restore original data ──
|
|
restore_payload = {
|
|
"rules": {
|
|
"type": target["rules"]["type"],
|
|
"pricing_zones": before_zones,
|
|
"lifecycle": before_lifecycle,
|
|
}
|
|
}
|
|
await client.patch(
|
|
f"/api/v1/admin/packages/{tier_id}",
|
|
headers=headers,
|
|
json=restore_payload,
|
|
)
|
|
print(" 🔄 Original data restored")
|
|
|
|
return all_preserved
|
|
|
|
|
|
async def main():
|
|
print("\n" + "=" * 70)
|
|
print("🔬 P0 CRITICAL FIX VERIFICATION: Deep Merge for JSONB")
|
|
print("=" * 70)
|
|
print()
|
|
|
|
# Test 1: Unit test
|
|
unit_ok = test_deep_merge_dict_unit()
|
|
|
|
# Test 2: API test
|
|
print()
|
|
api_ok = await test_api_patch_preserves_zones()
|
|
|
|
# Summary
|
|
print()
|
|
print("=" * 70)
|
|
print("📊 SUMMARY")
|
|
print("=" * 70)
|
|
print(f" Unit test: {'✅ PASS' if unit_ok else '❌ FAIL'}")
|
|
print(f" API test: {'✅ PASS' if api_ok else '❌ FAIL (or skipped)'}")
|
|
print()
|
|
|
|
if unit_ok:
|
|
print("✅ The deep_merge_dict() function is working correctly.")
|
|
print(" Multi-zone pricing data loss vulnerability is CLOSED.")
|
|
else:
|
|
print("❌ Fix verification FAILED!")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|