first prototype commit

This commit is contained in:
2026-08-09 21:45:06 -04:00
commit 40613a00d5
36 changed files with 4371 additions and 0 deletions
+32
View File
@@ -0,0 +1,32 @@
import os
import jwt
from datetime import datetime, timedelta, timezone
from typing import Optional
SECRET_KEY = os.getenv("JWT_SECRET", "choreus-super-secret-key-2026-family-chores")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_DAYS = 30
GOOGLE_CLIENT_ID = os.getenv("GOOGLE_CLIENT_ID", "")
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
to_encode = data.copy()
if expires_delta:
expire = datetime.now(timezone.utc) + expires_delta
else:
expire = datetime.now(timezone.utc) + timedelta(days=ACCESS_TOKEN_EXPIRE_DAYS)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
def decode_access_token(token: str) -> Optional[dict]:
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
except jwt.PyJWTError:
return None
def get_current_week_identifier() -> str:
now = datetime.now(timezone.utc)
year, week, _ = now.isocalendar()
return f"{year}-W{week:02d}"
+20
View File
@@ -0,0 +1,20 @@
import os
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
DB_PATH = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "choreus.db")
SQLALCHEMY_DATABASE_URL = f"sqlite:///{DB_PATH}"
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
+104
View File
@@ -0,0 +1,104 @@
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
from backend.database import engine, Base, SessionLocal
from backend.models import User, ChoreType, ChoreCompletion, UserRole, RecurrenceType
from backend.routers import auth, users, chores, reports
from backend.config import get_current_week_identifier
from datetime import datetime, timezone, timedelta
def seed_initial_data():
Base.metadata.create_all(bind=engine)
db = SessionLocal()
try:
# Seed Users if empty
if db.query(User).count() == 0:
admin_user = User(
email="admin@choreus.app",
name="Sarah (Admin)",
avatar_url="https://api.dicebear.com/7.x/bottts/svg?seed=SarahAdmin",
role=UserRole.ADMIN.value,
weekly_star_quota=20
)
kid_user1 = User(
email="leo@choreus.app",
name="Leo",
avatar_url="https://api.dicebear.com/7.x/bottts/svg?seed=LeoKid",
role=UserRole.REGULAR.value,
weekly_star_quota=15
)
kid_user2 = User(
email="maya@choreus.app",
name="Maya",
avatar_url="https://api.dicebear.com/7.x/bottts/svg?seed=MayaKid",
role=UserRole.REGULAR.value,
weekly_star_quota=15
)
db.add_all([admin_user, kid_user1, kid_user2])
db.commit()
# Seed Chores if empty
if db.query(ChoreType).count() == 0:
chores_list = [
ChoreType(title="Empty & fill dishwasher", description="Clear dishes, load dirty ones and start cycle", star_reward=1, recurrence=RecurrenceType.DAILY.value, icon="Utensils"),
ChoreType(title="Take out trash & recycling", description="Empty kitchen bin and carry bins to curb", star_reward=1, recurrence=RecurrenceType.DAILY.value, icon="Trash2"),
ChoreType(title="Vacuum living room", description="Vacuum carpet and couch cushions thoroughly", star_reward=3, recurrence=RecurrenceType.WEEKLY.value, icon="Broom"),
ChoreType(title="Clean bathroom", description="Scrub sink, mirror, toilet, and wipe floor", star_reward=4, recurrence=RecurrenceType.WEEKLY.value, icon="Sparkles"),
ChoreType(title="Clear snow from driveway", description="Shovel front driveway and salt walkways after snow fall", star_reward=5, recurrence=RecurrenceType.SPONTANEOUS.value, icon="Snowflake"),
ChoreType(title="Go buy groceries", description="Pick up weekly groceries from list", star_reward=4, recurrence=RecurrenceType.SPONTANEOUS.value, icon="ShoppingBag"),
]
db.add_all(chores_list)
db.commit()
# Seed sample completions if none exist
if db.query(ChoreCompletion).count() == 0:
users_list = db.query(User).all()
chores_types = db.query(ChoreType).all()
current_week = get_current_week_identifier()
if len(users_list) >= 3 and len(chores_types) >= 3:
c1 = ChoreCompletion(
chore_type_id=chores_types[0].id,
user_id=users_list[1].id, # Leo
stars_earned=chores_types[0].star_reward,
week_identifier=current_week,
completed_at=datetime.now(timezone.utc) - timedelta(hours=5),
notes="All clean!"
)
c2 = ChoreCompletion(
chore_type_id=chores_types[2].id,
user_id=users_list[2].id, # Maya
stars_earned=chores_types[2].star_reward,
week_identifier=current_week,
completed_at=datetime.now(timezone.utc) - timedelta(hours=2),
notes="Living room carpet looks great"
)
db.add_all([c1, c2])
db.commit()
finally:
db.close()
@asynccontextmanager
async def lifespan(app: FastAPI):
seed_initial_data()
yield
app = FastAPI(title="ChoreUS API", version="1.0.0", description="Home Chore Manager API", lifespan=lifespan)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(auth.router)
app.include_router(users.router)
app.include_router(chores.router)
app.include_router(reports.router)
@app.get("/")
def root():
return {"message": "Welcome to ChoreUS Home Chore Manager API", "status": "running"}
+56
View File
@@ -0,0 +1,56 @@
from datetime import datetime, timezone
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Boolean, Enum
from sqlalchemy.orm import relationship
import enum
from backend.database import Base
class UserRole(str, enum.Enum):
ADMIN = "admin"
REGULAR = "regular"
class RecurrenceType(str, enum.Enum):
DAILY = "daily"
WEEKLY = "weekly"
SPONTANEOUS = "spontaneous"
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True, nullable=False)
name = Column(String, nullable=False)
google_id = Column(String, unique=True, nullable=True, index=True)
avatar_url = Column(String, nullable=True)
role = Column(String, default=UserRole.REGULAR.value, nullable=False)
weekly_star_quota = Column(Integer, default=15, nullable=False)
is_active = Column(Boolean, default=True, nullable=False)
created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
completions = relationship("ChoreCompletion", back_populates="user", cascade="all, delete-orphan")
class ChoreType(Base):
__tablename__ = "chore_types"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, nullable=False)
description = Column(String, nullable=True)
star_reward = Column(Integer, default=1, nullable=False) # 1 to 5 stars
recurrence = Column(String, default=RecurrenceType.DAILY.value, nullable=False) # daily, weekly, spontaneous
icon = Column(String, default="CheckSquare", nullable=False)
created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
completions = relationship("ChoreCompletion", back_populates="chore_type", cascade="all, delete-orphan")
class ChoreCompletion(Base):
__tablename__ = "chore_completions"
id = Column(Integer, primary_key=True, index=True)
chore_type_id = Column(Integer, ForeignKey("chore_types.id", ondelete="CASCADE"), nullable=False)
user_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False)
completed_at = Column(DateTime, default=lambda: datetime.now(timezone.utc), nullable=False)
stars_earned = Column(Integer, nullable=False)
week_identifier = Column(String, nullable=False, index=True) # e.g., '2026-W32'
notes = Column(String, nullable=True)
user = relationship("User", back_populates="completions")
chore_type = relationship("ChoreType", back_populates="completions")
+113
View File
@@ -0,0 +1,113 @@
from fastapi import APIRouter, Depends, HTTPException, status, Header
from sqlalchemy.orm import Session
from backend.database import get_db
from backend.models import User, UserRole
from backend.schemas import UserOut, GoogleAuthRequest, DemoLoginRequest
from backend.config import create_access_token, decode_access_token, GOOGLE_CLIENT_ID
from typing import Optional
from google.oauth2 import id_token
from google.auth.transport import requests as google_requests
router = APIRouter(prefix="/api/auth", tags=["auth"])
def get_current_user(authorization: Optional[str] = Header(None), db: Session = Depends(get_db)) -> User:
if not authorization:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing authorization header")
parts = authorization.split()
if len(parts) != 2 or parts[0].lower() != "bearer":
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authorization header format")
token = parts[1]
payload = decode_access_token(token)
if not payload or "sub" not in payload:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired token")
user_id = int(payload["sub"])
user = db.query(User).filter(User.id == user_id, User.is_active == True).first()
if not user:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found or inactive")
return user
def get_admin_user(current_user: User = Depends(get_current_user)) -> User:
if current_user.role != UserRole.ADMIN.value:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin privileges required")
return current_user
@router.post("/google")
def google_auth(req: GoogleAuthRequest, db: Session = Depends(get_db)):
email = req.email
name = req.name or "Google User"
avatar_url = req.avatar_url or f"https://api.dicebear.com/7.x/bottts/svg?seed={email}"
google_sub = None
# Try verifying real Google token if client ID configured and credential provided
if req.credential and len(req.credential) > 50:
try:
if GOOGLE_CLIENT_ID:
id_info = id_token.verify_oauth2_token(req.credential, google_requests.Request(), GOOGLE_CLIENT_ID)
else:
id_info = id_token.verify_oauth2_token(req.credential, google_requests.Request())
email = id_info.get("email", email)
name = id_info.get("name", name)
avatar_url = id_info.get("picture", avatar_url)
google_sub = id_info.get("sub")
except Exception as e:
# Fallback to provided details if token decode fails in dev
if not email:
raise HTTPException(status_code=400, detail=f"Google token verification failed: {str(e)}")
if not email:
raise HTTPException(status_code=400, detail="Email is required")
user = db.query(User).filter(User.email == email).first()
# Check if this is the first user ever registered
total_users = db.query(User).count()
initial_role = UserRole.ADMIN.value if total_users == 0 else UserRole.REGULAR.value
if not user:
user = User(
email=email,
name=name,
avatar_url=avatar_url,
google_id=google_sub,
role=initial_role,
weekly_star_quota=15
)
db.add(user)
db.commit()
db.refresh(user)
else:
# Update existing user google_id or default avatar if missing
if google_sub and not user.google_id:
user.google_id = google_sub
if avatar_url and not user.avatar_url:
user.avatar_url = avatar_url
db.commit()
db.refresh(user)
token = create_access_token({"sub": str(user.id), "email": user.email, "role": user.role})
return {
"access_token": token,
"token_type": "bearer",
"user": UserOut.model_validate(user)
}
@router.post("/demo-login")
def demo_login(req: DemoLoginRequest, db: Session = Depends(get_db)):
user = db.query(User).filter(User.email == req.email).first()
if not user:
raise HTTPException(status_code=404, detail="Demo user not found. Run backend initialization first.")
token = create_access_token({"sub": str(user.id), "email": user.email, "role": user.role})
return {
"access_token": token,
"token_type": "bearer",
"user": UserOut.model_validate(user)
}
@router.get("/me", response_model=UserOut)
def get_me(current_user: User = Depends(get_current_user)):
return current_user
+132
View File
@@ -0,0 +1,132 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List, Optional
from datetime import datetime, timezone
from backend.database import get_db
from backend.models import User, ChoreType, ChoreCompletion, UserRole, RecurrenceType
from backend.schemas import ChoreTypeOut, ChoreTypeCreate, ChoreTypeUpdate, ChoreCompletionCreate, ChoreCompletionOut
from backend.routers.auth import get_current_user, get_admin_user
from backend.config import get_current_week_identifier
router = APIRouter(prefix="/api/chores", tags=["chores"])
@router.get("/types", response_model=List[ChoreTypeOut])
def list_chore_types(db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
return db.query(ChoreType).order_by(ChoreType.id.asc()).all()
@router.post("/types", response_model=ChoreTypeOut)
def create_chore_type(chore_in: ChoreTypeCreate, db: Session = Depends(get_db), admin: User = Depends(get_admin_user)):
if chore_in.star_reward < 1 or chore_in.star_reward > 5:
raise HTTPException(status_code=400, detail="Star reward must be between 1 and 5")
if chore_in.recurrence not in [r.value for r in RecurrenceType]:
raise HTTPException(status_code=400, detail=f"Invalid recurrence type. Must be daily, weekly, or spontaneous")
new_chore = ChoreType(
title=chore_in.title,
description=chore_in.description,
star_reward=chore_in.star_reward,
recurrence=chore_in.recurrence,
icon=chore_in.icon
)
db.add(new_chore)
db.commit()
db.refresh(new_chore)
return new_chore
@router.patch("/types/{chore_id}", response_model=ChoreTypeOut)
def update_chore_type(chore_id: int, chore_in: ChoreTypeUpdate, db: Session = Depends(get_db), admin: User = Depends(get_admin_user)):
chore = db.query(ChoreType).filter(ChoreType.id == chore_id).first()
if not chore:
raise HTTPException(status_code=404, detail="Chore type not found")
if chore_in.title is not None:
chore.title = chore_in.title
if chore_in.description is not None:
chore.description = chore_in.description
if chore_in.star_reward is not None:
if chore_in.star_reward < 1 or chore_in.star_reward > 5:
raise HTTPException(status_code=400, detail="Star reward must be between 1 and 5")
chore.star_reward = chore_in.star_reward
if chore_in.recurrence is not None:
if chore_in.recurrence not in [r.value for r in RecurrenceType]:
raise HTTPException(status_code=400, detail="Invalid recurrence type")
chore.recurrence = chore_in.recurrence
if chore_in.icon is not None:
chore.icon = chore_in.icon
db.commit()
db.refresh(chore)
return chore
@router.delete("/types/{chore_id}")
def delete_chore_type(chore_id: int, db: Session = Depends(get_db), admin: User = Depends(get_admin_user)):
chore = db.query(ChoreType).filter(ChoreType.id == chore_id).first()
if not chore:
raise HTTPException(status_code=404, detail="Chore type not found")
db.delete(chore)
db.commit()
return {"message": "Chore type deleted successfully", "id": chore_id}
@router.post("/complete", response_model=ChoreCompletionOut)
def complete_chore(payload: ChoreCompletionCreate, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
chore_type = db.query(ChoreType).filter(ChoreType.id == payload.chore_type_id).first()
if not chore_type:
raise HTTPException(status_code=404, detail="Chore type not found")
week_id = get_current_week_identifier()
now_utc = datetime.now(timezone.utc)
completion = ChoreCompletion(
chore_type_id=chore_type.id,
user_id=current_user.id,
completed_at=now_utc,
stars_earned=chore_type.star_reward,
week_identifier=week_id,
notes=payload.notes
)
db.add(completion)
db.commit()
db.refresh(completion)
return ChoreCompletionOut(
id=completion.id,
chore_type_id=completion.chore_type_id,
user_id=completion.user_id,
completed_at=completion.completed_at,
stars_earned=completion.stars_earned,
week_identifier=completion.week_identifier,
notes=completion.notes,
user_name=current_user.name,
user_avatar=current_user.avatar_url,
chore_title=chore_type.title,
chore_icon=chore_type.icon
)
@router.get("/completions", response_model=List[ChoreCompletionOut])
def get_recent_completions(week: Optional[str] = None, user_id: Optional[int] = None, limit: int = 50, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
query = db.query(ChoreCompletion)
if week:
query = query.filter(ChoreCompletion.week_identifier == week)
if user_id:
query = query.filter(ChoreCompletion.user_id == user_id)
completions = query.order_by(ChoreCompletion.completed_at.desc()).limit(limit).all()
result = []
for c in completions:
result.append(ChoreCompletionOut(
id=c.id,
chore_type_id=c.chore_type_id,
user_id=c.user_id,
completed_at=c.completed_at,
stars_earned=c.stars_earned,
week_identifier=c.week_identifier,
notes=c.notes,
user_name=c.user.name if c.user else "Unknown",
user_avatar=c.user.avatar_url if c.user else None,
chore_title=c.chore_type.title if c.chore_type else "Chore",
chore_icon=c.chore_type.icon if c.chore_type else "CheckSquare"
))
return result
+66
View File
@@ -0,0 +1,66 @@
from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session
from typing import List, Optional
from backend.database import get_db
from backend.models import User, ChoreCompletion, ChoreType
from backend.schemas import WeeklyReportOut, UserWeekProgress
from backend.routers.auth import get_current_user
from backend.config import get_current_week_identifier
router = APIRouter(prefix="/api/reports", tags=["reports"])
@router.get("/weeks")
def get_available_weeks(db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
# Returns all distinct week_identifiers present in database plus current week
weeks = db.query(ChoreCompletion.week_identifier).distinct().all()
week_set = {w[0] for w in weeks if w[0]}
current_week = get_current_week_identifier()
week_set.add(current_week)
sorted_weeks = sorted(list(week_set), reverse=True)
return {"current_week": current_week, "weeks": sorted_weeks}
@router.get("/weekly", response_model=WeeklyReportOut)
def get_weekly_report(week: Optional[str] = Query(None), db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
target_week = week if week else get_current_week_identifier()
users = db.query(User).filter(User.is_active == True).order_by(User.id.asc()).all()
user_progress_list: List[UserWeekProgress] = []
total_stars_earned = 0
total_completions = 0
for user in users:
completions = db.query(ChoreCompletion).filter(
ChoreCompletion.user_id == user.id,
ChoreCompletion.week_identifier == target_week
).all()
stars = sum(c.stars_earned for c in completions)
comp_count = len(completions)
total_stars_earned += stars
total_completions += comp_count
quota = user.weekly_star_quota if user.weekly_star_quota > 0 else 1
pct = round(min(100.0, (stars / quota) * 100), 1)
user_progress_list.append(UserWeekProgress(
user_id=user.id,
name=user.name,
email=user.email,
avatar_url=user.avatar_url,
role=user.role,
weekly_star_quota=user.weekly_star_quota,
stars_earned=stars,
percentage=pct,
completions_count=comp_count
))
# Sort users by stars earned descending
user_progress_list.sort(key=lambda u: u.stars_earned, reverse=True)
return WeeklyReportOut(
week_identifier=target_week,
user_progress=user_progress_list,
total_stars_earned=total_stars_earned,
total_completions=total_completions
)
+81
View File
@@ -0,0 +1,81 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List, Optional
from backend.database import get_db
from backend.models import User, UserRole
from backend.schemas import UserOut, UserCreate, UserUpdate
from backend.routers.auth import get_current_user, get_admin_user
router = APIRouter(prefix="/api/users", tags=["users"])
@router.get("", response_model=List[UserOut])
def get_all_users(db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
return db.query(User).order_by(User.id.asc()).all()
@router.post("", response_model=UserOut)
def create_user(user_in: UserCreate, db: Session = Depends(get_db), admin: User = Depends(get_admin_user)):
existing = db.query(User).filter(User.email == user_in.email).first()
if existing:
raise HTTPException(status_code=400, detail="User with this email already exists")
avatar = user_in.avatar_url or f"https://api.dicebear.com/7.x/bottts/svg?seed={user_in.email}"
new_user = User(
email=user_in.email,
name=user_in.name,
google_id=user_in.google_id,
avatar_url=avatar,
role=user_in.role,
weekly_star_quota=user_in.weekly_star_quota
)
db.add(new_user)
db.commit()
db.refresh(new_user)
return new_user
@router.get("/{user_id}", response_model=UserOut)
def get_user_by_id(user_id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
user = db.query(User).filter(User.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@router.patch("/{user_id}", response_model=UserOut)
def update_user(user_id: int, user_in: UserUpdate, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
# Regular users can only update their own avatar and name. Admins can update role, quota, active status.
if current_user.role != UserRole.ADMIN.value and current_user.id != user_id:
raise HTTPException(status_code=403, detail="Cannot edit another user's account")
user = db.query(User).filter(User.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
if user_in.name is not None:
user.name = user_in.name
if user_in.avatar_url is not None:
user.avatar_url = user_in.avatar_url
# Admin-only fields
if current_user.role == UserRole.ADMIN.value:
if user_in.role is not None:
user.role = user_in.role
if user_in.weekly_star_quota is not None:
user.weekly_star_quota = user_in.weekly_star_quota
if user_in.is_active is not None:
user.is_active = user_in.is_active
db.commit()
db.refresh(user)
return user
@router.delete("/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db), admin: User = Depends(get_admin_user)):
if admin.id == user_id:
raise HTTPException(status_code=400, detail="Cannot delete your own admin account")
user = db.query(User).filter(User.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
db.delete(user)
db.commit()
return {"message": "User deleted successfully", "id": user_id}
+96
View File
@@ -0,0 +1,96 @@
from pydantic import BaseModel, EmailStr, Field, ConfigDict
from typing import Optional, List
from datetime import datetime
class UserBase(BaseModel):
email: EmailStr
name: str
avatar_url: Optional[str] = None
role: str = "regular"
weekly_star_quota: int = Field(default=15, ge=1, le=100)
class UserCreate(UserBase):
google_id: Optional[str] = None
class UserUpdate(BaseModel):
name: Optional[str] = None
avatar_url: Optional[str] = None
role: Optional[str] = None
weekly_star_quota: Optional[int] = Field(default=None, ge=1, le=100)
is_active: Optional[bool] = None
class UserOut(UserBase):
id: int
google_id: Optional[str] = None
is_active: bool
created_at: datetime
model_config = ConfigDict(from_attributes=True)
class ChoreTypeBase(BaseModel):
title: str
description: Optional[str] = None
star_reward: int = Field(default=1, ge=1, le=5)
recurrence: str = Field(default="daily", description="daily, weekly, spontaneous")
icon: str = "CheckSquare"
class ChoreTypeCreate(ChoreTypeBase):
pass
class ChoreTypeUpdate(BaseModel):
title: Optional[str] = None
description: Optional[str] = None
star_reward: Optional[int] = Field(default=None, ge=1, le=5)
recurrence: Optional[str] = None
icon: Optional[str] = None
class ChoreTypeOut(ChoreTypeBase):
id: int
created_at: datetime
model_config = ConfigDict(from_attributes=True)
class ChoreCompletionCreate(BaseModel):
chore_type_id: int
notes: Optional[str] = None
class ChoreCompletionOut(BaseModel):
id: int
chore_type_id: int
user_id: int
completed_at: datetime
stars_earned: int
week_identifier: str
notes: Optional[str] = None
user_name: Optional[str] = None
user_avatar: Optional[str] = None
chore_title: Optional[str] = None
chore_icon: Optional[str] = None
model_config = ConfigDict(from_attributes=True)
class UserWeekProgress(BaseModel):
user_id: int
name: str
email: str
avatar_url: Optional[str]
role: str
weekly_star_quota: int
stars_earned: int
percentage: float
completions_count: int
class WeeklyReportOut(BaseModel):
week_identifier: str
user_progress: List[UserWeekProgress]
total_stars_earned: int
total_completions: int
class GoogleAuthRequest(BaseModel):
credential: Optional[str] = None
email: Optional[EmailStr] = None
name: Optional[str] = None
avatar_url: Optional[str] = None
class DemoLoginRequest(BaseModel):
email: EmailStr
+31
View File
@@ -0,0 +1,31 @@
import pytest
from fastapi.testclient import TestClient
from backend.main import app
def test_root_endpoint():
with TestClient(app) as client:
response = client.get("/")
assert response.status_code == 200
assert response.json()["status"] == "running"
def test_demo_login():
with TestClient(app) as client:
response = client.post("/api/auth/demo-login", json={"email": "admin@choreus.app"})
assert response.status_code == 200
data = response.json()
assert "access_token" in data
assert data["user"]["email"] == "admin@choreus.app"
assert data["user"]["role"] == "admin"
def test_google_auth_mock():
with TestClient(app) as client:
response = client.post("/api/auth/google", json={
"credential": "mock_token",
"email": "newuser@choreus.app",
"name": "New User",
"avatar_url": "https://api.dicebear.com/7.x/bottts/svg?seed=NewUser"
})
assert response.status_code == 200
data = response.json()
assert data["user"]["email"] == "newuser@choreus.app"
assert data["user"]["role"] == "regular"
+47
View File
@@ -0,0 +1,47 @@
import pytest
from fastapi.testclient import TestClient
from backend.main import app
def test_list_chores():
with TestClient(app) as client:
auth_res = client.post("/api/auth/demo-login", json={"email": "admin@choreus.app"})
token = auth_res.json()["access_token"]
headers = {"Authorization": f"Bearer {token}"}
response = client.get("/api/chores/types", headers=headers)
assert response.status_code == 200
chores = response.json()
assert len(chores) >= 1
assert "title" in chores[0]
def test_create_and_complete_chore():
with TestClient(app) as client:
admin_res = client.post("/api/auth/demo-login", json={"email": "admin@choreus.app"})
admin_token = admin_res.json()["access_token"]
admin_headers = {"Authorization": f"Bearer {admin_token}"}
# Create chore
chore_res = client.post("/api/chores/types", headers=admin_headers, json={
"title": "Clean Garage",
"description": "Organize tools and sweep floor",
"star_reward": 5,
"recurrence": "weekly",
"icon": "Wrench"
})
assert chore_res.status_code == 200
chore_data = chore_res.json()
chore_id = chore_data["id"]
# Complete chore as user
user_res = client.post("/api/auth/demo-login", json={"email": "leo@choreus.app"})
user_token = user_res.json()["access_token"]
user_headers = {"Authorization": f"Bearer {user_token}"}
comp_res = client.post("/api/chores/complete", headers=user_headers, json={
"chore_type_id": chore_id,
"notes": "Done with garage!"
})
assert comp_res.status_code == 200
comp_data = comp_res.json()
assert comp_data["stars_earned"] == 5
assert comp_data["chore_title"] == "Clean Garage"