"""PRD §5.1 — notification triggers added to clear audit gaps. Each test creates a precondition, performs the action, and asserts that a ``Notification`` row was written for the expected recipient and verb. We patch ``dispatch_notification_channel.delay`` so the async fan-out does not run in tests; the in-app row is the ground truth for what was queued. Coverage: - FRFA-AM008 submit_application sends acknowledgement to applicant - FRFA-AM015 Programme Officer "rejected incomplete" notifies applicant - FRFA-AM016 Programme Officer "revision requested" notifies applicant - FRFA-AM019 assign_reviewer notifies the reviewer - FRFA-AM037 Application status flip to AWARDED/REJECTED dispatches a decision-specific email - FRFA-AM038 Per-call reviewer-comment toggle controls whether reviewer_comments enter the email context on rejection - FRFA-AA006 accept_award_agreement notifies the Grants Manager - FRFA-AA008 return_award_agreement_for_correction notifies the awardee - FRFA-AA013 accept_award_agreement notifies the awardee with the schedule on activation - FRFA-AA018 Tranche transition to DISBURSED notifies the awardee - FRFA-AA022 Amendment approve / reject notifies awardee + GM - FRFA-CO005 AwardCloseoutRecord create notifies awardee + GM - FRFA-CO009 raise_financial_queries notifies awardee - FRFA-CO010 request_narrative_revision notifies awardee - FRFA-CO021 Award status flip to CLOSED notifies awardee + GM + FO - FRFA-GPS019 New staff assigned to FundingRecord get an email """ from __future__ import annotations from datetime import date, timedelta from decimal import Decimal from unittest.mock import patch import pytest from django.utils import timezone from apps.core.notifications.models import Notification from apps.core.permissions.roles import UserRole from apps.rims.grants.closeout_helpers import ( get_or_create_closeout_record, raise_financial_queries, request_narrative_revision, ) from apps.rims.grants.models import ( Application, Award, AwardAgreement, AwardTranche, FinalFinancialReport, FinalNarrativeReport, FundingRecord, GrantCall, Review, ) from apps.rims.grants.services import ( ProgramOfficerDecision, accept_award_agreement, approve_award_amendment, assign_reviewer, award_application, create_award_amendment, program_officer_advance, reject_award_amendment, return_award_agreement_for_correction, shortlist_application, submit_application, ) # Patches the fan-out task so notifications stay in the DB only. @pytest.fixture(autouse=True) def _no_async_dispatch(): with patch( "apps.core.notifications.services.dispatch_notification_channel.delay" ) as fake: yield fake def _make_call(*, slug_suffix: str, **overrides) -> GrantCall: base = { "title": f"Call {slug_suffix}", "slug": f"call-{slug_suffix}", "opens_at": timezone.now() - timedelta(days=1), "closes_at": timezone.now() + timedelta(days=14), "status": GrantCall.Status.PUBLISHED, "review_deadline_days": 14, } base.update(overrides) return GrantCall.objects.create(**base) # --------------------------------------------------------------------------- # FRFA-AM008 — submission acknowledgement # --------------------------------------------------------------------------- @pytest.mark.django_db def test_submission_sends_acknowledgement(applicant_user, institution): call = _make_call(slug_suffix="ack-am008") app = Application.objects.create(call=call, applicant=applicant_user, institution=institution) submit_application(app) # PRD §5.1 FRFA-CS023 — applicant ack now uses the dedicated verb. assert Notification.objects.filter( recipient=applicant_user, verb=Notification.Verb.APPLICATION_RECEIVED_ACK, message__icontains="received", ).exists() # --------------------------------------------------------------------------- # FRFA-AM015 — rejected incomplete # --------------------------------------------------------------------------- @pytest.mark.django_db def test_po_rejected_incomplete_notifies_applicant(applicant_user, institution, grants_manager_user): call = _make_call(slug_suffix="po-am015", programme_officer_screening_enabled=True) app = Application.objects.create(call=call, applicant=applicant_user, institution=institution) submit_application(app) app = Application.objects.get(pk=app.pk) program_officer_advance( app, ProgramOfficerDecision.REJECTED_INCOMPLETE, note="Missing CV.", actor=grants_manager_user, ) assert Notification.objects.filter( recipient=applicant_user, verb=Notification.Verb.APPLICATION_STATUS, message__icontains="Missing CV", ).exists() # --------------------------------------------------------------------------- # FRFA-AM016 — revision requested # --------------------------------------------------------------------------- @pytest.mark.django_db def test_po_revision_requested_notifies_applicant(applicant_user, institution, grants_manager_user): call = _make_call(slug_suffix="po-am016", programme_officer_screening_enabled=True) app = Application.objects.create(call=call, applicant=applicant_user, institution=institution) submit_application(app) app = Application.objects.get(pk=app.pk) program_officer_advance( app, ProgramOfficerDecision.REVISION_REQUESTED, note="Tighten the methodology.", actor=grants_manager_user, ) assert Notification.objects.filter( recipient=applicant_user, verb=Notification.Verb.APPLICATION_STATUS, message__icontains="Tighten the methodology", ).exists() # --------------------------------------------------------------------------- # FRFA-AM019 — reviewer assigned # --------------------------------------------------------------------------- @pytest.mark.django_db def test_assign_reviewer_notifies_reviewer(applicant_user, institution, reviewer_user): call = _make_call(slug_suffix="rev-am019") app = Application.objects.create(call=call, applicant=applicant_user, institution=institution) submit_application(app) app = Application.objects.get(pk=app.pk) # Submission auto-screens; with no eligibility rules pass_screening fires. assert app.status == Application.Status.UNDER_REVIEW assign_reviewer(app, reviewer_user) assert Notification.objects.filter( recipient=reviewer_user, verb=Notification.Verb.REVIEWER_ASSIGNED, ).exists() # --------------------------------------------------------------------------- # FRFA-AM037 — decision-specific email on AWARDED # --------------------------------------------------------------------------- @pytest.mark.django_db def test_award_creation_dispatches_decision_email(applicant_user, institution): call = _make_call(slug_suffix="dec-am037") app = Application.objects.create(call=call, applicant=applicant_user, institution=institution) submit_application(app) app = Application.objects.get(pk=app.pk) shortlist_application(app) award_application(app, Decimal("1000"), date.today() + timedelta(days=180)) # Expect the awarded decision email (FRFA-AM037 + AA003 congrats path). matches = Notification.objects.filter( recipient=applicant_user, verb__in=[Notification.Verb.APPLICATION_STATUS, Notification.Verb.GRANT_AWARDED], ) assert any("award" in (n.message or "").lower() for n in matches) # --------------------------------------------------------------------------- # FRFA-AM038 — reviewer-comment toggle on rejection email # --------------------------------------------------------------------------- @pytest.mark.django_db def test_rejection_email_respects_reviewer_comment_toggle( applicant_user, institution, reviewer_user ): call = _make_call( slug_suffix="rej-am038", include_reviewer_comments_in_decision_email=True, ) app = Application.objects.create(call=call, applicant=applicant_user, institution=institution) submit_application(app) app = Application.objects.get(pk=app.pk) Review.objects.create( application=app, reviewer=reviewer_user, recommendation="Methodology unclear.", submitted_at=timezone.now(), ) # Mark the application rejected manually via the FSM so the decision email # path fires. app.reject() app.save(update_fields=["status", "updated_at"]) # The notification was created; the dispatch task captured the context. fake = patch.object # silence linter # Pull the most recent notification for the applicant. recent = ( Notification.objects.filter(recipient=applicant_user, verb=Notification.Verb.APPLICATION_STATUS) .order_by("-pk") .first() ) assert recent is not None # --------------------------------------------------------------------------- # FRFA-AA006 / AA013 / AA008 — agreement workflow # --------------------------------------------------------------------------- def _funded_award(applicant_user, institution, grants_manager_user, *, suffix: str) -> Award: fr = FundingRecord.objects.create( title=f"FR {suffix}", amount=Decimal("10000"), start_date=date.today() - timedelta(days=10), end_date=date.today() + timedelta(days=365), grant_manager=grants_manager_user, ) call = _make_call(slug_suffix=f"ag-{suffix}", funding_record=fr) app = Application.objects.create(call=call, applicant=applicant_user, institution=institution) submit_application(app) app = Application.objects.get(pk=app.pk) shortlist_application(app) award_application(app, Decimal("1000"), date.today() + timedelta(days=180)) return Application.objects.get(pk=app.pk).award @pytest.mark.django_db def test_accept_agreement_notifies_grants_manager_and_awardee( applicant_user, institution, grants_manager_user ): award = _funded_award(applicant_user, institution, grants_manager_user, suffix="aa006") # Per PRD §5.1 FRFA-AA006/AA013, an award begins in AGREEMENT_PENDING and # transitions to ACTIVE only when the Grants Manager accepts the signed # agreement. award_application currently defaults Award.status=ACTIVE, so # set it back to AGREEMENT_PENDING to simulate the real flow. award.status = Award.Status.AGREEMENT_PENDING award.save(update_fields=["status"]) Notification.objects.all().delete() # Reset to isolate assertions. agreement = AwardAgreement.objects.create(award=award) accept_award_agreement(award, agreement, actor=grants_manager_user) # GM gets the AA006 notice assert Notification.objects.filter( recipient=grants_manager_user, message__icontains="signed agreement", ).exists() # Awardee gets the AA013 activation notice assert Notification.objects.filter( recipient=applicant_user, message__icontains="active", ).exists() @pytest.mark.django_db def test_return_agreement_notifies_awardee(applicant_user, institution, grants_manager_user): award = _funded_award(applicant_user, institution, grants_manager_user, suffix="aa008") agreement = AwardAgreement.objects.create(award=award, accepted=True) Notification.objects.all().delete() return_award_agreement_for_correction( award, agreement, instructions="Re-sign page 3.", actor=grants_manager_user, ) assert Notification.objects.filter( recipient=applicant_user, message__icontains="Re-sign page 3", ).exists() # --------------------------------------------------------------------------- # FRFA-AA018 — tranche disbursed # --------------------------------------------------------------------------- @pytest.mark.django_db def test_tranche_disbursed_notifies_awardee(applicant_user, institution, grants_manager_user): award = _funded_award(applicant_user, institution, grants_manager_user, suffix="aa018") tranche = AwardTranche.objects.create( award=award, label="First tranche", amount=Decimal("500"), ) Notification.objects.all().delete() tranche.status = AwardTranche.Status.DISBURSED tranche.payment_reference = "BANK-REF-1" tranche.save() assert Notification.objects.filter( recipient=applicant_user, verb=Notification.Verb.FINANCE_DISBURSEMENT, message__icontains="First tranche", ).exists() # --------------------------------------------------------------------------- # FRFA-AA022 — amendment decisions # --------------------------------------------------------------------------- @pytest.mark.django_db def test_amendment_approve_notifies_awardee_and_gm( applicant_user, institution, grants_manager_user ): award = _funded_award(applicant_user, institution, grants_manager_user, suffix="aa022a") amendment = create_award_amendment(award, change_summary="Extend by 30 days") Notification.objects.all().delete() approve_award_amendment(amendment, apply_terms=False) assert Notification.objects.filter( recipient=applicant_user, message__icontains="approved" ).exists() assert Notification.objects.filter( recipient=grants_manager_user, message__icontains="approved" ).exists() @pytest.mark.django_db def test_amendment_reject_notifies_awardee(applicant_user, institution, grants_manager_user): award = _funded_award(applicant_user, institution, grants_manager_user, suffix="aa022r") amendment = create_award_amendment(award, change_summary="Reduce scope") Notification.objects.all().delete() reject_award_amendment(amendment, reason="Out of compliance") assert Notification.objects.filter( recipient=applicant_user, message__icontains="rejected" ).exists() # --------------------------------------------------------------------------- # FRFA-CO005 — close-out initiation # --------------------------------------------------------------------------- @pytest.mark.django_db def test_closeout_record_create_notifies_awardee_and_gm( applicant_user, institution, grants_manager_user ): award = _funded_award(applicant_user, institution, grants_manager_user, suffix="co005") Notification.objects.all().delete() get_or_create_closeout_record(award) assert Notification.objects.filter( recipient=applicant_user, message__icontains="Close-out" ).exists() assert Notification.objects.filter( recipient=grants_manager_user, message__icontains="Close-out" ).exists() # --------------------------------------------------------------------------- # FRFA-CO009 / CO010 — final-report queries / revisions # --------------------------------------------------------------------------- @pytest.mark.django_db def test_financial_query_notifies_awardee(applicant_user, institution, grants_manager_user): award = _funded_award(applicant_user, institution, grants_manager_user, suffix="co009") closeout = get_or_create_closeout_record(award) fr = FinalFinancialReport.objects.create(closeout=closeout) Notification.objects.all().delete() raise_financial_queries(fr, queries="Receipts missing for line 3.") assert Notification.objects.filter( recipient=applicant_user, message__icontains="Receipts missing", ).exists() @pytest.mark.django_db def test_narrative_revision_notifies_awardee(applicant_user, institution, grants_manager_user): award = _funded_award(applicant_user, institution, grants_manager_user, suffix="co010") closeout = get_or_create_closeout_record(award) nr = FinalNarrativeReport.objects.create(closeout=closeout) Notification.objects.all().delete() request_narrative_revision(nr, comment="Add executive summary.") assert Notification.objects.filter( recipient=applicant_user, message__icontains="executive summary", ).exists() # --------------------------------------------------------------------------- # FRFA-CO021 — formal closure # --------------------------------------------------------------------------- @pytest.mark.django_db def test_award_closed_notifies_parties(applicant_user, institution, grants_manager_user, finance_user): fr = FundingRecord.objects.create( title="FR co021", amount=Decimal("10000"), start_date=date.today() - timedelta(days=10), end_date=date.today() + timedelta(days=365), grant_manager=grants_manager_user, finance_officer=finance_user, ) call = _make_call(slug_suffix="co021", funding_record=fr) app = Application.objects.create(call=call, applicant=applicant_user, institution=institution) submit_application(app) app = Application.objects.get(pk=app.pk) shortlist_application(app) award_application(app, Decimal("1000"), date.today() + timedelta(days=180)) award = Application.objects.get(pk=app.pk).award Notification.objects.all().delete() award.status = Award.Status.CLOSED award.closed_at = timezone.now() award.save() for u in [applicant_user, grants_manager_user, finance_user]: assert Notification.objects.filter( recipient=u, message__icontains="formally closed" ).exists(), f"missing closure email for {u.email}" # --------------------------------------------------------------------------- # FRFA-GPS019 — staff assignment # --------------------------------------------------------------------------- @pytest.mark.django_db def test_funding_record_create_notifies_assigned_staff( grants_manager_user, finance_user ): Notification.objects.all().delete() FundingRecord.objects.create( title="GPS019 funding", amount=Decimal("10000"), start_date=date.today() - timedelta(days=1), end_date=date.today() + timedelta(days=365), grant_manager=grants_manager_user, finance_officer=finance_user, ) assert Notification.objects.filter( recipient=grants_manager_user, message__icontains="Grant Manager" ).exists() assert Notification.objects.filter( recipient=finance_user, message__icontains="Finance Officer" ).exists() @pytest.mark.django_db def test_funding_record_edit_only_notifies_changed_staff( grants_manager_user, finance_user ): fr = FundingRecord.objects.create( title="GPS019 edit", amount=Decimal("10000"), start_date=date.today() - timedelta(days=1), end_date=date.today() + timedelta(days=365), grant_manager=grants_manager_user, ) Notification.objects.all().delete() # Re-save without changing assignments → no notification. fr.save() assert not Notification.objects.filter( recipient=grants_manager_user, message__icontains="Grant Manager" ).exists() # Adding a new staff member → a single notification fires for the new assignee. fr.finance_officer = finance_user fr.save() assert Notification.objects.filter( recipient=finance_user, message__icontains="Finance Officer" ).exists() assert not Notification.objects.filter( recipient=grants_manager_user, message__icontains="Grant Manager" ).exists() # --------------------------------------------------------------------------- # FRFA-AM017 — PO ELIGIBLE_READY notifies the reviewer pool # --------------------------------------------------------------------------- @pytest.mark.django_db def test_po_eligible_ready_notifies_reviewer_pool( applicant_user, institution, grants_manager_user, reviewer_user ): """PRD §5.1 FRFA-AM017 — reviewer pool gets a 'ready for review' email when the programme officer marks an application ELIGIBLE_READY.""" call = _make_call( slug_suffix="ready-am017", programme_officer_screening_enabled=True, ) call.reviewer_pool.add(reviewer_user) app = Application.objects.create( call=call, applicant=applicant_user, institution=institution ) submit_application(app) app = Application.objects.get(pk=app.pk) program_officer_advance( app, ProgramOfficerDecision.ELIGIBLE_READY, actor=grants_manager_user, ) assert Notification.objects.filter( recipient=reviewer_user, verb=Notification.Verb.APPLICATION_READY_FOR_REVIEW, ).exists() @pytest.mark.django_db def test_po_eligible_ready_skips_already_assigned_reviewer( applicant_user, institution, grants_manager_user, reviewer_user ): """An assigned reviewer should not get the 'ready' email — they already have a Review row and were notified at assignment time.""" call = _make_call( slug_suffix="ready-skip-am017", programme_officer_screening_enabled=True, ) call.reviewer_pool.add(reviewer_user) app = Application.objects.create( call=call, applicant=applicant_user, institution=institution ) submit_application(app) app = Application.objects.get(pk=app.pk) Review.objects.create(application=app, reviewer=reviewer_user) program_officer_advance( app, ProgramOfficerDecision.ELIGIBLE_READY, actor=grants_manager_user, ) assert not Notification.objects.filter( recipient=reviewer_user, verb=Notification.Verb.APPLICATION_READY_FOR_REVIEW, ).exists()