#!/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())