feat: teacher/company org-scoped access to /admin/sources with AI toggle

backend/content/management/commands/seed_lookups.py

@@ -252,6 +252,15 @@ def handle(self, *args, **options):
             },
         )
 
+        # Link org-specific sources to their organizations so Teacher users can see them
+        for source_key, org_token in [("unibz", "unibz"), ("mdu", "mdu")]:
+            org = organization_map.get(org_token)
+            if org:
+                ScrapingSource.objects.filter(key=source_key).update(
+                    organization=org,
+                    organization_token=str(org.id),
+                )
+
         # Single shared password for all loginable seeded users
         SEED_PASSWORD = "passw0rd"
 

backend/content/scrapers/service.py

@@ -555,7 +555,11 @@ def _upsert_offer(
         if offer_type is None:
             LOGGER.warning("[%s] offer_type %r not found in DB — discarding", source.key, resolved_offer_type_name)
             return "skipped", None, None
-        target_profile = TargetProfile.objects.get(name=source.target_profile)
+        try:
+            target_profile = TargetProfile.objects.get(name=source.target_profile)
+        except TargetProfile.DoesNotExist:
+            LOGGER.error("[%s] target_profile '%s' not in DB — skipping offer", source.key, source.target_profile)
+            return "skipped", None, None
 
         natural_key = (source.url, str(organization.id), str(offer_type.id))
 

backend/content/test_sources.py

@@ -2,7 +2,7 @@
 import json
 import uuid
 from django.test import TestCase, Client
-from content.models import Organization, ScrapingSource, User
+from content.models import Organization, ScrapingSource, User, UserOrganization, UserRole
 from content.auth import hash_password
 
 _TEST_ORG_ID = str(uuid.UUID("00000000-0000-0000-0000-000000000001"))
@@ -188,6 +188,35 @@ def test_create_missing_required_fields(self):
         )
         self.assertEqual(resp.status_code, 400)
 
