implement unpublish route and test it

README.md

@@ -285,6 +285,7 @@ Frameworks and technologies used in _Shortify_
 
 ## Roadmap
 - [ ] Implement a route to `unpublish` a post
+- [ ] Implement a route to `delete` a post
 - [ ] Use [Redis] for page caching
 - [ ] Introduce likes in _SRB_
 - [ ] Implement following and follower machnism

src/core/acl.py

@@ -6,7 +6,7 @@
 from src.core.enums import PermissionGrantsEnum, RoutesEnum, UserRolesEnum
 from src.core.exceptions import UnAuthorisedAccessError
 from src.core.schemas import UserSchema
-from src.repository.models import DraftModel, CommentModel
+from src.repository.models import DraftModel, CommentModel, PostModel
 from src.repository.user_repo import UserRepo
 from src.service.user_service import UserService
 
@@ -23,39 +23,49 @@ async def _user_not_self_not_allowed(
     username: str,
     req_username: int | str,
     *args,
-) -> bool:
+):
     """Allow if the user is requested for him/herself resource"""
-    if username == req_username:
-        return True
-    raise UnAuthorisedAccessError()
+    if username != req_username:
+        raise UnAuthorisedAccessError()
 
 
 async def _draft_not_self_not_allowed(
     username: str, draft_id: int | str, session: AsyncSession
-) -> bool:
+):
     stmt = (
         select(DraftModel.id)
         .where(DraftModel.username == username)
         .where(DraftModel.id == draft_id)
     )
     draft = (await session.execute(stmt)).first()
-    if draft is not None:
-        return True
-    raise UnAuthorisedAccessError()
+    if draft is None:
+        raise UnAuthorisedAccessError()
 
 
 async def _comment_not_self_not_allowed(
     username: str, comment_id: int | str, session: AsyncSession
-) -> bool:
+):
     stmt = (
         select(CommentModel.id)
         .where(CommentModel.username == username)
         .where(CommentModel.id == comment_id)
     )
     comment = (await session.execute(stmt)).first()
-    if comment is not None:
-        return True
-    raise UnAuthorisedAccessError()
+    if comment is None:
+        raise UnAuthorisedAccessError()
+
+
+async def _unpublish_not_self_not_allowed(
+    username: str, post_id: int, session: AsyncSession
+):
+    stmt = (
+        select(PostModel.id)
+        .where(PostModel.username == username)
+        .where(PostModel.id == post_id)
+    )
+    post = (await session.execute(stmt)).first()
+    if post is None:
+        raise UnAuthorisedAccessError()
 
 
 _GRANT_MAPPER = {
@@ -64,6 +74,7 @@ async def _comment_not_self_not_allowed(
     PermissionGrantsEnum.USER_NOT_SELF_NOT_ALLOWED: _user_not_self_not_allowed,
     PermissionGrantsEnum.DRAFT_NOT_SELF_NOT_ALLOWED: _draft_not_self_not_allowed,
     PermissionGrantsEnum.COMMENT_NOT_SELF_NOT_ALLOWED: _comment_not_self_not_allowed,
+    PermissionGrantsEnum.UNPUBLISH_NOT_SELF_NOT_ALLOWED: _unpublish_not_self_not_allowed,
 }
 
 ACLSetting: TypeAlias = dict[RoutesEnum, dict[UserRolesEnum, PermissionGrantsEnum]]
@@ -84,6 +95,10 @@ async def _comment_not_self_not_allowed(
         UserRolesEnum.ADMIN: PermissionGrantsEnum.IS_ALLOWED,
         UserRolesEnum.USER: PermissionGrantsEnum.COMMENT_NOT_SELF_NOT_ALLOWED,
     },
+    RoutesEnum.UNPUBLISH_POST: {
+        UserRolesEnum.ADMIN: PermissionGrantsEnum.IS_ALLOWED,
+        UserRolesEnum.USER: PermissionGrantsEnum.UNPUBLISH_NOT_SELF_NOT_ALLOWED,
+    },
 }
 
 

