Test6

.github/workflows/production.yml

@@ -11,6 +11,8 @@ on:
 jobs:
   test:
     runs-on: ubuntu-latest
+    env:
+      CI: true
     strategy:
       matrix:
         python-version: [3.10.12] # Define Python versions here
@@ -70,15 +72,15 @@ jobs:
         uses: docker/build-push-action@v5
         with:
           push: true
-          tags: kaw393939/wis_club_api:${{ github.sha }} # Uses the Git SHA for tagging
+          tags: ipl2/user_management_final:${{ github.sha }} # Uses the Git SHA for tagging
           platforms: linux/amd64,linux/arm64 # Multi-platform support
-          cache-from: type=registry,ref=kaw393939/wis_club_api:cache
+          cache-from: type=registry,ref=ipl2/user_management_final:cache
           cache-to: type=inline,mode=max
 
       - name: Scan the Docker image
         uses: aquasecurity/trivy-action@master
         with:
-          image-ref: 'kaw393939/wis_club_api:${{ github.sha }}'
+          image-ref: 'ipl2/user_management_final:${{ github.sha }}'
           format: 'table'
           exit-code: '1' # Fail the job if vulnerabilities are found
           ignore-unfixed: true

Dockerfile

@@ -15,7 +15,7 @@ WORKDIR /myapp
 RUN apt-get update && apt-get install -y --no-install-recommends \
     gcc \
     libpq-dev \
