Feature/pdct 1533 documents should only show documents in allowed corpora

app/api/api_v1/routers/documents.py

@@ -1,8 +1,8 @@
 import logging
 from http.client import NOT_FOUND
-from typing import Union
+from typing import Annotated, Union
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, Header, HTTPException, Request
 
 from app.clients.db.session import get_db
 from app.models.document import (
@@ -14,6 +14,7 @@
     get_family_document_and_context,
     get_slugged_objects,
 )
+from app.service.custom_app import AppTokenFactory
 
 _LOGGER = logging.getLogger(__file__)
 
@@ -28,20 +29,23 @@
     ],
 )
 async def family_or_document_detail(
-    slug: str,
-    db=Depends(get_db),
+    slug: str, request: Request, app_token: Annotated[str, Header()], db=Depends(get_db)
 ):
     """Get details of the family or document associated with the slug."""
     _LOGGER.info(
         f"Getting detailed information for family or document '{slug}'",
         extra={
-            "props": {
-                "import_id_or_slug": slug,
-            },
+            "props": {"import_id_or_slug": slug, "app_token": str(app_token)},
         },
     )
 
-    family_document_import_id, family_import_id = get_slugged_objects(db, slug)
+    # Decode the app token and validate it.
+    token = AppTokenFactory()
+    token.decode_and_validate(db, request, app_token)
+
+    family_document_import_id, family_import_id = get_slugged_objects(
+        db, slug, token.allowed_corpora_ids
+    )
     if family_document_import_id is None and family_import_id is None:
         raise HTTPException(status_code=NOT_FOUND, detail=f"Nothing found for {slug}")
 

app/api/api_v1/routers/search.py

@@ -277,8 +277,6 @@ def download_all_search_documents(
     request: Request, app_token: Annotated[str, Header()], db=Depends(get_db)
 ) -> RedirectResponse:
     """Download a CSV containing details of all the documents in the corpus."""
-    token = AppTokenFactory()
-
     _LOGGER.info(
         "Whole data download request",
         extra={

app/repository/document.py

@@ -40,20 +40,27 @@
 _LOGGER = logging.getLogger(__file__)
 
 
-def get_slugged_objects(db: Session, slug: str) -> tuple[Optional[str], Optional[str]]:
-    """
-    Matches the slug name to a FamilyDocument or Family import_id
+def get_slugged_objects(
+    db: Session, slug: str, allowed_corpora: Optional[list[str]] = None
+) -> tuple[Optional[str], Optional[str]]:
+    """Match the slug name to a FamilyDocument or Family import ID.
 
     :param Session db: connection to db
     :param str slug: slug name to match
-    :return tuple[Optional[str], Optional[str]]: the FamilyDocument import id or
-    the Family import_id
+    :param Optional[list[str]] allowed_corpora: The corpora IDs to look
+        for the slugged object in.
+    :return tuple[Optional[str], Optional[str]]: the FamilyDocument
+        import id or the Family import_id.
     """
-    result = (
-        db.query(Slug.family_document_import_id, Slug.family_import_id).filter(
-            Slug.name == slug
-        )
-    ).one_or_none()
+    query = db.query(Slug.family_document_import_id, Slug.family_import_id).filter(
+        Slug.name == slug
+    )
+
+    # # Only get slug for the fam/doc if it belongs to the list of allowed corpora.
+    # if allowed_corpora is not None:
+    #     query = query.filter()
+
+    result = query.one_or_none()
     if result is None:
         return (None, None)
     return result

tests/non_search/routers/documents/setup_doc_fam_lookup.py

@@ -0,0 +1,24 @@
+from typing import Optional
+
+from fastapi import status
+
+DOCUMENTS_ENDPOINT = "/api/v1/documents"
+TEST_HOST = "http://localhost:3000/"
+
+
+def _make_doc_fam_lookup_request(
+    client,
+    token,
+    slug: str,
+    expected_status_code: int = status.HTTP_200_OK,
+    origin: Optional[str] = TEST_HOST,
+):
+    headers = (
+        {"app-token": token}
+        if origin is None
+        else {"app-token": token, "origin": origin}
+    )
+
+    response = client.get(f"{DOCUMENTS_ENDPOINT}/{slug}", headers=headers)
+    assert response.status_code == expected_status_code, response.text
+    return response.json()