#!/usr/bin/env python """Comprehensive transfer feature test with pre/post transfer benefit verification.""" from django.contrib.auth import get_user_model from rest_framework.test import APIClient from apps.business.distributor.models import ( DistributorID, DistributorTransferRequest, DistributorTransferTerms, DistributorRewardOwnershipLedger, DistributorTransferHistory ) from apps.business.products.models import Product from apps.business.distributor.services.distributor_service import create_distributor_id from decimal import Decimal import uuid from django.utils import timezone User = get_user_model() def s(v): return str(v) if v is not None else None print("=" * 100) print("COMPREHENSIVE TRANSFER TEST: Pre/Post Benefit Verification") print("=" * 100) # Clean slate DistributorTransferHistory.objects.all().delete() DistributorTransferRequest.objects.all().delete() print("\n[SETUP] Cleaned transfer data") try: # Test scenario 1: PowerStream ID with pre-transfer bonuses print("\n" + "=" * 100) print("TEST SCENARIO 1: PowerStream-Active ID with Pre-Transfer Powerstream Bonuses") print("=" * 100) sfx1 = uuid.uuid4().hex[:8] ps_old_owner = User.objects.create_user( mobile=f'91{sfx1}01', email=f'ps_old_{sfx1}@test.com', password='x12345', first_name='PS_Old', last_name='Owner', referral_code=f'PS_OLD_{sfx1}'[:16] ) ps_new_owner = User.objects.create_user( mobile=f'91{sfx1}02', email=f'ps_new_{sfx1}@test.com', password='x12345', first_name='PS_New', last_name='Owner', referral_code=f'PS_NEW_{sfx1}'[:16] ) admin1 = User.objects.create_user( mobile=f'91{sfx1}03', email=f'admin1_{sfx1}@test.com', password='x12345', first_name='Admin1', last_name='User', is_staff=True, role='ADMIN' ) product = Product.objects.filter(is_active=True).first() ps_dist = create_distributor_id(user=ps_old_owner, product=product, sponsor_distributor=None) # Add pre-transfer powerstream bonus for old owner pre_transfer_ps_bonus = Decimal('500.00') DistributorRewardOwnershipLedger.objects.create( distributor=ps_dist, beneficiary_user=ps_old_owner, reward_type='POWER_STREAM', reward_reference=f'PS_BONUS_{ps_dist.id}', amount=pre_transfer_ps_bonus, reward_timestamp=timezone.now(), ownership_period_id=1 if hasattr(ps_dist, 'ownership_periods') else None, ) print(f"[PS-SETUP] Old owner: {ps_old_owner.mobile} ({ps_old_owner.referral_code})") print(f"[PS-SETUP] New owner: {ps_new_owner.mobile} ({ps_new_owner.referral_code})") print(f"[PS-SETUP] Distributor: {ps_dist.distributor_code}") print(f"[PS-SETUP] Pre-transfer PS bonus credited to old owner: {pre_transfer_ps_bonus}") # Initiate transfer terms = list(DistributorTransferTerms.objects.filter(is_active=True, mandatory=True)) mandatory_ids = [s(t.id) for t in terms] client = APIClient() client.force_authenticate(user=ps_old_owner) payload = { 'distributor_id': s(ps_dist.id), 'reason': 'PS-Scenario transfer', 'referral_code': ps_new_owner.referral_code, 'accepted_term_ids': mandatory_ids, } r = client.post('/api/v1/web/distributor/transfers/', payload, format='json') assert r.status_code == 201, f"PS transfer submit failed: {r.status_code} {r.data}" ps_req_id = s(r.data['data']['id']) # New owner confirm client.force_authenticate(user=ps_new_owner) r = client.post(f'/api/v1/web/distributor/transfers/{ps_req_id}/new-owner-consent/', {'accepted_term_ids': mandatory_ids}, format='json') assert r.status_code == 200, f"PS new owner consent failed: {r.status_code}" # Admin approve client.force_authenticate(user=admin1) r = client.post(f'/api/v1/admin/distributor/transfers/{ps_req_id}/action/', {'action': 'approve', 'admin_notes': 'Approved PS transfer'}, format='json') assert r.status_code == 200, f"PS admin approve failed: {r.status_code} {r.data}" print(f"[PS-APPROVED] Transfer request {ps_req_id} approved") # Verify old owner my IDs (should be empty or not show PS ID) client.force_authenticate(user=ps_old_owner) r = client.get('/api/v1/web/distributor/ids/') assert r.status_code == 200 old_owner_ids = r.data if isinstance(r.data, list) else r.data.get('data', []) ps_in_old_list = any(s(x.get('id')) == s(ps_dist.id) for x in old_owner_ids) print(f"[PS-CHECK-OLD] PS ID in old owner's my IDs: {ps_in_old_list} (expected: False)") # Verify new owner my IDs (should include PS ID with transfer tag) client.force_authenticate(user=ps_new_owner) r = client.get('/api/v1/web/distributor/ids/') assert r.status_code == 200 new_owner_ids = r.data if isinstance(r.data, list) else r.data.get('data', []) ps_in_new_list = next((x for x in new_owner_ids if s(x.get('id')) == s(ps_dist.id)), None) assert ps_in_new_list, "PS ID missing from new owner's my IDs" ps_transfer_info = ps_in_new_list.get('transfer_info') print(f"[PS-CHECK-NEW] PS ID in new owner's my IDs: True") print(f"[PS-TRANSFER-INFO] {ps_transfer_info}") assert ps_transfer_info and ps_transfer_info.get('is_transferred_in'), "PS transfer_info not marked as transferred_in" assert ps_transfer_info.get('pre_transfer_benefit_total_for_old_owner') == s(pre_transfer_ps_bonus), \ f"PS pre-transfer benefit mismatch: {ps_transfer_info.get('pre_transfer_benefit_total_for_old_owner')} != {pre_transfer_ps_bonus}" print(f"[PS-VERIFIED] Transfer tag and pre-transfer benefit match!") # Check old owner transferred-out tab client.force_authenticate(user=ps_old_owner) r = client.get('/api/v1/web/distributor/transfers/history/transferred-out-ids/') assert r.status_code == 200, f"Transferred-out list failed: {r.status_code} {r.data}" transferred_out = r.data.get('data', []) if isinstance(r.data, dict) else r.data ps_transferred_out = next((x for x in transferred_out if s(x.get('distributor_id')) == s(ps_dist.id)), None) assert ps_transferred_out, "PS ID missing from old owner's transferred-out list" print(f"[PS-OLD-OWNER-TRANSFERRED] {ps_transferred_out}") print("\n✅ TEST SCENARIO 1 PASSED\n") # Test scenario 2: Non-PowerStream ID with regular rewards print("=" * 100) print("TEST SCENARIO 2: Non-PowerStream ID with Pre-Transfer Level Rewards") print("=" * 100) sfx2 = uuid.uuid4().hex[:8] np_old_owner = User.objects.create_user( mobile=f'91{sfx2}01', email=f'np_old_{sfx2}@test.com', password='x12345', first_name='NP_Old', last_name='Owner', referral_code=f'NP_OLD_{sfx2}'[:16] ) np_new_owner = User.objects.create_user( mobile=f'91{sfx2}02', email=f'np_new_{sfx2}@test.com', password='x12345', first_name='NP_New', last_name='Owner', referral_code=f'NP_NEW_{sfx2}'[:16] ) admin2 = User.objects.create_user( mobile=f'91{sfx2}03', email=f'admin2_{sfx2}@test.com', password='x12345', first_name='Admin2', last_name='User', is_staff=True, role='ADMIN' ) np_dist = create_distributor_id(user=np_old_owner, product=product, sponsor_distributor=None) # Add pre-transfer level reward pre_transfer_reward = Decimal('250.00') DistributorRewardOwnershipLedger.objects.create( distributor=np_dist, beneficiary_user=np_old_owner, reward_type='LEVEL_REWARD', reward_reference=f'LEVEL_REWARD_{np_dist.id}', amount=pre_transfer_reward, reward_timestamp=timezone.now(), ownership_period_id=1 if hasattr(np_dist, 'ownership_periods') else None, ) print(f"[NP-SETUP] Old owner: {np_old_owner.mobile} ({np_old_owner.referral_code})") print(f"[NP-SETUP] New owner: {np_new_owner.mobile} ({np_new_owner.referral_code})") print(f"[NP-SETUP] Distributor: {np_dist.distributor_code}") print(f"[NP-SETUP] Pre-transfer level reward credited to old owner: {pre_transfer_reward}") # Initiate transfer client.force_authenticate(user=np_old_owner) payload = { 'distributor_id': s(np_dist.id), 'reason': 'NP-Scenario transfer', 'referral_code': np_new_owner.referral_code, 'accepted_term_ids': mandatory_ids, } r = client.post('/api/v1/web/distributor/transfers/', payload, format='json') assert r.status_code == 201 np_req_id = s(r.data['data']['id']) # New owner confirm client.force_authenticate(user=np_new_owner) r = client.post(f'/api/v1/web/distributor/transfers/{np_req_id}/new-owner-consent/', {'accepted_term_ids': mandatory_ids}, format='json') assert r.status_code == 200 # Admin approve client.force_authenticate(user=admin2) r = client.post(f'/api/v1/admin/distributor/transfers/{np_req_id}/action/', {'action': 'approve', 'admin_notes': 'Approved NP transfer'}, format='json') assert r.status_code == 200 print(f"[NP-APPROVED] Transfer request {np_req_id} approved") # Verify new owner receives ID with transfer tag and pre-transfer benefit client.force_authenticate(user=np_new_owner) r = client.get('/api/v1/web/distributor/ids/') assert r.status_code == 200 new_owner_ids = r.data if isinstance(r.data, list) else r.data.get('data', []) np_in_new_list = next((x for x in new_owner_ids if s(x.get('id')) == s(np_dist.id)), None) assert np_in_new_list, "NP ID missing from new owner's my IDs" np_transfer_info = np_in_new_list.get('transfer_info') print(f"[NP-CHECK-NEW] NP ID in new owner's my IDs: True") print(f"[NP-TRANSFER-INFO] {np_transfer_info}") assert np_transfer_info and np_transfer_info.get('is_transferred_in'), "NP transfer_info not marked" assert np_transfer_info.get('pre_transfer_benefit_total_for_old_owner') == s(pre_transfer_reward), \ f"NP pre-transfer benefit mismatch: {np_transfer_info.get('pre_transfer_benefit_total_for_old_owner')} != {pre_transfer_reward}" print(f"[NP-VERIFIED] Transfer tag and pre-transfer benefit match!") # Check old owner transferred-out list includes NP ID client.force_authenticate(user=np_old_owner) r = client.get('/api/v1/web/distributor/transfers/history/transferred-out-ids/') assert r.status_code == 200 transferred_out = r.data.get('data', []) if isinstance(r.data, dict) else r.data np_transferred_out = next((x for x in transferred_out if s(x.get('distributor_id')) == s(np_dist.id)), None) assert np_transferred_out, "NP ID missing from old owner's transferred-out list" print(f"[NP-OLD-OWNER-TRANSFERRED] {np_transferred_out}") print("\n✅ TEST SCENARIO 2 PASSED\n") # Final summary print("=" * 100) print("COMPREHENSIVE TEST SUMMARY") print("=" * 100) print("\n✅ PowerStream Scenario:") print(f" - Old owner had {pre_transfer_ps_bonus} PS bonus pre-transfer") print(f" - Transfer marked: is_transferred_in=True") print(f" - Pre-transfer benefit visible in new owner's ID card: {pre_transfer_ps_bonus}") print(f" - Old owner can see in 'Transferred IDs' tab: Yes") print("\n✅ Non-PowerStream Scenario:") print(f" - Old owner had {pre_transfer_reward} level reward pre-transfer") print(f" - Transfer marked: is_transferred_in=True") print(f" - Pre-transfer benefit visible in new owner's ID card: {pre_transfer_reward}") print(f" - Old owner can see in 'Transferred IDs' tab: Yes") print("\n✅ ALL TESTS PASSED - FEATURE 100% FUNCTIONAL") finally: # Cleanup DistributorTransferHistory.objects.all().delete() DistributorTransferRequest.objects.all().delete()