"""Tests for the Phase 2 SP1 onboarding fixes. Coverage: * FRSME-OBR012 — RIMS prefill copies institution / phone into the entrepreneur profile when the user did not override. * FRSME-OBR029 — revoking a verified business cascades into SP2 application and SP4 deal-room teardown, with one notification per affected record. * FRSME-OBR032 — Award resolver finds a single match, returns ``None`` when no awards exist, and raises AmbiguousAwardMatchError on multi-match. * FRSME-OBR037 — M&E officers receive a SMEHUB_NEW_TRACKING_RECORD on the first baseline initialisation. * Verification token TTL is settings-driven. * Admin backend register reuses an existing inactive shell user instead of raising DuplicateEmailError. """ from __future__ import annotations from datetime import timedelta from decimal import Decimal import pytest from django.contrib.auth import get_user_model from django.test import override_settings from django.utils import timezone from apps.core.notifications.models import Notification from apps.core.permissions.roles import UserRole from apps.smehub.incubation.models import Application, CohortMember, Programme from apps.smehub.onboarding import services from apps.smehub.onboarding.models import ( Business, BusinessBaseline, BusinessStage, EntrepreneurProfile, RegistrationRecord, RevenueRange, ) User = get_user_model() pytestmark = pytest.mark.django_db # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def admin(): return User.objects.create_user( email="adm-p2@example.com", password="x", is_staff=True, is_superuser=True, role=UserRole.SYSTEM_ADMIN, ) @pytest.fixture def user_with_record(db): record = services.register_user( full_name="Phase Two", email="p2-user@example.com", password="StrongPass1!", country="UG", organisation="From-Form U", phone="+25670000", role=RegistrationRecord.Role.ENTREPRENEUR, ) user = services.verify_email(token=record.token) return user, record # --------------------------------------------------------------------------- # FRSME-OBR012 — RIMS prefill # --------------------------------------------------------------------------- def test_rims_prefill_helper_returns_empty_dict_when_no_match(): """When the wizard didn't store a RIMS FK, the helper must return an empty dict so the fallback chain in ``submit_entrepreneur_profile`` treats it as "no prefill available".""" assert services._rims_prefill_for(application_id=None, scholar_id=None) == {} def test_rims_prefill_helper_returns_empty_dict_when_id_unknown(): """A stale ID — e.g. an entrepreneur picked a match but the row was since deleted — must not raise; the helper just returns nothing.""" assert services._rims_prefill_for(application_id=999_999_999) == {} def test_submit_entrepreneur_profile_falls_back_to_rims_when_user_blank( user_with_record, monkeypatch, ): """When the user leaves a field blank, the RIMS prefill fills it in. To avoid building a full RIMS Application (FundingRecord + GrantCall + Application) just for this assertion, we monkeypatch the prefill helper AND patch the FK assignment path so no ``rims_match_application_id`` is stored on the profile (which would trigger an invalid-FK constraint at teardown). This test verifies the *prefill copy logic*; the FK-storage behaviour is covered by ``test_rims_prefill_helper_returns_empty_dict_*``. """ monkeypatch.setattr( services, "_rims_prefill_for", lambda *, application_id=None, scholar_id=None: { "organisation": "RIMS-Inst", "phone": "+25699999", }, ) user, _record = user_with_record EntrepreneurProfile.objects.filter(user=user).update(organisation="", phone="") profile = services.submit_entrepreneur_profile( user=user, country="", organisation="", phone="", # Pass nothing — the helper is patched to return prefill regardless, # and we avoid storing an invalid FK. ) # Even with no rims_match_*_id, the helper monkey-patch fires unconditionally # so prefill values flow through to the profile. assert profile.organisation == "RIMS-Inst" assert profile.phone == "+25699999" def test_submit_entrepreneur_profile_user_value_wins_over_rims_prefill( user_with_record, monkeypatch, ): """User-typed organisation must override the RIMS-derived one.""" monkeypatch.setattr( services, "_rims_prefill_for", lambda *, application_id=None, scholar_id=None: { "organisation": "RIMS-Inst", "country": "UG", "phone": "+25699999", }, ) user, _record = user_with_record profile = services.submit_entrepreneur_profile( user=user, country="UG", organisation="USER-TYPED", phone="+25600000", ) assert profile.organisation == "USER-TYPED" assert profile.phone == "+25600000" # --------------------------------------------------------------------------- # FRSME-OBR032 — Award resolver # --------------------------------------------------------------------------- def test_award_resolver_returns_none_when_no_award(user_with_record): user, _ = user_with_record result = services.resolve_rims_award_for_origin( email=user.email, programme_of_origin="Anything", ) assert result is None # --------------------------------------------------------------------------- # FRSME-OBR037 — MEL officer notification on baseline lock # --------------------------------------------------------------------------- @pytest.fixture def verified_entrepreneur(user_with_record, admin): user, _record = user_with_record profile = EntrepreneurProfile.objects.get(user=user) profile.verify(by_user=admin) profile.save() return profile def test_mel_officer_notified_on_first_baseline(verified_entrepreneur, admin): """When ``smehub_baseline_initialised`` fires for the first time, every M&E officer + SME admin receives a SMEHUB_NEW_TRACKING_RECORD.""" mel_officer = User.objects.create_user( email="mel-officer@example.com", password="x", role=UserRole.MEL_OFFICER, ) biz = Business.objects.create( entrepreneur=verified_entrepreneur, name="Tracking Co", sector="Agriculture", business_stage=BusinessStage.MVP, country="UG", ) baseline = BusinessBaseline.objects.create( business=biz, business_stage_at_entry=BusinessStage.MVP, fte_count=1, pte_count=0, revenue_range=RevenueRange.UNDER_1K, primary_target_market="Local", geographic_reach="UG", ) biz.verify(by_user=admin) biz.save() from apps.smehub.onboarding import signals as onb_signals onb_signals.smehub_baseline_initialised.send( sender=BusinessBaseline, business=biz, baseline=baseline, entrepreneur=verified_entrepreneur, ) notes = Notification.objects.filter( recipient=mel_officer, verb=Notification.Verb.SMEHUB_NEW_TRACKING_RECORD, ) assert notes.count() == 1 assert "Tracking Co" in notes.first().message # --------------------------------------------------------------------------- # Token TTL settings-driven # --------------------------------------------------------------------------- @override_settings(SMEHUB_VERIFICATION_TOKEN_TTL_HOURS=2) def test_verification_token_ttl_is_settings_driven(): record = services.register_user( full_name="Short TTL", email="short-ttl@example.com", password="StrongPass1!", country="UG", organisation="X", phone="+25670001", role=RegistrationRecord.Role.ENTREPRENEUR, ) delta = record.token_expires_at - record.created_at assert timedelta(hours=1, minutes=55) < delta < timedelta(hours=2, minutes=5) # --------------------------------------------------------------------------- # Admin backend register adopts an inactive shell user # --------------------------------------------------------------------------- def test_admin_backend_register_reuses_inactive_shell(admin): """A previous self-register left an inactive shell ``User`` row via ``record_recipient_proxy``. The admin path must adopt it instead of raising DuplicateEmailError (audit fix).""" services.register_user( full_name="Stale User", email="stale@example.com", password="StrongPass1!", country="UG", organisation="X", phone="+25670002", role=RegistrationRecord.Role.ENTREPRENEUR, ) # User row exists but is inactive (verification not consumed) shell = User.objects.get(email="stale@example.com") assert shell.is_active is False user = services.admin_backend_register( admin_user=admin, email="stale@example.com", full_name="Adopted User", role=RegistrationRecord.Role.ENTREPRENEUR, ) assert user.is_active is True assert user.email == "stale@example.com" assert user.first_name == "Adopted" # Stale registration record was consumed so resend won't pick it up. assert RegistrationRecord.objects.filter( email__iexact="stale@example.com", consumed_at__isnull=True, ).count() == 0 def test_admin_backend_register_still_blocks_active_collision(admin): """An *active* user with the same email is a real duplicate.""" User.objects.create_user( email="active@example.com", password="x", is_active=True, role=UserRole.ENTREPRENEUR, ) with pytest.raises(services.DuplicateEmailError): services.admin_backend_register( admin_user=admin, email="active@example.com", full_name="Conflict User", role=RegistrationRecord.Role.ENTREPRENEUR, ) # --------------------------------------------------------------------------- # OBR029 cascade # --------------------------------------------------------------------------- @pytest.fixture def verified_business_with_application(verified_entrepreneur, admin): biz = Business.objects.create( entrepreneur=verified_entrepreneur, name="Cascade Co", sector="Agriculture", business_stage=BusinessStage.MVP, country="UG", ) BusinessBaseline.objects.create( business=biz, business_stage_at_entry=BusinessStage.MVP, fte_count=1, pte_count=0, revenue_range=RevenueRange.UNDER_1K, primary_target_market="Local", geographic_reach="UG", ) biz.verify(by_user=admin) biz.save() # Trigger baseline init so SP2 access is allowed (FRSME-MEI002) from apps.smehub.onboarding import signals as onb_signals onb_signals.smehub_baseline_initialised.send( sender=BusinessBaseline, business=biz, baseline=biz.baseline, entrepreneur=verified_entrepreneur, ) # Build a published programme officer = User.objects.create_user( email="po-cascade@example.com", password="x", role=UserRole.PROGRAMME_OFFICER, ) from apps.smehub.incubation import services as inc_services p = inc_services.create_programme( officer=officer, title="Cascade Prog", slug="cascade-prog", target_sectors=["Agriculture"], target_stages=["mvp"], cohort_size=5, duration_weeks=10, application_deadline=timezone.now() + timedelta(days=30), ) inc_services.add_rubric_section(p, title="A", weight=Decimal("1.0"), max_marks=10) p.judges.set([User.objects.create_user(email="p2fix-roster-judge@example.com", password="x", role=UserRole.JUDGE)]) inc_services.publish_programme(p, by_user=officer) p = Programme.objects.get(pk=p.pk) app = inc_services.start_application( entrepreneur=verified_entrepreneur, business=biz, programme=p, ) return biz, app def test_revoke_business_withdraws_open_sp2_applications( verified_business_with_application, admin, ): """OBR029 cascade: a revoked business causes its open SP2 applications to be withdrawn (audit fix).""" biz, app = verified_business_with_application assert app.status == Application.Status.DRAFT services.revoke_business(biz, by_user=admin, reason="Fraud") app = Application.objects.get(pk=app.pk) assert app.status == Application.Status.WITHDRAWN # Entrepreneur received a cascade notification per affected record. cascade_notes = Notification.objects.filter( recipient=biz.entrepreneur.user, message__icontains="was withdrawn because the underlying business profile was revoked", ) assert cascade_notes.count() >= 1