[Issue #7081] Create new endpoint for form instructions documents

api/src/api/form_alpha/form_route.py

@@ -2,12 +2,15 @@
 import uuid
 from typing import cast
 
+from flask import request
+
 import src.adapters.db as db
 import src.adapters.db.flask_db as flask_db
 import src.api.form_alpha.form_schema as form_schema
 import src.api.response as response
 from src.api.form_alpha.form_blueprint import form_blueprint
 from src.api.route_utils import raise_flask_error
+from src.api.schemas.response_schema import AbstractResponseSchema
 from src.auth.api_key_auth import ApiKeyUser
 from src.auth.endpoint_access_util import verify_access
 from src.auth.multi_auth import AuthType, api_key_multi_auth, api_key_multi_auth_security_schemes
@@ -67,3 +70,43 @@ def form_update(
         form = update_form(db_session, form_id, json_data)
 
     return response.ApiResponse(message="Success", data=form)
+
+
+@form_blueprint.put("/forms/<uuid:form_id>/form_instructions/<uuid:form_instruction_id>")
+@form_blueprint.output(AbstractResponseSchema)
+@api_key_multi_auth.login_required
+@flask_db.with_db_session()
+@form_blueprint.doc(security=api_key_multi_auth_security_schemes)
+def form_instruction_upsert(
+    db_session: db.Session, form_id: uuid.UUID, form_instruction_id: uuid.UUID
+) -> response.ApiResponse:
+    add_extra_data_to_current_request_logs(
+        {"form_id": form_id, "form_instruction_id": form_instruction_id}
+    )
+    logger.info("PUT /alpha/forms/:form_id/form_instructions/:form_instruction_id")
+
+    # Get the file from the request
+    # The request should be multipart/form-data
+    if "file" not in request.files:
+        raise_flask_error(400, "No file part in the request")
+
+    file_obj = request.files["file"]
+
+    with db_session.begin():
+        # Check auth
+        multi_auth_user = api_key_multi_auth.get_user()
+        if multi_auth_user.auth_type == AuthType.API_KEY_AUTH:
+            # Check if user is the internal admin user (auth_token_0)
+            if cast(ApiKeyUser, multi_auth_user.user).username != "auth_token_0":
+                raise_flask_error(403, "Only internal admin users can update form instructions")
+        else:
+            # Temporary auth check until legacy auth is removed
+            user = cast(UserApiKey, multi_auth_user.user).user
+            db_session.add(user)
+            verify_access(user, {Privilege.UPDATE_FORM}, None)
+
+        from src.services.form_alpha.upsert_form_instruction import upsert_form_instruction
+
+        upsert_form_instruction(db_session, form_id, form_instruction_id, file_obj)
+
+    return response.ApiResponse(message="Success")

api/src/services/form_alpha/upsert_form_instruction.py

@@ -0,0 +1,81 @@
+import logging
+import uuid
+
+from werkzeug.datastructures import FileStorage
+
+import src.adapters.db as db
+import src.util.file_util as file_util
+from src.adapters.aws import S3Config
+from src.api.route_utils import raise_flask_error
+from src.constants.lookup_constants import SubmissionIssue
+from src.db.models.competition_models import FormInstruction
+
+logger = logging.getLogger(__name__)
+
+
+def upsert_form_instruction(
+    db_session: db.Session,
+    form_id: uuid.UUID,
+    form_instruction_id: uuid.UUID,
+    file_obj: FileStorage,
+) -> FormInstruction:
+    """
+    Upsert a form instruction.
+    If the form instruction exists, update it.
+    If not, create it.
+    Upload the file to S3.
+    """
+    if file_obj.filename is None:
+        logger.info(
+            "Invalid file name, cannot parse",
+            extra={"submission_issue": SubmissionIssue.INVALID_FILE_NAME},
+        )
+        raise_flask_error(422, "Invalid file name, cannot parse")
+
+    secure_file_name = file_util.get_secure_file_name(file_obj.filename)
+
+    # Check if form instruction exists
+    form_instruction = db_session.get(FormInstruction, form_instruction_id)
+
+    if not form_instruction:
+        form_instruction = FormInstruction(
+            form_instruction_id=form_instruction_id,
+            file_name=file_util.get_file_name(file_obj.filename),
+            file_location="",  # Will be set below
+        )
+        db_session.add(form_instruction)
+
+    # Construct S3 path
+    # s3://{public_files_bucket_path}/forms/{form_id}/instructions/{file_name}
+    s3_config = S3Config()
+    base_path = s3_config.public_files_bucket_path
+
+    new_s3_location = file_util.join(
+        base_path, "forms", str(form_id), "instructions", secure_file_name
+    )
+
+    # If updating, check if we need to delete old file
+    if form_instruction.file_location and form_instruction.file_location != new_s3_location:
+        # Delete old file
+        logger.info(
+            "Deleting old form instruction file",
+            extra={"old_file_location": form_instruction.file_location},
+        )
+        try:
+            file_util.delete_file(form_instruction.file_location)
+        except Exception:
+            logger.exception(
+                "Failed to delete old form instruction file",
+                extra={"old_file_location": form_instruction.file_location},
+            )
+
+    # Upload to S3
+    with file_util.open_stream(
+        new_s3_location, mode="wb", content_type=file_obj.mimetype
+    ) as writefile:
+        file_obj.save(writefile)
+
+    form_instruction.file_location = new_s3_location
+    form_instruction.file_name = file_util.get_file_name(file_obj.filename)
+
+    return form_instruction

api/tests/src/api/form_alpha/test_form_instruction_route.py

@@ -0,0 +1,143 @@
+import io
+import uuid
+from unittest.mock import MagicMock, patch
+
+import pytest
+from werkzeug.datastructures import FileStorage
+
+import src.adapters.db as db
+from src.constants.lookup_constants import Privilege
+from src.db.models.competition_models import FormInstruction
+from tests.src.db.models.factories import (
+    FormFactory,
+    FormInstructionFactory,
+    InternalUserRoleFactory,
+    UserApiKeyFactory,
+    UserFactory,
+)
+
+
+@pytest.fixture
+def form_instruction_file():
+    return FileStorage(
+        stream=io.BytesIO(b"test content"),
+        filename="instructions.pdf",
+        content_type="application/pdf",
+    )
+
+
+def test_form_instruction_upsert_create_new(
+    client, db_session: db.Session, form_instruction_file, enable_factory_create
+):
+    # Setup user with privilege
+    user = UserFactory.create()
+    InternalUserRoleFactory.create(user=user, role__privileges=[Privilege.UPDATE_FORM])
+    api_key = UserApiKeyFactory.create(user=user)
+
+    form = FormFactory.create()
+    form_instruction_id = uuid.uuid4()
+
+    # Mock S3
+    with patch("src.util.file_util.open_stream") as mock_open_stream:
+        mock_file = MagicMock(spec=io.BytesIO)
+        mock_open_stream.return_value.__enter__.return_value = mock_file
+
+        # Execute
+        resp = client.put(
+            f"/alpha/forms/{form.form_id}/form_instructions/{form_instruction_id}",
+            data={"file": form_instruction_file},
+            content_type="multipart/form-data",
+            headers={"X-API-Key": api_key.key_id},
+        )
+
+        # Verify
+        assert resp.status_code == 200
+
+        # Check DB
+        instruction = db_session.get(FormInstruction, form_instruction_id)
+        assert instruction is not None
+        assert instruction.file_name == "instructions.pdf"
+        assert f"forms/{form.form_id}/instructions/instructions.pdf" in instruction.file_location
+
+        # Check S3 upload
+        mock_open_stream.assert_called()
+
+
+def test_form_instruction_upsert_update_existing(
+    client, db_session: db.Session, form_instruction_file, enable_factory_create
+):
+    # Setup user with privilege
+    user = UserFactory.create()
+    InternalUserRoleFactory.create(user=user, role__privileges=[Privilege.UPDATE_FORM])
+    api_key = UserApiKeyFactory.create(user=user)
+
+    form = FormFactory.create()
+    existing_instruction = FormInstructionFactory.create()
+    existing_instruction.file_location = "s3://bucket/old/path/file.pdf"
+    db_session.add(existing_instruction)
+    db_session.commit()
+
+    # Mock S3 and delete_file
+    with patch("src.util.file_util.open_stream") as mock_open_stream, patch(
+        "src.util.file_util.delete_file"
+    ) as mock_delete_file:
+
+        mock_file = MagicMock(spec=io.BytesIO)
+        mock_open_stream.return_value.__enter__.return_value = mock_file
+
+        # Execute
+        resp = client.put(
+            f"/alpha/forms/{form.form_id}/form_instructions/{existing_instruction.form_instruction_id}",
+            data={"file": form_instruction_file},
+            content_type="multipart/form-data",
+            headers={"X-API-Key": api_key.key_id},
+        )
+
+        # Verify
+        assert resp.status_code == 200
+
+        # Check DB
+        db_session.refresh(existing_instruction)
+        assert existing_instruction.file_name == "instructions.pdf"
+        assert (
+            f"forms/{form.form_id}/instructions/instructions.pdf"
+            in existing_instruction.file_location
+        )
+
+        # Check old file deleted
+        mock_delete_file.assert_called_with("s3://bucket/old/path/file.pdf")
+
+
+def test_form_instruction_upsert_no_auth(client, db_session: db.Session, form_instruction_file):
+    form_id = uuid.uuid4()
+    form_instruction_id = uuid.uuid4()
+
+    resp = client.put(
+        f"/alpha/forms/{form_id}/form_instructions/{form_instruction_id}",
+        data={"file": form_instruction_file},
+        content_type="multipart/form-data",
+    )
+
+    assert resp.status_code == 401
+
+
+def test_form_instruction_upsert_wrong_privilege(
+    client, db_session: db.Session, form_instruction_file, enable_factory_create
+):
+    # User without UPDATE_FORM
+    user = UserFactory.create()
+    # Give some other privilege
+    InternalUserRoleFactory.create(user=user, role__privileges=[Privilege.VIEW_APPLICATION])
+    api_key = UserApiKeyFactory.create(user=user)
+
+    form_id = uuid.uuid4()
+    form_instruction_id = uuid.uuid4()
+
+    resp = client.put(
+        f"/alpha/forms/{form_id}/form_instructions/{form_instruction_id}",
+        data={"file": form_instruction_file},
+        content_type="multipart/form-data",
+        headers={"X-API-Key": api_key.key_id},
+    )
+
+    assert resp.status_code == 403