#!/usr/bin/env python3 """ P0 DATA STRUCTURAL AUDIT: _extract_limit() Standalone Test Script This script simulates the exact _extract_limit() function and feeds it the ACTUAL JSON data from the database to verify if the limit extraction works correctly for corporate tiers. Author: System Architect Date: 2026-06-28 """ import json import sys from typing import Any # ============================================================================= # EXACT copy of _extract_limit() from admin_organizations.py:397 # ============================================================================= def _extract_limit( rules: dict, keys: list[str], default: int = 0, ) -> int: """ P0 ARCHITECTURE UNIFICATION: Egyetlen, robusztus függvény a limit értékek kinyerésére a SubscriptionTier.rules JSONB objektumból. Két rétegben keres: 1. Közvetlenül a rules objektumban: rules["max_vehicles"] 2. A rules["allowances"] nested objektumban: rules["allowances"]["max_vehicles"] Ez biztosítja, hogy a JSONB struktúra változásai (pl. 'allowances' nested object) ne törjék meg a limit számításokat. Args: rules: A SubscriptionTier.rules JSONB dict-je. keys: A keresendő kulcsok listája (pl. ["max_vehicles", "max_assets"]). Az első találat értéke kerül visszaadásra. default: Alapértelmezett érték, ha egyik kulcs sem található. Returns: int: A talált limit érték, vagy a default. """ if not rules or not isinstance(rules, dict): return default # 1. szint: Közvetlen keresés a rules objektumban for key in keys: value = rules.get(key) if value is not None and isinstance(value, (int, float)): return int(value) # 2. szint: Keresés a rules["allowances"] nested objektumban allowances = rules.get("allowances") if allowances and isinstance(allowances, dict): for key in keys: value = allowances.get(key) if value is not None and isinstance(value, (int, float)): return int(value) # 3. szint: Keresés a rules["limits"] nested objektumban (alternatív név) limits = rules.get("limits") if limits and isinstance(limits, dict): for key in keys: value = limits.get(key) if value is not None and isinstance(value, (int, float)): return int(value) return default # ============================================================================= # ACTUAL database JSON data (retrieved via PostgreSQL MCP query on 2026-06-28) # ============================================================================= # corp_premium_plus_v1 (id=17) — exact JSON as stored in system.subscription_tiers.rules CORP_PREMIUM_PLUS_V1_RAW = """{ "type": "corporate", "pricing": { "currency": "EUR", "credit_price": 6000, "yearly_price": 599.99, "monthly_price": 59.99 }, "duration": { "days": 30, "allow_stacking": true }, "ad_policy": { "show_ads": false, "ad_free_grace_days": 0, "max_daily_impressions": null }, "affiliate": { "referral_bonus_credits": 250, "commission_rate_percent": 20 }, "lifecycle": { "is_public": true }, "marketing": { "badge": "Plus", "subtitle": "Nagy flották prémium menedzsmentje.", "highlight_color": "#8B5CF6" }, "allowances": { "max_garages": 10, "max_vehicles": 50, "monthly_free_credits": 800 }, "display_name": "Céges Prémium Plus", "entitlements": [ "SRV_DATA_EXPORT", "SRV_AI_UPLOAD", "SRV_ACCOUNTING_SYNC" ], "pricing_zones": { "HU": { "currency": "HUF", "credit_price": 6000, "yearly_price": 239990, "monthly_price": 23990 }, "US": { "currency": "USD", "credit_price": 6000, "yearly_price": 699.99, "monthly_price": 69.99 }, "DEFAULT": { "currency": "EUR", "credit_price": 6000, "yearly_price": 599.99, "monthly_price": 59.99 } } }""" # corp_premium_v1 (id=16) — exact JSON as stored CORP_PREMIUM_V1_RAW = """{ "type": "corporate", "pricing": { "currency": "EUR", "credit_price": 3000, "yearly_price": 299.99, "monthly_price": 29.99 }, "duration": { "days": 30, "allow_stacking": true }, "ad_policy": { "show_ads": false, "ad_free_grace_days": 0, "max_daily_impressions": null }, "affiliate": { "referral_bonus_credits": 100, "commission_rate_percent": 15 }, "lifecycle": { "is_public": true }, "marketing": { "badge": "Prémium", "subtitle": "Közepes flották számára tervezve.", "highlight_color": "#3B82F6" }, "allowances": { "max_garages": 3, "max_vehicles": 20, "monthly_free_credits": 300 }, "display_name": "Céges Prémium", "entitlements": [ "SRV_DATA_EXPORT", "SRV_AI_UPLOAD" ], "pricing_zones": { "HU": { "currency": "HUF", "credit_price": 3000, "yearly_price": 119990, "monthly_price": 11990 }, "US": { "currency": "USD", "credit_price": 3000, "yearly_price": 349.99, "monthly_price": 34.99 }, "DEFAULT": { "currency": "EUR", "credit_price": 3000, "yearly_price": 299.99, "monthly_price": 29.99 } } }""" # private_pro_v1 (id=14) — exact JSON as stored PRIVATE_PRO_V1_RAW = """{ "type": "private", "lifecycle": { "is_public": true, "available_until": null }, "marketing": { "badge": null, "subtitle": null }, "allowances": { "max_garages": 1, "max_vehicles": 3, "monthly_free_credits": 0 }, "pricing_zones": { "HU": { "currency": "HUF", "credit_price": 25000, "yearly_price": 9990.0, "monthly_price": 990.0 }, "US": { "currency": "USD", "credit_price": 25000, "yearly_price": 35.9, "monthly_price": 3.5 }, "DEFAULT": { "currency": "EUR", "credit_price": 25000, "yearly_price": 29.9, "monthly_price": 2.99 } } }""" # feature_capabilities data (as stored) CORP_PREMIUM_PLUS_V1_FC_RAW = "{}" CORP_PREMIUM_V1_FC_RAW = "{}" PRIVATE_PRO_V1_FC_RAW = """{"export_data": false, "advanced_reports": true, "max_cost_category_depth": 3}""" # ============================================================================= # JSON double-encoding simulation # ============================================================================= DOUBLE_ENCODED_RAW = json.dumps(CORP_PREMIUM_PLUS_V1_RAW) # ============================================================================= # Test harness # ============================================================================= def print_separator(title: str): print(f"\n{'='*80}") print(f" {title}") print(f"{'='*80}") def test_single_tier( tier_name: str, rules_raw: str, fc_raw: str, vehicle_keys: list, branch_keys: list, user_keys: list, expected_vehicles: int, expected_branches: int, expected_users: int, ): """Test _extract_limit for a single tier, simulating the exact endpoint logic.""" print(f"\n ┌─ Tier: {tier_name}") rules = json.loads(rules_raw) fc = json.loads(fc_raw) print(f" ├─ rules type: {type(rules).__name__}") print(f" ├─ rules keys: {list(rules.keys())}") print(f" ├─ 'allowances' key present: {'allowances' in rules}") if 'allowances' in rules: print(f" ├─ allowances content: {json.dumps(rules['allowances'], indent=4)}") base_vehicles = _extract_limit( rules, vehicle_keys, _extract_limit(fc, vehicle_keys, 1), ) base_branches = _extract_limit( rules, branch_keys, _extract_limit(fc, branch_keys, 0), ) base_users = _extract_limit( rules, user_keys, _extract_limit(fc, user_keys, 0), ) print(f" ├─ RESULT: vehicles={base_vehicles} (expected={expected_vehicles})") print(f" ├─ RESULT: branches={base_branches} (expected={expected_branches})") print(f" └─ RESULT: users={base_users} (expected={expected_users})") status = "✅ PASS" if (base_vehicles == expected_vehicles and base_branches == expected_branches and base_users == expected_users) else "❌ FAIL" print(f" {status}") return status def test_double_encoding_scenario(): """Test what happens if the JSON is double-encoded.""" print_separator("SIMULATION: Double-encoded JSON (a known anti-pattern)") double_encoded = DOUBLE_ENCODED_RAW print(f"\n Double-encoded value preview: {double_encoded[:80]}...") print(f" Double-encoded type: {type(double_encoded).__name__}") parsed = json.loads(double_encoded) print(f" After json.loads(): type={type(parsed).__name__}") print(f" After json.loads(): is dict? {isinstance(parsed, dict)}") print(f" After json.loads(): is str? {isinstance(parsed, str)}") if isinstance(parsed, str): print(f"\n ❌ DOUBLE-ENCODING DETECTED!") print(f" _extract_limit receives a STRING, not a dict!") print(f" First check: 'if not rules or not isinstance(rules, dict):'") print(f" → Returns default=0 immediately!") result = _extract_limit(parsed, ["max_vehicles", "max_assets"], 0) print(f" _extract_limit(string, ...) = {result}") else: print(f"\n ✅ No double-encoding. Data is properly deserialized as dict.") def test_edge_cases(): """Test edge cases and potential failure modes.""" print_separator("EDGE CASE TESTS") result = _extract_limit({}, ["max_vehicles"], 5) print(f" Empty dict -> {result} (expected: 5) {'✅' if result == 5 else '❌'}") result = _extract_limit(None, ["max_vehicles"], 5) print(f" None -> {result} (expected: 5) {'✅' if result == 5 else '❌'}") result = _extract_limit("not a dict", ["max_vehicles"], 5) print(f" String input -> {result} (expected: 5) {'✅' if result == 5 else '❌'}") result = _extract_limit({"max_vehicles": 100}, ["max_vehicles", "max_assets"], 0) print(f" Direct key match -> {result} (expected: 100) {'✅' if result == 100 else '❌'}") result = _extract_limit({"allowances": {"max_vehicles": 50}}, ["max_vehicles", "max_assets"], 0) print(f" allowances['max_vehicles'] -> {result} (expected: 50) {'✅' if result == 50 else '❌'}") result = _extract_limit({"limits": {"max_vehicles": 25}}, ["max_vehicles", "max_assets"], 0) print(f" limits['max_vehicles'] -> {result} (expected: 25) {'✅' if result == 25 else '❌'}") # ============================================================================= # MAIN # ============================================================================= def main(): print_separator("P0 DATA STRUCTURAL AUDIT: _extract_limit() Standalone Test") print(" Date: 2026-06-28") print(" Source: system.subscription_tiers (PostgreSQL JSONB)") # --- Test 1: corp_premium_plus_v1 --- print_separator("TEST 1: corp_premium_plus_v1 (id=17)") print(" Expected: vehicles=50, branches=10, users=?") print(" Note: 'max_users'/'max_members' keys NOT present in rules JSON.") print(" The function will fall back to feature_capabilities (empty dict {}),") print(" then to the nested default (0).") test_single_tier( tier_name="corp_premium_plus_v1", rules_raw=CORP_PREMIUM_PLUS_V1_RAW, fc_raw=CORP_PREMIUM_PLUS_V1_FC_RAW, vehicle_keys=["max_vehicles", "max_assets"], branch_keys=["max_branches", "max_garages"], user_keys=["max_users", "max_members"], expected_vehicles=50, expected_branches=10, expected_users=0, ) # --- Test 2: corp_premium_v1 --- print_separator("TEST 2: corp_premium_v1 (id=16)") test_single_tier( tier_name="corp_premium_v1", rules_raw=CORP_PREMIUM_V1_RAW, fc_raw=CORP_PREMIUM_V1_FC_RAW, vehicle_keys=["max_vehicles", "max_assets"], branch_keys=["max_branches", "max_garages"], user_keys=["max_users", "max_members"], expected_vehicles=20, expected_branches=3, expected_users=0, ) # --- Test 3: private_pro_v1 --- print_separator("TEST 3: private_pro_v1 (id=14)") test_single_tier( tier_name="private_pro_v1", rules_raw=PRIVATE_PRO_V1_RAW, fc_raw=PRIVATE_PRO_V1_FC_RAW, vehicle_keys=["max_vehicles", "max_assets"], branch_keys=["max_branches", "max_garages"], user_keys=["max_users", "max_members"], expected_vehicles=3, expected_branches=1, expected_users=0, ) # --- Double-encoding simulation --- test_double_encoding_scenario() # --- Edge cases --- test_edge_cases() # --- Summary --- print_separator("SUMMARY") print(""" KEY FINDINGS FROM DB AUDIT: 1. All corporate tiers (corp_premium_plus_v1, corp_premium_v1) store their limits under rules['allowances'] nested object. 2. The _extract_limit() function correctly searches at 3 levels: Level 1: Direct keys in root (rules['max_vehicles']) Level 2: Nested under 'allowances' (rules['allowances']['max_vehicles']) Level 3: Nested under 'limits' (rules['limits']['max_vehicles']) 3. For corp_premium_plus_v1: allowances.max_vehicles = 50 Function should return 50, not 0. 4. ROOT CAUSE SUSPECTED: If _extract_limit returns 0 at runtime, the issue is likely: a) The 'rules' field is double-encoded (string inside JSONB) b) The wrong code path is being executed (e.g., no active_subs found) c) The rules dict is empty {} due to data migration issue """) if __name__ == "__main__": main()