-    && apt-get install -y libc-bin=2.36-9+deb12u7 \
+    libc-bin \
     && apt-get clean \
     && rm -rf /var/lib/apt/lists/*
 
@@ -30,7 +30,7 @@ RUN python -m venv /.venv \
 FROM python:3.12-slim-bookworm as final
 
 # Upgrade libc-bin in the final stage to ensure security patch is applied
-RUN apt-get update && apt-get install -y libc-bin=2.36-9+deb12u7 \
+RUN apt-get update && apt-get install -y libc-bin \
     && apt-get clean \
     && rm -rf /var/lib/apt/lists/*
 

app/routers/user_routes.py

@@ -27,7 +27,7 @@
 from app.dependencies import get_current_user, get_db, get_email_service, require_role
 from app.schemas.pagination_schema import EnhancedPagination
 from app.schemas.token_schema import TokenResponse
-from app.schemas.user_schemas import LoginRequest, UserBase, UserCreate, UserListResponse, UserResponse, UserUpdate
+from app.schemas.user_schemas import LoginRequest, UserBase, UserCreate, UserListResponse, UserResponse, UserUpdatePublic, UserUpdateAdmin
 from app.services.user_service import UserService
 from app.services.jwt_service import create_access_token
 from app.utils.link_generation import create_user_links, generate_pagination_links
@@ -78,15 +78,17 @@ async def get_user(user_id: UUID, request: Request, db: AsyncSession = Depends(g
 # This approach not only ensures that the API is secure and efficient but also promotes a better client
 # experience by adhering to REST principles and providing self-discoverable operations.
 
+# route only for users
 @router.put("/users/{user_id}", response_model=UserResponse, name="update_user", tags=["User Management Requires (Admin or Manager Roles)"])
-async def update_user(user_id: UUID, user_update: UserUpdate, request: Request, db: AsyncSession = Depends(get_db), token: str = Depends(oauth2_scheme), current_user: dict = Depends(require_role(["ADMIN", "MANAGER"]))):
+async def update_user(user_id: UUID, user_update: UserUpdatePublic, request: Request, db: AsyncSession = Depends(get_db), token: str = Depends(oauth2_scheme), current_user: dict = Depends(require_role(["ADMIN", "MANAGER"]))):
     """
     Update user information.
 
     - **user_id**: UUID of the user to update.
     - **user_update**: UserUpdate model with updated user information.
     """
     user_data = user_update.model_dump(exclude_unset=True)
+
     updated_user = await UserService.update(db, user_id, user_data)
     if not updated_user:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
@@ -108,6 +110,36 @@ async def update_user(user_id: UUID, user_update: UserUpdate, request: Request,
         links=create_user_links(updated_user.id, request)
     )
 
+# adding a new route for just admins and managers
+@router.put("/admin/users/{user_id}", response_model=UserResponse, tags=["Admin and Manager Only"])
+async def admin_update_user(
+    user_id: UUID,
+    update_data: UserUpdateAdmin,
+    request: Request,
+    session: AsyncSession = Depends(get_db),
+    email_service: EmailService = Depends(get_email_service),
+    token: str = Depends(oauth2_scheme),
+    current_user: dict = Depends(require_role(["ADMIN", "MANAGER"])),
+):
+    updated_user = await UserService.admin_update(session, user_id, update_data.model_dump(), email_service)
+    if not updated_user:
+        raise HTTPException(status_code=404, detail="User not found or update failed")
+    return UserResponse.model_construct(
+        id=updated_user.id,
+        bio=updated_user.bio,
+        first_name=updated_user.first_name,
+        last_name=updated_user.last_name,
+        nickname=updated_user.nickname,
+        email=updated_user.email,
+        role=updated_user.role,
+        last_login_at=updated_user.last_login_at,
+        profile_picture_url=updated_user.profile_picture_url,
+        github_profile_url=updated_user.github_profile_url,
+        linkedin_profile_url=updated_user.linkedin_profile_url,
+        created_at=updated_user.created_at,
+        updated_at=updated_user.updated_at,
+        links=create_user_links(updated_user.id, request)
+    )
 
 @router.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT, name="delete_user", tags=["User Management Requires (Admin or Manager Roles)"])
 async def delete_user(user_id: UUID, db: AsyncSession = Depends(get_db), token: str = Depends(oauth2_scheme), current_user: dict = Depends(require_role(["ADMIN", "MANAGER"]))):
@@ -164,6 +196,24 @@ async def create_user(user: UserCreate, request: Request, db: AsyncSession = Dep
         links=create_user_links(created_user.id, request)
     )
 
+# adding a new route to only admins-only to unlock a locked user account
+@router.post("/admin/users/{user_id}/unlock", status_code=status.HTTP_200_OK, tags=["Admin and Manager Only"])
+async def unlock_user_account(
+    user_id: UUID,
+    db: AsyncSession = Depends(get_db),
+    token: str = Depends(oauth2_scheme),
+    current_user: dict = Depends(require_role(["ADMIN", "MANAGER"]))
+):
+    '''Admin/manager can unlock a user account by resetting lock status and login attempts'''
+    success = await UserService.unlock_user_account(db, user_id)
+    if not success:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found or not locked")
+    updated_user = await UserService.get_by_id(db, user_id)
+    return {
+        "message": f"User account {user_id} has been unlocked successfully.",
+        "is_locked": updated_user.is_locked,
+        "failed_login_attempts": updated_user.failed_login_attempts,
+    }
 
 @router.get("/users/", response_model=UserListResponse, tags=["User Management Requires (Admin or Manager Roles)"])
 async def list_users(
@@ -233,6 +283,14 @@ async def login(form_data: OAuth2PasswordRequestForm = Depends(), session: Async
         return {"access_token": access_token, "token_type": "bearer"}
     raise HTTPException(status_code=401, detail="Incorrect email or password.")
 
+# adding new route for registering users
+@router.post("/register")
+async def register_user_route(
+    user_data: UserCreate,
+    db: AsyncSession = Depends(get_db),
+    email_service: EmailService = Depends(get_email_service),
+):
+    return await UserService.register_user(db, user_data.model_dump(), email_service)
 
 @router.get("/verify-email/{user_id}/{token}", status_code=status.HTTP_200_OK, name="verify_email", tags=["Login and Registration"])
 async def verify_email(user_id: UUID, token: str, db: AsyncSession = Depends(get_db), email_service: EmailService = Depends(get_email_service)):

app/schemas/user_schemas.py

@@ -7,6 +7,7 @@
 import re
 from app.models.user_model import UserRole
 from app.utils.nickname_gen import generate_nickname
+from uuid import UUID
 
 
 def validate_url(url: Optional[str]) -> Optional[str]:
@@ -17,6 +18,27 @@ def validate_url(url: Optional[str]) -> Optional[str]:
         raise ValueError('Invalid URL format')
     return url
 
+# adding class. allowed updated fields for public users
+class UserUpdatePublic(BaseModel):
+    first_name: Optional[str] = Field(None, example="John")
+    last_name: Optional[str] = Field(None, example="Doe")
+    password: Optional[str] = Field(None, min_length=8)
+    phone_number: Optional[str] = None
+    email: Optional[EmailStr] = Field(None, example="john.doe@example.com")
+    github_profile_url: Optional[str] = Field(None, example="https://github.com/johndoe")
+    linkedin_profile_url: Optional[str] =Field(None, example="https://linkedin.com/in/johndoe")
+
+    class Config:
+        from_attributes = True
+
+# adding class. allowed updated fields for admins only
+class UserUpdateAdmin(UserUpdatePublic):
+    role: Optional[UserRole] = None
+    email_verified: Optional[bool] = None
+    is_locked: Optional[bool] = None
+    verification_token: Optional[str] = None
+    is_professional: Optional[bool] = None
+
 class UserBase(BaseModel):
     email: EmailStr = Field(..., example="john.doe@example.com")
     nickname: Optional[str] = Field(None, min_length=3, pattern=r'^[\w-]+$', example=generate_nickname())
@@ -26,7 +48,7 @@ class UserBase(BaseModel):
     profile_picture_url: Optional[str] = Field(None, example="https://example.com/profiles/john.jpg")
     linkedin_profile_url: Optional[str] =Field(None, example="https://linkedin.com/in/johndoe")
     github_profile_url: Optional[str] = Field(None, example="https://github.com/johndoe")
-    role: UserRole
+    role: Optional[UserRole] = UserRole.AUTHENTICATED
 
     _validate_urls = validator('profile_picture_url', 'linkedin_profile_url', 'github_profile_url', pre=True, allow_reuse=True)(validate_url)
 

app/services/email_service.py

@@ -6,8 +6,8 @@
 from app.models.user_model import User
 
 class EmailService:
-    def __init__(self, template_manager: TemplateManager):
-        self.smtp_client = SMTPClient(
+    def __init__(self, template_manager: TemplateManager, smtp_client: SMTPClient = None):
+        self.smtp_client = smtp_client or SMTPClient(
             server=settings.smtp_server,
             port=settings.smtp_port,
             username=settings.smtp_username,
@@ -34,4 +34,13 @@ async def send_verification_email(self, user: User):
             "name": user.first_name,
             "verification_url": verification_url,
             "email": user.email
-        }, 'email_verification')
+        }, 'email_verification')
+
+    async def send_email(self, to_email: str, subject: str, body: str):
+        self.smtp_client.send_email(subject, body, to_email)
+
+
+    async def send_professional_status_upgraded_email(self, user: User):
+        subject = "Professional Status Upgraded."
+        body = f"Dear {user.first_name or user.nickname},\n\nYour account has been upgraded to professional status.\n\nYour Team"
+        await self.send_email(user.email, subject, body)