446 lines
14 KiB
Python
446 lines
14 KiB
Python
#!/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()
|