src/core/config.py

@@ -3,7 +3,7 @@
 
 class Settings(BaseSettings):
     SRB_API_VERSION: str = "v1"
-    SRB_DEBUG: bool = False
+    SRB_DEBUG: bool = True
     SRB_PG_URL: str = "postgresql+asyncpg://postgres:[email protected]:5432"
     SRB_TEST_DATABASE_URL: str | None = None
     SRB_REDIS_CACHE_URL: str = "redis://@0.0.0.0:6379/0"

src/core/depends.py

@@ -42,7 +42,7 @@ async def get_current_username_with_access(
         raise ForbiddenError("Invalid Header")
 
     access_token = request.cookies.get("Access-Token")
-    if not access_token:
+    if access_token is None:
         raise ForbiddenError("Access-Token is not provided")
 
     access_token = decode_access_token(access_token)
@@ -53,6 +53,25 @@ async def get_current_username_with_access(
     return access_token.username
 
 
+async def get_tokens_from_cookies(
+    request: Request,
+    credentials: Annotated[HTTPAuthorizationCredentials, Depends(http_bearer)],
+):
+    if credentials.scheme != "Bearer":
+        raise ForbiddenError("Invalid Header")
+
+    access_token = request.cookies.get("Access-Token")
+    if access_token is None:
+        raise ForbiddenError("Access-Token is not provided")
+
+    access_token_d = decode_access_token(access_token)
+    csrf_token_d = decode_csrf_token(credentials.credentials)
+    if csrf_token_d.access_token != access_token_d:
+        CredentialsError()
+
+    return access_token, credentials.credentials
+
+
 async def get_current_user_from_db(
     session_maker: Annotated[async_sessionmaker, Depends(get_db_sessionmaker)],
     username: Annotated[str, Depends(get_current_username_with_access)],

src/core/enums.py

@@ -10,6 +10,7 @@ class RoutesEnum(StrEnum):
     GET_ALL_DRAFTS_BY_USERNAME = "GET_ALL_DRAFTS_BY_USERNAME"
     GET_ONE_DRAFT_BY_USERNAME = "GET_ONE_DRAFT_BY_USERNAME"
     DELETE_COMMENT = "DELETE_COMMENT"
+    UNPUBLISH_POST = "UNPUBLISH_POST"
 
 
 class APIPrefixesEnum(StrEnum):
@@ -31,6 +32,7 @@ class PermissionGrantsEnum(StrEnum):
     USER_NOT_SELF_NOT_ALLOWED = "USER_NOT_SELF_NOT_ALLOWED"
     DRAFT_NOT_SELF_NOT_ALLOWED = "DRAFT_NOT_SELF_NOT_ALLOWED"
     COMMENT_NOT_SELF_NOT_ALLOWED = "COMMENT_NOT_SELF_NOT_ALLOWED"
+    UNPUBLISH_NOT_SELF_NOT_ALLOWED = "UNPUBLISH_NOT_SELF_NOT_ALLOWED"
 
 
 class APIMethodsEnum(StrEnum):

src/core/exceptions.py

@@ -25,8 +25,8 @@ def __init__(self, draft_id):
 
 
 class PostNotFoundError(ResourceNotFoundError):
-    def __init__(self, link):
-        self.message = f"<Post:{link!r} is not found!"
+    def __init__(self, link_or_id):
+        self.message = f"<Post:{link_or_id!r} is not found!"
 
 
 class CommentNotFoundError(ResourceNotFoundError):

src/repository/draft_repo.py

@@ -106,6 +106,12 @@ async def get_by_link(self, username: str, slug: str) -> DraftSchema:
             return DraftSchema(**draft)
         raise ResourceNotFoundError("Draft is not Found!")
 
+    async def unpublish(self, draft_id):
+        stmt = (
+            update(self.model).where(self.model.id == draft_id).values(is_published=False)
+        )
+        await self.session.execute(stmt)
+
     def _select_all_columns(self) -> Select:
         return select(
             self.model.id,

src/repository/post_repo.py

@@ -7,6 +7,7 @@
 )
 from src.core.schemas import PostSchema, LittlePostSchema
 from src.repository import BaseRepo
+from src.repository.draft_repo import DraftRepo
 from src.repository.models import (
     DraftModel,
     PostModel,
@@ -101,3 +102,15 @@ async def get_all(self, username: str) -> list[LittlePostSchema]:
 
         posts = await self.execute_mappings_fetchall(stmt)
         return [LittlePostSchema(**p) for p in posts]
+
+    async def unpublish(self, post_id: int) -> int:
+        draft_repo = DraftRepo(self.session)
+
+        post = (
+            await self.session.execute(select(self.model).where(self.model.id == post_id))
+        ).scalar_one_or_none()
+        if post is None:
+            raise PostNotFoundError(post_id)
+        await self.session.delete(post)
+        await draft_repo.unpublish(post.draft_id)
+        return post.draft_id

src/service/post_service.py

@@ -1,3 +1,5 @@
+from src.core.config import settings
+from src.core.enums import APIPrefixesEnum
 from src.core.schemas import PublishDraftSchema, PostSchema, LittlePostSchema
 from src.repository.post_repo import PostRepo
 from src.service import Service
@@ -21,3 +23,8 @@ async def get_global_post(self, username: str, link: str) -> PostSchema:
 
     async def get_all_posts(self, username: str) -> list[LittlePostSchema]:
         return await self.repo.get_all(username)
+
+    async def unpublish_post(self, post_id: int) -> str:
+        draft_id = await self.repo.unpublish(post_id)
+        url = f"{settings.PREFIX}/{APIPrefixesEnum.DRAFTS.value}/{draft_id}"
+        return url[1:]

src/web/posts.py

@@ -2,17 +2,33 @@
 
 from fastapi import APIRouter, Depends
 from sqlalchemy.ext.asyncio import async_sessionmaker
+from starlette.requests import Request
+from starlette.responses import RedirectResponse
 
+from src.core.acl import check_permission, ACLSetting, get_permission_setting
 from src.core.database import get_db_sessionmaker
-from src.core.enums import APIPrefixesEnum
-from src.core.schemas import LittlePostSchema
+from src.core.depends import get_current_user_from_db, get_tokens_from_cookies
+from src.core.enums import APIPrefixesEnum, RoutesEnum
+from src.core.schemas import LittlePostSchema, UserSchema
 from src.repository.post_repo import PostRepo
 from src.repository.unitofwork import UnitOfWork
 from src.service.post_service import PostService
 
 router = APIRouter(prefix=f"/{APIPrefixesEnum.POSTS.value}")
 
 
+@router.get("/", response_model=list[LittlePostSchema])
+async def get_self_posts(
+    session_maker: Annotated[async_sessionmaker, Depends(get_db_sessionmaker)],
+    user: Annotated[UserSchema, Depends(get_current_user_from_db)],
+):
+    async with UnitOfWork(session_maker) as session:
+        repo = PostRepo(session)
+        service = PostService(repo)
+        posts = await service.get_all_posts(user.username)
+    return posts
+
+
 @router.get("/{username}", response_model=list[LittlePostSchema])
 async def get_posts(
     username: str,
@@ -23,3 +39,34 @@ async def get_posts(
         service = PostService(repo)
         posts = await service.get_all_posts(username)
     return posts
+
+
+@router.post("/unpublish/{post_id}", response_class=RedirectResponse)
+async def unpublish_post(
+    request: Request,
+    post_id: int,
+    session_maker: Annotated[async_sessionmaker, Depends(get_db_sessionmaker)],
+    user: Annotated[UserSchema, Depends(get_current_user_from_db)],
+    permission_setting: Annotated[ACLSetting, Depends(get_permission_setting)],
+    tokens: Annotated[str, Depends(get_tokens_from_cookies)],
+):
+    async with UnitOfWork(session_maker) as session:
+        await check_permission(
+            session, user, post_id, RoutesEnum.UNPUBLISH_POST, permission_setting
+        )
+        repo = PostRepo(session)
+        service = PostService(repo)
+        draft_url = await service.unpublish_post(post_id)
+
+    rr = RedirectResponse(
+        url=f"{request.base_url}{draft_url}",
+        status_code=303,
+        headers={"X-CSRF-TOKEN": tokens[1]},
+    )
+    rr.set_cookie(
+        key="Access-Token",
+        value=tokens[0],
+        httponly=True,
+        samesite="strict",
+    )
+    return rr

tests/conftest.py

@@ -83,7 +83,7 @@ def headers_cookies_tuple(self, refreshed_namedtuple):
 
 users_basic_url = base_url + f"{APIPrefixesEnum.USERS.value}"
 drafts_basic_url = base_url + f"{APIPrefixesEnum.DRAFTS.value}"
-post_basic_url = base_url + f"{APIPrefixesEnum.POSTS.value}"
+posts_basic_url = base_url + f"{APIPrefixesEnum.POSTS.value}"
 comments_basic_url = base_url + f"{APIPrefixesEnum.COMMENTS.value}"
 
 draft_data = {"title": "title", "body": "body"}

tests/test_posts.py

@@ -1,11 +1,44 @@
 from src.core.config import settings
 from src.core.enums import APIPrefixesEnum
-from tests.conftest import BaseTest, create_draft, create_post
+from tests.conftest import BaseTest, create_draft, create_post, posts_basic_url
 
 bt = BaseTest()
 
 
-def test_get_posts(client, refreshed_mahdi):
+def test_get_self_posts(client, refreshed_mahdi):
+    h, c = bt.headers_cookies_tuple(refreshed_mahdi)
+
+    draft_id = create_draft(client, h, c)
+    draft_id2 = create_draft(client, h, c, "title2")
+    create_post(client, h, c, draft_id)
+    create_post(client, h, c, draft_id2)
+
+    response = client.get(
+        f"{settings.PREFIX}/{APIPrefixesEnum.POSTS.value}/", headers=h, cookies=c
+    )
+    assert response.status_code == 200, response.text
+
+    data = response.json()
+    assert len(data) == 2
+
+    data1, data2 = data
+    assert data1["title"] == "title"
+    assert data1["slug"]
+    assert data1["published"]
+    assert data2["title"] == "title2"
+    assert data1["published"] != data2["published"]
+
+    response = client.get(f"/@mahdi/{data1['slug']}")
+    assert response.status_code == 200, response.text
+
+    data = response.json()
+    assert data["title"] == "title"
+    assert data["body"] == "body"
+    assert set(data["tags"]) == {"tag1", "tag2"}
+    assert not data["comments_count"]
+
+
+def test_get_posts_with_username(client, refreshed_mahdi):
     h, c = bt.headers_cookies_tuple(refreshed_mahdi)
 
     draft_id = create_draft(client, h, c)
@@ -34,3 +67,20 @@ def test_get_posts(client, refreshed_mahdi):
     assert data["body"] == "body"
     assert set(data["tags"]) == {"tag1", "tag2"}
     assert not data["comments_count"]
+
+
+def test_unpublish_post(client, refreshed_mahdi):
+    h, c = bt.headers_cookies_tuple(refreshed_mahdi)
+    draft_id = create_draft(client, h, c)
+    post_id = create_post(client, h, c, draft_id)
+
+    response = client.post(f"{posts_basic_url}/unpublish/{post_id}", cookies=c, headers=h)
+    assert response.status_code == 200, response.text
+
+    data = response.json()
+    assert data["title"] == "title"
+    assert data["body"] == "body"
+    assert data["username"] == "mahdi"
+
+    response = client.get(f"{posts_basic_url}", headers=h, cookies=c)
+    assert not response.json()