195 lines
6.9 KiB
Python
195 lines
6.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Commission Distribution Verification Test
|
|
Tests the 2-Level MLM commission distribution logic end-to-end.
|
|
|
|
Prerequisites:
|
|
- Seed data created via seed_commission_test_data.py
|
|
|
|
Usage:
|
|
docker compose exec sf_api python /app/backend/tests/active/test_commission_distribution.py
|
|
"""
|
|
|
|
import asyncio
|
|
import httpx
|
|
import sys
|
|
import os
|
|
|
|
BASE_URL = "http://sf_api:8000/api/v1"
|
|
ADMIN_EMAIL = "admin@profibot.hu"
|
|
ADMIN_PASSWORD = "Admin123!"
|
|
|
|
|
|
async def main():
|
|
passed = 0
|
|
failed = 0
|
|
|
|
async with httpx.AsyncClient(base_url=BASE_URL) as client:
|
|
# ── Step 1: Login ──
|
|
print("=" * 60)
|
|
print("STEP 1: Login with admin user")
|
|
print("=" * 60)
|
|
|
|
login_data = {
|
|
"username": ADMIN_EMAIL,
|
|
"password": ADMIN_PASSWORD,
|
|
"grant_type": "password",
|
|
}
|
|
|
|
login_resp = await client.post(
|
|
"/auth/login",
|
|
data=login_data,
|
|
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
|
)
|
|
|
|
if login_resp.status_code != 200:
|
|
print(f"❌ Login failed: {login_resp.status_code} {login_resp.text}")
|
|
sys.exit(1)
|
|
|
|
token_data = login_resp.json()
|
|
access_token = token_data.get("access_token")
|
|
if not access_token:
|
|
print(f"❌ No access_token in response: {token_data}")
|
|
sys.exit(1)
|
|
|
|
print(f"✅ Login successful, token obtained")
|
|
headers = {"Authorization": f"Bearer {access_token}"}
|
|
passed += 1
|
|
|
|
# ── Step 2: Check referral chain via DB ──
|
|
print("\n" + "=" * 60)
|
|
print("STEP 2: Verify referral chain exists")
|
|
print("=" * 60)
|
|
|
|
# We'll check via the users list endpoint
|
|
resp = await client.get(
|
|
"/admin/users?limit=50",
|
|
headers=headers,
|
|
)
|
|
if resp.status_code != 200:
|
|
print(f"❌ Failed to list users: {resp.status_code} {resp.text}")
|
|
failed += 1
|
|
else:
|
|
users = resp.json()
|
|
# Find our test users
|
|
test_users = [u for u in (users.get("users") or users.get("data") or users if isinstance(users, list) else [])
|
|
if isinstance(u, dict) and "commission_test" in str(u.get("email", ""))]
|
|
# Try to find from response
|
|
print(f"✅ Users endpoint responded (status={resp.status_code})")
|
|
passed += 1
|
|
|
|
# ── Step 3: Test commission distribution ──
|
|
print("\n" + "=" * 60)
|
|
print("STEP 3: Test commission distribution API")
|
|
print("=" * 60)
|
|
|
|
# Use the known test user IDs from seed data
|
|
# Buyer (Gen0) = ID 155, Gen1 = 154, Gen2 = 153
|
|
# Rule ID 4: STANDARD, HU, 5% Gen1, 2% Gen2, max 50000
|
|
|
|
distribute_payload = {
|
|
"buyer_user_id": 155,
|
|
"transaction_amount": 100000.00,
|
|
"transaction_date": "2026-07-24",
|
|
"region_code": "HU",
|
|
}
|
|
|
|
print(f"Request: POST /admin/commission-rules/distribute")
|
|
print(f"Payload: {distribute_payload}")
|
|
|
|
resp = await client.post(
|
|
"/admin/commission-rules/distribute",
|
|
json=distribute_payload,
|
|
headers=headers,
|
|
)
|
|
|
|
print(f"Response status: {resp.status_code}")
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
print(f"Response body: {data}")
|
|
print()
|
|
|
|
# Validate response structure
|
|
assert "transaction_amount" in data, "Missing transaction_amount"
|
|
assert "items" in data, "Missing items"
|
|
assert "total_commission" in data, "Missing total_commission"
|
|
assert len(data["items"]) > 0, "No commission items returned"
|
|
|
|
# Check Gen1 commission (5% of 100000 = 5000)
|
|
gen1_item = next((i for i in data["items"] if i["level"] == 1), None)
|
|
assert gen1_item is not None, "Missing Gen1 commission item"
|
|
assert gen1_item["user_id"] == 154, f"Expected Gen1 user_id=154, got {gen1_item['user_id']}"
|
|
assert gen1_item["commission_amount"] == 5000.00, \
|
|
f"Expected Gen1 commission=5000.00, got {gen1_item['commission_amount']}"
|
|
|
|
# Check Gen2 commission (2% of 100000 = 2000)
|
|
gen2_item = next((i for i in data["items"] if i["level"] == 2), None)
|
|
assert gen2_item is not None, "Missing Gen2 commission item"
|
|
assert gen2_item["user_id"] == 153, f"Expected Gen2 user_id=153, got {gen2_item['user_id']}"
|
|
assert gen2_item["commission_amount"] == 2000.00, \
|
|
f"Expected Gen2 commission=2000.00, got {gen2_item['commission_amount']}"
|
|
|
|
# Check total
|
|
assert data["total_commission"] == 7000.00, \
|
|
f"Expected total_commission=7000.00, got {data['total_commission']}"
|
|
|
|
print("✅ Commission distribution test PASSED!")
|
|
print(f" Gen1 (User 154): 5% of 100000 = 5000.00 ✅")
|
|
print(f" Gen2 (User 153): 2% of 100000 = 2000.00 ✅")
|
|
print(f" Total: 7000.00 ✅")
|
|
passed += 1
|
|
elif resp.status_code == 404:
|
|
print(f"⚠️ No matching rule found: {resp.text}")
|
|
print(" This means the rule resolution didn't match. Check tier/region.")
|
|
else:
|
|
print(f"❌ Distribution failed: {resp.text}")
|
|
failed += 1
|
|
|
|
# ── Step 4: Test with max_amount cap ──
|
|
print("\n" + "=" * 60)
|
|
print("STEP 4: Test commission with max_amount cap")
|
|
print("=" * 60)
|
|
|
|
# With a very large transaction, Gen1 should be capped at 50000
|
|
# 5% of 2000000 = 100000, but max is 50000
|
|
cap_payload = {
|
|
"buyer_user_id": 155,
|
|
"transaction_amount": 2000000.00,
|
|
"transaction_date": "2026-07-24",
|
|
"region_code": "HU",
|
|
}
|
|
|
|
resp = await client.post(
|
|
"/admin/commission-rules/distribute",
|
|
json=cap_payload,
|
|
headers=headers,
|
|
)
|
|
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
gen1_item = next((i for i in data["items"] if i["level"] == 1), None)
|
|
if gen1_item:
|
|
# 5% of 2M = 100000, capped at 50000
|
|
assert gen1_item["commission_amount"] <= 50000.00, \
|
|
f"Gen1 commission {gen1_item['commission_amount']} exceeds max 50000"
|
|
print(f"✅ Max amount cap verified: Gen1 commission={gen1_item['commission_amount']} (capped at 50000)")
|
|
passed += 1
|
|
else:
|
|
print("⚠️ No Gen1 item in capped test")
|
|
else:
|
|
print(f"⚠️ Cap test skipped (status={resp.status_code})")
|
|
|
|
# ── Summary ──
|
|
print("\n" + "=" * 60)
|
|
print(f"RESULTS: {passed} passed, {failed} failed")
|
|
print("=" * 60)
|
|
|
|
if failed > 0:
|
|
sys.exit(1)
|
|
else:
|
|
print("\n✅ All commission distribution tests PASSED!")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|