"""Tests for SME-Hub investment services (SP5). Coverage targets the key PRD acceptance points: • FRSME-INV001–003 — investor directory + verification FSM • FRSME-INV004–005 — funding call publish + eligible-entrepreneur fan-out • FRSME-INV006–007 — funding application FSM (draft → submitted → decided) • FRSME-INV008–009 — investor matchmaking refresh idempotency • FRSME-INV010 — readiness flag flips when thresholds satisfied • FRSME-INV011–012 — performance snapshot + dashboard share creates access log • FRSME-INV013–015 — investor connection request → accept opens a SP4 DealRoom • FRSME-INV016 — manual funding record creates audit log + emits funding_secured signal • FRSME-INV017 — domain signals fire on every milestone """ from __future__ import annotations from datetime import timedelta from decimal import Decimal import pytest from django.contrib.auth import get_user_model from django.utils import timezone from apps.core.audit.models import AuditLog from apps.core.notifications.models import Notification from apps.core.permissions.roles import UserRole from apps.smehub.investment import services from apps.smehub.investment import signals as inv_signals from apps.smehub.investment.models import ( DashboardAccessLog, FundingApplication, FundingCall, InvestmentReadiness, Investor, InvestorConnectionRequest, InvestorMatch, ManualFundingRecord, SMEPerformanceSnapshot, ) from apps.smehub.onboarding.models import ( Business, BusinessBaseline, BusinessStage, EntrepreneurProfile, ) from apps.smehub.showcasing.models import DealRoom User = get_user_model() pytestmark = pytest.mark.django_db # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def admin(): return User.objects.create_user( email="sp5-admin@example.com", password="x", is_superuser=True, is_staff=True, role=UserRole.SYSTEM_ADMIN, ) @pytest.fixture def entrepreneur(admin): user = User.objects.create_user( email="sp5-ent@example.com", password="x", role=UserRole.ENTREPRENEUR, ) profile = EntrepreneurProfile.objects.create(user=user, country="KE") profile.verify(by_user=admin) profile.save() return profile @pytest.fixture def verified_business(entrepreneur, admin): biz = Business.objects.create( entrepreneur=entrepreneur, name="AgriCo", sector="agriculture", business_stage=BusinessStage.MVP, country="KE", ) biz.verify(by_user=admin) biz.save() return biz @pytest.fixture def baseline(verified_business): return BusinessBaseline.objects.create( business=verified_business, business_stage_at_entry=BusinessStage.MVP, ) @pytest.fixture def investor_user(): return User.objects.create_user( email="sp5-inv@example.com", password="x", role=UserRole.INVESTOR, ) @pytest.fixture def verified_investor(admin, investor_user): inv = services.create_investor( by_user=admin, org_name="Acme Capital", type=Investor.Type.VC, investment_focus=["agriculture", "agritech"], geographic_preference=["KE", "UG"], ticket_size_min=Decimal("10000"), ticket_size_max=Decimal("250000"), contact_email="acme@example.com", contact_user=investor_user, auto_verify=True, ) return Investor.objects.get(pk=inv.pk) @pytest.fixture def published_call(admin, verified_investor): call = services.create_funding_call( by_user=admin, title="Smallholder Resilience Round", type=FundingCall.Type.SEED, application_deadline=timezone.now() + timedelta(days=30), target_sectors=["agriculture"], target_stages=[BusinessStage.MVP], target_countries=["KE"], amount_min=Decimal("5000"), amount_max=Decimal("25000"), investor=verified_investor, ) return services.publish_funding_call(call, by_user=admin) # --------------------------------------------------------------------------- # Investor directory (FRSME-INV001–003) # --------------------------------------------------------------------------- def test_create_investor_lands_pending_when_self_registered(admin, investor_user): inv = services.register_investor_self( by_user=investor_user, org_name="Self Reg Capital", type=Investor.Type.ANGEL, contact_email=investor_user.email, ) inv = Investor.objects.get(pk=inv.pk) assert inv.verification_status == Investor.Status.PENDING assert inv.contact_user == investor_user assert AuditLog.objects.filter( target_app="smehub_investment", target_model="investor", action="CREATE", ).exists() def test_verify_investor_transitions_pending_to_verified(admin): inv = services.create_investor( by_user=admin, org_name="Pending Co", type=Investor.Type.ANGEL, ) services.verify_investor(inv, by_user=admin) inv = Investor.objects.get(pk=inv.pk) assert inv.verification_status == Investor.Status.VERIFIED assert inv.verified_at is not None def test_reject_investor_records_reason(admin): inv = services.create_investor( by_user=admin, org_name="Reject Co", type=Investor.Type.VC, ) services.reject_investor(inv, by_user=admin, reason="No track record.") inv = Investor.objects.get(pk=inv.pk) assert inv.verification_status == Investor.Status.REJECTED assert inv.rejection_reason == "No track record." def test_deactivate_then_reactivate_round_trip(admin): inv = services.create_investor( by_user=admin, org_name="Cycle Co", type=Investor.Type.DFI, auto_verify=True, ) inv = Investor.objects.get(pk=inv.pk) services.deactivate_investor(inv, by_user=admin, reason="Pause") inv = Investor.objects.get(pk=inv.pk) assert inv.verification_status == Investor.Status.DEACTIVATED services.reactivate_investor(inv, by_user=admin) inv = Investor.objects.get(pk=inv.pk) assert inv.verification_status == Investor.Status.VERIFIED # --------------------------------------------------------------------------- # Funding calls (FRSME-INV004–005) # --------------------------------------------------------------------------- def test_publish_funding_call_notifies_eligible_entrepreneur(admin, verified_business, verified_investor): """Eligible entrepreneur receives notification on publish (FRSME-INV005).""" Notification.objects.filter(recipient=verified_business.entrepreneur.user).delete() call = services.create_funding_call( by_user=admin, title="Targeted call", type=FundingCall.Type.SEED, application_deadline=timezone.now() + timedelta(days=30), target_sectors=["agriculture"], target_stages=[BusinessStage.MVP], target_countries=["KE"], ) services.publish_funding_call(call, by_user=admin) assert Notification.objects.filter( recipient=verified_business.entrepreneur.user, verb=Notification.Verb.SMEHUB_FUNDING_CALL_PUBLISHED, ).exists() def test_publish_only_works_from_draft(admin): call = services.create_funding_call( by_user=admin, title="One", type=FundingCall.Type.SEED, application_deadline=timezone.now() + timedelta(days=10), ) services.publish_funding_call(call, by_user=admin) call = FundingCall.objects.get(pk=call.pk) with pytest.raises(services.InvestmentStateError): services.publish_funding_call(call, by_user=admin) # --------------------------------------------------------------------------- # Funding applications (FRSME-INV006–007) # --------------------------------------------------------------------------- def test_funding_application_full_fsm(admin, entrepreneur, verified_business, published_call): application = services.start_funding_application( funding_call=published_call, entrepreneur=entrepreneur, business=verified_business, ) assert application.status == FundingApplication.Status.DRAFT services.auto_save_funding_application(application, payload={"pitch_summary": "draft text"}) application = FundingApplication.objects.get(pk=application.pk) assert application.auto_saved_payload == {"pitch_summary": "draft text"} application.funding_request_amount = Decimal("12000") application.pitch_summary = "We sell millet varieties." application.save() services.submit_funding_application(application, by_user=entrepreneur.user) application = FundingApplication.objects.get(pk=application.pk) assert application.status == FundingApplication.Status.SUBMITTED services.decide_funding_application( application, action="accept", by_user=admin, note="Yes", ) application = FundingApplication.objects.get(pk=application.pk) assert application.status == FundingApplication.Status.ACCEPTED def test_decision_emits_funding_secured_signal(admin, entrepreneur, verified_business, published_call): received = [] def _handler(sender, **kwargs): received.append(kwargs) inv_signals.smehub_funding_secured.connect(_handler) try: application = services.start_funding_application( funding_call=published_call, entrepreneur=entrepreneur, business=verified_business, ) application.funding_request_amount = Decimal("15000") application.save() services.submit_funding_application(application, by_user=entrepreneur.user) application = FundingApplication.objects.get(pk=application.pk) services.decide_funding_application( application, action="accept", by_user=admin, ) finally: inv_signals.smehub_funding_secured.disconnect(_handler) assert any(r.get("source") == "application" for r in received) def test_application_blocked_after_deadline(admin, entrepreneur, verified_business, verified_investor): call = services.create_funding_call( by_user=admin, title="Deadline Past", type=FundingCall.Type.SEED, application_deadline=timezone.now() + timedelta(days=1), ) services.publish_funding_call(call, by_user=admin) # Push the deadline to the past via raw update to simulate timeout. FundingCall.objects.filter(pk=call.pk).update( application_deadline=timezone.now() - timedelta(days=1), ) call = FundingCall.objects.get(pk=call.pk) with pytest.raises(services.InvestmentStateError): services.start_funding_application( funding_call=call, entrepreneur=entrepreneur, business=verified_business, ) # --------------------------------------------------------------------------- # Performance snapshot (FRSME-INV011–012) # --------------------------------------------------------------------------- def test_snapshot_aggregates_cross_sp_data(entrepreneur, verified_business, baseline): snapshot = services.compile_performance_snapshot(entrepreneur, trigger="test") assert snapshot.payload["sp1"]["verified_businesses"] == 1 assert snapshot.payload["sp1"]["has_baseline"] is True assert snapshot.payload["sp5"]["funding_events"] == 0 def test_snapshot_is_idempotent_and_versioned(entrepreneur, verified_business): s1 = services.compile_performance_snapshot(entrepreneur, trigger="t1") v1 = s1.version s2 = services.compile_performance_snapshot(entrepreneur, trigger="t2") assert s2.pk == s1.pk assert s2.version == v1 + 1 # --------------------------------------------------------------------------- # Readiness (FRSME-INV010) # --------------------------------------------------------------------------- def test_readiness_flag_starts_false_for_new_entrepreneur(entrepreneur): services.compile_performance_snapshot(entrepreneur) record = services.compute_readiness(entrepreneur) assert record.is_ready is False assert any(a.get("action") for a in record.recommended_actions) def test_readiness_flag_flips_when_thresholds_relaxed(entrepreneur, verified_business, baseline): """Override thresholds to require only baseline + business — flag flips.""" services.compile_performance_snapshot(entrepreneur) record = services.compute_readiness( entrepreneur, thresholds={ "min_partnerships": 0, "min_showcase_or_catalogue": 0, "min_funding_events": 0, }, ) assert record.is_ready is True def test_readiness_uses_active_threshold_config(entrepreneur, verified_business, baseline): """FRSME-INV010 — compute_readiness reads the admin-configured thresholds (no per-call override needed).""" from apps.smehub.investment.models import ReadinessThresholdConfig ReadinessThresholdConfig.objects.create( name="Relaxed", min_partnerships=0, min_showcase_or_catalogue=0, min_funding_events=0, is_active=True, ) services.compile_performance_snapshot(entrepreneur) record = services.compute_readiness(entrepreneur) assert record.is_ready is True def test_structured_eligibility_min_staff_filters(admin, verified_business, baseline): """FRSME-INV004 — a min-full-time-staff criterion narrows the match.""" call = services.create_funding_call( by_user=admin, title="Staffed call", type=FundingCall.Type.SEED, application_deadline=timezone.now() + timedelta(days=30), target_sectors=["agriculture"], min_full_time_staff=5, ) user = verified_business.entrepreneur.user # baseline fte_count defaults to 0 → below the bar. assert user not in services._eligible_entrepreneur_users(call) baseline.fte_count = 6 baseline.save() assert user in services._eligible_entrepreneur_users(call) # --------------------------------------------------------------------------- # Investor matchmaking (FRSME-INV008–009) # --------------------------------------------------------------------------- def test_match_refresh_is_idempotent(entrepreneur, verified_business, verified_investor): services.compile_performance_snapshot(entrepreneur) n1 = services.refresh_matches_for(entrepreneur, trigger="t1") n2 = services.refresh_matches_for(entrepreneur, trigger="t2") assert n1 == n2 assert InvestorMatch.objects.filter(entrepreneur=entrepreneur).count() == n1 def test_match_drops_deactivated_investor(admin, entrepreneur, verified_business, verified_investor): services.compile_performance_snapshot(entrepreneur) services.refresh_matches_for(entrepreneur) assert InvestorMatch.objects.filter( entrepreneur=entrepreneur, investor=verified_investor, ).exists() services.deactivate_investor(verified_investor, by_user=admin, reason="Pause") services.refresh_matches_for(entrepreneur) assert not InvestorMatch.objects.filter( entrepreneur=entrepreneur, investor=verified_investor, ).exists() # --------------------------------------------------------------------------- # Investor pitch flow (FRSME-INV013–015) # --------------------------------------------------------------------------- def test_request_then_accept_opens_deal_room(admin, entrepreneur, verified_business, verified_investor): request = services.request_investor_connection( entrepreneur=entrepreneur, business=verified_business, investor=verified_investor, message="Pitch us please.", ) assert request.status == InvestorConnectionRequest.Status.PENDING services.accept_investor_request(request, by_user=admin) request = InvestorConnectionRequest.objects.get(pk=request.pk) assert request.status == InvestorConnectionRequest.Status.ACCEPTED assert request.deal_room is not None assert request.deal_room.entrepreneur == entrepreneur # Deal room is the SP4 model, not a duplicate assert isinstance(request.deal_room, DealRoom) def test_duplicate_active_request_returns_existing(admin, entrepreneur, verified_business, verified_investor): r1 = services.request_investor_connection( entrepreneur=entrepreneur, business=verified_business, investor=verified_investor, ) r2 = services.request_investor_connection( entrepreneur=entrepreneur, business=verified_business, investor=verified_investor, ) assert r1.pk == r2.pk def test_decline_request_records_reason(admin, entrepreneur, verified_business, verified_investor): request = services.request_investor_connection( entrepreneur=entrepreneur, business=verified_business, investor=verified_investor, ) services.decline_investor_request(request, by_user=admin, reason="No fit") request = InvestorConnectionRequest.objects.get(pk=request.pk) assert request.status == InvestorConnectionRequest.Status.DECLINED assert request.decline_reason == "No fit" def test_request_blocked_for_unverified_investor(admin, entrepreneur, verified_business): inv = services.create_investor( by_user=admin, org_name="Pending Inv", type=Investor.Type.ANGEL, ) with pytest.raises(services.InvestmentStateError): services.request_investor_connection( entrepreneur=entrepreneur, business=verified_business, investor=inv, ) # --------------------------------------------------------------------------- # Dashboard sharing (FRSME-INV012) # --------------------------------------------------------------------------- def test_share_dashboard_creates_access_log(entrepreneur, verified_investor): log = services.share_dashboard_with_investor( entrepreneur=entrepreneur, investor=verified_investor, scope=DashboardAccessLog.Scope.SUMMARY, note="Q2 update", ) assert DashboardAccessLog.objects.filter(pk=log.pk).exists() assert log.share_token is not None assert log.investor == verified_investor # --------------------------------------------------------------------------- # Manual funding records (FRSME-INV016) # --------------------------------------------------------------------------- def test_manual_funding_record_creates_audit_log_and_emits_signal( admin, entrepreneur, verified_business, ): received = [] def _handler(sender, **kwargs): received.append(kwargs) inv_signals.smehub_funding_secured.connect(_handler) try: record = services.record_manual_funding( by_admin=admin, entrepreneur=entrepreneur, business=verified_business, funder_name="Off-platform donor", amount=Decimal("12000"), funded_at=timezone.now().date(), ) finally: inv_signals.smehub_funding_secured.disconnect(_handler) assert ManualFundingRecord.objects.filter(pk=record.pk).exists() assert AuditLog.objects.filter( target_app="smehub_investment", target_model="manualfundingrecord", action="CREATE", object_id=str(record.pk), ).exists() assert any(r.get("source") == "manual" for r in received) def test_manual_funding_blocks_business_belonging_to_someone_else( admin, verified_business, ): other_user = User.objects.create_user( email="other-ent@example.com", password="x", role=UserRole.ENTREPRENEUR, ) other_profile = EntrepreneurProfile.objects.create(user=other_user, country="UG") other_profile.verify(by_user=admin) other_profile.save() with pytest.raises(services.InvestmentStateError): services.record_manual_funding( by_admin=admin, entrepreneur=other_profile, business=verified_business, funder_name="Mismatch", amount=Decimal("5000"), funded_at=timezone.now().date(), ) # --------------------------------------------------------------------------- # Cross-cutting M&EL signal coverage (FRSME-INV017) # --------------------------------------------------------------------------- def test_investor_match_refresh_emits_signal(entrepreneur, verified_business, verified_investor): received = [] def _handler(sender, **kwargs): received.append(kwargs) inv_signals.smehub_investor_match_refreshed.connect(_handler) try: services.refresh_matches_for(entrepreneur, trigger="signal-test") finally: inv_signals.smehub_investor_match_refreshed.disconnect(_handler) assert received assert received[0].get("trigger") == "signal-test"