"""Tests for the daily reward and commission pipeline.""" from __future__ import annotations from datetime import timedelta from decimal import Decimal from django.core.management import call_command from django.test import TestCase from django.utils import timezone from apps.investments.cron_tasks import DistributeRewardsTask from apps.investments.models import Investment, Reward from apps.referrals.models import Commission from apps.transactions.models import Transaction from apps.users.models import User class CalculateRewardsCommandTests(TestCase): def setUp(self) -> None: """Create a referral chain with one downline and three uplines.""" self.upline1 = User.objects.create_user( telegram_user_id=1001, first_name="U1", username="u1", ) self.upline2 = User.objects.create_user( telegram_user_id=1002, first_name="U2", username="u2", referred_by=self.upline1, ) self.upline3 = User.objects.create_user( telegram_user_id=1003, first_name="U3", username="u3", referred_by=self.upline2, ) self.downline = User.objects.create_user( telegram_user_id=2001, first_name="Downline", username="downline", referred_by=self.upline3, ) tier_info = Investment.get_tier_info(Decimal("1000")) now = timezone.now() self.investment = Investment.objects.create( user=self.downline, amount=Decimal("1000"), tier=tier_info["tier"], daily_reward_rate=tier_info["rate"], duration_days=tier_info["duration"], status=Investment.STATUS_PENDING, start_date=now - timedelta(days=2), end_date=now + timedelta(days=10), ) def test_creates_reward_commissions_and_transactions(self): """Command credits reward to investor and commissions up the chain.""" target_date = (self.investment.start_date + timedelta(days=1)).date() call_command("calculate_rewards", date=target_date.isoformat()) DistributeRewardsTask(target_date).execute(target_date) reward = Reward.objects.get(investment=self.investment, reward_date=target_date) self.assertEqual(reward.amount, Decimal("20.00000000")) commissions = Commission.objects.filter(reward=reward).order_by("level") self.assertEqual(commissions.count(), 3) self.assertListEqual( [c.level for c in commissions], [1, 2, 3], ) self.assertListEqual( [c.amount for c in commissions], [ Decimal("10.00000000"), # 50% of 20 Decimal("6.00000000"), # 30% of 20 Decimal("4.00000000"), # 20% of 20 ], ) # Balances updated for all parties (reward + 3 commissions). self.downline.refresh_from_db() self.upline1.refresh_from_db() self.upline2.refresh_from_db() self.upline3.refresh_from_db() self.assertEqual(self.downline.credit_balance, Decimal("20.00000000")) self.assertEqual(self.upline1.credit_balance, Decimal("4.00000000")) self.assertEqual(self.upline2.credit_balance, Decimal("6.00000000")) self.assertEqual(self.upline3.credit_balance, Decimal("10.00000000")) # Transactions recorded with correct types. transactions = Transaction.objects.order_by("created_at") self.assertEqual(transactions.count(), 4) self.assertEqual( [t.transaction_type for t in transactions], [ Transaction.TYPE_REWARD, Transaction.TYPE_COMMISSION, Transaction.TYPE_COMMISSION, Transaction.TYPE_COMMISSION, ], ) # Investment status should flip to active (not yet completed). self.investment.refresh_from_db() self.assertEqual(self.investment.status, Investment.STATUS_ACTIVE) def test_idempotent_for_existing_reward(self): """Running the command twice should not duplicate rewards or commissions.""" target_date = (self.investment.start_date + timedelta(days=1)).date() call_command("calculate_rewards", date=target_date.isoformat()) DistributeRewardsTask(target_date).execute(target_date) call_command("calculate_rewards", date=target_date.isoformat()) DistributeRewardsTask(target_date).execute(target_date) self.assertEqual( Reward.objects.filter(investment=self.investment, reward_date=target_date).count(), 1, ) self.assertEqual(Commission.objects.count(), 3) self.assertEqual(Transaction.objects.count(), 4) # Balances remain the same after the second run. self.downline.refresh_from_db() self.upline1.refresh_from_db() self.upline2.refresh_from_db() self.upline3.refresh_from_db() self.assertEqual(self.downline.credit_balance, Decimal("20.00000000")) self.assertEqual(self.upline1.credit_balance, Decimal("4.00000000")) self.assertEqual(self.upline2.credit_balance, Decimal("6.00000000")) self.assertEqual(self.upline3.credit_balance, Decimal("10.00000000")) def test_skips_before_first_reward_day(self): """No reward/commission is created on the investment start date.""" target_date = self.investment.start_date.date() call_command("calculate_rewards", date=target_date.isoformat()) self.assertFalse( Reward.objects.filter(investment=self.investment, reward_date=target_date).exists() ) self.assertEqual(Commission.objects.count(), 0) self.assertEqual(Transaction.objects.count(), 0) def test_marks_investment_completed_on_end_date(self): """Status flips to completed when target_date reaches end_date.""" target_date = self.investment.end_date.date() call_command("calculate_rewards", date=target_date.isoformat()) reward = Reward.objects.get(investment=self.investment, reward_date=target_date) self.assertEqual(reward.amount, Decimal("20.00000000")) self.investment.refresh_from_db() self.assertEqual(self.investment.status, Investment.STATUS_COMPLETED) def test_skips_completed_investment_status(self): """Completed investments are ignored even if the date is in range.""" target_date = (self.investment.start_date + timedelta(days=1)).date() self.investment.status = Investment.STATUS_COMPLETED self.investment.save(update_fields=["status"]) call_command("calculate_rewards", date=target_date.isoformat()) self.assertFalse( Reward.objects.filter(investment=self.investment, reward_date=target_date).exists() ) self.assertEqual(Commission.objects.count(), 0) self.assertEqual(Transaction.objects.count(), 0) def test_creates_reward_without_commissions_when_no_upline(self): """Rewards still post when the investor has no referrer; no commissions.""" target_date = (timezone.now() + timedelta(days=5)).date() self.investment.status = Investment.STATUS_COMPLETED self.investment.save(update_fields=["status"]) solo_user = User.objects.create_user( telegram_user_id=3001, first_name="Solo", username="solo" ) tier_info = Investment.get_tier_info(Decimal("100")) solo_investment = Investment.objects.create( user=solo_user, amount=Decimal("100"), tier=tier_info["tier"], daily_reward_rate=tier_info["rate"], duration_days=tier_info["duration"], status=Investment.STATUS_PENDING, start_date=timezone.now(), end_date=timezone.now() + timedelta(days=30), ) call_command("calculate_rewards", date=target_date.isoformat()) DistributeRewardsTask(target_date).execute(target_date) reward = Reward.objects.get(investment=solo_investment, reward_date=target_date) self.assertEqual(reward.amount, Decimal("1.00000000")) self.assertEqual(Commission.objects.count(), 0) transactions = Transaction.objects.filter(user=solo_user) self.assertEqual(transactions.count(), 1) self.assertEqual(transactions.first().transaction_type, Transaction.TYPE_REWARD) solo_user.refresh_from_db() self.assertEqual(solo_user.credit_balance, Decimal("1.00000000")) def test_partial_referral_chain_only_credits_available_uplines(self): """Only existing uplines receive commissions when the chain is shorter than 3.""" target_date = (timezone.now() + timedelta(days=7)).date() self.investment.status = Investment.STATUS_COMPLETED self.investment.save(update_fields=["status"]) upline = User.objects.create_user( telegram_user_id=4001, first_name="Top", username="top" ) downline = User.objects.create_user( telegram_user_id=4002, first_name="Down", username="down", referred_by=upline ) tier_info = Investment.get_tier_info(Decimal("1000")) short_chain_investment = Investment.objects.create( user=downline, amount=Decimal("1000"), tier=tier_info["tier"], daily_reward_rate=tier_info["rate"], duration_days=tier_info["duration"], status=Investment.STATUS_PENDING, start_date=timezone.now(), end_date=timezone.now() + timedelta(days=20), ) call_command("calculate_rewards", date=target_date.isoformat()) DistributeRewardsTask(target_date).execute(target_date) reward = Reward.objects.get(investment=short_chain_investment, reward_date=target_date) self.assertEqual(reward.amount, Decimal("20.00000000")) commissions = Commission.objects.filter(reward=reward) self.assertEqual(commissions.count(), 1) commission = commissions.first() self.assertEqual(commission.level, 1) self.assertEqual(commission.amount, Decimal("10.00000000")) self.assertEqual(Transaction.objects.count(), 2) upline.refresh_from_db() downline.refresh_from_db() self.assertEqual(upline.credit_balance, Decimal("10.00000000")) self.assertEqual(downline.credit_balance, Decimal("20.00000000")) def test_quantizes_reward_and_commissions_round_half_up(self): """Reward and commission amounts are rounded to 8 decimal places.""" target_date = (timezone.now() + timedelta(days=9)).date() self.investment.status = Investment.STATUS_COMPLETED self.investment.save(update_fields=["status"]) precise_user = User.objects.create_user( telegram_user_id=5001, first_name="Precise", username="precise" ) precise_investment = Investment.objects.create( user=precise_user, amount=Decimal("1234.56789012"), tier=1, daily_reward_rate=Decimal("1.25"), # custom rate to exercise rounding duration_days=30, status=Investment.STATUS_PENDING, start_date=timezone.now(), end_date=timezone.now() + timedelta(days=40), ) call_command("calculate_rewards", date=target_date.isoformat()) DistributeRewardsTask(target_date).execute(target_date) reward = Reward.objects.get(investment=precise_investment, reward_date=target_date) self.assertEqual(reward.amount, Decimal("15.43209863")) commissions = Commission.objects.filter(reward=reward).order_by("level") # No uplines, so no commissions. self.assertEqual(commissions.count(), 0) precise_user.refresh_from_db() self.assertEqual(precise_user.credit_balance, Decimal("15.43209863"))