forked from jamesp/sasa-membership
Add member profile questions, admin tooling, legal pages, and fast tests
- Add configurable profile questions with conditional visibility, admin-only fields, user answers, and seeded onboarding/volunteer questions
- Add admin UI for managing profile questions and member profile answers
- Add volunteer level/profile data support across backend schemas, models, API, and migration
- Update dashboard/profile UI, super admin menu, membership service types, and related styling
- Add privacy policy, terms of service, cookie notice, and footer links
- Add frontend Vitest coverage for profile question logic
- Add backend pytest coverage for profile answer normalization and validation
- Update restart.sh to build, run frontend/backend unit tests, and restart only after tests pass
- Refresh README, quickstart, project structure, instructions, and Square docs to match current app features
- Protect feature flag reload behind super-admin access
- Restrict admin-triggered password resets so admins can only reset member accounts
- Replace email template HTML preview rendering with escaped text preview
- Update docs for feature flag reload access, password reset scope, and email template preview safety
-- test user questions are also made by AI and not very useful. but i didn't know what to put there so its good enough for a test
This commit is contained in:
@@ -0,0 +1,77 @@
|
||||
"""Add volunteer level and dynamic profile questions
|
||||
|
||||
Revision ID: 2e8a0f9d4b31
|
||||
Revises: b583fd2cf202
|
||||
Create Date: 2026-05-04 17:50:00.000000
|
||||
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = '2e8a0f9d4b31'
|
||||
down_revision: Union[str, None] = 'b583fd2cf202'
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.add_column('users', sa.Column('volunteer_level', sa.String(length=50), nullable=True))
|
||||
|
||||
op.create_table(
|
||||
'profile_questions',
|
||||
sa.Column('id', sa.Integer(), nullable=False),
|
||||
sa.Column('key', sa.String(length=100), nullable=False),
|
||||
sa.Column('label', sa.String(length=255), nullable=False),
|
||||
sa.Column('help_text', sa.Text(), nullable=True),
|
||||
sa.Column('input_type', sa.String(length=30), nullable=False),
|
||||
sa.Column('placeholder', sa.String(length=255), nullable=True),
|
||||
sa.Column('options_json', sa.Text(), nullable=True),
|
||||
sa.Column('is_required', sa.Boolean(), nullable=False),
|
||||
sa.Column('is_active', sa.Boolean(), nullable=False),
|
||||
sa.Column('admin_only_edit', sa.Boolean(), nullable=False),
|
||||
sa.Column('display_order', sa.Integer(), nullable=False),
|
||||
sa.Column('depends_on_question_id', sa.Integer(), nullable=True),
|
||||
sa.Column('depends_on_value', sa.String(length=255), nullable=True),
|
||||
sa.Column('created_at', sa.DateTime(), nullable=False),
|
||||
sa.Column('updated_at', sa.DateTime(), nullable=False),
|
||||
sa.ForeignKeyConstraint(['depends_on_question_id'], ['profile_questions.id']),
|
||||
sa.PrimaryKeyConstraint('id'),
|
||||
)
|
||||
op.create_index(op.f('ix_profile_questions_id'), 'profile_questions', ['id'], unique=False)
|
||||
op.create_index(op.f('ix_profile_questions_key'), 'profile_questions', ['key'], unique=True)
|
||||
|
||||
op.create_table(
|
||||
'user_profile_answers',
|
||||
sa.Column('id', sa.Integer(), nullable=False),
|
||||
sa.Column('user_id', sa.Integer(), nullable=False),
|
||||
sa.Column('question_id', sa.Integer(), nullable=False),
|
||||
sa.Column('value_text', sa.Text(), nullable=True),
|
||||
sa.Column('updated_by_user_id', sa.Integer(), nullable=True),
|
||||
sa.Column('created_at', sa.DateTime(), nullable=False),
|
||||
sa.Column('updated_at', sa.DateTime(), nullable=False),
|
||||
sa.ForeignKeyConstraint(['question_id'], ['profile_questions.id']),
|
||||
sa.ForeignKeyConstraint(['updated_by_user_id'], ['users.id']),
|
||||
sa.ForeignKeyConstraint(['user_id'], ['users.id']),
|
||||
sa.PrimaryKeyConstraint('id'),
|
||||
sa.UniqueConstraint('user_id', 'question_id', name='uq_user_profile_answer'),
|
||||
)
|
||||
op.create_index(op.f('ix_user_profile_answers_id'), 'user_profile_answers', ['id'], unique=False)
|
||||
op.create_index(op.f('ix_user_profile_answers_question_id'), 'user_profile_answers', ['question_id'], unique=False)
|
||||
op.create_index(op.f('ix_user_profile_answers_user_id'), 'user_profile_answers', ['user_id'], unique=False)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_index(op.f('ix_user_profile_answers_user_id'), table_name='user_profile_answers')
|
||||
op.drop_index(op.f('ix_user_profile_answers_question_id'), table_name='user_profile_answers')
|
||||
op.drop_index(op.f('ix_user_profile_answers_id'), table_name='user_profile_answers')
|
||||
op.drop_table('user_profile_answers')
|
||||
|
||||
op.drop_index(op.f('ix_profile_questions_key'), table_name='profile_questions')
|
||||
op.drop_index(op.f('ix_profile_questions_id'), table_name='profile_questions')
|
||||
op.drop_table('profile_questions')
|
||||
|
||||
op.drop_column('users', 'volunteer_level')
|
||||
@@ -1,7 +1,7 @@
|
||||
from fastapi import APIRouter, Depends
|
||||
from typing import Dict, Any
|
||||
from app.services.feature_flag_service import feature_flags
|
||||
from app.schemas.feature_flags import FeatureFlagsResponse, FeatureFlagResponse
|
||||
from app.api.dependencies import get_super_admin_user
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
@@ -38,10 +38,11 @@ async def get_feature_flag(flag_name: str) -> FeatureFlagResponse:
|
||||
|
||||
|
||||
@router.post("/flags/reload")
|
||||
async def reload_feature_flags():
|
||||
async def reload_feature_flags(
|
||||
current_user = Depends(get_super_admin_user),
|
||||
):
|
||||
"""
|
||||
Reload feature flags from environment variables
|
||||
This could be protected with admin permissions in production
|
||||
Reload feature flags from environment variables.
|
||||
"""
|
||||
feature_flags.reload_flags()
|
||||
return {"message": "Feature flags reloaded successfully"}
|
||||
return {"message": "Feature flags reloaded successfully"}
|
||||
|
||||
+641
-17
@@ -1,16 +1,188 @@
|
||||
import json
|
||||
from datetime import date, datetime, timedelta
|
||||
from typing import Any, List, Optional
|
||||
import uuid
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from sqlalchemy.orm import Session
|
||||
from typing import List
|
||||
|
||||
from ...core.database import get_db
|
||||
from ...core.security import get_password_hash
|
||||
from ...models.models import User
|
||||
from ...schemas import UserResponse, UserUpdate, MessageResponse
|
||||
from ...models.models import ProfileQuestion, User, UserProfileAnswer, UserRole, PasswordResetToken
|
||||
from ...schemas import (
|
||||
MessageResponse,
|
||||
ProfileAnswersUpdateRequest,
|
||||
ProfileQuestionCreate,
|
||||
ProfileQuestionForUser,
|
||||
ProfileQuestionResponse,
|
||||
ProfileQuestionUpdate,
|
||||
UserResponse,
|
||||
UserUpdate,
|
||||
)
|
||||
from ...api.dependencies import get_current_active_user, get_admin_user
|
||||
from ...services.email_service import email_service
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
def _parse_options(options_json: Optional[str]) -> list[dict[str, str]]:
|
||||
if not options_json:
|
||||
return []
|
||||
try:
|
||||
parsed = json.loads(options_json)
|
||||
except (TypeError, json.JSONDecodeError):
|
||||
return []
|
||||
|
||||
if not isinstance(parsed, list):
|
||||
return []
|
||||
|
||||
normalized: list[dict[str, str]] = []
|
||||
for item in parsed:
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
label = str(item.get("label", "")).strip()
|
||||
value = str(item.get("value", "")).strip()
|
||||
if label and value:
|
||||
normalized.append({"label": label, "value": value})
|
||||
return normalized
|
||||
|
||||
|
||||
def _serialize_options(options: Optional[list[Any]]) -> Optional[str]:
|
||||
if not options:
|
||||
return None
|
||||
normalized = []
|
||||
for item in options:
|
||||
data = item.model_dump() if hasattr(item, "model_dump") else item
|
||||
normalized.append({"label": str(data["label"]), "value": str(data["value"])})
|
||||
return json.dumps(normalized)
|
||||
|
||||
|
||||
def _normalize_answer_value(question: ProfileQuestion, value: Any) -> Optional[str]:
|
||||
if value is None:
|
||||
return None
|
||||
|
||||
if isinstance(value, str) and value.strip() == "":
|
||||
return None
|
||||
|
||||
input_type = question.input_type
|
||||
|
||||
if input_type == "boolean":
|
||||
if isinstance(value, bool):
|
||||
return "true" if value else "false"
|
||||
|
||||
text = str(value).strip().lower()
|
||||
if text in {"true", "1", "yes", "y"}:
|
||||
return "true"
|
||||
if text in {"false", "0", "no", "n"}:
|
||||
return "false"
|
||||
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"Invalid boolean answer for question '{question.key}'"
|
||||
)
|
||||
|
||||
if input_type == "number":
|
||||
try:
|
||||
number = float(value)
|
||||
return str(int(number)) if number.is_integer() else str(number)
|
||||
except (TypeError, ValueError):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"Invalid number answer for question '{question.key}'"
|
||||
)
|
||||
|
||||
if input_type == "date":
|
||||
if isinstance(value, datetime):
|
||||
return value.date().isoformat()
|
||||
if isinstance(value, date):
|
||||
return value.isoformat()
|
||||
|
||||
text = str(value).strip()
|
||||
try:
|
||||
parsed = datetime.strptime(text, "%Y-%m-%d")
|
||||
return parsed.date().isoformat()
|
||||
except ValueError:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"Invalid date answer for question '{question.key}'. Use YYYY-MM-DD"
|
||||
)
|
||||
|
||||
if input_type == "select":
|
||||
selected = str(value).strip()
|
||||
option_values = {opt["value"] for opt in _parse_options(question.options_json)}
|
||||
if selected not in option_values:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"Invalid selection for question '{question.key}'"
|
||||
)
|
||||
return selected
|
||||
|
||||
return str(value).strip()
|
||||
|
||||
|
||||
def _deserialize_answer_value(question: ProfileQuestion, value_text: Optional[str]) -> Any:
|
||||
if value_text is None:
|
||||
return None
|
||||
|
||||
if question.input_type == "boolean":
|
||||
return value_text.lower() == "true"
|
||||
|
||||
if question.input_type == "number":
|
||||
try:
|
||||
number = float(value_text)
|
||||
return int(number) if number.is_integer() else number
|
||||
except ValueError:
|
||||
return value_text
|
||||
|
||||
return value_text
|
||||
|
||||
|
||||
def _validate_question_dependencies(
|
||||
db: Session,
|
||||
depends_on_question_id: Optional[int],
|
||||
depends_on_value: Optional[str],
|
||||
current_question_id: Optional[int] = None,
|
||||
) -> None:
|
||||
if depends_on_question_id is None:
|
||||
if depends_on_value is not None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="depends_on_value requires depends_on_question_id"
|
||||
)
|
||||
return
|
||||
|
||||
dependent_question = db.query(ProfileQuestion).filter(ProfileQuestion.id == depends_on_question_id).first()
|
||||
if not dependent_question:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="depends_on_question_id does not exist"
|
||||
)
|
||||
|
||||
if current_question_id is not None and current_question_id == depends_on_question_id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="A question cannot depend on itself"
|
||||
)
|
||||
|
||||
|
||||
def _normalize_volunteer_level(value: Optional[str]) -> Optional[str]:
|
||||
if value is None:
|
||||
return None
|
||||
|
||||
normalized = str(value).strip().lower()
|
||||
if normalized == "":
|
||||
return None
|
||||
|
||||
if normalized in {"yes", "true", "1"}:
|
||||
return "yes"
|
||||
if normalized in {"no", "false", "0"}:
|
||||
return "no"
|
||||
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Volunteer flag must be yes or no"
|
||||
)
|
||||
|
||||
|
||||
@router.get("/me", response_model=UserResponse)
|
||||
async def get_current_user_profile(
|
||||
current_user: User = Depends(get_current_active_user)
|
||||
@@ -27,25 +199,123 @@ async def update_current_user_profile(
|
||||
):
|
||||
"""Update current user's profile"""
|
||||
update_data = user_update.model_dump(exclude_unset=True)
|
||||
|
||||
# Check email uniqueness if email is being updated
|
||||
if 'email' in update_data and update_data['email'] != current_user.email:
|
||||
existing_user = db.query(User).filter(User.email == update_data['email']).first()
|
||||
|
||||
# Prevent privilege and volunteer-level edits through self-service profile endpoint.
|
||||
update_data.pop("role", None)
|
||||
update_data.pop("volunteer_level", None)
|
||||
|
||||
if "email" in update_data and update_data["email"] != current_user.email:
|
||||
existing_user = db.query(User).filter(User.email == update_data["email"]).first()
|
||||
if existing_user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Email already registered"
|
||||
)
|
||||
|
||||
|
||||
for field, value in update_data.items():
|
||||
setattr(current_user, field, value)
|
||||
|
||||
|
||||
db.commit()
|
||||
db.refresh(current_user)
|
||||
|
||||
|
||||
return current_user
|
||||
|
||||
|
||||
@router.get("/me/profile-questions", response_model=List[ProfileQuestionForUser])
|
||||
async def list_my_profile_questions(
|
||||
current_user: User = Depends(get_current_active_user),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
questions = db.query(ProfileQuestion).filter(ProfileQuestion.is_active == True).order_by(ProfileQuestion.display_order.asc(), ProfileQuestion.id.asc()).all()
|
||||
|
||||
answers = db.query(UserProfileAnswer).filter(UserProfileAnswer.user_id == current_user.id).all()
|
||||
answers_by_question = {answer.question_id: answer for answer in answers}
|
||||
|
||||
response: list[ProfileQuestionForUser] = []
|
||||
for question in questions:
|
||||
user_answer = answers_by_question.get(question.id)
|
||||
can_edit = (not question.admin_only_edit) or (current_user.role in [UserRole.ADMIN, UserRole.SUPER_ADMIN])
|
||||
|
||||
response.append(
|
||||
ProfileQuestionForUser(
|
||||
id=question.id,
|
||||
key=question.key,
|
||||
label=question.label,
|
||||
help_text=question.help_text,
|
||||
input_type=question.input_type,
|
||||
placeholder=question.placeholder,
|
||||
options=_parse_options(question.options_json),
|
||||
is_required=question.is_required,
|
||||
is_active=question.is_active,
|
||||
admin_only_edit=question.admin_only_edit,
|
||||
display_order=question.display_order,
|
||||
depends_on_question_id=question.depends_on_question_id,
|
||||
depends_on_value=question.depends_on_value,
|
||||
created_at=question.created_at,
|
||||
updated_at=question.updated_at,
|
||||
answer=_deserialize_answer_value(question, user_answer.value_text if user_answer else None),
|
||||
can_edit=can_edit,
|
||||
)
|
||||
)
|
||||
|
||||
return response
|
||||
|
||||
|
||||
@router.put("/me/profile-answers", response_model=MessageResponse)
|
||||
async def update_my_profile_answers(
|
||||
payload: ProfileAnswersUpdateRequest,
|
||||
current_user: User = Depends(get_current_active_user),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
if not payload.answers:
|
||||
return {"message": "No changes submitted"}
|
||||
|
||||
question_ids = {item.question_id for item in payload.answers}
|
||||
questions = db.query(ProfileQuestion).filter(ProfileQuestion.id.in_(question_ids), ProfileQuestion.is_active == True).all()
|
||||
questions_by_id = {question.id: question for question in questions}
|
||||
|
||||
missing_ids = [q_id for q_id in question_ids if q_id not in questions_by_id]
|
||||
if missing_ids:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Questions not found: {missing_ids}"
|
||||
)
|
||||
|
||||
for item in payload.answers:
|
||||
question = questions_by_id[item.question_id]
|
||||
if question.admin_only_edit:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail=f"Question '{question.label}' can only be changed by admins"
|
||||
)
|
||||
|
||||
normalized_value = _normalize_answer_value(question, item.value)
|
||||
|
||||
answer = db.query(UserProfileAnswer).filter(
|
||||
UserProfileAnswer.user_id == current_user.id,
|
||||
UserProfileAnswer.question_id == question.id
|
||||
).first()
|
||||
|
||||
if normalized_value is None:
|
||||
if answer:
|
||||
db.delete(answer)
|
||||
continue
|
||||
|
||||
if answer:
|
||||
answer.value_text = normalized_value
|
||||
answer.updated_by_user_id = current_user.id
|
||||
else:
|
||||
db.add(UserProfileAnswer(
|
||||
user_id=current_user.id,
|
||||
question_id=question.id,
|
||||
value_text=normalized_value,
|
||||
updated_by_user_id=current_user.id,
|
||||
))
|
||||
|
||||
db.commit()
|
||||
return {"message": "Profile answers updated successfully"}
|
||||
|
||||
|
||||
@router.get("/", response_model=List[UserResponse])
|
||||
async def list_users(
|
||||
skip: int = 0,
|
||||
@@ -58,6 +328,281 @@ async def list_users(
|
||||
return users
|
||||
|
||||
|
||||
@router.get("/admin/profile-questions", response_model=List[ProfileQuestionResponse])
|
||||
async def list_profile_questions_admin(
|
||||
include_inactive: bool = True,
|
||||
current_user: User = Depends(get_admin_user),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
query = db.query(ProfileQuestion)
|
||||
if not include_inactive:
|
||||
query = query.filter(ProfileQuestion.is_active == True)
|
||||
|
||||
questions = query.order_by(ProfileQuestion.display_order.asc(), ProfileQuestion.id.asc()).all()
|
||||
|
||||
return [
|
||||
ProfileQuestionResponse(
|
||||
id=question.id,
|
||||
key=question.key,
|
||||
label=question.label,
|
||||
help_text=question.help_text,
|
||||
input_type=question.input_type,
|
||||
placeholder=question.placeholder,
|
||||
options=_parse_options(question.options_json),
|
||||
is_required=question.is_required,
|
||||
is_active=question.is_active,
|
||||
admin_only_edit=question.admin_only_edit,
|
||||
display_order=question.display_order,
|
||||
depends_on_question_id=question.depends_on_question_id,
|
||||
depends_on_value=question.depends_on_value,
|
||||
created_at=question.created_at,
|
||||
updated_at=question.updated_at,
|
||||
)
|
||||
for question in questions
|
||||
]
|
||||
|
||||
|
||||
@router.post("/admin/profile-questions", response_model=ProfileQuestionResponse)
|
||||
async def create_profile_question_admin(
|
||||
payload: ProfileQuestionCreate,
|
||||
current_user: User = Depends(get_admin_user),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
if payload.input_type == "select" and not payload.options:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Select questions require options"
|
||||
)
|
||||
|
||||
_validate_question_dependencies(db, payload.depends_on_question_id, payload.depends_on_value)
|
||||
|
||||
existing = db.query(ProfileQuestion).filter(ProfileQuestion.key == payload.key).first()
|
||||
if existing:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Question key already exists"
|
||||
)
|
||||
|
||||
question = ProfileQuestion(
|
||||
key=payload.key,
|
||||
label=payload.label,
|
||||
help_text=payload.help_text,
|
||||
input_type=payload.input_type,
|
||||
placeholder=payload.placeholder,
|
||||
options_json=_serialize_options(payload.options),
|
||||
is_required=payload.is_required,
|
||||
is_active=payload.is_active,
|
||||
admin_only_edit=payload.admin_only_edit,
|
||||
display_order=payload.display_order,
|
||||
depends_on_question_id=payload.depends_on_question_id,
|
||||
depends_on_value=payload.depends_on_value,
|
||||
)
|
||||
|
||||
db.add(question)
|
||||
db.commit()
|
||||
db.refresh(question)
|
||||
|
||||
return ProfileQuestionResponse(
|
||||
id=question.id,
|
||||
key=question.key,
|
||||
label=question.label,
|
||||
help_text=question.help_text,
|
||||
input_type=question.input_type,
|
||||
placeholder=question.placeholder,
|
||||
options=_parse_options(question.options_json),
|
||||
is_required=question.is_required,
|
||||
is_active=question.is_active,
|
||||
admin_only_edit=question.admin_only_edit,
|
||||
display_order=question.display_order,
|
||||
depends_on_question_id=question.depends_on_question_id,
|
||||
depends_on_value=question.depends_on_value,
|
||||
created_at=question.created_at,
|
||||
updated_at=question.updated_at,
|
||||
)
|
||||
|
||||
|
||||
@router.put("/admin/profile-questions/{question_id}", response_model=ProfileQuestionResponse)
|
||||
async def update_profile_question_admin(
|
||||
question_id: int,
|
||||
payload: ProfileQuestionUpdate,
|
||||
current_user: User = Depends(get_admin_user),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
question = db.query(ProfileQuestion).filter(ProfileQuestion.id == question_id).first()
|
||||
if not question:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Question not found"
|
||||
)
|
||||
|
||||
update_data = payload.model_dump(exclude_unset=True)
|
||||
|
||||
if "key" in update_data and update_data["key"] != question.key:
|
||||
existing = db.query(ProfileQuestion).filter(ProfileQuestion.key == update_data["key"]).first()
|
||||
if existing:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Question key already exists"
|
||||
)
|
||||
|
||||
input_type = update_data.get("input_type", question.input_type)
|
||||
options = update_data.get("options")
|
||||
options_to_validate = options if options is not None else _parse_options(question.options_json)
|
||||
if input_type == "select" and not options_to_validate:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Select questions require options"
|
||||
)
|
||||
|
||||
depends_on_question_id = update_data.get("depends_on_question_id", question.depends_on_question_id)
|
||||
depends_on_value = update_data.get("depends_on_value", question.depends_on_value)
|
||||
_validate_question_dependencies(db, depends_on_question_id, depends_on_value, current_question_id=question.id)
|
||||
|
||||
for field, value in update_data.items():
|
||||
if field == "options":
|
||||
question.options_json = _serialize_options(value)
|
||||
else:
|
||||
setattr(question, field, value)
|
||||
|
||||
db.commit()
|
||||
db.refresh(question)
|
||||
|
||||
return ProfileQuestionResponse(
|
||||
id=question.id,
|
||||
key=question.key,
|
||||
label=question.label,
|
||||
help_text=question.help_text,
|
||||
input_type=question.input_type,
|
||||
placeholder=question.placeholder,
|
||||
options=_parse_options(question.options_json),
|
||||
is_required=question.is_required,
|
||||
is_active=question.is_active,
|
||||
admin_only_edit=question.admin_only_edit,
|
||||
display_order=question.display_order,
|
||||
depends_on_question_id=question.depends_on_question_id,
|
||||
depends_on_value=question.depends_on_value,
|
||||
created_at=question.created_at,
|
||||
updated_at=question.updated_at,
|
||||
)
|
||||
|
||||
|
||||
@router.delete("/admin/profile-questions/{question_id}", response_model=MessageResponse)
|
||||
async def deactivate_profile_question_admin(
|
||||
question_id: int,
|
||||
current_user: User = Depends(get_admin_user),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
question = db.query(ProfileQuestion).filter(ProfileQuestion.id == question_id).first()
|
||||
if not question:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Question not found"
|
||||
)
|
||||
|
||||
question.is_active = False
|
||||
db.commit()
|
||||
|
||||
return {"message": "Question deactivated successfully"}
|
||||
|
||||
|
||||
@router.get("/admin/users/{user_id}/profile-answers", response_model=List[ProfileQuestionForUser])
|
||||
async def get_user_profile_answers_admin(
|
||||
user_id: int,
|
||||
current_user: User = Depends(get_admin_user),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
user = db.query(User).filter(User.id == user_id).first()
|
||||
if not user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="User not found"
|
||||
)
|
||||
|
||||
questions = db.query(ProfileQuestion).order_by(ProfileQuestion.display_order.asc(), ProfileQuestion.id.asc()).all()
|
||||
answers = db.query(UserProfileAnswer).filter(UserProfileAnswer.user_id == user_id).all()
|
||||
answers_by_question = {answer.question_id: answer for answer in answers}
|
||||
|
||||
return [
|
||||
ProfileQuestionForUser(
|
||||
id=question.id,
|
||||
key=question.key,
|
||||
label=question.label,
|
||||
help_text=question.help_text,
|
||||
input_type=question.input_type,
|
||||
placeholder=question.placeholder,
|
||||
options=_parse_options(question.options_json),
|
||||
is_required=question.is_required,
|
||||
is_active=question.is_active,
|
||||
admin_only_edit=question.admin_only_edit,
|
||||
display_order=question.display_order,
|
||||
depends_on_question_id=question.depends_on_question_id,
|
||||
depends_on_value=question.depends_on_value,
|
||||
created_at=question.created_at,
|
||||
updated_at=question.updated_at,
|
||||
answer=_deserialize_answer_value(question, answers_by_question.get(question.id).value_text if answers_by_question.get(question.id) else None),
|
||||
can_edit=True,
|
||||
)
|
||||
for question in questions
|
||||
]
|
||||
|
||||
|
||||
@router.put("/admin/users/{user_id}/profile-answers", response_model=MessageResponse)
|
||||
async def update_user_profile_answers_admin(
|
||||
user_id: int,
|
||||
payload: ProfileAnswersUpdateRequest,
|
||||
current_user: User = Depends(get_admin_user),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
user = db.query(User).filter(User.id == user_id).first()
|
||||
if not user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="User not found"
|
||||
)
|
||||
|
||||
if not payload.answers:
|
||||
return {"message": "No changes submitted"}
|
||||
|
||||
question_ids = {item.question_id for item in payload.answers}
|
||||
questions = db.query(ProfileQuestion).filter(ProfileQuestion.id.in_(question_ids)).all()
|
||||
questions_by_id = {question.id: question for question in questions}
|
||||
|
||||
missing_ids = [q_id for q_id in question_ids if q_id not in questions_by_id]
|
||||
if missing_ids:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Questions not found: {missing_ids}"
|
||||
)
|
||||
|
||||
for item in payload.answers:
|
||||
question = questions_by_id[item.question_id]
|
||||
normalized_value = _normalize_answer_value(question, item.value)
|
||||
|
||||
answer = db.query(UserProfileAnswer).filter(
|
||||
UserProfileAnswer.user_id == user_id,
|
||||
UserProfileAnswer.question_id == question.id
|
||||
).first()
|
||||
|
||||
if normalized_value is None:
|
||||
if answer:
|
||||
db.delete(answer)
|
||||
continue
|
||||
|
||||
if answer:
|
||||
answer.value_text = normalized_value
|
||||
answer.updated_by_user_id = current_user.id
|
||||
else:
|
||||
db.add(UserProfileAnswer(
|
||||
user_id=user_id,
|
||||
question_id=question.id,
|
||||
value_text=normalized_value,
|
||||
updated_by_user_id=current_user.id,
|
||||
))
|
||||
|
||||
db.commit()
|
||||
return {"message": "User profile answers updated successfully"}
|
||||
|
||||
|
||||
@router.get("/{user_id}", response_model=UserResponse)
|
||||
async def get_user(
|
||||
user_id: int,
|
||||
@@ -88,18 +633,97 @@ async def update_user(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="User not found"
|
||||
)
|
||||
|
||||
|
||||
update_data = user_update.model_dump(exclude_unset=True)
|
||||
|
||||
|
||||
if "email" in update_data and update_data["email"] != user.email:
|
||||
existing_user = db.query(User).filter(User.email == update_data["email"]).first()
|
||||
if existing_user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Email already registered"
|
||||
)
|
||||
|
||||
if "role" in update_data and update_data["role"] == UserRole.SUPER_ADMIN and current_user.role != UserRole.SUPER_ADMIN:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Only super admins can assign super admin role"
|
||||
)
|
||||
|
||||
if "volunteer_level" in update_data:
|
||||
update_data["volunteer_level"] = _normalize_volunteer_level(update_data["volunteer_level"])
|
||||
|
||||
for field, value in update_data.items():
|
||||
setattr(user, field, value)
|
||||
|
||||
|
||||
db.commit()
|
||||
db.refresh(user)
|
||||
|
||||
|
||||
return user
|
||||
|
||||
|
||||
@router.post("/{user_id}/send-password-reset", response_model=MessageResponse)
|
||||
async def send_user_password_reset(
|
||||
user_id: int,
|
||||
current_user: User = Depends(get_admin_user),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
"""Send a one-time password reset link email for a user."""
|
||||
user = db.query(User).filter(User.id == user_id).first()
|
||||
if not user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="User not found"
|
||||
)
|
||||
|
||||
if user.role in [UserRole.ADMIN, UserRole.SUPER_ADMIN] and current_user.role != UserRole.SUPER_ADMIN:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Only super admins can send password reset emails for admin users"
|
||||
)
|
||||
|
||||
if not user.is_active:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Cannot reset password for inactive user"
|
||||
)
|
||||
|
||||
db.query(PasswordResetToken).filter(
|
||||
PasswordResetToken.user_id == user.id,
|
||||
PasswordResetToken.used == False,
|
||||
PasswordResetToken.expires_at > datetime.utcnow()
|
||||
).update({"used": True})
|
||||
|
||||
reset_token = str(uuid.uuid4())
|
||||
expires_at = datetime.utcnow() + timedelta(hours=1)
|
||||
|
||||
db_token = PasswordResetToken(
|
||||
user_id=user.id,
|
||||
token=reset_token,
|
||||
expires_at=expires_at,
|
||||
used=False
|
||||
)
|
||||
|
||||
db.add(db_token)
|
||||
db.commit()
|
||||
|
||||
try:
|
||||
await email_service.send_password_reset_email(
|
||||
to_email=user.email,
|
||||
first_name=user.first_name,
|
||||
reset_token=reset_token,
|
||||
db=db
|
||||
)
|
||||
except Exception as exc:
|
||||
print(f"Failed to send admin password reset email: {exc}")
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail="Failed to send reset email"
|
||||
)
|
||||
|
||||
return {"message": "One-time password reset email sent successfully"}
|
||||
|
||||
|
||||
@router.delete("/{user_id}", response_model=MessageResponse)
|
||||
async def delete_user(
|
||||
user_id: int,
|
||||
@@ -113,8 +737,8 @@ async def delete_user(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="User not found"
|
||||
)
|
||||
|
||||
|
||||
db.delete(user)
|
||||
db.commit()
|
||||
|
||||
|
||||
return {"message": "User deleted successfully"}
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
from sqlalchemy.orm import Session
|
||||
from ..models.models import MembershipTier, User, UserRole, EmailTemplate
|
||||
import json
|
||||
|
||||
from ..models.models import MembershipTier, User, UserRole, EmailTemplate, ProfileQuestion
|
||||
from .security import get_password_hash
|
||||
from datetime import datetime
|
||||
|
||||
@@ -70,3 +72,99 @@ def init_default_data(db: Session):
|
||||
db.add_all(default_templates)
|
||||
db.commit()
|
||||
print(f"✓ Created {len(default_templates)} default email templates")
|
||||
|
||||
# Seed default profile questions for onboarding and profile attributes
|
||||
existing_questions = db.query(ProfileQuestion).count()
|
||||
if existing_questions == 0:
|
||||
print("Creating default profile questions...")
|
||||
default_questions = [
|
||||
ProfileQuestion(
|
||||
key="has_professional_license",
|
||||
label="Do you hold a professional aviation-related license?",
|
||||
help_text="Select your current license status.",
|
||||
input_type="select",
|
||||
options_json=json.dumps([
|
||||
{"label": "No", "value": "none"},
|
||||
{"label": "Student", "value": "student"},
|
||||
{"label": "Private Pilot", "value": "ppl"},
|
||||
{"label": "Commercial Pilot", "value": "cpl"},
|
||||
{"label": "ATPL", "value": "atpl"},
|
||||
{"label": "Instructor", "value": "instructor"},
|
||||
]),
|
||||
is_required=False,
|
||||
is_active=True,
|
||||
admin_only_edit=False,
|
||||
display_order=10,
|
||||
),
|
||||
ProfileQuestion(
|
||||
key="license_number",
|
||||
label="License number",
|
||||
help_text="Optional: your current license number.",
|
||||
input_type="text",
|
||||
placeholder="e.g. UK.FCL.123456",
|
||||
is_required=False,
|
||||
is_active=True,
|
||||
admin_only_edit=False,
|
||||
display_order=20,
|
||||
depends_on_value="ppl",
|
||||
),
|
||||
ProfileQuestion(
|
||||
key="can_support_events",
|
||||
label="Can you support airport or membership events?",
|
||||
help_text="Choose yes if you're open to helping with events.",
|
||||
input_type="boolean",
|
||||
is_required=False,
|
||||
is_active=True,
|
||||
admin_only_edit=False,
|
||||
display_order=30,
|
||||
),
|
||||
ProfileQuestion(
|
||||
key="event_support_notes",
|
||||
label="What support can you offer?",
|
||||
help_text="Examples: stewarding, admin desk, setup/packdown, mentoring.",
|
||||
input_type="text",
|
||||
placeholder="Type details here",
|
||||
is_required=False,
|
||||
is_active=True,
|
||||
admin_only_edit=False,
|
||||
display_order=40,
|
||||
depends_on_value="true",
|
||||
),
|
||||
ProfileQuestion(
|
||||
key="hours_available_monthly",
|
||||
label="Approximate volunteer hours available each month",
|
||||
help_text="Optional estimate in hours.",
|
||||
input_type="number",
|
||||
is_required=False,
|
||||
is_active=True,
|
||||
admin_only_edit=False,
|
||||
display_order=50,
|
||||
),
|
||||
ProfileQuestion(
|
||||
key="medical_expiry_date",
|
||||
label="Medical certificate expiry date",
|
||||
help_text="Optional date in YYYY-MM-DD format.",
|
||||
input_type="date",
|
||||
is_required=False,
|
||||
is_active=True,
|
||||
admin_only_edit=False,
|
||||
display_order=60,
|
||||
),
|
||||
ProfileQuestion(
|
||||
key="completed_training_x",
|
||||
label="Completed Training X",
|
||||
help_text="This is set by admins once verified.",
|
||||
input_type="boolean",
|
||||
is_required=False,
|
||||
is_active=True,
|
||||
admin_only_edit=True,
|
||||
display_order=70,
|
||||
),
|
||||
]
|
||||
db.add_all(default_questions)
|
||||
db.commit()
|
||||
question_by_key = {question.key: question for question in db.query(ProfileQuestion).all()}
|
||||
question_by_key["license_number"].depends_on_question_id = question_by_key["has_professional_license"].id
|
||||
question_by_key["event_support_notes"].depends_on_question_id = question_by_key["can_support_events"].id
|
||||
db.commit()
|
||||
print(f"✓ Created {len(default_questions)} default profile questions")
|
||||
|
||||
@@ -15,6 +15,8 @@ from .models import (
|
||||
VolunteerRole,
|
||||
VolunteerAssignment,
|
||||
VolunteerSchedule,
|
||||
ProfileQuestion,
|
||||
UserProfileAnswer,
|
||||
Certificate,
|
||||
File,
|
||||
Notification,
|
||||
@@ -36,6 +38,8 @@ __all__ = [
|
||||
"VolunteerRole",
|
||||
"VolunteerAssignment",
|
||||
"VolunteerSchedule",
|
||||
"ProfileQuestion",
|
||||
"UserProfileAnswer",
|
||||
"Certificate",
|
||||
"File",
|
||||
"Notification",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
from sqlalchemy import (
|
||||
Column, Integer, String, DateTime, Boolean, Enum as SQLEnum,
|
||||
Float, Text, ForeignKey, Date
|
||||
Float, Text, ForeignKey, Date, UniqueConstraint
|
||||
)
|
||||
from sqlalchemy.orm import relationship
|
||||
from datetime import datetime
|
||||
@@ -60,6 +60,7 @@ class User(Base):
|
||||
phone = Column(String(20), nullable=True)
|
||||
address = Column(Text, nullable=True)
|
||||
role = Column(SQLEnum(UserRole, values_callable=lambda x: [e.value for e in x]), default=UserRole.MEMBER, nullable=False)
|
||||
volunteer_level = Column(String(50), nullable=True)
|
||||
is_active = Column(Boolean, default=True, nullable=False)
|
||||
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
|
||||
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False)
|
||||
@@ -71,6 +72,54 @@ class User(Base):
|
||||
event_rsvps = relationship("EventRSVP", back_populates="user", cascade="all, delete-orphan")
|
||||
volunteer_assignments = relationship("VolunteerAssignment", back_populates="user", cascade="all, delete-orphan")
|
||||
certificates = relationship("Certificate", back_populates="user", cascade="all, delete-orphan")
|
||||
profile_answers = relationship(
|
||||
"UserProfileAnswer",
|
||||
back_populates="user",
|
||||
cascade="all, delete-orphan",
|
||||
foreign_keys="UserProfileAnswer.user_id"
|
||||
)
|
||||
|
||||
|
||||
class ProfileQuestion(Base):
|
||||
__tablename__ = "profile_questions"
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
key = Column(String(100), unique=True, nullable=False, index=True)
|
||||
label = Column(String(255), nullable=False)
|
||||
help_text = Column(Text, nullable=True)
|
||||
input_type = Column(String(30), nullable=False) # text, number, boolean, date, select
|
||||
placeholder = Column(String(255), nullable=True)
|
||||
options_json = Column(Text, nullable=True) # JSON array: [{"label":"Yes","value":"true"}]
|
||||
is_required = Column(Boolean, default=False, nullable=False)
|
||||
is_active = Column(Boolean, default=True, nullable=False)
|
||||
admin_only_edit = Column(Boolean, default=False, nullable=False)
|
||||
display_order = Column(Integer, default=0, nullable=False)
|
||||
depends_on_question_id = Column(Integer, ForeignKey("profile_questions.id"), nullable=True)
|
||||
depends_on_value = Column(String(255), nullable=True)
|
||||
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
|
||||
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False)
|
||||
|
||||
depends_on_question = relationship("ProfileQuestion", remote_side=[id], backref="dependent_questions")
|
||||
answers = relationship("UserProfileAnswer", back_populates="question", cascade="all, delete-orphan")
|
||||
|
||||
|
||||
class UserProfileAnswer(Base):
|
||||
__tablename__ = "user_profile_answers"
|
||||
__table_args__ = (
|
||||
UniqueConstraint("user_id", "question_id", name="uq_user_profile_answer"),
|
||||
)
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
|
||||
question_id = Column(Integer, ForeignKey("profile_questions.id"), nullable=False, index=True)
|
||||
value_text = Column(Text, nullable=True)
|
||||
updated_by_user_id = Column(Integer, ForeignKey("users.id"), nullable=True)
|
||||
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
|
||||
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False)
|
||||
|
||||
user = relationship("User", foreign_keys=[user_id], back_populates="profile_answers")
|
||||
question = relationship("ProfileQuestion", back_populates="answers")
|
||||
updated_by_user = relationship("User", foreign_keys=[updated_by_user_id])
|
||||
|
||||
|
||||
class MembershipTier(Base):
|
||||
|
||||
@@ -37,6 +37,13 @@ from .schemas import (
|
||||
EventRSVPBase,
|
||||
EventRSVPUpdate,
|
||||
EventRSVPResponse,
|
||||
QuestionOption,
|
||||
ProfileQuestionCreate,
|
||||
ProfileQuestionUpdate,
|
||||
ProfileQuestionResponse,
|
||||
ProfileQuestionForUser,
|
||||
ProfileAnswerUpdate,
|
||||
ProfileAnswersUpdateRequest,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
@@ -78,4 +85,11 @@ __all__ = [
|
||||
"EventRSVPBase",
|
||||
"EventRSVPUpdate",
|
||||
"EventRSVPResponse",
|
||||
"QuestionOption",
|
||||
"ProfileQuestionCreate",
|
||||
"ProfileQuestionUpdate",
|
||||
"ProfileQuestionResponse",
|
||||
"ProfileQuestionForUser",
|
||||
"ProfileAnswerUpdate",
|
||||
"ProfileAnswersUpdateRequest",
|
||||
]
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
from pydantic import BaseModel, EmailStr, Field, ConfigDict
|
||||
from typing import Optional
|
||||
from typing import Optional, Literal, Any
|
||||
from datetime import datetime, date
|
||||
from ..models.models import UserRole, MembershipStatus, PaymentStatus, PaymentMethod
|
||||
|
||||
@@ -24,6 +24,7 @@ class UserUpdate(BaseModel):
|
||||
phone: Optional[str] = None
|
||||
address: Optional[str] = None
|
||||
role: Optional[UserRole] = None
|
||||
volunteer_level: Optional[str] = Field(None, max_length=50)
|
||||
|
||||
|
||||
class UserResponse(UserBase):
|
||||
@@ -31,6 +32,7 @@ class UserResponse(UserBase):
|
||||
|
||||
id: int
|
||||
role: UserRole
|
||||
volunteer_level: Optional[str] = None
|
||||
is_active: bool
|
||||
created_at: datetime
|
||||
last_login: Optional[datetime] = None
|
||||
@@ -285,3 +287,80 @@ class EventRSVPResponse(EventRSVPBase):
|
||||
attended: bool
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
|
||||
|
||||
# Profile Question Schemas
|
||||
ProfileQuestionInputType = Literal["text", "number", "boolean", "date", "select"]
|
||||
|
||||
|
||||
class QuestionOption(BaseModel):
|
||||
label: str = Field(..., min_length=1, max_length=100)
|
||||
value: str = Field(..., min_length=1, max_length=100)
|
||||
|
||||
|
||||
class ProfileQuestionBase(BaseModel):
|
||||
key: str = Field(..., min_length=2, max_length=100, pattern=r"^[a-z0-9_]+$")
|
||||
label: str = Field(..., min_length=2, max_length=255)
|
||||
help_text: Optional[str] = None
|
||||
input_type: ProfileQuestionInputType
|
||||
placeholder: Optional[str] = Field(None, max_length=255)
|
||||
options: Optional[list[QuestionOption]] = None
|
||||
is_required: bool = False
|
||||
is_active: bool = True
|
||||
admin_only_edit: bool = False
|
||||
display_order: int = 0
|
||||
depends_on_question_id: Optional[int] = None
|
||||
depends_on_value: Optional[str] = Field(None, max_length=255)
|
||||
|
||||
|
||||
class ProfileQuestionCreate(ProfileQuestionBase):
|
||||
pass
|
||||
|
||||
|
||||
class ProfileQuestionUpdate(BaseModel):
|
||||
key: Optional[str] = Field(None, min_length=2, max_length=100, pattern=r"^[a-z0-9_]+$")
|
||||
label: Optional[str] = Field(None, min_length=2, max_length=255)
|
||||
help_text: Optional[str] = None
|
||||
input_type: Optional[ProfileQuestionInputType] = None
|
||||
placeholder: Optional[str] = Field(None, max_length=255)
|
||||
options: Optional[list[QuestionOption]] = None
|
||||
is_required: Optional[bool] = None
|
||||
is_active: Optional[bool] = None
|
||||
admin_only_edit: Optional[bool] = None
|
||||
display_order: Optional[int] = None
|
||||
depends_on_question_id: Optional[int] = None
|
||||
depends_on_value: Optional[str] = Field(None, max_length=255)
|
||||
|
||||
|
||||
class ProfileQuestionResponse(BaseModel):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
id: int
|
||||
key: str
|
||||
label: str
|
||||
help_text: Optional[str] = None
|
||||
input_type: ProfileQuestionInputType
|
||||
placeholder: Optional[str] = None
|
||||
options: list[QuestionOption] = []
|
||||
is_required: bool
|
||||
is_active: bool
|
||||
admin_only_edit: bool
|
||||
display_order: int
|
||||
depends_on_question_id: Optional[int] = None
|
||||
depends_on_value: Optional[str] = None
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
|
||||
|
||||
class ProfileQuestionForUser(ProfileQuestionResponse):
|
||||
answer: Optional[Any] = None
|
||||
can_edit: bool = True
|
||||
|
||||
|
||||
class ProfileAnswerUpdate(BaseModel):
|
||||
question_id: int
|
||||
value: Optional[Any] = None
|
||||
|
||||
|
||||
class ProfileAnswersUpdateRequest(BaseModel):
|
||||
answers: list[ProfileAnswerUpdate]
|
||||
|
||||
@@ -0,0 +1,7 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
APP_ROOT = Path(__file__).resolve().parents[2]
|
||||
|
||||
if str(APP_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(APP_ROOT))
|
||||
@@ -0,0 +1,113 @@
|
||||
from datetime import date, datetime
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
|
||||
from app.api.v1.users import (
|
||||
_deserialize_answer_value,
|
||||
_normalize_answer_value,
|
||||
_normalize_volunteer_level,
|
||||
_parse_options,
|
||||
_serialize_options,
|
||||
)
|
||||
from app.models.models import ProfileQuestion
|
||||
from app.schemas import QuestionOption
|
||||
|
||||
|
||||
def make_question(input_type: str, options_json: str | None = None) -> ProfileQuestion:
|
||||
return ProfileQuestion(
|
||||
key=f"{input_type}_question",
|
||||
label=f"{input_type.title()} Question",
|
||||
input_type=input_type,
|
||||
options_json=options_json,
|
||||
)
|
||||
|
||||
|
||||
def test_option_parsing_and_serialization_filters_invalid_items() -> None:
|
||||
assert _parse_options('[{"label":" Yes ","value":" yes "}, {"label":"","value":"no"}, "bad"]') == [
|
||||
{"label": "Yes", "value": "yes"}
|
||||
]
|
||||
assert _parse_options("not-json") == []
|
||||
|
||||
serialized = _serialize_options([QuestionOption(label="Private Pilot", value="ppl")])
|
||||
|
||||
assert _parse_options(serialized) == [{"label": "Private Pilot", "value": "ppl"}]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("value", "expected"),
|
||||
[
|
||||
(True, "true"),
|
||||
("yes", "true"),
|
||||
("0", "false"),
|
||||
(False, "false"),
|
||||
],
|
||||
)
|
||||
def test_boolean_answers_are_normalized(value: object, expected: str) -> None:
|
||||
assert _normalize_answer_value(make_question("boolean"), value) == expected
|
||||
|
||||
|
||||
def test_invalid_boolean_answer_raises_400() -> None:
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
_normalize_answer_value(make_question("boolean"), "maybe")
|
||||
|
||||
assert exc.value.status_code == 400
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("value", "expected"),
|
||||
[
|
||||
(3, "3"),
|
||||
("3.50", "3.5"),
|
||||
(date(2026, 5, 4), "2026-05-04"),
|
||||
(datetime(2026, 5, 4, 12, 30), "2026-05-04"),
|
||||
],
|
||||
)
|
||||
def test_number_and_date_answers_are_normalized(value: object, expected: str) -> None:
|
||||
input_type = "date" if isinstance(value, (date, datetime)) else "number"
|
||||
|
||||
assert _normalize_answer_value(make_question(input_type), value) == expected
|
||||
|
||||
|
||||
def test_select_answers_must_match_configured_options() -> None:
|
||||
question = make_question("select", '[{"label":"Private Pilot","value":"ppl"}]')
|
||||
|
||||
assert _normalize_answer_value(question, "ppl") == "ppl"
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
_normalize_answer_value(question, "cpl")
|
||||
|
||||
assert exc.value.status_code == 400
|
||||
|
||||
|
||||
def test_empty_answers_clear_existing_values() -> None:
|
||||
assert _normalize_answer_value(make_question("text"), "") is None
|
||||
assert _normalize_answer_value(make_question("text"), None) is None
|
||||
|
||||
|
||||
def test_answer_deserialization_restores_frontend_types() -> None:
|
||||
assert _deserialize_answer_value(make_question("boolean"), "true") is True
|
||||
assert _deserialize_answer_value(make_question("boolean"), "false") is False
|
||||
assert _deserialize_answer_value(make_question("number"), "10") == 10
|
||||
assert _deserialize_answer_value(make_question("number"), "10.5") == 10.5
|
||||
assert _deserialize_answer_value(make_question("text"), "SASA") == "SASA"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("value", "expected"),
|
||||
[
|
||||
("yes", "yes"),
|
||||
("true", "yes"),
|
||||
("0", "no"),
|
||||
("", None),
|
||||
(None, None),
|
||||
],
|
||||
)
|
||||
def test_volunteer_level_accepts_boolean_like_values(value: str | None, expected: str | None) -> None:
|
||||
assert _normalize_volunteer_level(value) == expected
|
||||
|
||||
|
||||
def test_invalid_volunteer_level_raises_400() -> None:
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
_normalize_volunteer_level("sometimes")
|
||||
|
||||
assert exc.value.status_code == 400
|
||||
@@ -28,3 +28,6 @@ email-validator==2.1.0
|
||||
aiofiles==23.2.1
|
||||
Jinja2==3.1.2
|
||||
python-dateutil==2.8.2
|
||||
|
||||
# Tests
|
||||
pytest==8.3.4
|
||||
|
||||
Reference in New Issue
Block a user