CDR: Add self-answerable questions (#793)

### Description

Please explain the changes you made here.

### Checklist

- [ ] Created tests which fail without the change (if possible)
- [ ] All tests passing
- [ ] Extended the documentation, if necessary

Author: Malem38

app/modules/cdr/cruds_cdr.py

@@ -867,6 +867,20 @@ def create_customdata_field(db: AsyncSession, datafield: models_cdr.CustomDataFi
     db.add(datafield)
 
 
+async def update_customdata_field(
+    db: AsyncSession,
+    field_id: UUID,
+    datafield: schemas_cdr.CustomDataFieldBase,
+):
+    await db.execute(
+        update(models_cdr.CustomDataField)
+        .where(
+            models_cdr.CustomDataField.id == field_id,
+        )
+        .values(**datafield.model_dump(exclude_none=True)),
+    )
+
+
 async def get_customdata_field(
     db: AsyncSession,
     field_id: UUID,

app/modules/cdr/endpoints_cdr.py

@@ -3020,13 +3020,56 @@ async def create_custom_data_field(
         id=uuid4(),
         product_id=product_id,
         name=custom_data_field.name,
+        can_user_answer=custom_data_field.can_user_answer,
     )
 
     cruds_cdr.create_customdata_field(db, db_data)
     await db.flush()
     return db_data
 
 
+@module.router.patch(
+    "/cdr/sellers/{seller_id}/products/{product_id}/data/{field_id}/",
+    status_code=204,
+)
+async def update_custom_data_field(
+    seller_id: UUID,
+    product_id: UUID,
+    field_id: UUID,
+    custom_data_field: schemas_cdr.CustomDataFieldBase,
+    db: AsyncSession = Depends(get_db),
+    user: models_users.CoreUser = Depends(is_user_a_member),
+):
+    await is_user_in_a_seller_group(
+        seller_id,
+        user,
+        db=db,
+    )
+    await check_request_consistency(db=db, seller_id=seller_id, product_id=product_id)
+    db_datafield = await cruds_cdr.get_customdata_field(db=db, field_id=field_id)
+    if db_datafield is None:
+        raise HTTPException(
+            status_code=404,
+            detail="Field not found.",
+        )
+    if db_datafield.product_id != product_id:
+        raise HTTPException(
+            status_code=403,
+            detail="Field does not belong to this product.",
+        )
+
+    datafield = schemas_cdr.CustomDataFieldBase(
+        name=custom_data_field.name,
+        can_user_answer=custom_data_field.can_user_answer,
+    )
+
+    await cruds_cdr.update_customdata_field(
+        db,
+        field_id=field_id,
+        datafield=datafield,
+    )
+
+
 @module.router.delete(
     "/cdr/sellers/{seller_id}/products/{product_id}/data/{field_id}/",
     status_code=204,
@@ -3105,18 +3148,29 @@ async def create_custom_data(
     db: AsyncSession = Depends(get_db),
     user: models_users.CoreUser = Depends(is_user_a_member),
 ):
-    await is_user_in_a_seller_group(
-        seller_id,
-        user,
-        db=db,
-    )
     await check_request_consistency(db=db, seller_id=seller_id, product_id=product_id)
     db_field = await cruds_cdr.get_customdata_field(db=db, field_id=field_id)
     if db_field is None:
         raise HTTPException(
             status_code=404,
             detail="Field not found.",
         )
+    if not (
+        is_user_member_of_any_group(user, [GroupType.admin_cdr])
+        or seller_id
+        in [
+            s.id
+            for s in await cruds_cdr.get_sellers_by_group_ids(
+                db=db,
+                group_ids=[g.id for g in user.groups],
+            )
+        ]
+    ) and not (db_field.can_user_answer and user_id == user.id):
+        raise HTTPException(
+            status_code=403,
+            detail="You are not authorized to add data for this field.",
+        )
+
     if db_field.product_id != product_id:
         raise HTTPException(
             status_code=403,
@@ -3147,18 +3201,28 @@ async def update_custom_data(
     db: AsyncSession = Depends(get_db),
     user: models_users.CoreUser = Depends(is_user_a_member),
 ):
-    await is_user_in_a_seller_group(
-        seller_id,
-        user,
-        db=db,
-    )
     await check_request_consistency(db=db, seller_id=seller_id, product_id=product_id)
     db_data = await cruds_cdr.get_customdata(db=db, field_id=field_id, user_id=user_id)
     if db_data is None:
         raise HTTPException(
             status_code=404,
             detail="Field Data not found.",
         )
+    if not (
+        is_user_member_of_any_group(user, [GroupType.admin_cdr])
+        or seller_id
+        in [
+            s.id
+            for s in await cruds_cdr.get_sellers_by_group_ids(
+                db=db,
+                group_ids=[g.id for g in user.groups],
+            )
+        ]
+    ) and not (db_data.field.can_user_answer and user_id == user.id):
+        raise HTTPException(
+            status_code=403,
+            detail="You are not authorized to edit data for this field.",
+        )
     if db_data.field.product_id != product_id:
         raise HTTPException(
             status_code=403,
@@ -3185,18 +3249,28 @@ async def delete_customdata(
     db: AsyncSession = Depends(get_db),
     user: models_users.CoreUser = Depends(is_user_a_member),
 ):
-    await is_user_in_a_seller_group(
-        seller_id,
-        user,
-        db=db,
-    )
     await check_request_consistency(db=db, seller_id=seller_id, product_id=product_id)
     db_data = await cruds_cdr.get_customdata(db=db, field_id=field_id, user_id=user_id)
     if db_data is None:
         raise HTTPException(
             status_code=404,
             detail="Field Data not found.",
         )
+    if not (
+        is_user_member_of_any_group(user, [GroupType.admin_cdr])
+        or seller_id
+        in [
+            s.id
+            for s in await cruds_cdr.get_sellers_by_group_ids(
+                db=db,
+                group_ids=[g.id for g in user.groups],
+            )
+        ]
+    ) and not (db_data.field.can_user_answer and user_id == user.id):
+        raise HTTPException(
+            status_code=403,
+            detail="You are not authorized to delete data for this field.",
+        )
     if db_data.field.product_id != product_id:
         raise HTTPException(
             status_code=403,

app/modules/cdr/models_cdr.py

@@ -305,6 +305,7 @@ class CustomDataField(Base):
         ForeignKey("cdr_product.id"),
     )
     name: Mapped[str]
+    can_user_answer: Mapped[bool]
 
 
 class CustomData(Base):

app/modules/cdr/schemas_cdr.py

@@ -301,6 +301,7 @@ class UpdateUserWSMessageModel(WSMessageModel):
 
 class CustomDataFieldBase(BaseModel):
     name: str
+    can_user_answer: bool
 
 
 class CustomDataFieldComplete(CustomDataFieldBase):

migrations/versions/38_auto_answerable_customdata.py

@@ -0,0 +1,46 @@
+"""auto-answerable customdata
+
+Create Date: 2025-08-20 14:06:53.519826
+"""
+
+from collections.abc import Sequence
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from pytest_alembic import MigrationContext
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "06c94803745b"
+down_revision: str | None = "1294f07e0c02"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "cdr_customdata_field",
+        sa.Column("can_user_answer", sa.Boolean(), nullable=True),
+    )
+    op.execute("UPDATE cdr_customdata_field SET can_user_answer = false")
+    op.alter_column("cdr_customdata_field", "can_user_answer", nullable=False)
+
+
+def downgrade() -> None:
+    op.drop_column("cdr_customdata_field", "can_user_answer")
+
+
+def pre_test_upgrade(
+    alembic_runner: "MigrationContext",
+    alembic_connection: sa.Connection,
+) -> None:
+    pass
+
+
+def test_upgrade(
+    alembic_runner: "MigrationContext",
+    alembic_connection: sa.Connection,
+) -> None:
+    pass

tests/test_cdr.py

@@ -2569,7 +2569,7 @@ async def test_validate_purchase(client: TestClient):
 def test_create_customdata_field(client: TestClient):
     response = client.post(
         f"/cdr/sellers/{seller.id}/products/{product.id}/data/",
-        json={"name": "Chambre"},
+        json={"name": "Chambre", "can_user_answer": False},
         headers={"Authorization": f"Bearer {token_bde}"},
     )
     assert response.status_code == 201
@@ -2578,7 +2578,7 @@ def test_create_customdata_field(client: TestClient):
 def test_create_customdata_field_user(client: TestClient):
     response = client.post(
         f"/cdr/sellers/{seller.id}/products/{product.id}/data/",
-        json={"name": "Chambre"},
+        json={"name": "Chambre", "can_user_answer": False},
         headers={"Authorization": f"Bearer {token_user}"},
     )
     assert response.status_code == 403
@@ -2589,6 +2589,7 @@ async def test_delete_customdata_field(client: TestClient):
         id=uuid.uuid4(),
         product_id=product.id,
         name="Supprime",
+        can_user_answer=False,
     )
     await add_object_to_db(field)
 
@@ -2610,6 +2611,7 @@ async def test_create_customdata(client: TestClient):
         id=uuid.uuid4(),
         product_id=product.id,
         name="Field",
+        can_user_answer=False,
     )
     await add_object_to_db(customdata_field)
     response = client.post(
@@ -2625,6 +2627,7 @@ async def test_create_customdata_user(client: TestClient):
         id=uuid.uuid4(),
         product_id=product.id,
         name="Field",
+        can_user_answer=False,
     )
     await add_object_to_db(customdata_field)
     response = client.post(
@@ -2640,6 +2643,7 @@ async def test_update_customdata(client: TestClient):
         id=uuid.uuid4(),
         product_id=product.id,
         name="Edit",
+        can_user_answer=False,
     )
     await add_object_to_db(field)
     customdata = models_cdr.CustomData(
@@ -2669,6 +2673,7 @@ async def test_update_customdata_user(client: TestClient):
         id=uuid.uuid4(),
         product_id=product.id,
         name="Edit",
+        can_user_answer=False,
     )
     await add_object_to_db(field)
     customdata = models_cdr.CustomData(
@@ -2691,6 +2696,7 @@ async def test_delete_customdata(client: TestClient):
         id=uuid.uuid4(),
         product_id=product.id,
         name="Supprime",
+        can_user_answer=False,
     )
     await add_object_to_db(field)
     customdata = models_cdr.CustomData(
@@ -2718,6 +2724,7 @@ async def test_customdata_deletion_on_purchase_deletion(client: TestClient):
         id=uuid.uuid4(),
         product_id=product.id,
         name="Supprime",
+        can_user_answer=False,
     )
     await add_object_to_db(field)
     customdata = models_cdr.CustomData(

tests/test_cdr_result.py

@@ -141,13 +141,15 @@ async def init_objects():
         id=uuid.uuid4(),
         product_id=product1.id,
         name="Champ 1",
+        can_user_answer=False,
     )
 
     global customdata_field2
     customdata_field2 = models_cdr.CustomDataField(
         id=uuid.uuid4(),
         product_id=product2.id,
         name="Champ 2",
+        can_user_answer=False,
     )
 
     global product1_variant1