Merge pull request #1192 from eniiku/feat/update_squeezepage_endpoint

feat: update squeeze-page endpoint

Author: joboy-dev

api/v1/routes/squeeze.py

@@ -1,12 +1,13 @@
-from fastapi import APIRouter, Depends, status, BackgroundTasks
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, status
 from sqlalchemy.orm import Session
-from api.db.database import get_db
+
 from api.core.responses import SUCCESS
+from api.db.database import get_db
 from api.utils.success_response import success_response
+from api.v1.models import *
+from api.v1.schemas.squeeze import CreateSqueeze, FilterSqueeze, UpdateSqueeze
 from api.v1.services.squeeze import squeeze_service
-from api.v1.schemas.squeeze import CreateSqueeze, FilterSqueeze
 from api.v1.services.user import user_service
-from api.v1.models import *
 
 squeeze = APIRouter(prefix="/squeeze", tags=["Squeeze Page"])
 
@@ -52,7 +53,34 @@ def get_squeeze(
         return success_response(status.HTTP_404_NOT_FOUND, "Squeeze page not found!")
     return success_response(status.HTTP_200_OK, SUCCESS, squeeze_page)
 
+
 @squeeze.delete("/{squeeze_id}", status_code=status.HTTP_204_NO_CONTENT)
-def delete_squeeze(squeeze_id: str, db: Session = Depends(get_db), current_user: User = Depends(user_service.get_current_super_admin)):
+def delete_squeeze(
+    squeeze_id: str,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(user_service.get_current_super_admin),
+):
     """Delete a squeeze page"""
-    squeeze_service.delete(db, squeeze_id)
+    squeeze_service.delete(db, squeeze_id)
+
+
+@squeeze.put(
+    "/{squeeze_id}", response_model=success_response, status_code=status.HTTP_200_OK
+)
+def update_squeeze(
+    squeeze_id: str,
+    data: UpdateSqueeze,
+    db: Session = Depends(get_db),
+    authorized_user: User = Depends(user_service.get_current_super_admin),
+):
+    """Update a squeeze page"""
+
+    if not authorized_user:
+        raise HTTPException(status_code=401, detail="You are not Authorized")
+
+    updated_squeeze = squeeze_service.update(db, squeeze_id, data)
+    return success_response(
+        status.HTTP_200_OK,
+        "Squeeze page updated successfully",
+        updated_squeeze.to_dict(),
+    )

api/v1/schemas/squeeze.py

@@ -1,5 +1,8 @@
-from pydantic import BaseModel, EmailStr
 from enum import Enum
+from typing import Optional
+
+from pydantic import BaseModel, EmailStr
+
 
 class SqueezeStatusEnum(str, Enum):
     online = "online"
@@ -21,3 +24,13 @@ class CreateSqueeze(BaseModel):
 
 class FilterSqueeze(BaseModel):
     status: SqueezeStatusEnum = None
+
+
+class UpdateSqueeze(BaseModel):
+    title: Optional[str] = None
+    headline: Optional[str] = None
+    sub_headline: Optional[str] = None
+    body: Optional[str] = None
+    type: Optional[str] = None
+    status: Optional[SqueezeStatusEnum] = None
+    full_name: Optional[str] = None

api/v1/services/squeeze.py

@@ -1,15 +1,18 @@
-from fastapi import HTTPException, BackgroundTasks
+from fastapi import BackgroundTasks, HTTPException
 from sqlalchemy.orm import Session
+
 from api.core.base.services import Service
-from api.v1.models.squeeze import Squeeze
 from api.core.dependencies.email_sender import send_email
-from api.v1.schemas.squeeze import CreateSqueeze, FilterSqueeze
+from api.v1.models.squeeze import Squeeze
+from api.v1.schemas.squeeze import CreateSqueeze, FilterSqueeze, UpdateSqueeze
 
 
 class SqueezeService(Service):
     """Squeeze service"""
 
-    def create(self, background_tasks: BackgroundTasks, db: Session, data: CreateSqueeze):
+    def create(
+        self, background_tasks: BackgroundTasks, db: Session, data: CreateSqueeze
+    ):
         """Create squeeze page"""
         new_squeeze = Squeeze(
             title=data.title,
@@ -26,16 +29,13 @@ def create(self, background_tasks: BackgroundTasks, db: Session, data: CreateSqu
         db.add(new_squeeze)
         db.commit()
         db.refresh(new_squeeze)
-        cta_link = 'https://anchor-python.teams.hng.tech/about-us'
+        cta_link = "https://anchor-python.teams.hng.tech/about-us"
         background_tasks.add_task(
-            send_email, 
+            send_email,
             recipient=data.email,
-            template_name='squeeze.html',
-            subject='Welcome to HNG Squeeze',
-            context={
-                'name': data.full_name,
-                'cta_link': cta_link
-            }
+            template_name="squeeze.html",
+            subject="Welcome to HNG Squeeze",
+            context={"name": data.full_name, "cta_link": cta_link},
         )
 
         return new_squeeze
@@ -62,17 +62,36 @@ def fetch(self, db: Session, id: str, filter: FilterSqueeze = None):
             squeeze = db.query(Squeeze).filter(Squeeze.id == id).first()
         return squeeze
 
-    def update(self, db: Session, id: str, schema):
+    def update(self, db: Session, id: str, data: UpdateSqueeze):
         """Update a specific squeeze page"""
-        pass
+        squeeze = db.query(Squeeze).filter(Squeeze.id == id).first()
+
+        if not squeeze:
+            raise HTTPException(status_code=404, detail="Squeeze page not found")
+
+        # Update only the fields that are provided in the update data
+        for field, value in data.dict(exclude_unset=True).items():
+            setattr(squeeze, field, value)
+
+        try:
+            db.commit()
+            db.refresh(squeeze)
+        except Exception as e:
+            db.rollback()
+            raise HTTPException(
+                status_code=500,
+                detail=f"Unexpected error occurred while updating the squeeze page: {e}",
+            )
+
+        return squeeze
 
     def delete(self, db: Session, id: str):
         """Delete a specific squeeze page"""
         squeeze = db.query(Squeeze).filter(Squeeze.id == id).first()
 
         if not squeeze:
             raise HTTPException(status_code=404, detail="Squeeze page not found")
-        
+
         db.delete(squeeze)
         db.commit()
         db.refresh()

tests/v1/squeeze_page/test_update_squeeze.py

@@ -0,0 +1,151 @@
+from datetime import datetime, timezone
+from unittest.mock import MagicMock
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+from uuid_extensions import uuid7
+
+from api.db.database import get_db
+from api.v1.models.squeeze import Squeeze
+from api.v1.models.user import User
+from api.v1.schemas.squeeze import UpdateSqueeze
+from api.v1.services.user import user_service
+from main import app
+
+
+@pytest.fixture(scope="module")
+def client():
+    with TestClient(app) as c:
+        yield c
+
+
+@pytest.fixture
+def mock_db_session():
+    db = MagicMock(spec=Session)
+    yield db
+
+
+@pytest.fixture
+def mock_user():
+    return User(
+        id=f"{uuid7()}",
+        email="test@example.com",
+        password="hashedpassword1",
+        first_name="test",
+        last_name="user",
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+
+
+@pytest.fixture()
+def mock_get_current_admin():
+    return User(
+        id=str(uuid7()),
+        email="admin@gmail.com",
+        password=user_service.hash_password("Testadmin@123"),
+        first_name="Admin",
+        last_name="User",
+        is_active=True,
+        is_superadmin=True,
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+
+
+@pytest.fixture
+def valid_squeeze():
+    return UpdateSqueeze(
+        title="Updated Squeeze Page", headline="This is the new headline"
+    )
+
+
+@pytest.fixture
+def exisiting_squeeze(mock_db_session):
+    squeeze = Squeeze(
+        id=str(uuid7()),
+        title="My Squeeze Page",
+        email="user1@example.com",
+        headline="My Headline 1",
+        sub_headline="My Sub Headline 1",
+        body="My Body 1",
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+
+    mock_db_session.query(Squeeze).filter(
+        Squeeze.id == squeeze.id
+    ).first.return_value = squeeze
+    return squeeze
+
+
+def test_update_squeeze_success(
+    client, mock_db_session, exisiting_squeeze, valid_squeeze, mock_get_current_admin
+):
+    """Test to successfully update a squeeze page"""
+
+    # Mock the dependencies
+    app.dependency_overrides[get_db] = lambda: mock_db_session
+    app.dependency_overrides[user_service.get_current_super_admin] = (
+        lambda: mock_get_current_admin
+    )
+
+    response = client.put(
+        f"api/v1/squeeze/{exisiting_squeeze.id}",
+        headers={"Authorization": "Bearer token"},
+        json=valid_squeeze.model_dump(),
+    )
+
+    assert response.status_code == 200
+    response_data = response.json()
+    assert response_data["data"]["title"] == valid_squeeze.title
+    assert response_data["data"]["headline"] == valid_squeeze.headline
+
+
+def test_update_squeeze_not_found(
+    client, mock_db_session, valid_squeeze, mock_get_current_admin
+):
+    """Test updating a non-existent squeeze page"""
+
+    non_existent_id = str(uuid7())
+    mock_db_session.query(Squeeze).filter(
+        Squeeze.id == non_existent_id
+    ).first.return_value = None
+
+    # Mock the dependencies
+    app.dependency_overrides[get_db] = lambda: mock_db_session
+    app.dependency_overrides[user_service.get_current_super_admin] = (
+        lambda: mock_get_current_admin
+    )
+
+    response = client.put(
+        f"api/v1/squeeze/{non_existent_id}",
+        headers={"Authorization": "Bearer token"},
+        json=valid_squeeze.model_dump(),
+    )
+
+    assert response.status_code == 404
+
+
+def test_update_squeeze_internal_error(
+    client, mock_db_session, exisiting_squeeze, valid_squeeze, mock_get_current_admin
+):
+    """Test for internal server error during update"""
+
+    # Simulate a database error
+    mock_db_session.commit.side_effect = Exception("Database error")
+
+    # Mock the dependencies
+    app.dependency_overrides[get_db] = lambda: mock_db_session
+    app.dependency_overrides[user_service.get_current_super_admin] = (
+        lambda: mock_get_current_admin
+    )
+
+    response = client.put(
+        f"api/v1/squeeze/{exisiting_squeeze.id}",
+        headers={"Authorization": "Bearer token"},
+        json=valid_squeeze.model_dump(),
+    )
+
+    assert response.status_code == 500