@patch('communicate.mailgun_api.MailgunAPI.send_email') @patch('billing.notifications.emails.emit_task_limit_event') def test_handle_plan_limits_with_workspaces(self, mock_emit_event, mock_send_email): """ Test that emails are sent and events are emitted when a workspace reaches 90% of its task limit 1. Create an HQ account with a non-metered plan 2. Create a workspace account with the HQ account as parent 3. Record 90 tasks for the workspace (90% of 100 task limit) 4. Call handle_plan_limits and verify email and event are emitted """ # Freeze time at a specific date for testing frozen_time = datetime(2025, 5, 4, tzinfo=pytz.UTC) with freeze_time(frozen_time): user, customuser, account = self.create_user('test-hq@example.com') # Create a plan group and non-metered plan with 100 task limit plan_group = PlanGroup.objects.create(name='Test Plan Group', default=True) plan = Plan.objects.create( name='Test Non-metered Plan', amount=0, actions=100, zaps=5, current_default=True, polling_interval_minutes=5, metered_task_pricing=False, ) plan_group.plans.add(plan) account.plan_group = plan_group account.switch_plan(plan) account.trial_end_date = None account.save() account = Account.objects.get(id=account.id) # Create a workspace with the account as parent workspace = create_workspace( account.owner, account, account.owner.id, field_values={"name": "test workspace 1"}, ) # Verify the workspace account has 100 included tasks self.assertEqual(workspace.get_included_tasks_per_month(), 100) # Create a node for the workspace for testing read_node, write_node = self.create_nodes_multi( customuser, account=workspace ) read_node.paused = False read_node.subscription_claim_id = str(uuid.uuid4()) read_node.save() # Record 90 tasks for the workspace (90% of the limit) workspace.record_task_for_customuser(customuser, amount=90) handle_plan_limits(write_node) # Verify the warning email was sent mock_send_email.assert_called_once() args, kwargs = mock_send_email.call_args self.assertEqual(kwargs['recipient_email'], workspace.owner.email) self.assertEqual(kwargs['template_name'], "task_limit_warning") self.assertEqual( kwargs['mailgun_variables']['billable_tasks_this_period'], 90 ) self.assertEqual(kwargs['mailgun_variables']['action_limit'], 100) # No event is emitted at 90% threshold for workspaces mock_emit_event.assert_not_called()