"""Tests for SME-Hub linkage services (SP3a). Coverage targets the key PRD acceptance points: * FRSME-MPL003 — directory entry verification flips status + emits signal * FRSME-MPL008 — duplicate-pending connection request returns the existing row * FRSME-MPL010 — accepting a connection creates a Connection object * FRSME-MPL013 / A5 — MoU formalisation + amendment chain stores versions * FRSME-MPL017 / MPL019 — advisory session cancellation marks status + reason * FRSME-MPL005 / MPL006 — recommendation snapshot refreshes on business update """ from __future__ import annotations import pytest from django.contrib.auth import get_user_model from django.core.files.base import ContentFile from django.utils import timezone from apps.core.permissions.roles import UserRole from apps.smehub.linkage import services from apps.smehub.linkage.ai_matchmaking import refresh_recommendations from apps.smehub.linkage.models import ( AdvisorySession, Connection, ConnectionRequest, MoU, PartnerDirectoryEntry, RecommendationSnapshot, ServiceProviderEntry, ) from apps.smehub.onboarding.models import ( Business, BusinessStage, EntrepreneurProfile, ) User = get_user_model() pytestmark = pytest.mark.django_db @pytest.fixture def admin(): return User.objects.create_user( email="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="ent@example.com", password="x", role=UserRole.ENTREPRENEUR, ) profile = EntrepreneurProfile.objects.create( user=user, country="UG", organisation="Test Co", ) 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="UG", ) biz.verify(by_user=admin) biz.save() return biz @pytest.fixture def partner_user(): return User.objects.create_user(email="partner@example.com", password="x", role=UserRole.PARTNER) @pytest.fixture def partner_entry(partner_user): return services.create_directory_entry( kind="partner", managed_by=partner_user, created_by=partner_user, organisation_name="Test Partner", org_type=PartnerDirectoryEntry.OrgType.NGO, sector_focus=["Agriculture"], geographic_coverage=["UG", "KE"], contact_email="hello@partner.example", description="Demo partner.", areas_of_interest="Agritech sourcing.", ) @pytest.fixture def verified_partner(partner_entry, admin): services.verify_directory_entry(partner_entry, by_user=admin) return PartnerDirectoryEntry.objects.get(pk=partner_entry.pk) @pytest.fixture def provider_user(): return User.objects.create_user(email="provider@example.com", password="x", role=UserRole.SERVICE_PROVIDER) @pytest.fixture def verified_provider(provider_user, admin): entry = services.create_directory_entry( kind="service_provider", managed_by=provider_user, created_by=provider_user, organisation_name="Test Provider", org_type=ServiceProviderEntry.OrgType.CORPORATE, sector_focus=["Agriculture"], geographic_coverage=["UG"], service_offerings=["legal"], contact_email="hi@provider.example", description="Legal services.", areas_of_interest="Compliance for agritech.", ) services.verify_directory_entry(entry, by_user=admin) return ServiceProviderEntry.objects.get(pk=entry.pk) # --------------------------------------------------------------------------- # Directory verification # --------------------------------------------------------------------------- def test_directory_entry_verification_flips_status(partner_entry, admin): assert partner_entry.verification_status == PartnerDirectoryEntry.Status.PENDING services.verify_directory_entry(partner_entry, by_user=admin) refreshed = PartnerDirectoryEntry.objects.get(pk=partner_entry.pk) assert refreshed.verification_status == PartnerDirectoryEntry.Status.VERIFIED assert refreshed.verified_by_id == admin.pk def test_directory_rejection_records_reason(partner_entry, admin): services.reject_directory_entry(partner_entry, by_user=admin, reason="Out of scope.") refreshed = PartnerDirectoryEntry.objects.get(pk=partner_entry.pk) assert refreshed.verification_status == PartnerDirectoryEntry.Status.REJECTED assert "Out of scope" in refreshed.rejection_reason # --------------------------------------------------------------------------- # Connection requests (FRSME-MPL007–010) # --------------------------------------------------------------------------- def test_request_connection_creates_pending_request(entrepreneur, verified_partner): cr = services.request_connection( entrepreneur=entrepreneur, target=verified_partner, message="Let's talk.", ) assert cr.status == ConnectionRequest.Status.PENDING assert cr.entrepreneur_id == entrepreneur.pk def test_duplicate_pending_request_returns_existing(entrepreneur, verified_partner): first = services.request_connection(entrepreneur=entrepreneur, target=verified_partner, message="Hi") second = services.request_connection(entrepreneur=entrepreneur, target=verified_partner, message="Hi again") assert second.pk == first.pk assert ConnectionRequest.objects.filter(entrepreneur=entrepreneur).count() == 1 def test_duplicate_active_request_returns_existing(entrepreneur, verified_partner, partner_user): """FRSME-MPL008 — block when an accepted (active) connection already exists.""" first = services.request_connection(entrepreneur=entrepreneur, target=verified_partner) services.accept_connection(first, by_user=partner_user) # Try a fresh request after the original was accepted — should surface the existing. second = services.request_connection(entrepreneur=entrepreneur, target=verified_partner, message="Again") assert second.pk == first.pk assert ConnectionRequest.objects.filter(entrepreneur=entrepreneur).count() == 1 def test_accept_connection_creates_connection_row(entrepreneur, verified_partner, partner_user): cr = services.request_connection(entrepreneur=entrepreneur, target=verified_partner) connection = services.accept_connection(cr, by_user=partner_user) assert isinstance(connection, Connection) assert connection.entrepreneur_id == entrepreneur.pk refreshed = ConnectionRequest.objects.get(pk=cr.pk) assert refreshed.status == ConnectionRequest.Status.ACCEPTED def test_decline_connection_records_reason(entrepreneur, verified_partner, partner_user): cr = services.request_connection(entrepreneur=entrepreneur, target=verified_partner) services.decline_connection(cr, by_user=partner_user, reason="Wrong sector.") refreshed = ConnectionRequest.objects.get(pk=cr.pk) assert refreshed.status == ConnectionRequest.Status.DECLINED assert "Wrong sector" in refreshed.decline_reason # --------------------------------------------------------------------------- # MoU (FRSME-MPL011–013, A5) # --------------------------------------------------------------------------- def test_init_mou_writes_initial_version(entrepreneur, verified_partner, partner_user): cr = services.request_connection(entrepreneur=entrepreneur, target=verified_partner) connection = services.accept_connection(cr, by_user=partner_user) mou = services.init_mou( connection=connection, title="Pilot agreement", summary="Pilot supply chain.", by_user=entrepreneur.user, ) assert mou.status == MoU.Status.DRAFT assert mou.versions.count() == 1 assert mou.versions.first().notes.startswith("Initial") def test_formalise_mou_records_signed_file_and_version(entrepreneur, verified_partner, partner_user, admin): cr = services.request_connection(entrepreneur=entrepreneur, target=verified_partner) connection = services.accept_connection(cr, by_user=partner_user) mou = services.init_mou(connection=connection, title="A", by_user=entrepreneur.user) # Phase 6 audit fix: both parties must confirm before formalisation. services.confirm_mou_party(mou, party="entrepreneur", by_user=entrepreneur.user) services.confirm_mou_party(mou, party="counterparty", by_user=partner_user) services.formalise_mou( mou, signed_file=ContentFile(b"signed", name="signed.pdf"), by_user=admin, ) refreshed = MoU.objects.get(pk=mou.pk) assert refreshed.status == MoU.Status.FORMALISED assert refreshed.signed_file assert refreshed.versions.count() >= 2 def test_amend_mou_marks_parent_amended_and_links_chain(entrepreneur, verified_partner, partner_user, admin): cr = services.request_connection(entrepreneur=entrepreneur, target=verified_partner) connection = services.accept_connection(cr, by_user=partner_user) mou = services.init_mou(connection=connection, title="Original", by_user=entrepreneur.user) services.confirm_mou_party(mou, party="entrepreneur", by_user=entrepreneur.user) services.confirm_mou_party(mou, party="counterparty", by_user=partner_user) services.formalise_mou(mou, signed_file=ContentFile(b"s", name="s.pdf"), by_user=admin) refreshed = MoU.objects.get(pk=mou.pk) amended = services.amend_mou(refreshed, title="Original v2", by_user=entrepreneur.user) parent = MoU.objects.get(pk=mou.pk) assert parent.status == MoU.Status.AMENDED assert amended.parent_mou_id == parent.pk assert amended.connection_id == connection.pk def test_mou_comment_thread_creates_comment(entrepreneur, verified_partner, partner_user): cr = services.request_connection(entrepreneur=entrepreneur, target=verified_partner) connection = services.accept_connection(cr, by_user=partner_user) mou = services.init_mou(connection=connection, title="X", by_user=entrepreneur.user) comment = services.comment_on_mou(mou=mou, author=partner_user, body="Looks good.") assert mou.comments.count() == 1 reply = services.comment_on_mou(mou=mou, author=entrepreneur.user, body="Thanks!", parent_id=comment.pk) assert reply.parent_id == comment.pk # --------------------------------------------------------------------------- # Advisory sessions (FRSME-MPL017–019) # --------------------------------------------------------------------------- @pytest.fixture def provider_connection(entrepreneur, verified_provider, provider_user): """An accepted connection between entrepreneur and verified_provider — required by MPL017.""" cr = services.request_connection(entrepreneur=entrepreneur, target=verified_provider) return services.accept_connection(cr, by_user=provider_user) def test_book_advisory_session_creates_scheduled_row(entrepreneur, verified_provider, provider_connection): when = timezone.now() + timezone.timedelta(days=3) session = services.book_advisory_session( entrepreneur=entrepreneur, service_provider=verified_provider, starts_at=when, agenda="Compliance review.", ) assert session.status == AdvisorySession.Status.SCHEDULED assert session.starts_at == when def test_book_advisory_blocked_without_connection(entrepreneur, verified_provider): """FRSME-MPL017 — booking requires an active Connection.""" when = timezone.now() + timezone.timedelta(days=3) with pytest.raises(services.LinkageStateError) as exc: services.book_advisory_session( entrepreneur=entrepreneur, service_provider=verified_provider, starts_at=when, ) assert "connection" in str(exc.value).lower() def test_cancel_advisory_session_records_reason(entrepreneur, verified_provider, provider_connection): when = timezone.now() + timezone.timedelta(days=3) session = services.book_advisory_session( entrepreneur=entrepreneur, service_provider=verified_provider, starts_at=when, ) services.cancel_advisory_session(session, reason="Provider sick.", by_user=verified_provider.managed_by) refreshed = AdvisorySession.objects.get(pk=session.pk) assert refreshed.status == AdvisorySession.Status.CANCELLED assert "Provider sick" in refreshed.cancellation_reason def test_complete_advisory_session_sets_completed_at(entrepreneur, verified_provider, provider_connection): when = timezone.now() - timezone.timedelta(days=1) session = services.book_advisory_session( entrepreneur=entrepreneur, service_provider=verified_provider, starts_at=when, ) services.complete_advisory_session(session, notes="All actioned.") refreshed = AdvisorySession.objects.get(pk=session.pk) assert refreshed.status == AdvisorySession.Status.COMPLETED assert refreshed.completed_at is not None def test_book_by_slot_consumes_and_cancel_frees( entrepreneur, verified_provider, provider_connection, ): """FRSME-MPL017/MPL018 — booking a slot consumes it; cancelling frees it.""" from apps.smehub.linkage.models import AdvisorySlot slot = AdvisorySlot.objects.create( service_provider=verified_provider, starts_at=timezone.now() + timezone.timedelta(days=2), duration_minutes=45, ) session = services.book_advisory_session( entrepreneur=entrepreneur, service_provider=verified_provider, slot=slot, agenda="Legal review.", ) # Slot time/duration win over any free-text values. assert session.starts_at == slot.starts_at assert session.duration_minutes == 45 slot.refresh_from_db() assert slot.session_id == session.pk assert slot.is_booked is True # Cancelling frees the slot. services.cancel_advisory_session(session, reason="Reschedule.", by_user=entrepreneur.user) slot.refresh_from_db() assert slot.session_id is None assert slot.is_booked is False def test_book_rejects_slot_from_another_provider( entrepreneur, verified_provider, provider_connection, admin, ): """A slot that doesn't belong to the target provider is rejected.""" from apps.smehub.linkage.models import AdvisorySlot other_user = User.objects.create_user( email="other-provider@example.com", password="x", role=UserRole.SERVICE_PROVIDER, ) other = services.create_directory_entry( kind="service_provider", managed_by=other_user, organisation_name="Other Provider", org_type=ServiceProviderEntry.OrgType.CORPORATE, contact_email="o@o.example", description="x", areas_of_interest="x", ) services.verify_directory_entry(other, by_user=admin) foreign_slot = AdvisorySlot.objects.create( service_provider=ServiceProviderEntry.objects.get(pk=other.pk), starts_at=timezone.now() + timezone.timedelta(days=2), ) with pytest.raises(services.LinkageStateError): services.book_advisory_session( entrepreneur=entrepreneur, service_provider=verified_provider, slot=foreign_slot, ) def test_book_advisory_blocked_on_unverified_provider(entrepreneur, provider_user): entry = services.create_directory_entry( kind="service_provider", managed_by=provider_user, organisation_name="Pending Pro", org_type=ServiceProviderEntry.OrgType.CORPORATE, contact_email="x@x.example", description="x", areas_of_interest="x", ) with pytest.raises(services.LinkageStateError): services.book_advisory_session( entrepreneur=entrepreneur, service_provider=entry, starts_at=timezone.now() + timezone.timedelta(days=2), ) # --------------------------------------------------------------------------- # AI matchmaking (FRSME-MPL005, MPL006) # --------------------------------------------------------------------------- def test_recommendation_snapshot_refreshes_after_business_update( entrepreneur, verified_partner, verified_business, ): snapshot = refresh_recommendations(entrepreneur, trigger="initial") initial_first_score = snapshot.recommendations[0]["score"] if snapshot.recommendations else 0 initial_refreshed_at = snapshot.refreshed_at # Add a second matching partner — should improve coverage services.verify_directory_entry( services.create_directory_entry( kind="partner", managed_by=None, organisation_name="Another Partner", org_type=PartnerDirectoryEntry.OrgType.CORPORATE, sector_focus=["Agriculture"], geographic_coverage=["UG"], contact_email="y@y.example", description="x", areas_of_interest="x", ), by_user=User.objects.filter(is_superuser=True).first(), ) snapshot2 = refresh_recommendations(entrepreneur, trigger="post_update") assert snapshot2.refreshed_at >= initial_refreshed_at assert RecommendationSnapshot.objects.filter(entrepreneur=entrepreneur).count() == 1 assert len(snapshot2.recommendations) >= 1 assert snapshot2.recommendations[0]["score"] >= 0 def test_recommendation_returns_partners_matching_business_sector(entrepreneur, verified_business, verified_partner): snapshot = refresh_recommendations(entrepreneur, trigger="initial") ids = [r["id"] for r in snapshot.recommendations if r.get("kind") == "partner"] assert verified_partner.pk in ids # --------------------------------------------------------------------------- # Suggestions (PRD A2) # --------------------------------------------------------------------------- def test_suggest_organisation_records_submission(entrepreneur): rec = services.suggest_organisation( suggested_by=entrepreneur.user, kind="partner", organisation_name="Suggested Co", rationale="Great fit for agritech.", ) assert rec.organisation_name == "Suggested Co" assert rec.status == "submitted" assert rec.suggested_by_id == entrepreneur.user.pk