from django.test import TestCase from django.contrib.auth.models import User from rest_framework.test import APIClient from django.utils import timezone from api.models import Profile, Opportunity, IntroductionRequest, BenefitEvent from api.services.matching import ( jaccard, cosine, tokenize_interests_text, rank_users_for, rank_opportunities_for, ) from api.services.benefit import log_benefit_event, build_leaderboard_queryset class SimilarityUtilsTests(TestCase): def test_tokenize_and_jaccard(self): a = tokenize_interests_text("AI, Data Science") b = tokenize_interests_text("data science, analytics") self.assertGreater(jaccard(a, b), 0) self.assertLessEqual(jaccard(a, b), 1) def test_cosine(self): self.assertAlmostEqual(cosine(["ai", "ai"], ["ai"]) > 0, True) self.assertEqual(cosine(["a"], ["b"]), 0.0) class MatchingServiceTests(TestCase): def setUp(self): # Users self.u1 = User.objects.create_user(username="alice", password="x") self.u2 = User.objects.create_user(username="bob", password="x") self.u3 = User.objects.create_user(username="carol", password="x") # Profiles (user post_save signal creates a default Profile; update it) self.p1 = Profile.objects.get(user=self.u1) self.p1.bio = "Loves AI and data" self.p1.interests = "AI, Data" self.p1.is_verified = True self.p1.save() self.p2 = Profile.objects.get(user=self.u2) self.p2.bio = "AI researcher" self.p2.interests = "AI, ML" self.p2.save() self.p3 = Profile.objects.get(user=self.u3) self.p3.bio = "Designer" self.p3.interests = "UX, UI" self.p3.save() # Taggit tags (best-effort; fine if taggit not fully migrated yet) try: self.p1.tags.set(["ai", "data"]) # prefer tags in matching self.p2.tags.set(["ai", "ml"]) self.p3.tags.set(["ux", "ui"]) except Exception: pass self.p1.industry = "Technology" self.p2.industry = "Technology" self.p3.industry = "Design" self.p1.save(); self.p2.save(); self.p3.save() # Opportunity self.o1 = Opportunity.objects.create(title="Build Analytics Dashboard", description="Work with data and AI dashboards") def test_rank_users_for_returns_scored_results(self): ranked = rank_users_for(self.u1, k=5) self.assertTrue(any(r.profile.user == self.u2 for r in ranked)) # Existing request excludes candidate IntroductionRequest.objects.create(from_user=self.u1, to_user=self.u2, message="hi") ranked2 = rank_users_for(self.u1, k=5) self.assertFalse(any(r.profile.user == self.u2 for r in ranked2)) def test_rank_opportunities_for_uses_text_similarity(self): ranked = rank_opportunities_for(self.u1, k=5) self.assertTrue(any(r.opportunity == self.o1 and r.score > 0 for r in ranked)) class MatchingEndpointsTests(TestCase): def setUp(self): self.client = APIClient() self.user = User.objects.create_user(username="dave", password="x") self.other = User.objects.create_user(username="erin", password="x") self.pu = Profile.objects.get(user=self.user) self.pu.bio = "Data person" self.pu.interests = "Data, Analytics" self.pu.save() self.po = Profile.objects.get(user=self.other) self.po.bio = "Analytics work" self.po.interests = "Analytics, BI" self.po.save() self.pu.industry = "Technology" self.po.industry = "Technology" self.pu.save(); self.po.save() try: self.pu.tags.set(["data", "analytics"]) # optional self.po.tags.set(["analytics", "bi"]) # optional except Exception: pass Opportunity.objects.create(title="Analytics Role", description="Looking for data analytics help") def test_profiles_match_requires_auth(self): resp = self.client.get("/api/profiles/match/") self.assertIn(resp.status_code, (401, 403)) def test_profiles_match_returns_results_for_authed_user(self): self.client.force_authenticate(user=self.user) resp = self.client.get("/api/profiles/match/?k=10") self.assertEqual(resp.status_code, 200) self.assertIsInstance(resp.json(), list) class BenefitAndLeaderboardTests(TestCase): def setUp(self): self.client = APIClient() self.a = User.objects.create_user(username="alice", password="x") self.b = User.objects.create_user(username="bob", password="x") self.c = User.objects.create_user(username="carol", password="x") # Default user for authenticated endpoints in this suite self.user = self.a def test_benefit_event_disallows_self(self): with self.assertRaises(Exception): BenefitEvent.objects.create( benefactor=self.a, beneficiary=self.a, kind=BenefitEvent.KIND_REFERRAL, points=10, ) def test_leaderboard_all_time_and_weekly(self): # a helps b twice, c once log_benefit_event(benefactor=self.a, beneficiary=self.b, kind=BenefitEvent.KIND_REFERRAL) log_benefit_event(benefactor=self.a, beneficiary=self.c, kind=BenefitEvent.KIND_ACCEPTED_ANSWER) # c older event outside 7d window old = BenefitEvent( benefactor=self.c, beneficiary=self.b, kind=BenefitEvent.KIND_RECOMMENDATION, points=5, ) old.created_at = timezone.now() - timezone.timedelta(days=60) old.save() all_qs = build_leaderboard_queryset("all") usernames = [u.username for u in all_qs] self.assertIn("alice", usernames) weekly_qs = build_leaderboard_queryset("weekly") w_usernames = [u.username for u in weekly_qs] self.assertIn("alice", w_usernames) # carol's only event is too old for weekly self.assertNotIn("carol", w_usernames) def test_leaderboard_endpoints(self): log_benefit_event(benefactor=self.a, beneficiary=self.b, kind=BenefitEvent.KIND_REFERRAL) # Public leaderboard resp = self.client.get("/api/leaderboard/?period=all&limit=10") self.assertEqual(resp.status_code, 200) self.assertIn("results", resp.json()) # Me endpoint requires auth resp2 = self.client.get("/api/leaderboard/me/?period=all") self.assertIn(resp2.status_code, (401, 403)) self.client.force_authenticate(user=self.a) resp3 = self.client.get("/api/leaderboard/me/?period=all") self.assertEqual(resp3.status_code, 200) self.assertIsNotNone(resp3.json().get("result")) def test_opportunities_recommend_returns_results_for_authed_user(self): self.client.force_authenticate(user=self.user) resp = self.client.get("/api/opportunities/recommend/?k=10") self.assertEqual(resp.status_code, 200) self.assertIsInstance(resp.json(), list)