"""Tests for the individual cron tasks (R001, D001).""" from datetime import date, timedelta from decimal import Decimal from django.test import TestCase from django.utils import timezone from django_cronjob_utils.models import CronExecution from apps.investments.cron_tasks import ( CalculateRewardsTask, DistributeRewardsTask, CALCULATE_TASK_CODE, DISTRIBUTE_TASK_CODE, ) 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 CronTasksTest(TestCase): def setUp(self): # Create users self.upline = User.objects.create_user( telegram_user_id=1001, first_name="Upline", username="upline" ) self.downline = User.objects.create_user( telegram_user_id=2001, first_name="Downline", username="downline", referred_by=self.upline, ) now = timezone.now() # Create investment starting 2 days ago self.start_date = now - timedelta(days=2) tier_info = Investment.get_tier_info(Decimal("1000")) 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_ACTIVE, start_date=self.start_date, end_date=self.start_date + timedelta(days=120), ) # Target date for calculation/distribution # Investment on Day 0. First reward on Day 1. # We start_date = now - 2 days. # So we can calculate for now - 1 day (Yesterday). self.target_date = self.start_date.date() + timedelta(days=1) def test_calculate_task_creates_pending_rewards(self): """R001 should create rewards but NOT distribute them.""" task = CalculateRewardsTask(self.target_date) result = task.run() # Use .run() to simulate real cron execution (logs to DB) self.assertTrue(result.success) # Verify Reward created reward = Reward.objects.get(investment=self.investment, reward_date=self.target_date) self.assertEqual(reward.amount, Decimal("20.00000000")) # 2% of 1000 self.assertIsNone(reward.distributed_at) # Should be pending # Verify Investment Status self.investment.refresh_from_db() self.assertEqual(self.investment.status, Investment.STATUS_ACTIVE) # Verify No Commission yet self.assertFalse(Commission.objects.exists()) # Verify No Transaction yet self.assertFalse(Transaction.objects.exists()) # Verify CronExecution record execution = CronExecution.objects.get( task_code=CALCULATE_TASK_CODE, execution_date=self.target_date ) self.assertTrue(execution.success) def test_distribute_task_skips_if_calculate_not_done(self): """D001 should fail/skip if R001 hasn't run successfully for the same date.""" task = DistributeRewardsTask(self.target_date) # We can check should_run() or run() # run() calls should_run() internally and handles the skipping logic result = task.run() # It should skip because R001 record is missing self.assertTrue(result.skipped) # We can't easily check log buffer without internals, but skipped=True implies the check passed. # In current implementation, should_run returns False, so run() returns skipped=True. def test_distribute_task_distributes_rewards_after_calculation(self): """D001 should distribute rewards if R001 success record exists.""" # 1. Run Calculate Task calc_task = CalculateRewardsTask(self.target_date) calc_task.run() # Verify pending reward exists reward = Reward.objects.get(investment=self.investment, reward_date=self.target_date) self.assertIsNone(reward.distributed_at) # 2. Run Distribute Task dist_task = DistributeRewardsTask(self.target_date) result = dist_task.run() self.assertTrue(result.success) self.assertFalse(result.skipped) # Verify Reward Distributed reward.refresh_from_db() self.assertIsNotNone(reward.distributed_at) # Verify Transaction Created (Reward) tx = Transaction.objects.filter( user=self.downline, transaction_type=Transaction.TYPE_REWARD ).first() self.assertIsNotNone(tx) self.assertEqual(tx.amount, reward.amount) # Verify Commission Created & Distributed comm = Commission.objects.filter(reward=reward).first() self.assertIsNotNone(comm) self.assertEqual(comm.level, 1) self.assertEqual(comm.upline_user, self.upline) self.assertEqual(comm.status, Commission.STATUS_DISTRIBUTED) # Verify Commission Transaction ctx = Transaction.objects.filter( user=self.upline, transaction_type=Transaction.TYPE_COMMISSION ).first() self.assertIsNotNone(ctx) def test_calculate_task_idempotency(self): """R001 should not duplicate rewards if run multiple times.""" task = CalculateRewardsTask(self.target_date) task.run() # Run again CronExecution.objects.all().delete() # Clear history to allow re-run logic check # (Though CronTask checks history, we simulating force run or fresh run behavior for logic) # Actually CalculateRewardsTask uses get_or_create, so it handles duplication at model level too. task.execute(self.target_date) # Call execute directly to bypass CronExecution check self.assertEqual(Reward.objects.count(), 1) def test_distribute_task_idempotency(self): """D001 should not re-distribute already distributed rewards.""" # Setup: Calculate and Distribute CalculateRewardsTask(self.target_date).run() DistributeRewardsTask(self.target_date).run() # Verify state reward = Reward.objects.first() self.assertIsNotNone(reward.distributed_at) initial_tx_count = Transaction.objects.count() # Run Distribute again (simulate force run / bypass cron check) task = DistributeRewardsTask(self.target_date) CronExecution.objects.filter(task_code=DISTRIBUTE_TASK_CODE).delete() # allow run result = task.run() # Should process 0 rewards self.assertIn("Distributed 0 rewards", result.message) self.assertEqual(Transaction.objects.count(), initial_tx_count) def test_investment_completion_logic(self): """Investment should be marked pending -> active -> completed.""" # Setup investment that ends on target_date short_inv = Investment.objects.create( user=self.downline, amount=Decimal("100"), tier=1, daily_reward_rate=1.0, duration_days=1, status=Investment.STATUS_PENDING, start_date=self.target_date - timedelta(days=1), # Started yesterday end_date=self.target_date, # Ends today ) # Run Calculate for target_date (today, which is end_date) # It's day 1 of investment. Created yesterday. # target_date = end_date. task = CalculateRewardsTask(self.target_date) task.run() short_inv.refresh_from_db() self.assertEqual(short_inv.status, Investment.STATUS_COMPLETED) # Reward should exist self.assertTrue( Reward.objects.filter(investment=short_inv, reward_date=self.target_date).exists() )