+    def test_create_auto_generates_key_when_absent(self):
+        payload = {
+            "name": "Auto Key Source",
+            "url": "https://autokey.example.com",
+            "organization_id": _TEST_ORG_ID,
+        }
+        resp = self.client.post(
+            self.url,
+            data=json.dumps(payload),
+            content_type="application/json",
+            **self._auth(),
+        )
+        self.assertEqual(resp.status_code, 201)
+        data = json.loads(resp.content)
+        key = data["key"]
+        self.assertTrue(len(key) > 0)
+        import uuid as _uuid
+        _uuid.UUID(key)  # raises ValueError if not a valid UUID
+
+    def test_create_invalid_target_profile(self):
+        payload = _source_payload(target_profile="invalid_profile", key="bad_tp_src")
+        resp = self.client.post(
+            self.url,
+            data=json.dumps(payload),
+            content_type="application/json",
+            **self._auth(),
+        )
+        self.assertEqual(resp.status_code, 400)
+
     def test_create_duplicate_key(self):
         ScrapingSource.objects.create(
             key="test_source", name="Existing", url="https://x.com", organization_token="o",
@@ -459,3 +488,201 @@ def test_get_sources_empty_db(self):
         from content.scrapers.source_registry import get_sources
         results = get_sources()
         self.assertEqual(results, [])
+
+
+_OTHER_ORG_ID = str(uuid.UUID("00000000-0000-0000-0000-000000000002"))
+
+
+def _make_teacher(username="teacher_test", password="Teacher123!"):
+    user = User.objects.create(
+        username=username,
+        email=f"{username}@example.com",
+        password_hash=hash_password(password),
+        profile=User.ProfileType.TEACHER,
+        email_verified=True,
+        approval_status="approved",
+        is_active=True,
+    )
+    return user, password
+
+
+def _make_company(username="company_test", password="Company123!"):
+    user = User.objects.create(
+        username=username,
+        email=f"{username}@example.com",
+        password_hash=hash_password(password),
+        profile=User.ProfileType.COMPANY,
+        email_verified=True,
+        approval_status="approved",
+        is_active=True,
+    )
+    return user, password
+
+
+def _make_other_org():
+    org, _ = Organization.objects.get_or_create(
+        id=_OTHER_ORG_ID,
+        defaults={
+            "name": "Other Org",
+            "type": Organization.OrganizationType.UNIVERSITY,
+            "country": "SE",
+            "website": "https://other.example.com",
+        },
+    )
+    return org
+
+
+class SourcesOrgScopeTestCase(TestCase):
+    """Teachers can only access sources belonging to their own organization."""
+
+    def setUp(self):
+        self.client = Client()
+        ScrapingSource.objects.all().delete()
+        self.own_org = _make_org()
+        self.other_org = _make_other_org()
+        self.role, _ = UserRole.objects.get_or_create(
+            name="researcher", defaults={"description": "Researcher role"}
+        )
+        self.teacher, self.teacher_pw = _make_teacher()
+        UserOrganization.objects.create(
+            user=self.teacher, organization=self.own_org, role=self.role
+        )
+        self.teacher_token = _token(self.client, self.teacher.username, self.teacher_pw)
+        self.own_source = ScrapingSource.objects.create(
+            key="own_src",
+            name="Own Source",
+            url="https://own.example.com",
+            organization=self.own_org,
+            organization_token=str(self.own_org.id),
+            llm_fallback_enabled=True,
+            enabled=True,
+        )
+        self.other_source = ScrapingSource.objects.create(
+            key="other_src",
+            name="Other Source",
+            url="https://other.example.com",
+            organization=self.other_org,
+            organization_token=str(self.other_org.id),
+            llm_fallback_enabled=True,
+            enabled=True,
+        )
+
+    def _auth(self, token=None):
+        return {"HTTP_AUTHORIZATION": f"Bearer {token or self.teacher_token}"}
+
+    def test_teacher_can_list_own_org_sources(self):
+        resp = self.client.get("/api/admin/sources", **self._auth())
+        self.assertEqual(resp.status_code, 200)
+        data = json.loads(resp.content)
+        keys = {s["key"] for s in data["results"]}
+        self.assertIn("own_src", keys)
+
+    def test_teacher_cannot_see_other_org_sources(self):
+        resp = self.client.get("/api/admin/sources", **self._auth())
+        self.assertEqual(resp.status_code, 200)
+        data = json.loads(resp.content)
+        keys = {s["key"] for s in data["results"]}
+        self.assertNotIn("other_src", keys)
+        self.assertEqual(data["count"], 1)
+
+    def test_teacher_can_create_source_org_auto_assigned(self):
+        payload = {
+            "name": "New Teacher Source",
+            "url": "https://new.example.com",
+        }
+        resp = self.client.post(
+            "/api/admin/sources",
+            data=json.dumps(payload),
+            content_type="application/json",
+            **self._auth(),
+        )
+        self.assertEqual(resp.status_code, 201)
+        data = json.loads(resp.content)
+        self.assertEqual(data["organization_id"], str(self.own_org.id))
+        import uuid as _uuid
+        _uuid.UUID(data["key"])  # auto-generated key must be a valid UUID
+        src = ScrapingSource.objects.get(key=data["key"])
+        self.assertEqual(str(src.organization_id), str(self.own_org.id))
+
+    def test_teacher_can_toggle_llm_fallback_own_source(self):
+        resp = self.client.patch(
+            f"/api/admin/sources/{self.own_source.key}",
+            data=json.dumps({"llm_fallback_enabled": False}),
+            content_type="application/json",
+            **self._auth(),
+        )
+        self.assertEqual(resp.status_code, 200)
+        data = json.loads(resp.content)
+        self.assertFalse(data["llm_fallback_enabled"])
+        self.own_source.refresh_from_db()
+        self.assertFalse(self.own_source.llm_fallback_enabled)
+
+    def test_teacher_cannot_change_source_organization(self):
+        resp = self.client.patch(
+            f"/api/admin/sources/{self.own_source.key}",
+            data=json.dumps({"organization_id": str(self.other_org.id)}),
+            content_type="application/json",
+            **self._auth(),
+        )
+        self.assertEqual(resp.status_code, 200)
+        self.own_source.refresh_from_db()
+        self.assertEqual(str(self.own_source.organization_id), str(self.own_org.id))
+
+    def test_teacher_cannot_access_other_org_source_detail(self):
+        resp = self.client.get(
+            f"/api/admin/sources/{self.other_source.key}", **self._auth()
+        )
+        self.assertEqual(resp.status_code, 404)
+
+    def test_teacher_cannot_patch_other_org_source(self):
+        resp = self.client.patch(
+            f"/api/admin/sources/{self.other_source.key}",
+            data=json.dumps({"name": "Hacked"}),
+            content_type="application/json",
+            **self._auth(),
+        )
+        self.assertEqual(resp.status_code, 404)
+
+    def test_teacher_cannot_delete_other_org_source(self):
+        resp = self.client.delete(
+            f"/api/admin/sources/{self.other_source.key}", **self._auth()
+        )
+        self.assertEqual(resp.status_code, 404)
+        self.assertTrue(ScrapingSource.objects.filter(key="other_src").exists())
+
+    def test_teacher_can_delete_own_org_source(self):
+        resp = self.client.delete(
+            f"/api/admin/sources/{self.own_source.key}", **self._auth()
+        )
+        self.assertEqual(resp.status_code, 204)
+        self.assertFalse(ScrapingSource.objects.filter(key="own_src").exists())
+
+    def test_teacher_with_no_org_cannot_create_source(self):
+        teacher_no_org, pw = _make_teacher(username="teacher_no_org", password="Teacher123!")
+        token = _token(self.client, "teacher_no_org", pw)
+        resp = self.client.post(
+            "/api/admin/sources",
+            data=json.dumps({"name": "New Source", "url": "https://new.example.com"}),
+            content_type="application/json",
+            HTTP_AUTHORIZATION=f"Bearer {token}",
+        )
+        self.assertEqual(resp.status_code, 403)
+        data = json.loads(resp.content)
+        self.assertIn("No organization linked", data["detail"])
+
+    def test_patch_invalid_target_profile_returns_400(self):
+        resp = self.client.patch(
+            f"/api/admin/sources/{self.own_source.key}",
+            data=json.dumps({"target_profile": "invalid_profile"}),
+            content_type="application/json",
+            **self._auth(),
+        )
+        self.assertEqual(resp.status_code, 400)
+        data = json.loads(resp.content)
+        self.assertIn("target_profile", data["detail"])
+
+    def test_company_cannot_access_sources(self):
+        company, company_pw = _make_company()
+        company_token = _token(self.client, company.username, company_pw)
+        resp = self.client.get("/api/admin/sources", HTTP_AUTHORIZATION=f"Bearer {company_token}")
+        self.assertEqual(resp.status_code, 403)