from __future__ import annotations import gc import json import logging import os import random import time import tracemalloc from collections import Counter, defaultdict, deque from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError, as_completed from datetime import timedelta from decimal import Decimal from pathlib import Path from uuid import uuid4 from django.contrib.auth import get_user_model from django.db import IntegrityError, close_old_connections, connection from django.db.models import Count, Max, Min, Q, Sum from django.utils import timezone from apps.business.binary_tree.services.binary_service import ( build_binary_subtree_dict, get_binary_ancestor_ids, verify_binary_integrity, ) from apps.business.binary_tree.services.metrics_service import ( compute_and_persist_all, get_binary_dashboard_data, ) from apps.business.distributor.models import ( AutoIDAllocation, DistributorID, DistributorOwnershipPeriod, DistributorQualificationState, DistributorRewardOwnershipLedger, DistributorTransferHistory, DistributorTransferRequest, ) from apps.business.distributor.services.hybrid_service import ( activate_auto_ids, approve_transfer_request, ensure_current_ownership, refresh_qualification_for_distributor, submit_transfer_request, ) from apps.business.orders.models import Order from apps.business.orders.services.order_service import ( create_pending_payment_order, execute_purchase, fulfill_pending_payment_order, ) from apps.business.pool.models import PoolEntry from apps.business.power_stream.models import PowerStreamBonusRecord, PowerStreamMonthlyClaim from apps.business.power_stream.services.power_stream_service import ( get_user_power_stream_summary, process_monthly_claim, process_power_stream_bonus, ) from apps.business.products.models import Product from apps.business.rewards.models import DistributorLevelProgress from apps.business.rewards.services.level_service import ( calculate_level_progress, process_auto_id_generation, process_level_rewards, ) from apps.business.schemes.models import OpportunityBundleLevel from apps.business.system_stats_service import SystemStatsService from apps.business.wallet.models import SuperCoinTransaction, SuperCoinWallet, Wallet, WalletLedger from apps.business.wallet.services.wallet_service import WalletService from apps.business.withdrawals.models import BankAccount, WithdrawalRequest from apps.business.withdrawals.services.withdrawal_service import ( create_withdrawal_request, process_withdrawal, ) User = get_user_model() logger = logging.getLogger(__name__) class FullSystemValidationService: PASSWORD = 'SysVal@Test123' def __init__( self, *, scales, output_root, special_mobile='8668192080', special_main_balance=Decimal('150000.00'), special_super_balance=Decimal('150000.00'), concurrency_workers=8, concurrency_purchases=24, task_mode='sync', ): self.scales = scales self.output_root = Path(output_root) self.special_mobile = special_mobile self.special_main_balance = Decimal(str(special_main_balance)) self.special_super_balance = Decimal(str(special_super_balance)) self.concurrency_workers = concurrency_workers self.concurrency_purchases = concurrency_purchases self.task_mode = (task_mode or 'sync').lower() self.started_at = timezone.now() self.run_seed = int(self.started_at.strftime('%j%H%M')) % 1000 self.run_tag = f"SYSVAL-{self.started_at.strftime('%Y%m%d%H%M%S')}" self.random = random.Random(self.run_seed) self.report_dir = self.output_root / 'docs' / 'validation' / self.run_tag.lower() self.report_dir.mkdir(parents=True, exist_ok=True) self.checkpoint_file = self.report_dir / 'checkpoint.json' self.product = None self.root_distributor = None self.admin_user = None self.special_user = None self.special_distributor = None self.standalone_product = None self.bundle_levels = [] self.auto_id_enabled_levels = [] self.run_user_ids = set() self.run_distributor_ids = set() self.run_order_ids = set() self.run_withdrawal_ids = set() self.transfer_history_ids = set() self.scenario_results = [] self.edge_case_results = [] self.concurrency_results = [] self.known_risks = [] self.notes = [] self.binary_snapshots = [] self.step_results = [] self.execution_metrics = defaultdict(list) self.purchase_durations = [] self.runtime_stats = {} self.validation_summary = {} def _log_event(self, category, event, **payload): envelope = { 'run_tag': self.run_tag, 'category': category, 'event': event, 'timestamp': timezone.now().isoformat(), **payload, } logger.info(json.dumps(envelope, default=str)) def _record_metric(self, metric_name, seconds, **labels): entry = {'seconds': round(float(seconds), 6), **labels} self.execution_metrics[metric_name].append(entry) def _checkpoint(self, state, **extra): payload = { 'run_tag': self.run_tag, 'state': state, 'updated_at': timezone.now().isoformat(), 'steps': self.step_results, 'scenarios': self.scenario_results, 'known_risks': self.known_risks, **extra, } try: self.checkpoint_file.write_text(json.dumps(payload, indent=2, default=str), encoding='utf-8') except Exception as exc: self.notes.append(f'Checkpoint write failed: {exc}') def _run_db_with_retry(self, name, callback, *, attempts=4): for attempt in range(1, attempts + 1): try: return callback() except Exception as exc: if self._is_db_locked_error(exc) and attempt < attempts: close_old_connections() sleep_s = min(1.0, 0.2 * attempt) self._log_event('runtime', 'db_retry', operation=name, attempt=attempt, sleep_seconds=sleep_s, error=str(exc)) time.sleep(sleep_s) continue raise def _should_run_async_tasks(self): if self.task_mode == 'async': return True if self.task_mode == 'sync': return False # auto mode: keep deterministic sync on sqlite, async otherwise return connection.vendor != 'sqlite' def _collect_runtime_stats(self): current_mem, peak_mem = (0, 0) if tracemalloc.is_tracing(): current_mem, peak_mem = tracemalloc.get_traced_memory() return { 'pid': os.getpid(), 'db_vendor': connection.vendor, 'gc_counts': list(gc.get_count()), 'tracemalloc_current_bytes': int(current_mem), 'tracemalloc_peak_bytes': int(peak_mem), 'purchase_count': len(self.purchase_durations), 'purchase_total_seconds': round(sum(self.purchase_durations), 4), 'purchase_avg_seconds': round((sum(self.purchase_durations) / max(len(self.purchase_durations), 1)), 4), 'purchase_throughput_per_second': round((len(self.purchase_durations) / max(sum(self.purchase_durations), 0.001)), 4), } def _build_validation_summary(self, integrity, financial_summary, wallet_summary): checks = [] for scenario in self.scenario_results: checks.append({'name': f"scenario_{scenario.get('label')}", 'passed': bool(scenario.get('passed', False)), 'partial': bool(scenario.get('passed', False)) and bool(scenario.get('integrity_issues_found'))}) for result in self.edge_case_results: checks.append({'name': result.get('name'), 'passed': bool(result.get('passed', False)), 'partial': False}) for result in self.concurrency_results: partial = bool(result.get('passed')) and int(result.get('failure_count') or 0) > 0 checks.append({'name': result.get('name'), 'passed': bool(result.get('passed', False)), 'partial': partial}) checks.append({'name': 'binary_integrity', 'passed': not integrity.get('issues_found', True), 'partial': False}) checks.append({'name': 'wallet_reconciliation', 'passed': int(wallet_summary.get('mismatch_count', 0)) == 0, 'partial': False}) checks.append({'name': 'withdrawal_vs_growth_pool', 'passed': bool(financial_summary.get('actual_withdrawals_lt_net_growth_pool', False)), 'partial': False}) total_passed = sum(1 for check in checks if check['passed']) partial_passes = sum(1 for check in checks if check['partial']) failures = sum(1 for check in checks if not check['passed']) warnings = len(self.known_risks) return { 'total_checks': len(checks), 'total_passed': total_passed, 'partial_passes': partial_passes, 'warnings': warnings, 'failures': failures, 'unresolved_risks': self.known_risks, } def run(self): tracemalloc.start() self._configure_sqlite_session() self._checkpoint('starting') self._discover_environment() self._ensure_special_account() self._execute_step('registration_tests', self._run_registration_tests) self._execute_step('special_account_purchase_tests', self._run_special_account_purchase_tests) for label, scale in self.scales: deep_checks = label == 'small' scenario_name = f'scenario_{label}' self._checkpoint('scenario_in_progress', current_scenario=scenario_name, target_size=scale) try: scenario_result = self._simulate_network(label=label, size=scale, deep_checks=deep_checks) scenario_result['passed'] = True scenario_result['error'] = None self.scenario_results.append(scenario_result) self._log_event('scenario', 'completed', name=scenario_name, size=scale, duration_seconds=scenario_result.get('duration_seconds')) except BaseException as exc: message = f'{scenario_name} failed: {exc}' self.known_risks.append(message) self.edge_case_results.append({'name': scenario_name, 'passed': False, 'detail': str(exc)}) self.scenario_results.append( { 'label': label, 'size': scale, 'passed': False, 'error': str(exc), } ) self._log_event('scenario', 'failed', name=scenario_name, size=scale, error=str(exc)) self._checkpoint('scenario_completed', current_scenario=scenario_name) self._execute_step('transfer_tests', self._run_transfer_tests) self._execute_step('withdrawal_tests', self._run_withdrawal_tests) self._execute_step('power_stream_claim_test', self._run_power_stream_claim_test) self._execute_step('admin_controls_validation', self._run_admin_controls_validation) self._execute_step('edge_case_tests', self._run_edge_case_tests) self._execute_step('concurrency_tests', self._run_concurrency_tests) report_bundle = {} final_integrity = {} financial_summary = {} wallet_summary = {} production_readiness_score = 0 try: final_integrity = verify_binary_integrity(bundle_id=self.product.opportunity_bundle_id) financial_summary = self._build_financial_summary() wallet_summary = self._build_wallet_reconciliation() self.validation_summary = self._build_validation_summary(final_integrity, financial_summary, wallet_summary) production_readiness_score = self._score_readiness(final_integrity, financial_summary, wallet_summary) report_bundle = self._write_reports(final_integrity, financial_summary, wallet_summary) except Exception as exc: self.known_risks.append(f'finalization failed: {exc}') self._log_event('runtime', 'finalization_failed', error=str(exc)) final_integrity = final_integrity or {'issues_found': True} financial_summary = financial_summary or {'actual_withdrawals_lt_net_growth_pool': False} wallet_summary = wallet_summary or {'mismatch_count': -1, 'checked_wallets': 0, 'mismatches': []} self.validation_summary = self._build_validation_summary(final_integrity, financial_summary, wallet_summary) production_readiness_score = self._score_readiness(final_integrity, financial_summary, wallet_summary) finally: self.runtime_stats = self._collect_runtime_stats() self._checkpoint('finalizing') summary = { 'run_tag': self.run_tag, 'report_dir': str(self.report_dir), 'task_mode': self.task_mode, 'scenarios': self.scenario_results, 'final_integrity': final_integrity, 'financial_summary': financial_summary, 'wallet_summary': wallet_summary, 'edge_case_results': self.edge_case_results, 'concurrency_results': self.concurrency_results, 'known_risks': self.known_risks, 'notes': self.notes, 'execution_metrics': dict(self.execution_metrics), 'runtime_stats': self.runtime_stats, 'validation_summary': self.validation_summary, 'production_readiness_score': production_readiness_score, 'reports': report_bundle, } self._write_json('summary.json', summary) if 'final_acceptance_report' not in report_bundle: report_bundle['final_acceptance_report'] = self._write_text('FINAL_ACCEPTANCE_REPORT.md', self._final_acceptance_markdown(summary)) summary['reports'] = report_bundle self._write_json('summary.json', summary) self._checkpoint('completed', final_acceptance_report=report_bundle.get('final_acceptance_report')) return summary def _execute_step(self, name, callback): started = time.perf_counter() self._log_event('step', 'started', name=name) attempts = 3 for attempt in range(1, attempts + 1): try: result = callback() duration_seconds = round(time.perf_counter() - started, 4) self.step_results.append({'name': name, 'passed': True, 'attempt': attempt, 'duration_seconds': duration_seconds}) self._record_metric('step_duration_seconds', duration_seconds, step=name) self._log_event('step', 'completed', name=name, attempt=attempt, duration_seconds=duration_seconds) self._checkpoint('step_completed', current_step=name) return result except BaseException as exc: if self._is_db_locked_error(exc) and attempt < attempts: close_old_connections() time.sleep(0.25 * attempt) continue message = f'{name} failed: {exc}' self.known_risks.append(message) self.edge_case_results.append({'name': name, 'passed': False, 'detail': str(exc)}) duration_seconds = round(time.perf_counter() - started, 4) self.step_results.append({'name': name, 'passed': False, 'attempt': attempt, 'duration_seconds': duration_seconds, 'error': str(exc)}) self._record_metric('step_duration_seconds', duration_seconds, step=name) self._log_event('step', 'failed', name=name, attempt=attempt, duration_seconds=duration_seconds, error=str(exc)) self._checkpoint('step_failed', current_step=name, error=str(exc)) return None def _configure_sqlite_session(self): if connection.vendor != 'sqlite': return try: with connection.cursor() as cursor: cursor.execute('PRAGMA busy_timeout = 15000') cursor.execute('PRAGMA journal_mode = WAL') except Exception as exc: self.notes.append(f'SQLite session tuning skipped: {exc}') def _is_db_locked_error(self, exc): return 'database is locked' in str(exc).lower() def _discover_environment(self): self.admin_user = User.objects.filter(is_superuser=True).order_by('date_joined').first() or User.objects.filter(is_staff=True).order_by('date_joined').first() if self.admin_user is None: raise RuntimeError('No admin/superuser available for validation flows.') root = DistributorID.objects.filter(is_company_root=True).select_related('product__opportunity_bundle', 'user').order_by('created_at').first() if root is None: raise RuntimeError('No company root distributor found for validation.') bundle_products = Product.objects.filter( is_active=True, opportunity_bundle_id=root.product.opportunity_bundle_id, ).select_related('opportunity_bundle').order_by('base_cost', 'created_at') self.product = bundle_products.first() if self.product is None: raise RuntimeError('No active bundle product found for validation.') self.root_distributor = DistributorID.objects.filter( is_company_root=True, product__opportunity_bundle_id=self.product.opportunity_bundle_id, ).select_related('user', 'product__opportunity_bundle').order_by('created_at').first() if self.root_distributor is None: raise RuntimeError('No company root distributor found for selected bundle.') self.standalone_product = Product.objects.filter(is_active=True, opportunity_bundle__isnull=True).order_by('base_cost', 'created_at').first() self.bundle_levels = list( OpportunityBundleLevel.objects.filter(opportunity_bundle_id=self.product.opportunity_bundle_id) .order_by('level_number') ) self.auto_id_enabled_levels = [level for level in self.bundle_levels if level.auto_id_enabled] if self.auto_id_enabled_levels and not any(level.auto_id_product_id for level in self.auto_id_enabled_levels): self.known_risks.append( 'Auto-ID levels are configured, but no auto_id_product is mapped; auto-ID allocation cannot be exercised end-to-end.' ) if connection.vendor == 'sqlite': original_workers = self.concurrency_workers original_purchases = self.concurrency_purchases self.concurrency_workers = min(self.concurrency_workers, 2) self.concurrency_purchases = min(self.concurrency_purchases, 8) if self.concurrency_workers != original_workers or self.concurrency_purchases != original_purchases: self.notes.append( f'SQLite concurrency guard applied: workers {original_workers}->{self.concurrency_workers}, purchases {original_purchases}->{self.concurrency_purchases}.' ) def _ensure_special_account(self): special_user = User.objects.filter(mobile=self.special_mobile).first() if special_user is None: special_user = User.objects.create_user( mobile=self.special_mobile, email=f'{self.run_tag.lower()}-special@sysval.test', password=self.PASSWORD, login_pin='123456', first_name='TEST', last_name='NON-PRODUCTION', is_active=True, is_mobile_verified=True, is_email_verified=True, is_onboarding_complete=True, accept_terms=True, ) self.special_user = special_user self._ensure_main_wallet_balance(self.special_user, self.special_main_balance, label='SPECIAL-MAIN') self._ensure_supercoin_balance(self.special_user, self.special_super_balance, label='SPECIAL-SUPER') self._ensure_verified_bank_account(self.special_user) special_dist = DistributorID.objects.filter( user=self.special_user, product__opportunity_bundle_id=self.product.opportunity_bundle_id, ).select_related('product__opportunity_bundle').order_by('created_at').first() if special_dist is None: special_dist = self._execute_wallet_purchase( user=self.special_user, sponsor=self.root_distributor, scenario='special', index=1, label='SPECIAL-ROOT-PURCHASE', ) self._consume_async_business([special_dist]) self.special_distributor = special_dist self.run_user_ids.add(self.special_user.id) def _run_registration_tests(self): try: dup_email = f'{self.run_tag.lower()}-duplicate@sysval.test' user = User.objects.create_user( mobile=f'79{self.run_seed:03d}99999', email=dup_email, password=self.PASSWORD, login_pin='123456', referred_by=self.special_user, is_active=True, is_mobile_verified=True, is_email_verified=True, ) self.edge_case_results.append({'name': 'referral_registration', 'passed': user.referred_by_id == self.special_user.id, 'detail': f'user={user.id}'}) self.run_user_ids.add(user.id) except Exception as exc: self.edge_case_results.append({'name': 'referral_registration', 'passed': False, 'detail': str(exc)}) try: User.objects.create_user( mobile=self.special_mobile, email=f'{self.run_tag.lower()}-dup-mobile@sysval.test', password=self.PASSWORD, login_pin='123456', ) self.edge_case_results.append({'name': 'duplicate_registration_prevention', 'passed': False, 'detail': 'Duplicate mobile unexpectedly allowed.'}) except Exception as exc: self.edge_case_results.append({'name': 'duplicate_registration_prevention', 'passed': True, 'detail': str(exc)}) def _run_special_account_purchase_tests(self): self._execute_wallet_purchase( user=self.special_user, sponsor=self.root_distributor, scenario='special', index=2, label='SPECIAL-MAIN-WALLET-PURCHASE', ) self._execute_wallet_purchase( user=self.special_user, sponsor=self.root_distributor, scenario='special', index=3, label='SPECIAL-SUPERCOIN-PURCHASE', use_supercoins=True, super_coin_amount=min(self.product.base_cost, Decimal('4000.00')), ) manual_user = self._create_test_user('special', 4, referred_by=self.root_distributor.user) pending_order = create_pending_payment_order( user=manual_user, product=self.product, sponsor_distributor=self.root_distributor, client_reference_id=self._make_reference('SPECIAL-PENDING'), payment_method='qr_scan', qr_reference=self._make_reference('UTR'), amount=self.product.base_cost, ) fulfilled_order, fulfilled_dist, _ = fulfill_pending_payment_order(pending_order, self.admin_user) self.run_order_ids.add(fulfilled_order.id) self.run_distributor_ids.add(fulfilled_dist.id) self.run_user_ids.add(manual_user.id) self._consume_async_business([fulfilled_dist]) def _simulate_network(self, *, label, size, deep_checks=False): started = time.perf_counter() scenario_start = timezone.now() created_users = [] created_distributors = [] special_directs = [] sponsor_candidates = [self.special_distributor, self.root_distributor] self._log_event('scenario', 'started', label=label, size=size) for index in range(1, size + 1): sponsor = self._pick_sponsor(label, index, sponsor_candidates, special_directs) user = self._run_db_with_retry( f'create_test_user_{label}_{index}', lambda current_label=label, current_index=index, current_sponsor=sponsor: self._create_test_user(current_label, current_index, referred_by=current_sponsor.user), ) created_users.append(user) distributor = self._run_db_with_retry( f'wallet_purchase_{label}_{index}', lambda current_user=user, current_sponsor=sponsor, current_label=label, current_index=index: self._execute_wallet_purchase( user=current_user, sponsor=current_sponsor, scenario=current_label, index=current_index, label=f'{current_label.upper()}-PURCHASE-{current_index}', ), ) created_distributors.append(distributor) sponsor_candidates.append(distributor) if sponsor.id == self.special_distributor.id: special_directs.append(distributor) if index % 100 == 0 or index == size: progress = round(index / max(size, 1) * 100, 2) self._checkpoint( 'scenario_progress', current_scenario=label, processed=index, target=size, progress_percent=progress, runtime=self._collect_runtime_stats(), ) gc.collect() post_processing = self._consume_async_business(created_distributors) integrity = verify_binary_integrity(bundle_id=self.product.opportunity_bundle_id) compute_and_persist_all(bundle_id=self.product.opportunity_bundle_id, only_stale=False) if deep_checks: snapshot = build_binary_subtree_dict(self.root_distributor.id, max_depth=4) self.binary_snapshots.append({'label': f'{label}-root-snapshot', 'tree': snapshot}) level_distribution = Counter( DistributorLevelProgress.objects.filter(distributor_id__in=[dist.id for dist in created_distributors]).values_list('current_level', flat=True) ) duration_seconds = round(time.perf_counter() - started, 2) sponsor_direct_counts = list( DistributorID.objects.filter(id__in=[dist.id for dist in created_distributors]) .annotate(directs=Count('direct_referrals')) .values_list('directs', flat=True) ) result = { 'label': label, 'size': size, 'duration_seconds': duration_seconds, 'created_users': len(created_users), 'created_distributors': len(created_distributors), 'purchase_throughput_per_second': round(len(created_distributors) / max(duration_seconds, 0.001), 4), 'scenario_started_at': scenario_start.isoformat(), 'global_position_min': self._aggregate_dist(created_distributors, Min('global_position')), 'global_position_max': self._aggregate_dist(created_distributors, Max('global_position')), 'power_stream_records_created': post_processing['power_stream_records_created'], 'level_rewards_processed': post_processing['level_rewards_processed'], 'auto_id_allocations_created': post_processing['auto_id_allocations_created'], 'integrity_issues_found': integrity['issues_found'], 'level_distribution': dict(sorted(level_distribution.items())), 'max_direct_sponsor_referrals': max(sponsor_direct_counts or [0]), 'direct_sponsor_referral_avg': round(sum(sponsor_direct_counts) / max(len(sponsor_direct_counts), 1), 2), } self._record_metric('scenario_duration_seconds', duration_seconds, scenario=label) self._log_event('scenario', 'finished', label=label, duration_seconds=duration_seconds, created_distributors=len(created_distributors)) return result def _run_transfer_tests(self): candidate = DistributorID.objects.filter(id__in=self.run_distributor_ids).exclude(user=self.special_user).order_by('created_at').first() new_owner = User.objects.filter(id__in=self.run_user_ids).exclude(id=candidate.user_id if candidate else None).order_by('created_at').first() final_owner = User.objects.filter(id__in=self.run_user_ids).exclude(id__in=[candidate.user_id if candidate else None, new_owner.id if new_owner else None]).order_by('created_at').first() if not candidate or not new_owner or not final_owner: self.notes.append('Transfer tests skipped because insufficient simulated distributors/users were available.') return first_started = time.perf_counter() first_request = self._run_db_with_retry( 'transfer_submit_first', lambda: submit_transfer_request( requested_by=new_owner, target_distributor=candidate, reason='TEST / NON-PRODUCTION first transfer', referral_code=new_owner.referral_code or '', ), ) first_history = self._run_db_with_retry( 'transfer_approve_first', lambda: approve_transfer_request(transfer_request=first_request, approved_by=self.admin_user, notes='Approved during system validation'), ) first_duration = round(time.perf_counter() - first_started, 4) self._record_metric('transfer_processing_seconds', first_duration, stage='first_transfer') self._log_event('transfer', 'first_transfer_completed', distributor=str(candidate.id), duration_seconds=first_duration) self.transfer_history_ids.add(first_history.id) second_started = time.perf_counter() second_request = self._run_db_with_retry( 'transfer_submit_second', lambda: submit_transfer_request( requested_by=final_owner, target_distributor=candidate, reason='TEST / NON-PRODUCTION chained transfer', referral_code=final_owner.referral_code or '', ), ) second_history = self._run_db_with_retry( 'transfer_approve_second', lambda: approve_transfer_request(transfer_request=second_request, approved_by=self.admin_user, notes='Approved during chained transfer validation'), ) second_duration = round(time.perf_counter() - second_started, 4) self._record_metric('transfer_processing_seconds', second_duration, stage='second_transfer') self._log_event('transfer', 'second_transfer_completed', distributor=str(candidate.id), duration_seconds=second_duration) self.transfer_history_ids.add(second_history.id) periods = list(DistributorOwnershipPeriod.objects.filter(distributor=candidate).order_by('ownership_effective_from')) qualification = refresh_qualification_for_distributor(candidate) reward_ledger_count = DistributorRewardOwnershipLedger.objects.filter(distributor=candidate).count() if reward_ledger_count == 0: self.known_risks.append( f'Reward ownership ledger remained empty for transferred distributor {candidate.distributor_code}; historical reward preservation is not auditable end-to-end.' ) self.edge_case_results.append( { 'name': 'transfer_chain', 'passed': len(periods) >= 3 and qualification.ownership_period.is_current, 'detail': { 'distributor_code': candidate.distributor_code, 'period_count': len(periods), 'current_owner': str(candidate.user_id), 'qualification_reset_directs': qualification.achieved_directs, 'reward_ownership_entries': reward_ledger_count, }, } ) def _run_withdrawal_tests(self): withdrawal_user = self._create_test_user('edge', 90, referred_by=self.special_user) self._ensure_main_wallet_balance(withdrawal_user, Decimal('7000.00'), label='WITHDRAWAL-TEST') bank_account = self._ensure_verified_bank_account(withdrawal_user) try: create_withdrawal_request(withdrawal_user, Decimal('1000.00'), bank_account=bank_account, enforce_last_day=True) self.edge_case_results.append({'name': 'withdrawal_last_day_rule', 'passed': False, 'detail': 'Withdrawal unexpectedly allowed before last day of month.'}) except Exception as exc: self.edge_case_results.append({'name': 'withdrawal_last_day_rule', 'passed': True, 'detail': str(exc)}) withdraw_started = time.perf_counter() withdrawal = self._run_db_with_retry( 'withdrawal_create', lambda: create_withdrawal_request( withdrawal_user, Decimal('5000.00'), bank_account=bank_account, enforce_last_day=False, ), ) self.run_withdrawal_ids.add(withdrawal.id) self._run_db_with_retry('withdrawal_approve', lambda: process_withdrawal(withdrawal.id, 'APPROVE', remarks='TEST / NON-PRODUCTION approval')) self._run_db_with_retry('withdrawal_paid', lambda: process_withdrawal(withdrawal.id, 'PAID', remarks='TEST / NON-PRODUCTION payout')) withdraw_duration = round(time.perf_counter() - withdraw_started, 4) self._record_metric('withdrawal_processing_seconds', withdraw_duration, stage='create_to_paid') self._log_event('withdrawal', 'completed', withdrawal_id=str(withdrawal.id), duration_seconds=withdraw_duration) def _run_power_stream_claim_test(self): summary_before = get_user_power_stream_summary(self.special_user) self._log_event('power_stream', 'claim_attempt', user_id=str(self.special_user.id)) try: result = process_monthly_claim(self.special_user) self.edge_case_results.append({'name': 'power_stream_claim', 'passed': True, 'detail': result}) self._log_event('power_stream', 'claim_success', user_id=str(self.special_user.id), detail=result) try: process_monthly_claim(self.special_user) self.edge_case_results.append({'name': 'duplicate_power_stream_claim_prevention', 'passed': False, 'detail': 'Second monthly claim unexpectedly succeeded.'}) except Exception as exc: self.edge_case_results.append({'name': 'duplicate_power_stream_claim_prevention', 'passed': True, 'detail': str(exc)}) self._log_event('power_stream', 'duplicate_claim_blocked', user_id=str(self.special_user.id), error=str(exc)) except Exception as exc: if summary_before.get('monthly', {}).get('already_claimed_this_month'): self.edge_case_results.append( { 'name': 'power_stream_claim', 'passed': True, 'detail': { 'summary_before': summary_before, 'note': 'Claim already completed this month; guard path confirmed.', 'error': str(exc), }, } ) else: self.edge_case_results.append({'name': 'power_stream_claim', 'passed': False, 'detail': {'summary_before': summary_before, 'error': str(exc)}}) def _run_admin_controls_validation(self): stats = SystemStatsService.get_stats() binary_dashboard = get_binary_dashboard_data(self.special_distributor) integrity = verify_binary_integrity(bundle_id=self.product.opportunity_bundle_id) self.notes.append( f"Admin validation sampled system stats, binary dashboard data, and bundle integrity. completed_orders={stats['orders']['completed']} integrity_issues={integrity['issues_found']}" ) self.binary_snapshots.append({'label': 'special-distributor-dashboard', 'dashboard': binary_dashboard}) def _run_edge_case_tests(self): client_reference = self._make_reference('DUPLICATE-PURCHASE') dupe_user = self._create_test_user('edge', 1, referred_by=self.special_user) self._ensure_main_wallet_balance(dupe_user, self.product.base_cost + Decimal('1000.00'), label='EDGE-DUPLICATE') first_order, first_dist, _, _ = execute_purchase( user=dupe_user, product=self.product, sponsor_distributor=self.special_distributor, client_reference_id=client_reference, amount=self.product.base_cost, payment_method='main_wallet', ) second_order, second_dist, _, _ = execute_purchase( user=dupe_user, product=self.product, sponsor_distributor=self.special_distributor, client_reference_id=client_reference, amount=self.product.base_cost, payment_method='main_wallet', ) self.edge_case_results.append( { 'name': 'duplicate_purchase_idempotency', 'passed': first_order.id == second_order.id and first_dist.id == second_dist.id, 'detail': {'order_id': str(first_order.id), 'distributor_id': str(first_dist.id)}, } ) self.run_order_ids.update({first_order.id, second_order.id}) self.run_distributor_ids.update({first_dist.id, second_dist.id}) self.run_user_ids.add(dupe_user.id) if self.standalone_product is not None: invalid_user = self._create_test_user('edge', 2, referred_by=self.special_user) self._ensure_main_wallet_balance(invalid_user, self.standalone_product.base_cost + Decimal('1000.00'), label='EDGE-FUND') try: execute_purchase( user=invalid_user, product=self.standalone_product, sponsor_distributor=self.special_distributor, client_reference_id=self._make_reference('INVALID-SPONSOR'), amount=self.standalone_product.base_cost, payment_method='main_wallet', ) self.edge_case_results.append({'name': 'bundle_mismatch_sponsor_prevention', 'passed': False, 'detail': 'Invalid sponsor mismatch purchase unexpectedly succeeded.'}) except Exception as exc: self.edge_case_results.append({'name': 'bundle_mismatch_sponsor_prevention', 'passed': True, 'detail': str(exc)}) def _run_concurrency_tests(self): concurrency_users = [] for index in range(1, self.concurrency_purchases + 1): sponsor = self.special_distributor if index <= 6 else self.root_distributor user = self._create_test_user('conc', index, referred_by=sponsor.user) concurrency_users.append((user, sponsor)) started = time.perf_counter() purchase_results = [] def _purchase_in_thread(user_id, sponsor_id, ordinal): close_old_connections() if connection.vendor == 'sqlite': with connection.cursor() as cursor: cursor.execute('PRAGMA busy_timeout = 15000') user = User.objects.get(pk=user_id) sponsor = DistributorID.objects.get(pk=sponsor_id) self._ensure_main_wallet_balance(user, self.product.base_cost + Decimal('1000.00'), label=f'CONC-{ordinal}') try: order, distributor, _, _ = execute_purchase( user=user, product=self.product, sponsor_distributor=sponsor, client_reference_id=self._make_reference(f'CONC-{ordinal}'), amount=self.product.base_cost, payment_method='main_wallet', ) return {'passed': True, 'order_id': str(order.id), 'distributor_id': str(distributor.id)} except Exception as exc: return {'passed': False, 'error': str(exc)} with ThreadPoolExecutor(max_workers=self.concurrency_workers) as executor: futures = [ executor.submit(_purchase_in_thread, user.id, sponsor.id, ordinal) for ordinal, (user, sponsor) in enumerate(concurrency_users, start=1) ] try: for future in as_completed(futures, timeout=180): purchase_results.append(future.result(timeout=15)) except FuturesTimeoutError: for future in futures: if not future.done(): future.cancel() purchase_results.append({'passed': False, 'error': 'Concurrency purchase execution timed out.'}) duration_seconds = round(time.perf_counter() - started, 2) succeeded = [item for item in purchase_results if item['passed']] for item in succeeded: if item.get('distributor_id'): self.run_distributor_ids.add(item['distributor_id']) integrity_after = verify_binary_integrity(bundle_id=self.product.opportunity_bundle_id) self.concurrency_results.append( { 'name': 'simultaneous_purchases', 'passed': not integrity_after['issues_found'], 'duration_seconds': duration_seconds, 'success_count': len(succeeded), 'failure_count': len(purchase_results) - len(succeeded), 'details': purchase_results[:10], } ) transfer_target = DistributorID.objects.filter(id__in=self.run_distributor_ids).exclude(user=self.special_user).order_by('created_at').first() concurrent_owner = User.objects.filter(id__in=self.run_user_ids).exclude(id=transfer_target.user_id if transfer_target else None).order_by('created_at').first() if transfer_target and concurrent_owner: request = submit_transfer_request( requested_by=concurrent_owner, target_distributor=transfer_target, reason='TEST / NON-PRODUCTION concurrent transfer approval', referral_code=concurrent_owner.referral_code or '', ) def _approve_transfer(request_id): close_old_connections() req = DistributorTransferRequest.objects.get(pk=request_id) history = approve_transfer_request(transfer_request=req, approved_by=self.admin_user, notes='Concurrent approval test') return str(history.id) with ThreadPoolExecutor(max_workers=2) as executor: future_ids = [executor.submit(_approve_transfer, request.id) for _ in range(2)] try: history_ids = [future.result(timeout=20) for future in as_completed(future_ids, timeout=40)] except FuturesTimeoutError: history_ids = ['timeout'] self.concurrency_results.append( { 'name': 'simultaneous_transfer_approvals', 'passed': len(set(history_ids)) == 1, 'details': history_ids, } ) allocation = AutoIDAllocation.objects.filter(owner_user=self.special_user, remaining_count__gte=2).order_by('-created_at').first() if allocation is None: self.concurrency_results.append( { 'name': 'simultaneous_auto_id_activation', 'passed': False, 'details': 'Skipped: no AutoIDAllocation with remaining_count >= 2 is available under current master configuration.', } ) else: def _activate_allocation(allocation_id, ordinal): close_old_connections() local_allocation = AutoIDAllocation.objects.get(pk=allocation_id) return activate_auto_ids(user=self.special_user, allocation=local_allocation, count=1) with ThreadPoolExecutor(max_workers=2) as executor: futures = [executor.submit(_activate_allocation, allocation.id, ordinal) for ordinal in range(1, 3)] activation_results = [] for future in as_completed(futures): try: activation_results.append({'passed': True, 'detail': future.result()}) except Exception as exc: activation_results.append({'passed': False, 'detail': str(exc)}) self.concurrency_results.append( { 'name': 'simultaneous_auto_id_activation', 'passed': any(item['passed'] for item in activation_results), 'details': activation_results, } ) def _pick_sponsor(self, label, index, sponsor_candidates, special_directs): if label == 'small' and index <= 5: return self.special_distributor if label == 'small' and 6 <= index <= 12 and special_directs: return special_directs[(index - 6) % len(special_directs)] preferred_window = sponsor_candidates[: max(8, min(len(sponsor_candidates), 50))] weights = list(range(len(preferred_window), 0, -1)) return self.random.choices(preferred_window, weights=weights, k=1)[0] def _create_test_user(self, scenario, index, *, referred_by): scenario_digits = { 'special': '9', 'small': '1', 'medium': '2', 'large': '3', 'edge': '4', 'conc': '5', } scenario_digit = scenario_digits.get(scenario, '8') attempt = index while True: mobile = f"7{scenario_digit}{self.run_seed:03d}{attempt:05d}" email = f"{self.run_tag.lower()}.{scenario}.{attempt}@sysval.test" existing = User.objects.filter(Q(mobile=mobile) | Q(email=email)).first() if existing is None: break attempt += 1 user = User.objects.create_user( mobile=mobile, email=email, password=self.PASSWORD, login_pin='123456', first_name='TEST', last_name=f'NON-PRODUCTION {scenario.upper()} {attempt}', referred_by=referred_by, is_active=True, is_mobile_verified=True, is_email_verified=True, is_onboarding_complete=True, accept_terms=True, ) self.run_user_ids.add(user.id) return user def _ensure_main_wallet_balance(self, user, required_balance, *, label): required_balance = Decimal(str(required_balance)) wallet, _ = Wallet.objects.get_or_create(user=user) if wallet.main_balance >= required_balance: return wallet topup = required_balance - wallet.main_balance WalletService.credit_main_wallet( user=user, amount=topup, source_type='ADMIN_CREDIT', reference_id=self._make_reference(f'{label}-MAIN'), ) self._log_event('wallet', 'main_balance_topped_up', user_id=str(user.id), topup=str(topup), label=label) return Wallet.objects.get(user=user) def _ensure_supercoin_balance(self, user, required_balance, *, label): required_balance = Decimal(str(required_balance)) wallet, _ = SuperCoinWallet.objects.get_or_create(user=user) if wallet.balance >= required_balance: return wallet topup = required_balance - wallet.balance WalletService.credit_supercoins( user=user, amount=topup, source_type='ADMIN_CREDIT', reference_id=self._make_reference(f'{label}-SUPER'), metadata={'note': 'TEST / NON-PRODUCTION DATA'}, ) self._log_event('wallet', 'supercoin_balance_topped_up', user_id=str(user.id), topup=str(topup), label=label) return SuperCoinWallet.objects.get(user=user) def _ensure_verified_bank_account(self, user): account = BankAccount.objects.filter(user=user, is_active=True, is_verified=True).order_by('-is_primary', '-created_at').first() if account is not None: return account account = BankAccount.objects.create( user=user, account_holder_name=f'{user.first_name or "TEST"} {user.last_name or "USER"}'.strip(), account_number=f'0000{str(user.mobile)[-6:]}', ifsc_code='SBIN0001234', bank_name='State Bank of India', is_verified=True, is_primary=True, ) return account def _execute_wallet_purchase(self, *, user, sponsor, scenario, index, label, use_supercoins=False, super_coin_amount=None): reserve = self.product.base_cost + Decimal('1000.00') self._ensure_main_wallet_balance(user, reserve, label=label) if use_supercoins: self._ensure_supercoin_balance(user, Decimal(str(super_coin_amount or self.product.base_cost)), label=label) purchase_started = time.perf_counter() self._log_event('purchase', 'started', user_id=str(user.id), sponsor_id=str(sponsor.id), scenario=scenario, index=index) order, distributor, _, _ = self._run_db_with_retry( f'execute_purchase_{scenario}_{index}', lambda: execute_purchase( user=user, product=self.product, sponsor_distributor=sponsor, client_reference_id=self._make_reference(f'{scenario}-{index}'), amount=self.product.base_cost, payment_method='main_wallet', use_super_coins=use_supercoins, super_coin_amount=super_coin_amount, ), ) purchase_duration = round(time.perf_counter() - purchase_started, 6) self.purchase_durations.append(purchase_duration) self._record_metric('purchase_pipeline_seconds', purchase_duration, scenario=scenario) # Distributor creation includes company BFS placement in the current runtime path. self._record_metric('bfs_placement_timing_seconds', purchase_duration, scenario=scenario) self._log_event( 'purchase', 'completed', order_id=str(order.id), distributor_id=str(distributor.id), sponsor_id=str(sponsor.id), binary_parent_id=str(distributor.binary_parent_distributor_id) if distributor.binary_parent_distributor_id else None, binary_position=distributor.binary_position, duration_seconds=purchase_duration, ) self.run_order_ids.add(order.id) self.run_distributor_ids.add(distributor.id) self.run_user_ids.add(user.id) return distributor def _consume_async_business(self, distributors): power_stream_records_created = 0 level_rewards_processed = 0 auto_id_allocations_created_before = AutoIDAllocation.objects.count() run_async = self._should_run_async_tasks() if run_async: from apps.business.power_stream.tasks import process_power_stream_bonus_task from apps.business.distributor.tasks import process_auto_id_generation_task for distributor in distributors: process_power_stream_bonus_task.delay(str(distributor.id)) process_auto_id_generation_task.delay(str(distributor.id)) self._log_event('power_stream', 'async_task_queued', distributor_id=str(distributor.id)) self.notes.append('Async task mode enabled: background task queues were used for power-stream and auto-id tasks.') else: for distributor in distributors: started = time.perf_counter() records = process_power_stream_bonus(distributor) duration = round(time.perf_counter() - started, 6) power_stream_records_created += len(records) self._record_metric('power_stream_processing_seconds', duration, distributor_id=str(distributor.id)) self._log_event('power_stream', 'processed_sync', distributor_id=str(distributor.id), records_created=len(records), duration_seconds=duration) compute_and_persist_all(bundle_id=self.product.opportunity_bundle_id, only_stale=False) relevant_ids = set() for distributor in distributors: relevant_ids.add(distributor.id) relevant_ids.update(get_binary_ancestor_ids(distributor)) relevant_distributors = list( DistributorID.objects.filter(id__in=relevant_ids) .select_related('product__opportunity_bundle', 'user') .order_by('global_position') ) for distributor in relevant_distributors: reward_started = time.perf_counter() reward_result = process_level_rewards(distributor) reward_duration = round(time.perf_counter() - reward_started, 6) self._record_metric('reward_processing_seconds', reward_duration, distributor_id=str(distributor.id)) if reward_result.get('levels_rewarded'): level_rewards_processed += len(reward_result['levels_rewarded']) self._log_event('rewards', 'processed', distributor_id=str(distributor.id), levels_rewarded=reward_result.get('levels_rewarded', []), duration_seconds=reward_duration) if not run_async: self._run_auto_id_generation_sync(distributor) auto_id_allocations_created_after = AutoIDAllocation.objects.count() return { 'power_stream_records_created': power_stream_records_created, 'level_rewards_processed': level_rewards_processed, 'auto_id_allocations_created': auto_id_allocations_created_after - auto_id_allocations_created_before, } def _run_auto_id_generation_sync(self, distributor): progress_result = calculate_level_progress(distributor) current_level = int(progress_result.get('current_level') or 0) if current_level <= 0: return {'auto_ids_created': 0, 'total_cost': Decimal('0.00')} levels = OpportunityBundleLevel.objects.filter( opportunity_bundle=distributor.product.opportunity_bundle, level_number__lte=current_level, auto_id_enabled=True, ).order_by('level_number') total_auto_ids = 0 total_cost = Decimal('0.00') for level in levels: result = process_auto_id_generation(distributor, level) total_auto_ids += int(result.get('auto_ids_created') or 0) total_cost += Decimal(str(result.get('total_cost') or Decimal('0.00'))) return {'auto_ids_created': total_auto_ids, 'total_cost': total_cost} def _aggregate_dist(self, distributors, aggregate): ids = [dist.id for dist in distributors] if not ids: return None key = next(iter(aggregate.default_alias.split('__') for _ in [0])) data = DistributorID.objects.filter(id__in=ids).aggregate(value=aggregate) return data['value'] def _build_financial_summary(self): involved_user_ids = set(self.run_user_ids) involved_user_ids.update({self.root_distributor.user_id, self.special_user.id}) involved_dist_ids = set(self.run_distributor_ids) involved_dist_ids.update({self.root_distributor.id, self.special_distributor.id}) ledger_scope = WalletLedger.objects.filter(user_id__in=involved_user_ids, created_at__gte=self.started_at) super_scope = SuperCoinTransaction.objects.filter(user_id__in=involved_user_ids, created_at__gte=self.started_at) order_scope = Order.objects.filter(id__in=self.run_order_ids) pool_scope = PoolEntry.objects.filter(distributor_id__in=involved_dist_ids, created_at__gte=self.started_at) ps_scope = PowerStreamBonusRecord.objects.filter(Q(trigger_distributor_id__in=involved_dist_ids) | Q(earner_id__in=involved_user_ids), created_at__gte=self.started_at) withdrawals_scope = WithdrawalRequest.objects.filter(id__in=self.run_withdrawal_ids) auto_scope = AutoIDAllocation.objects.filter(Q(owner_user_id__in=involved_user_ids) | Q(source_distributor_id__in=involved_dist_ids)) sponsor_payout_total = ledger_scope.filter(source_type='DIRECT_INCENTIVE', transaction_type=WalletLedger.TransactionType.CREDIT).aggregate(total=Sum('amount'))['total'] or Decimal('0.00') level_reward_total = ledger_scope.filter(source_type='LEVEL_REWARD', transaction_type=WalletLedger.TransactionType.CREDIT).aggregate(total=Sum('amount'))['total'] or Decimal('0.00') power_stream_credit_total = ledger_scope.filter(source_type='POWER_STREAM', transaction_type=WalletLedger.TransactionType.CREDIT).aggregate(total=Sum('amount'))['total'] or Decimal('0.00') growth_pool_inflow = pool_scope.aggregate(total=Sum('amount'))['total'] or Decimal('0.00') paid_withdrawals = withdrawals_scope.filter(status=WithdrawalRequest.Status.PAID).aggregate(total=Sum('net_amount'))['total'] or Decimal('0.00') pending_withdrawals = withdrawals_scope.filter(status__in=[WithdrawalRequest.Status.PENDING, WithdrawalRequest.Status.APPROVED]).aggregate(total=Sum('net_amount'))['total'] or Decimal('0.00') pending_power_stream = ps_scope.filter(status=PowerStreamBonusRecord.Status.PENDING).aggregate(total=Sum('net_amount'))['total'] or Decimal('0.00') forfeited_power_stream = ps_scope.filter(status__in=[PowerStreamBonusRecord.Status.FORFEITED, PowerStreamBonusRecord.Status.EXPIRED]).aggregate(total=Sum('net_amount'))['total'] or Decimal('0.00') pending_auto_id_liability = auto_scope.aggregate(total=Sum('auto_id_balance'))['total'] or Decimal('0.00') supercoin_circulation = SuperCoinWallet.objects.filter(user_id__in=involved_user_ids).aggregate(total=Sum('balance'))['total'] or Decimal('0.00') main_totals = Wallet.objects.filter(user_id__in=involved_user_ids).aggregate( main_total=Sum('main_balance'), repurchase_total=Sum('repurchase_balance'), voucher_total=Sum('voucher_balance'), locked_total=Sum('locked_amount'), ) total_wallet_balance = sum(Decimal(str(main_totals.get(key) or Decimal('0.00'))) for key in ('main_total', 'repurchase_total', 'voucher_total')) lock_percentage = Decimal('0.00') if total_wallet_balance > Decimal('0.00'): lock_percentage = (Decimal(str(main_totals.get('locked_total') or Decimal('0.00'))) / total_wallet_balance * Decimal('100')).quantize(Decimal('0.01')) return { 'completed_order_inflow': str(order_scope.filter(status='COMPLETED').aggregate(total=Sum('amount'))['total'] or Decimal('0.00')), 'sponsor_payout_totals': str(sponsor_payout_total), 'level_reward_totals': str(level_reward_total), 'power_stream_totals': str(power_stream_credit_total), 'growth_pool_inflow': str(growth_pool_inflow), 'wallet_lock_percentages': str(lock_percentage), 'super_coin_circulation': str(supercoin_circulation), 'forfeitures': str(forfeited_power_stream), 'withdrawal_outflow': str(paid_withdrawals), 'pending_liabilities': str(pending_withdrawals + pending_power_stream + pending_auto_id_liability), 'pending_withdrawals': str(pending_withdrawals), 'pending_power_stream': str(pending_power_stream), 'pending_auto_id_balance': str(pending_auto_id_liability), 'actual_withdrawals_lt_net_growth_pool': paid_withdrawals < growth_pool_inflow, } def _build_wallet_reconciliation(self): user_ids = set(self.run_user_ids) user_ids.update({self.root_distributor.user_id, self.special_user.id}) ledger_rows = WalletLedger.objects.filter(user_id__in=user_ids).values('user_id', 'wallet_type', 'transaction_type').annotate(total=Sum('amount')) ledger_map = defaultdict(lambda: {'main_credit': Decimal('0.00'), 'main_debit': Decimal('0.00'), 'rep_credit': Decimal('0.00'), 'rep_debit': Decimal('0.00')}) for row in ledger_rows: entry = ledger_map[row['user_id']] amount = Decimal(str(row.get('total') or Decimal('0.00'))) if row['wallet_type'] == WalletLedger.WalletType.EARNINGS: if row['transaction_type'] == WalletLedger.TransactionType.CREDIT: entry['main_credit'] = amount else: entry['main_debit'] = amount elif row['wallet_type'] == WalletLedger.WalletType.REPURCHASE: if row['transaction_type'] == WalletLedger.TransactionType.CREDIT: entry['rep_credit'] = amount else: entry['rep_debit'] = amount mismatches = [] checked = 0 for wallet in Wallet.objects.filter(user_id__in=user_ids): checked += 1 entry = ledger_map[wallet.user_id] main_expected = entry['main_credit'] - entry['main_debit'] rep_expected = entry['rep_credit'] - entry['rep_debit'] if main_expected != wallet.main_balance or rep_expected != wallet.repurchase_balance: mismatches.append( { 'user_id': str(wallet.user_id), 'main_expected': str(main_expected), 'main_balance': str(wallet.main_balance), 'repurchase_expected': str(rep_expected), 'repurchase_balance': str(wallet.repurchase_balance), } ) return { 'checked_wallets': checked, 'mismatch_count': len(mismatches), 'mismatches': mismatches[:50], } def _score_readiness(self, integrity, financial_summary, wallet_summary): score = 100 if integrity['issues_found']: score -= 30 if wallet_summary['mismatch_count']: score -= 20 if not financial_summary['actual_withdrawals_lt_net_growth_pool']: score -= 20 if any('Auto-ID' in risk for risk in self.known_risks): score -= 10 if any('Reward ownership ledger' in risk for risk in self.known_risks): score -= 15 return max(score, 0) def _write_reports(self, integrity, financial_summary, wallet_summary): reports = {} reports['technical_validation_report'] = self._write_text('01_technical_validation_report.md', self._technical_validation_markdown(integrity)) reports['business_validation_report'] = self._write_text('02_business_validation_report.md', self._business_validation_markdown()) reports['financial_simulation_report'] = self._write_text('03_financial_simulation_report.md', self._financial_validation_markdown(financial_summary)) reports['architecture_integrity_report'] = self._write_text('04_architecture_integrity_report.md', self._architecture_integrity_markdown(integrity)) reports['binary_placement_audit'] = self._write_text('05_binary_placement_audit.md', self._binary_placement_markdown(integrity)) reports['transfer_ownership_audit'] = self._write_text('06_transfer_ownership_audit.md', self._transfer_audit_markdown()) reports['wallet_reconciliation_report'] = self._write_text('07_wallet_reconciliation_report.md', self._wallet_reconciliation_markdown(wallet_summary)) reports['stress_test_report'] = self._write_text('08_stress_test_report.md', self._stress_test_markdown()) reports['edge_case_report'] = self._write_text('09_edge_case_report.md', self._edge_case_markdown()) reports['known_risks_and_recommendations'] = self._write_text('10_known_risks_and_recommendations.md', self._risk_markdown()) reports['final_status_summary'] = self._write_text('11_final_status_summary.md', self._final_status_markdown(integrity, financial_summary, wallet_summary)) reports['final_acceptance_report'] = self._write_text('FINAL_ACCEPTANCE_REPORT.md', self._final_acceptance_markdown({ 'run_tag': self.run_tag, 'task_mode': self.task_mode, 'scenarios': self.scenario_results, 'edge_case_results': self.edge_case_results, 'concurrency_results': self.concurrency_results, 'known_risks': self.known_risks, 'runtime_stats': self.runtime_stats, 'execution_metrics': dict(self.execution_metrics), 'validation_summary': self.validation_summary, })) return reports def _technical_validation_markdown(self, integrity): scenario_lines = '\n'.join( f"- {item['label']}: users={item['created_users']} distributors={item['created_distributors']} duration={item['duration_seconds']}s integrity_issues={item['integrity_issues_found']}" for item in self.scenario_results ) return ( '# Technical Validation Report\n\n' 'TEST / NON-PRODUCTION DATA\n\n' f'- Run tag: {self.run_tag}\n' f'- Started at: {self.started_at.isoformat()}\n' f'- Bundle product: {self.product.name} ({self.product.id})\n' f'- Company root: {self.root_distributor.distributor_code}\n' f'- Special mobile: {self.special_mobile}\n\n' '## Scenario Results\n' f'{scenario_lines}\n\n' '## Final Integrity Snapshot\n' f"- Issues found: {integrity['issues_found']}\n" f"- Invalid BFS placements: {len(integrity['bfs_inconsistencies'])}\n" f"- Duplicate child slots: {len(integrity['duplicate_child_slots'])}\n" f"- Cycles: {len(integrity['cycles'])}\n" ) def _business_validation_markdown(self): lines = ['# Business Validation Report', '', 'TEST / NON-PRODUCTION DATA', '', '## Validated Flows'] lines.extend( [ '- User registration with referral linkage', '- Wallet-funded purchase lifecycle', '- Manual payment review lifecycle', '- Sponsor income crediting', '- Company-root BFS placement', '- Level progression snapshots and reward attempts', '- Withdrawal request, approval, and payout', '- Transfer approval with ownership period rollover', '- Power stream claim attempt and duplicate-claim prevention', ] ) lines.append('') lines.append('## Acceptance Notes') for result in self.edge_case_results: status = 'PASS' if result['passed'] else 'FAIL' lines.append(f"- {status}: {result['name']} -> {result['detail']}") return '\n'.join(lines) + '\n' def _financial_validation_markdown(self, financial_summary): return ( '# Financial Simulation Report\n\n' 'TEST / NON-PRODUCTION DATA\n\n' '## Totals\n' f"- Completed order inflow: {financial_summary['completed_order_inflow']}\n" f"- Sponsor payout totals: {financial_summary['sponsor_payout_totals']}\n" f"- Level reward totals: {financial_summary['level_reward_totals']}\n" f"- Power stream totals: {financial_summary['power_stream_totals']}\n" f"- Growth pool inflow: {financial_summary['growth_pool_inflow']}\n" f"- Withdrawal outflow: {financial_summary['withdrawal_outflow']}\n" f"- Pending liabilities: {financial_summary['pending_liabilities']}\n" f"- Super coin circulation: {financial_summary['super_coin_circulation']}\n" f"- Wallet lock percentages: {financial_summary['wallet_lock_percentages']}\n\n" '## Control Check\n' f"- Actual withdrawals < Net growth pool: {financial_summary['actual_withdrawals_lt_net_growth_pool']}\n\n" '## Wallet Flow Diagram\n' '```mermaid\n' 'flowchart LR\n' ' Purchases[Completed Purchases] --> Direct[Direct Incentive Credits]\n' ' Purchases --> Pool[Growth Pool Entries]\n' ' Purchases --> Levels[Level Reward Credits]\n' ' Purchases --> Power[Power Stream Pending]\n' ' Power --> Claims[Monthly Claims]\n' ' Claims --> MainWallet[Main Wallet]\n' ' MainWallet --> Withdrawals[Withdrawal Requests / Paid Outflow]\n' '```\n' ) def _architecture_integrity_markdown(self, integrity): return ( '# Architecture Integrity Report\n\n' 'TEST / NON-PRODUCTION DATA\n\n' '## Separation Layers\n' '- Referral layer: user referral_code / referred_by\n' '- Sponsor genealogy layer: sponsor_distributor\n' '- Company binary layer: binary_parent_distributor + binary_position\n' '- Reward propagation layer: binary ancestry for level rewards; sponsor ancestry for sponsor income and power stream only\n' '- Ownership resolution layer: DistributorOwnershipPeriod / DistributorQualificationState\n\n' '## Integrity Results\n' f"- Cycles: {len(integrity['cycles'])}\n" f"- Duplicate child slots: {len(integrity['duplicate_child_slots'])}\n" f"- Orphan nodes: {len(integrity['orphan_nodes'])}\n" f"- Invalid BFS placements: {len(integrity['bfs_inconsistencies'])}\n" f"- Sponsor/Binary mismatches (informational): {len(integrity['sponsor_binary_mismatches'])}\n\n" '## Sponsor vs Binary Illustration\n' '```mermaid\n' 'flowchart TD\n' f" Root[{self.root_distributor.distributor_code}] --> Binary1[Company BFS Child]\n" f" Sponsor[{self.special_distributor.distributor_code}] --> DirectA[Sponsor Direct]\n" ' Binary1 -. can differ from sponsor genealogy .- DirectA\n' '```\n' ) def _binary_placement_markdown(self, integrity): snapshot_block = '' if self.binary_snapshots: tree = self.binary_snapshots[0].get('tree') snapshot_block = self._tree_to_mermaid(tree) return ( '# Binary Placement Audit\n\n' 'TEST / NON-PRODUCTION DATA\n\n' '## Placement Checks\n' f"- Invalid BFS placements: {len(integrity['bfs_inconsistencies'])}\n" f"- Duplicate slots: {len(integrity['duplicate_child_slots'])}\n" f"- Missing parent references: {len(integrity['missing_parent_references'])}\n" f"- Multiple roots: {len(integrity['multiple_roots'])}\n\n" '## Binary Snapshot\n' f'{snapshot_block}\n' ) def _transfer_audit_markdown(self): histories = DistributorTransferHistory.objects.filter(id__in=self.transfer_history_ids).select_related('distributor', 'old_owner_user', 'new_owner_user').order_by('approved_at') lines = ['# Transfer Ownership Audit', '', 'TEST / NON-PRODUCTION DATA', '', '## Transfer Histories'] for history in histories: periods = list(DistributorOwnershipPeriod.objects.filter(distributor=history.distributor).order_by('ownership_effective_from')) lines.append( f"- Distributor {history.distributor.distributor_code}: {history.old_owner_user_id} -> {history.new_owner_user_id} at {history.approved_at.isoformat()} (periods={len(periods)})" ) if periods: lines.append('```mermaid') lines.append('timeline') lines.append(f" title Ownership Timeline for {history.distributor.distributor_code}") for index, period in enumerate(periods, start=1): start = period.ownership_effective_from.isoformat() end = period.ownership_effective_to.isoformat() if period.ownership_effective_to else 'CURRENT' lines.append(f" Period {index} : owner {period.owner_user_id} from {start} to {end}") lines.append('```') return '\n'.join(lines) + '\n' def _wallet_reconciliation_markdown(self, wallet_summary): lines = ['# Wallet Reconciliation Report', '', 'TEST / NON-PRODUCTION DATA', '', '## Summary'] lines.append(f"- Checked wallets: {wallet_summary['checked_wallets']}") lines.append(f"- Mismatch count: {wallet_summary['mismatch_count']}") if wallet_summary['mismatch_count']: lines.append('') lines.append('## Mismatches') for mismatch in wallet_summary['mismatches']: lines.append(f"- {mismatch}") return '\n'.join(lines) + '\n' def _stress_test_markdown(self): lines = ['# Stress Test Report', '', 'TEST / NON-PRODUCTION DATA'] for result in self.concurrency_results: status = 'PASS' if result['passed'] else 'FAIL' lines.append(f"- {status}: {result['name']} -> {result['details']}") return '\n'.join(lines) + '\n' def _edge_case_markdown(self): lines = ['# Edge Case Report', '', 'TEST / NON-PRODUCTION DATA'] for result in self.edge_case_results: status = 'PASS' if result['passed'] else 'FAIL' lines.append(f"- {status}: {result['name']} -> {result['detail']}") return '\n'.join(lines) + '\n' def _risk_markdown(self): lines = ['# Known Risks & Recommendations', '', 'TEST / NON-PRODUCTION DATA', '', '## Risks'] if not self.known_risks: lines.append('- No material runtime risks were detected during this validation run.') else: lines.extend(f'- {risk}' for risk in self.known_risks) lines.append('') lines.append('## Recommendations') lines.append('- Keep the strict binary integrity command in release checklists.') lines.append('- Run the full validation command after any reward, transfer, or wallet logic change.') lines.append('- Fill missing auto-ID master configuration before treating auto-ID activation as production ready.') lines.append('- Add ownership-ledger writes if transfer-era reward attribution must be auditable post-factum.') return '\n'.join(lines) + '\n' def _final_status_markdown(self, integrity, financial_summary, wallet_summary): safe_areas = [] unsafe_areas = [] if not integrity['issues_found']: safe_areas.append('Company-root FIFO/BFS placement integrity') safe_areas.append('Binary subtree structure and slot uniqueness') else: unsafe_areas.append('Binary integrity') if wallet_summary['mismatch_count'] == 0: safe_areas.append('Wallet ledger vs stored balance reconciliation for checked wallets') else: unsafe_areas.append('Wallet reconciliation mismatches') if financial_summary['actual_withdrawals_lt_net_growth_pool']: safe_areas.append('Withdrawal outflow vs growth pool control') else: unsafe_areas.append('Withdrawal outflow exceeds growth pool control threshold') if any('Auto-ID' in risk for risk in self.known_risks): unsafe_areas.append('Auto-ID configuration readiness') if any('Reward ownership ledger' in risk for risk in self.known_risks): unsafe_areas.append('Transfer reward ownership auditability') readiness_score = self._score_readiness(integrity, financial_summary, wallet_summary) return ( '# Final Status Summary\n\n' 'TEST / NON-PRODUCTION DATA\n\n' f'- Production readiness score: {readiness_score}/100\n' f"- Validation summary: passed={self.validation_summary.get('total_passed', 0)} partial={self.validation_summary.get('partial_passes', 0)} warnings={self.validation_summary.get('warnings', 0)} failures={self.validation_summary.get('failures', 0)}\n" f"- SAFE AREAS: {', '.join(safe_areas) if safe_areas else 'None'}\n" f"- UNSAFE AREAS: {', '.join(unsafe_areas) if unsafe_areas else 'None'}\n" '- Architecture readiness: strict company binary enforced; simulation executed against real purchase lifecycle\n' '- Scalability concerns: local SQLite concurrency remains a limiting environment for high-contention tests\n' '- Unresolved coupling: none found in binary placement runtime; sponsor/binary mismatches are expected informational artifacts\n' '- Financial risks: see financial simulation report and wallet reconciliation report\n' '- Business logic gaps: auto-ID master mapping and transfer reward ownership ledger auditability\n' ) def _final_acceptance_markdown(self, summary): validation = summary.get('validation_summary') or {} runtime = summary.get('runtime_stats') or {} scenarios = summary.get('scenarios') or [] lines = [ '# FINAL ACCEPTANCE REPORT', '', 'TEST / NON-PRODUCTION DATA', '', '## Run Context', f"- Run tag: {summary.get('run_tag')}", f"- Task mode: {summary.get('task_mode')}", f"- Report directory: {self.report_dir}", '', '## Validation Summary', f"- Total passed: {validation.get('total_passed', 0)}", f"- Partial passes: {validation.get('partial_passes', 0)}", f"- Warnings: {validation.get('warnings', 0)}", f"- Failures: {validation.get('failures', 0)}", ] unresolved = validation.get('unresolved_risks') or [] lines.append('') lines.append('## Unresolved Risks') if unresolved: lines.extend(f'- {risk}' for risk in unresolved) else: lines.append('- None') lines.append('') lines.append('## Scenario Outcomes') for item in scenarios: lines.append( f"- {item.get('label')}: passed={item.get('passed')} size={item.get('size')} duration={item.get('duration_seconds')} error={item.get('error')}" ) lines.append('') lines.append('## Runtime & Throughput') lines.append(f"- Purchase count: {runtime.get('purchase_count', 0)}") lines.append(f"- Purchase throughput/sec: {runtime.get('purchase_throughput_per_second', 0)}") lines.append(f"- Purchase avg seconds: {runtime.get('purchase_avg_seconds', 0)}") lines.append(f"- Tracemalloc current bytes: {runtime.get('tracemalloc_current_bytes', 0)}") lines.append(f"- Tracemalloc peak bytes: {runtime.get('tracemalloc_peak_bytes', 0)}") metrics = summary.get('execution_metrics') or {} lines.append('') lines.append('## Timing Metrics') for metric_name in ('purchase_pipeline_seconds', 'bfs_placement_timing_seconds', 'reward_processing_seconds', 'transfer_processing_seconds'): entries = metrics.get(metric_name, []) if not entries: lines.append(f'- {metric_name}: no samples') continue avg_s = round(sum(item.get('seconds', 0.0) for item in entries) / max(len(entries), 1), 6) lines.append(f'- {metric_name}: samples={len(entries)} avg_seconds={avg_s}') lines.append('') lines.append('## Conclusion') lines.append('- This phase focused on reliability, correctness, reproducibility, reporting, observability, and runtime consistency only.') return '\n'.join(lines) + '\n' def _tree_to_mermaid(self, tree): if not tree: return 'No tree snapshot available.' lines = ['```mermaid', 'flowchart TD'] queue = deque([tree]) seen = set() while queue: node = queue.popleft() if not node: continue node_id = node['id'] if node_id in seen: continue seen.add(node_id) label = f"{node['distributor_code']}\\nGP {node.get('global_position')}" lines.append(f" {self._node_key(node_id)}[{label}]") for side in ('left', 'right'): child = node.get(side) if child: lines.append(f" {self._node_key(node_id)} -->|{side.upper()}| {self._node_key(child['id'])}") queue.append(child) lines.append('```') return '\n'.join(lines) def _node_key(self, node_id): return f"N{str(node_id).replace('-', '')[:10]}" def _make_reference(self, label): return f"{self.run_tag}-{label}-{uuid4().hex[:8].upper()}" def _write_text(self, name, content): path = self.report_dir / name path.write_text(content, encoding='utf-8') return str(path) def _write_json(self, name, content): path = self.report_dir / name path.write_text(json.dumps(content, indent=2, default=str), encoding='utf-8') return str(path)