diff --git a/config/env_config.py b/config/env_config.py index 029a7974..93c29f8a 100644 --- a/config/env_config.py +++ b/config/env_config.py @@ -28,9 +28,14 @@ def _get_env(self, name: str) -> str: Raises: RuntimeError: If the mandatory environment variable is missing. """ + # Intentionally NOT in exceptions.py: startup-only crash path. + # Never caught by application code. Builtin RuntimeError sufficient. + # Do not move to exceptions.py. value = os.getenv(name) if not value: - raise RuntimeError(f"Infrastructure Error : Missing environment variable '{name}'") + raise RuntimeError( + f"Infrastructure Error : Missing environment variable '{name}'" + ) return value @property diff --git a/exceptions.py b/exceptions.py new file mode 100644 index 00000000..cc5672ea --- /dev/null +++ b/exceptions.py @@ -0,0 +1,103 @@ +class BlogCommentError(Exception): + """Base for all application and infrastructure exceptions.""" + pass + + +class AccountAlreadyExistsError(BlogCommentError): + """Raised when username or email already exists.""" + pass + + +class FileTooLargeError(BlogCommentError): + """Raised when uploaded file exceeds max size.""" + pass + + +class FileTypeError(BlogCommentError): + """Raised when uploaded file has unsupported type/extension.""" + pass + + +class AuthenticationError(BlogCommentError): + """Raised when credentials are invalid or user is not authenticated.""" + pass + + +class AuthorizationError(BlogCommentError): + """Raised when user lacks permission for the requested action.""" + pass + + +class InsufficientPermissionsError(AuthorizationError): + """Raised when account role does not have the required permission level.""" + pass + + +class AccountNotFoundError(BlogCommentError): + """Raised when an account is not found by id, username, or email.""" + pass + + +class AccountBannedError(BlogCommentError): + """Raised when an account is banned and cannot perform actions.""" + pass + + +class UsernameAlreadyTakenError(BlogCommentError): + """Raised when trying to register with an existing username.""" + pass + + +class EmailAlreadyTakenError(BlogCommentError): + """Raised when trying to register or update with an existing email.""" + pass + + +class ArticleNotFoundError(BlogCommentError): + """Raised when an article is not found.""" + pass + + +class OwnershipError(BlogCommentError): + """Raised when a user is not the owner of the resource.""" + pass + + +class CommentNotFoundError(BlogCommentError): + """Raised when a comment is not found.""" + pass + + +class CommentAuthorizationError(AuthorizationError): + """Raised when a user is not authorized for a comment action.""" + pass + + +class CommentDeletedError(BlogCommentError): + """Raised when attempting to modify a deleted comment.""" + pass + + +class CommentValidationError(BlogCommentError): + """Raised when comment content is invalid (empty, too long, etc.).""" + pass + + +class DatabaseError(BlogCommentError): + """Raised when a database operation fails (connection, query, commit).""" + pass + + +class PasswordsDoNotMatchError(ValueError): + """Raised when password and confirmation do not match.""" + pass + + +class CommentEmptyError(ValueError): + """Raised when comment content is empty.""" + pass + + +class CommentTooLongError(ValueError): + """Raised when comment content exceeds max length.""" + pass diff --git a/src/application/application_exceptions.py b/src/application/application_exceptions.py deleted file mode 100644 index 6483fd61..00000000 --- a/src/application/application_exceptions.py +++ /dev/null @@ -1,30 +0,0 @@ -class ApplicationError(Exception): - """ - Base class for all business/application layer exceptions. - """ - pass - - -class AccountAlreadyExistsError(ApplicationError): - """ - Raised when attempting to create an account with a username or email - that already exists in the database. - - This typically occurs during concurrent registration requests where - a unique constraint violation is detected at the database level. - """ - pass - - -class FileTooLargeError(ApplicationError): - """ - Raised when an uploaded file exceeds the maximum allowed size. - """ - pass - - -class FileTypeError(ApplicationError): - """ - Raised when an uploaded file has an unsupported MIME type or extension. - """ - pass diff --git a/src/application/input_ports/account_session_management.py b/src/application/input_ports/account_session_management.py index 81cb1694..2b0d0523 100644 --- a/src/application/input_ports/account_session_management.py +++ b/src/application/input_ports/account_session_management.py @@ -66,35 +66,30 @@ def update_avatar(self, avatar_file_id: str | None) -> None: pass @abstractmethod - def update_email(self, new_email: str) -> str | None: + def update_email(self, new_email: str) -> None: """ Updates the email address for the currently authenticated account. - Validates that the new email is not already in use by another account - before persisting the change. - Args: new_email: The new email address to set. - Returns: - str | None: None on success, or an error message string if - the email is already taken or the user is not authenticated. + Raises: + AuthenticationError: If the user is not signed in. + EmailAlreadyTakenError: If the email is already in use by another account. """ pass @abstractmethod - def update_password(self, new_password: str) -> str | None: + def update_password(self, new_password: str) -> None: """ Updates the password for the currently authenticated account. - Hashes the new password and persists it via the account repository. Args: new_password: The new plaintext password to set. - Returns: - str | None: None on success, or an error message string if - the user is not authenticated. + Raises: + AuthenticationError: If the user is not signed in. """ pass @@ -164,7 +159,7 @@ def count_search_accounts(self, query: str) -> int: pass @abstractmethod - def ban_account(self, admin_id: int, target_account_id: int, ban_reason: str | None) -> str | None: + def ban_account(self, admin_id: int, target_account_id: int, ban_reason: str | None) -> None: """ Bans a user account. Only admins can ban non-admin accounts. @@ -173,13 +168,15 @@ def ban_account(self, admin_id: int, target_account_id: int, ban_reason: str | N target_account_id: The unique identifier of the account to ban. ban_reason: Optional reason for the ban. - Returns: - str | None: None on success, or an error message string if the operation fails. + Raises: + AuthorizationError: If the requester is not an admin. + AccountNotFoundError: If the target account is not found. + AuthorizationError: If the target is another admin. """ pass @abstractmethod - def unban_account(self, admin_id: int, target_account_id: int) -> str | None: + def unban_account(self, admin_id: int, target_account_id: int) -> None: """ Unbans a user account. Only admins can unban accounts. @@ -187,8 +184,9 @@ def unban_account(self, admin_id: int, target_account_id: int) -> str | None: admin_id: The unique identifier of the admin performing the action. target_account_id: The unique identifier of the account to unban. - Returns: - str | None: None on success, or an error message string if the operation fails. + Raises: + AuthorizationError: If the requester is not an admin. + AccountNotFoundError: If the target account is not found. """ pass @@ -203,11 +201,14 @@ def delete_account(self, account_id: int) -> None: Args: account_id: The unique identifier of the account to delete. + + Raises: + AccountNotFoundError: If no account exists with the given id. """ pass @abstractmethod - def update_account_role(self, admin_id: int, target_id: int, new_role: str) -> str | None: + def update_account_role(self, admin_id: int, target_id: int, new_role: str) -> None: """ Allows an admin user to update the role of another user account. @@ -216,7 +217,9 @@ def update_account_role(self, admin_id: int, target_id: int, new_role: str) -> s target_id: The unique identifier of the account whose role is to be updated. new_role: The new role string ("user" or "author"). - Returns: - str | None: None on success, or an error message string if the operation fails. + Raises: + AuthorizationError: If the requester is not an admin. + AccountNotFoundError: If the target account is not found. + AuthorizationError: If the target is another admin. """ pass diff --git a/src/application/input_ports/article_management.py b/src/application/input_ports/article_management.py index 7a22a34b..a8af0333 100644 --- a/src/application/input_ports/article_management.py +++ b/src/application/input_ports/article_management.py @@ -10,7 +10,7 @@ class ArticleManagementPort(ABC): """ @abstractmethod - def create_article(self, title: str, content: str, author_id: int, author_role: str, description: str = "") -> Article | str: + def create_article(self, title: str, content: str, author_id: int, author_role: str, description: str = "") -> Article: """ Creates a new article if the user has sufficient permissions. @@ -22,8 +22,12 @@ def create_article(self, title: str, content: str, author_id: int, author_role: description (str): Short description displayed in article list. Optional. Returns: - Article | str: The newly created Article domain entity, - or an error message string if unauthorized or account not found. + Article: The newly created Article domain entity. + + Raises: + AccountNotFoundError: If the author account is not found. + InsufficientPermissionsError: If the user is not an author or admin. + AccountBannedError: If the account is banned. """ pass @@ -51,7 +55,7 @@ def get_by_id(self, article_id: int) -> Article | None: pass @abstractmethod - def update_article(self, article_id: int, user_id: int, title: str, content: str, description: str = "") -> Article | str: + def update_article(self, article_id: int, user_id: int, title: str, content: str, description: str = "") -> Article: """ Updates an existing article ensuring the requester is the original author. @@ -63,13 +67,19 @@ def update_article(self, article_id: int, user_id: int, title: str, content: str description (str): Short description displayed in article list. Optional. Returns: - Article | str: The updated Article domain entity, - or an error message string if not found or unauthorized. + Article: The updated Article domain entity. + + Raises: + AccountNotFoundError: If the user account is not found. + InsufficientPermissionsError: If the user is not an author or admin. + AccountBannedError: If the account is banned. + ArticleNotFoundError: If the article does not exist. + OwnershipError: If the user is not the author (and not admin). """ pass @abstractmethod - def delete_article(self, article_id: int, user_id: int) -> bool | str: + def delete_article(self, article_id: int, user_id: int) -> bool: """ Deletes an article. Only the original author or an admin can delete it. @@ -78,7 +88,14 @@ def delete_article(self, article_id: int, user_id: int) -> bool | str: user_id (int): ID of the user requesting the deletion. Returns: - bool | str: True if deletion was successful, or an error message string. + bool: True if deletion was successful. + + Raises: + AccountNotFoundError: If the user account is not found. + InsufficientPermissionsError: If the user is not an author or admin. + AccountBannedError: If the account is banned. + ArticleNotFoundError: If the article does not exist. + OwnershipError: If the user is not the author (and not admin). """ pass @@ -122,17 +139,18 @@ def get_author_name(self, author_id: int | None) -> str: pass @abstractmethod - def get_article_with_comments(self, article_id: int) -> ArticleDetailView | str: + def get_article_with_comments(self, article_id: int) -> ArticleDetailView: """ Orchestrates the retrieval of an article, its associated comments, and author information. - Respects the separation of concerns by using the comment management port. Args: article_id (int): ID of the article to retrieve. Returns: - ArticleDetailView | str: A Read Model for the complete article detail page, - or an error message string if the article is missing. + ArticleDetailView: A Read Model for the complete article detail page. + + Raises: + ArticleNotFoundError: If the article does not exist. """ pass @@ -166,4 +184,3 @@ def count_search(self, query: str) -> int: The total number of matching articles. """ pass - diff --git a/src/application/input_ports/comment_management.py b/src/application/input_ports/comment_management.py index 2cf39e2c..ec3de1e5 100644 --- a/src/application/input_ports/comment_management.py +++ b/src/application/input_ports/comment_management.py @@ -10,7 +10,7 @@ class CommentManagementPort(ABC): """ @abstractmethod - def create_comment(self, article_id: int, user_id: int, content: str) -> Comment | str: + def create_comment(self, article_id: int, user_id: int, content: str) -> Comment: """ Creates a top-level comment on an article. @@ -20,12 +20,18 @@ def create_comment(self, article_id: int, user_id: int, content: str) -> Comment content (str): Text content of the comment. Returns: - Comment | str: The created Comment entity, or an error message string. + Comment: The created Comment entity. + + Raises: + AccountNotFoundError: If the account is not found. + AccountBannedError: If the account is banned. + ArticleNotFoundError: If the article does not exist. + CommentValidationError: If the content is empty after sanitization. """ pass @abstractmethod - def create_reply(self, parent_comment_id: int, user_id: int, content: str) -> Comment | str: + def create_reply(self, parent_comment_id: int, user_id: int, content: str) -> Comment: """ Creates a reply directly to a parent comment. @@ -35,13 +41,19 @@ def create_reply(self, parent_comment_id: int, user_id: int, content: str) -> Co content (str): The text content of the reply. Returns: - Comment | str: The new Comment domain entity if successful, - or an error message string if unauthorized or parent not found. + Comment: The new Comment domain entity if successful. + + Raises: + AccountNotFoundError: If the account is not found. + AccountBannedError: If the account is banned. + CommentNotFoundError: If the parent comment does not exist. + CommentDeletedError: If the parent comment is deleted. + CommentValidationError: If the content is empty or max depth exceeded. """ pass @abstractmethod - def get_comments_for_article(self, article_id: int) -> list[CommentNode] | str: + def get_comments_for_article(self, article_id: int) -> list[CommentNode]: """ Retrieves all comments for a specific article and structures them into a nested tree for display, along with associated author names. @@ -50,8 +62,10 @@ def get_comments_for_article(self, article_id: int) -> list[CommentNode] | str: article_id (int): ID of the article. Returns: - list[CommentNode] | str: The nested tree root nodes, - or an error message string if the article is not found. + list[CommentNode]: The nested tree root nodes. + + Raises: + ArticleNotFoundError: If the article does not exist. """ pass @@ -70,30 +84,33 @@ def mask_comments_by_account_id(self, account_id: int) -> None: pass @abstractmethod - def delete_comment(self, comment_id: int, user_id: int) -> bool | str: + def delete_comment(self, comment_id: int, user_id: int) -> bool: """ Soft-deletes a comment. Sets is_deleted=True and deleted_at=now. Author and admin can soft-delete. Content preserved in DB but display shows "Comment removed". - Author displayed as "Anonymous" after deletion. Args: comment_id (int): ID of the comment to delete. user_id (int): ID of the user requesting the deletion. Returns: - bool | str: True if deletion was successful, or an error message string. + bool: True if deletion was successful. + + Raises: + AccountNotFoundError: If the account is not found. + AccountBannedError: If the account is banned. + CommentNotFoundError: If the comment does not exist. + CommentAuthorizationError: If the user is not the author nor admin. """ pass @abstractmethod - def edit_comment(self, comment_id: int, user_id: int, content: str) -> Comment | str: + def edit_comment(self, comment_id: int, user_id: int, content: str) -> Comment: """ - Edits a comment's content. - Author only (not admin). - Updates content and sets edited_at=now. - Cannot edit a deleted comment. + Edits a comment's content. Author only (not admin). + Updates content and sets edited_at=now. Cannot edit a deleted comment. Args: comment_id (int): ID of the comment to edit. @@ -101,22 +118,36 @@ def edit_comment(self, comment_id: int, user_id: int, content: str) -> Comment | content (str): New text content of the comment. Returns: - Comment | str: The updated Comment entity, or an error message string. + Comment: The updated Comment entity. + + Raises: + AccountNotFoundError: If the account is not found. + AccountBannedError: If the account is banned. + CommentNotFoundError: If the comment does not exist. + CommentAuthorizationError: If the user is not the comment author. + CommentDeletedError: If the comment has been deleted. + CommentValidationError: If the content is empty after sanitization. """ pass @abstractmethod - def hard_delete_comment(self, comment_id: int, user_id: int) -> bool | str: + def hard_delete_comment(self, comment_id: int, user_id: int) -> bool: """ Permanently deletes a comment from the database. Admin only. - Intended for removing already soft-deleted comments. - Children comments get comment_reply_to set to NULL via FK. + Only allowed on already soft-deleted comments. Args: comment_id (int): ID of the comment to permanently delete. user_id (int): ID of the requesting user (must be admin). Returns: - bool | str: True if deletion was successful, or an error message string. + bool: True if deletion was successful. + + Raises: + AccountNotFoundError: If the account is not found. + AccountBannedError: If the account is banned. + CommentNotFoundError: If the comment does not exist. + CommentAuthorizationError: If the user is not an admin. + CommentValidationError: If the comment is not soft-deleted first. """ pass diff --git a/src/application/input_ports/login_management.py b/src/application/input_ports/login_management.py index 4849afa7..5203bfb7 100644 --- a/src/application/input_ports/login_management.py +++ b/src/application/input_ports/login_management.py @@ -9,7 +9,7 @@ class LoginManagementPort(ABC): """ @abstractmethod - def authenticate_user(self, username: str, password: str) -> Account | str: + def authenticate_user(self, username: str, password: str) -> Account: """ Validates the user's credentials. @@ -18,7 +18,10 @@ def authenticate_user(self, username: str, password: str) -> Account | str: password (str): The plaintext password provided by the user. Returns: - Account | str: The authenticated Account instance if - credentials match, or an error message string if it fails. + Account: The authenticated Account instance. + + Raises: + AuthenticationError: If the username or password is invalid. + AccountBannedError: If the account is banned. """ pass diff --git a/src/application/input_ports/registration_management.py b/src/application/input_ports/registration_management.py index ba8db903..a89f9c80 100644 --- a/src/application/input_ports/registration_management.py +++ b/src/application/input_ports/registration_management.py @@ -9,7 +9,7 @@ class RegistrationManagementPort(ABC): """ @abstractmethod - def create_account(self, username: str, password: str, email: str) -> Account | str: + def create_account(self, username: str, password: str, email: str) -> Account: """ Creates a new user account. @@ -19,7 +19,12 @@ def create_account(self, username: str, password: str, email: str) -> Account | email (str): The email address for the new account. Returns: - Account | str: The newly created Account domain entity, or an - error message string if creation fails. + Account: The newly created Account domain entity. + + Raises: + UsernameAlreadyTakenError: If the username already exists. + EmailAlreadyTakenError: If the email already exists. + AccountAlreadyExistsError: If a race condition causes a unique + constraint violation at the database level. """ pass diff --git a/src/application/services/article_service.py b/src/application/services/article_service.py index 0660442d..43dd41c7 100644 --- a/src/application/services/article_service.py +++ b/src/application/services/article_service.py @@ -2,6 +2,13 @@ import re from datetime import UTC, datetime +from exceptions import ( + AccountBannedError, + AccountNotFoundError, + ArticleNotFoundError, + InsufficientPermissionsError, + OwnershipError, +) from src.application.domain.account import Account, AccountRole from src.application.domain.article import Article, ArticleDetailView, ArticleWithAuthor from src.application.input_ports.article_management import ArticleManagementPort @@ -32,6 +39,8 @@ def _extract_image_uuids(content: str) -> set[str]: return set() try: data = json.loads(content) + # Python builtin — safety net for json.loads on non-string input. + # Not in exceptions.py. Do not move it there. except (json.JSONDecodeError, TypeError): return set() uuids: set[str] = set() @@ -96,7 +105,7 @@ def __init__( self.comment_repository = comment_repository self.file_service = file_service - def _get_account_if_author_or_admin(self, user_id: int) -> Account | str: + def _get_account_if_author_or_admin(self, user_id: int) -> Account: """ Checks if a user exists and has the required permissions (admin or author). @@ -104,23 +113,26 @@ def _get_account_if_author_or_admin(self, user_id: int) -> Account | str: user_id (int): The unique identifier of the user. Returns: - Account | str: The Account domain entity if authorized, or an error message string. + Account: The Account domain entity if authorized. + + Raises: + AccountNotFoundError: If the account does not exist. + InsufficientPermissionsError: If the user is not an author or admin. + AccountBannedError: If the account is banned. """ account = self.account_repository.get_by_id(user_id) if not account: - # TODO: Raise AccountNotFoundException - return "Account not found." + raise AccountNotFoundError("Account not found.") if account.account_role not in [AccountRole.ADMIN, AccountRole.AUTHOR]: - # TODO: Raise InsufficientPermissionsException - return "Insufficient permissions." + raise InsufficientPermissionsError("Insufficient permissions.") if account.is_banned: - return "Account is banned." + raise AccountBannedError("Account is banned.") return account - def create_article(self, title: str, content: str, author_id: int, author_role: str, description: str = "") -> Article | str: + def create_article(self, title: str, content: str, author_id: int, author_role: str, description: str = "") -> Article: """ Creates a new article and saves it via the repository if the account exists and the user has the correct permissions. @@ -133,12 +145,14 @@ def create_article(self, title: str, content: str, author_id: int, author_role: description (str): Short description displayed in article list. Optional. Returns: - Article | str: The newly created Article domain entity, - or an error message string if unauthorized or account not found. + Article: The newly created Article domain entity. + + Raises: + AccountNotFoundError: If the account does not exist. + InsufficientPermissionsError: If the user is not an author or admin. + AccountBannedError: If the account is banned. """ - account_or_error = self._get_account_if_author_or_admin(author_id) - if isinstance(account_or_error, str): - return account_or_error + self._get_account_if_author_or_admin(author_id) new_article = Article( article_id=0, @@ -173,7 +187,7 @@ def get_by_id(self, article_id: int) -> Article | None: """ return self.article_repository.get_by_id(article_id) - def update_article(self, article_id: int, user_id: int, title: str, content: str, description: str = "") -> Article | str: + def update_article(self, article_id: int, user_id: int, title: str, content: str, description: str = "") -> Article: """ Updates an existing article. Only the original author or an admin can edit (admins can also edit anonymous articles whose author account was deleted). @@ -186,21 +200,23 @@ def update_article(self, article_id: int, user_id: int, title: str, content: str description (str): Short description displayed in article list. Optional. Returns: - Article | str: The updated Article domain entity, - or an error message string if not found or unauthorized. + Article: The updated Article domain entity. + + Raises: + AccountNotFoundError: If the account does not exist. + InsufficientPermissionsError: If the user is not an author or admin. + AccountBannedError: If the account is banned. + ArticleNotFoundError: If the article does not exist. + OwnershipError: If the user is not the author (and not admin). """ - account_or_error = self._get_account_if_author_or_admin(user_id) - if isinstance(account_or_error, str): - return account_or_error + account = self._get_account_if_author_or_admin(user_id) article = self.article_repository.get_by_id(article_id) if not article: - # TODO: Raise ArticleNotFoundException - return "Article not found." + raise ArticleNotFoundError("Article not found.") - if account_or_error.account_role != AccountRole.ADMIN and article.article_author_id != user_id: - # TODO: Raise OwnershipException - return "Unauthorized : You are not the author of this article." + if account.account_role != AccountRole.ADMIN and article.article_author_id != user_id: + raise OwnershipError("Unauthorized: You are not the author of this article.") old_content = article.article_content article.article_title = title @@ -218,7 +234,7 @@ def update_article(self, article_id: int, user_id: int, title: str, content: str return article - def delete_article(self, article_id: int, user_id: int) -> bool | str: + def delete_article(self, article_id: int, user_id: int) -> bool: """ Deletes an article. Only the original author or an admin can delete it. @@ -227,21 +243,23 @@ def delete_article(self, article_id: int, user_id: int) -> bool | str: user_id (int): ID of the user requesting the deletion. Returns: - bool | str: True if deletion was successful, or an error message string. + bool: True if deletion was successful. + + Raises: + AccountNotFoundError: If the account does not exist. + InsufficientPermissionsError: If the user is not an author or admin. + AccountBannedError: If the account is banned. + ArticleNotFoundError: If the article does not exist. + OwnershipError: If the user is not the author (and not admin). """ - account_or_error = self._get_account_if_author_or_admin(user_id) - if isinstance(account_or_error, str): - return account_or_error + account = self._get_account_if_author_or_admin(user_id) - account: Account = account_or_error article = self.article_repository.get_by_id(article_id) if not article: - # TODO: Raise ArticleNotFoundException - return "Article not found." + raise ArticleNotFoundError("Article not found.") if account.account_role != AccountRole.ADMIN and article.article_author_id != user_id: - # TODO: Raise OwnershipException - return "Unauthorized : Only authors or admins can delete articles." + raise OwnershipError("Unauthorized: Only authors or admins can delete articles.") if self.file_service: for uuid in _extract_image_uuids(article.article_content): @@ -302,7 +320,7 @@ def get_author_name(self, author_id: int | None) -> str: account = self.account_repository.get_by_id(author_id) return account.account_username if account else "Unknown" - def get_article_with_comments(self, article_id: int) -> ArticleDetailView | str: + def get_article_with_comments(self, article_id: int) -> ArticleDetailView: """ Orchestrates the retrieval of an article, its associated threaded comments, and all involved author names in optimized batches. @@ -311,12 +329,14 @@ def get_article_with_comments(self, article_id: int) -> ArticleDetailView | str: article_id (int): ID of the article to retrieve. Returns: - ArticleDetailView | str: A Read Model for the complete article detail page, - or an error message string if the article is not found. + ArticleDetailView: A Read Model for the complete article detail page. + + Raises: + ArticleNotFoundError: If the article is not found. """ article = self.article_repository.get_by_id(article_id) if not article: - return "Article not found." + raise ArticleNotFoundError("Article not found.") all_comments = self.comment_repository.get_all_by_article_id(article_id) known_ids = {article.article_author_id} if article.article_author_id is not None else set() diff --git a/src/application/services/comment_service.py b/src/application/services/comment_service.py index 4bdf32bc..e17f2591 100644 --- a/src/application/services/comment_service.py +++ b/src/application/services/comment_service.py @@ -2,6 +2,15 @@ import nh3 +from exceptions import ( + AccountBannedError, + AccountNotFoundError, + ArticleNotFoundError, + CommentAuthorizationError, + CommentDeletedError, + CommentNotFoundError, + CommentValidationError, +) from src.application.domain.account import Account, AccountRole from src.application.domain.comment import Comment, CommentNode from src.application.input_ports.comment_management import CommentManagementPort @@ -36,21 +45,25 @@ def __init__( self.article_repository = article_repository self.account_repository = account_repository - def _get_account_if_exists(self, user_id: int) -> Account | str: + def _get_account_if_exists(self, user_id: int) -> Account: """ - Retrieves an account by user ID. Returns error string if not found or banned. + Retrieves an account by user ID. Args: user_id (int): The ID of the user to look up. Returns: - Account | str: The Account domain entity, or an error message string. + Account: The Account domain entity. + + Raises: + AccountNotFoundError: If the account does not exist. + AccountBannedError: If the account is banned. """ account = self.account_repository.get_by_id(user_id) if not account: - return "Account not found." + raise AccountNotFoundError("Account not found.") if account.is_banned: - return "Account is banned." + raise AccountBannedError("Account is banned.") return account @staticmethod @@ -66,7 +79,7 @@ def _get_comment_depth(comment_id: int, comment_repo: CommentRepository) -> int: return depth - def create_comment(self, article_id: int, user_id: int, content: str) -> Comment | str: + def create_comment(self, article_id: int, user_id: int, content: str) -> Comment: """ Creates a new top-level comment on an article. @@ -79,17 +92,19 @@ def create_comment(self, article_id: int, user_id: int, content: str) -> Comment content (str): Raw comment text (may contain limited HTML). Returns: - Comment | str: The created Comment domain entity, or an error message. - """ - account_or_error = self._get_account_if_exists(user_id) - if isinstance(account_or_error, str): - return account_or_error + Comment: The created Comment domain entity. - account: Account = account_or_error + Raises: + AccountNotFoundError: If the account does not exist. + AccountBannedError: If the account is banned. + ArticleNotFoundError: If the article does not exist. + CommentValidationError: If the content is empty after sanitization. + """ + account = self._get_account_if_exists(user_id) article = self.article_repository.get_by_id(article_id) if not article: - return "Article not found." + raise ArticleNotFoundError("Article not found.") sanitized = nh3.clean( content, @@ -98,7 +113,7 @@ def create_comment(self, article_id: int, user_id: int, content: str) -> Comment link_rel="noopener noreferrer", ) if not sanitized.strip(): - return "Comment cannot be empty." + raise CommentValidationError("Comment cannot be empty.") fake_comment_id = 0 new_comment = Comment( comment_id=fake_comment_id, @@ -112,7 +127,7 @@ def create_comment(self, article_id: int, user_id: int, content: str) -> Comment self.comment_repository.save(new_comment) return new_comment - def create_reply(self, parent_comment_id: int, user_id: int, content: str) -> Comment | str: + def create_reply(self, parent_comment_id: int, user_id: int, content: str) -> Comment: """ Creates a reply to an existing comment. @@ -124,24 +139,27 @@ def create_reply(self, parent_comment_id: int, user_id: int, content: str) -> Co content (str): Raw reply text (may contain limited HTML). Returns: - Comment | str: The created Comment domain entity, or an error message. + Comment: The created Comment domain entity. + + Raises: + AccountNotFoundError: If the account does not exist. + AccountBannedError: If the account is banned. + CommentNotFoundError: If the parent comment does not exist. + CommentDeletedError: If the parent comment is deleted. + CommentValidationError: If the content is empty or max depth exceeded. """ - account_or_error = self._get_account_if_exists(user_id) - if isinstance(account_or_error, str): - return account_or_error - - account: Account = account_or_error + account = self._get_account_if_exists(user_id) parent_comment = self.comment_repository.get_by_id(parent_comment_id) if not parent_comment: - return "Parent comment not found." + raise CommentNotFoundError("Parent comment not found.") if parent_comment.is_deleted: - return "Cannot reply to a deleted comment." + raise CommentDeletedError("Cannot reply to a deleted comment.") parent_depth = self._get_comment_depth(parent_comment.comment_id, self.comment_repository) if parent_depth >= MAX_REPLY_DEPTH: - return "Cannot reply to a comment at maximum nesting depth." + raise CommentValidationError("Cannot reply to a comment at maximum nesting depth.") sanitized = nh3.clean( content, @@ -150,7 +168,7 @@ def create_reply(self, parent_comment_id: int, user_id: int, content: str) -> Co link_rel="noopener noreferrer", ) if not sanitized.strip(): - return "Comment cannot be empty." + raise CommentValidationError("Comment cannot be empty.") fake_comment_id = 0 new_reply = Comment( comment_id=fake_comment_id, @@ -164,7 +182,7 @@ def create_reply(self, parent_comment_id: int, user_id: int, content: str) -> Co self.comment_repository.save(new_reply) return new_reply - def get_comments_for_article(self, article_id: int) -> list[CommentNode] | str: + def get_comments_for_article(self, article_id: int) -> list[CommentNode]: """ Retrieves all comments for an article as a nested tree. @@ -172,12 +190,14 @@ def get_comments_for_article(self, article_id: int) -> list[CommentNode] | str: article_id (int): ID of the article. Returns: - list[CommentNode] | str: List of root CommentNode objects with nested replies, - or an error message if the article is not found. + list[CommentNode]: List of root CommentNode objects with nested replies. + + Raises: + ArticleNotFoundError: If the article does not exist. """ article = self.article_repository.get_by_id(article_id) if not article: - return "Article not found." + raise ArticleNotFoundError("Article not found.") all_comments = self.comment_repository.get_all_by_article_id(article_id) author_ids = {c.comment_written_account_id for c in all_comments if c.comment_written_account_id is not None} @@ -202,7 +222,7 @@ def mask_comments_by_account_id(self, account_id: int) -> None: comment.deleted_at = datetime.now(UTC) self.comment_repository.save(comment) - def delete_comment(self, comment_id: int, user_id: int) -> bool | str: + def delete_comment(self, comment_id: int, user_id: int) -> bool: """ Soft-deletes a comment. Author or admin only. Idempotent if already deleted. @@ -211,21 +231,23 @@ def delete_comment(self, comment_id: int, user_id: int) -> bool | str: user_id (int): ID of the requesting user. Returns: - bool | str: True on success, or an error message string. - """ - account_or_error = self._get_account_if_exists(user_id) - if isinstance(account_or_error, str): - return account_or_error + bool: True on success. - account: Account = account_or_error + Raises: + AccountNotFoundError: If the account does not exist. + AccountBannedError: If the account is banned. + CommentNotFoundError: If the comment does not exist. + CommentAuthorizationError: If the user is not the author nor admin. + """ + account = self._get_account_if_exists(user_id) comment = self.comment_repository.get_by_id(comment_id) if not comment: - return "Comment not found." + raise CommentNotFoundError("Comment not found.") is_author = comment.comment_written_account_id == account.account_id is_admin = account.account_role == AccountRole.ADMIN if not is_author and not is_admin: - return "Unauthorized: You can only delete your own comments." + raise CommentAuthorizationError("Unauthorized: You can only delete your own comments.") if comment.is_deleted: return True @@ -235,7 +257,7 @@ def delete_comment(self, comment_id: int, user_id: int) -> bool | str: self.comment_repository.save(comment) return True - def edit_comment(self, comment_id: int, user_id: int, content: str) -> Comment | str: + def edit_comment(self, comment_id: int, user_id: int, content: str) -> Comment: """ Edits a comment's content. Author only (not admin). Cannot edit a deleted comment. @@ -245,22 +267,26 @@ def edit_comment(self, comment_id: int, user_id: int, content: str) -> Comment | content (str): New comment text (may contain limited HTML). Returns: - Comment | str: The updated Comment domain entity, or an error message. + Comment: The updated Comment domain entity. + + Raises: + AccountNotFoundError: If the account does not exist. + AccountBannedError: If the account is banned. + CommentNotFoundError: If the comment does not exist. + CommentAuthorizationError: If the user is not the comment author. + CommentDeletedError: If the comment has been deleted. + CommentValidationError: If the content is empty after sanitization. """ - account_or_error = self._get_account_if_exists(user_id) - if isinstance(account_or_error, str): - return account_or_error - - account: Account = account_or_error + account = self._get_account_if_exists(user_id) comment = self.comment_repository.get_by_id(comment_id) if not comment: - return "Comment not found." + raise CommentNotFoundError("Comment not found.") if comment.comment_written_account_id != account.account_id: - return "Unauthorized: You can only edit your own comments." + raise CommentAuthorizationError("Unauthorized: You can only edit your own comments.") if comment.is_deleted: - return "Cannot edit a deleted comment." + raise CommentDeletedError("Cannot edit a deleted comment.") sanitized = nh3.clean( content, @@ -269,14 +295,14 @@ def edit_comment(self, comment_id: int, user_id: int, content: str) -> Comment | link_rel="noopener noreferrer", ) if not sanitized.strip(): - return "Comment cannot be empty." + raise CommentValidationError("Comment cannot be empty.") comment.comment_content = sanitized comment.edited_at = datetime.now(UTC) self.comment_repository.save(comment) return comment - def hard_delete_comment(self, comment_id: int, user_id: int) -> bool | str: + def hard_delete_comment(self, comment_id: int, user_id: int) -> bool: """ Permanently deletes a comment from the database. Admin only. Only allowed on already soft-deleted comments. @@ -287,22 +313,25 @@ def hard_delete_comment(self, comment_id: int, user_id: int) -> bool | str: user_id (int): ID of the requesting user (must be admin). Returns: - bool | str: True on success, or an error message string. + bool: True on success. + + Raises: + AccountNotFoundError: If the account does not exist. + AccountBannedError: If the account is banned. + CommentNotFoundError: If the comment does not exist. + CommentAuthorizationError: If the user is not an admin. + CommentValidationError: If the comment is not soft-deleted first. """ - account_or_error = self._get_account_if_exists(user_id) - if isinstance(account_or_error, str): - return account_or_error - - account: Account = account_or_error + account = self._get_account_if_exists(user_id) if account.account_role != AccountRole.ADMIN: - return "Unauthorized: Only admins can permanently delete comments." + raise CommentAuthorizationError("Unauthorized: Only admins can permanently delete comments.") comment = self.comment_repository.get_by_id(comment_id) if not comment: - return "Comment not found." + raise CommentNotFoundError("Comment not found.") if not comment.is_deleted: - return "Comment is not soft-deleted. Use soft-delete first." + raise CommentValidationError("Comment is not soft-deleted. Use soft-delete first.") self.comment_repository.delete(comment_id) return True diff --git a/src/application/services/file_service.py b/src/application/services/file_service.py index 8bc80d2b..668f17b9 100644 --- a/src/application/services/file_service.py +++ b/src/application/services/file_service.py @@ -1,7 +1,7 @@ from datetime import datetime from uuid import uuid4 -from src.application.application_exceptions import FileTooLargeError, FileTypeError +from exceptions import FileTooLargeError, FileTypeError from src.application.domain.file_record import FileRecord from src.application.input_ports.file_management import FileManagementPort from src.application.output_ports.file_storage_repository import FileStorageRepository diff --git a/src/application/services/login_service.py b/src/application/services/login_service.py index a41a02e6..b43b75bb 100644 --- a/src/application/services/login_service.py +++ b/src/application/services/login_service.py @@ -1,3 +1,10 @@ +from exceptions import ( + AccountBannedError, + AccountNotFoundError, + AuthenticationError, + AuthorizationError, + EmailAlreadyTakenError, +) from src.application.domain.account import Account, AccountRole from src.application.input_ports.account_session_management import AccountSessionManagementPort from src.application.input_ports.login_management import LoginManagementPort @@ -32,7 +39,7 @@ def __init__( self.session_repository = session_repository self.password_hasher_repository = password_hasher_repository - def authenticate_user(self, username: str, password: str) -> Account | str: + def authenticate_user(self, username: str, password: str) -> Account: """ Authenticates a user by verifying credentials. @@ -45,17 +52,20 @@ def authenticate_user(self, username: str, password: str) -> Account | str: password (str): The plaintext password provided by the user. Returns: - Account | str: The authenticated Account instance if - credentials match, or an error message string if it fails. + Account: The authenticated Account instance. + + Raises: + AuthenticationError: If the username or password is invalid. + AccountBannedError: If the account is banned. """ account = self.account_repository.find_by_username(username) if not account: - return "Invalid username or password." + raise AuthenticationError("Invalid username or password.") if self.password_hasher_repository.verify(password, account.account_password): if account.is_banned: - return "This account has been banned." + raise AccountBannedError("This account has been banned.") if self.password_hasher_repository.check_needs_rehash(account.account_password): new_hash = self.password_hasher_repository.hash(password) @@ -65,7 +75,7 @@ def authenticate_user(self, username: str, password: str) -> Account | str: self.session_repository.save_account(account) return account - return "Invalid username or password." + raise AuthenticationError("Invalid username or password.") def get_current_account(self) -> Account | None: """ @@ -124,7 +134,7 @@ def update_avatar(self, avatar_file_id: str | None) -> None: return self.account_repository.update_avatar(account.account_id, avatar_file_id) - def update_email(self, new_email: str) -> str | None: + def update_email(self, new_email: str) -> None: """ Updates the email address for the currently logged-in account. @@ -135,25 +145,24 @@ def update_email(self, new_email: str) -> str | None: Args: new_email: The new email address to set. - Returns: - str | None: None on success, or an error message string if - the email is already taken or the user is unauthenticated. + Raises: + AuthenticationError: If the user is not signed in. + EmailAlreadyTakenError: If the email is already in use by another account. """ account = self.get_current_account() if not account: - return "You must be signed in to update your email." + raise AuthenticationError("You must be signed in to update your email.") if new_email == account.account_email: - return None + return existing = self.account_repository.find_by_email(new_email) if existing and existing.account_id != account.account_id: - return "This email is already taken." + raise EmailAlreadyTakenError("This email is already taken.") self.account_repository.update_email(account.account_id, new_email) - return None - def update_password(self, new_password: str) -> str | None: + def update_password(self, new_password: str) -> None: """ Updates the password for the currently logged-in account. @@ -164,20 +173,18 @@ def update_password(self, new_password: str) -> str | None: Args: new_password: The new plaintext password to set. - Returns: - str | None: None on success, or an error message string if - the user is not authenticated or the password is empty. + Raises: + AuthenticationError: If the user is not signed in. """ account = self.get_current_account() if not account: - return "You must be signed in to update your password." + raise AuthenticationError("You must be signed in to update your password.") if not new_password: - return "Password is required." + return new_hash = self.password_hasher_repository.hash(new_password) self.account_repository.update_password(account.account_id, new_hash) - return None def get_all_accounts(self, page: int = 1, per_page: int = 20) -> list[Account]: """ @@ -241,14 +248,14 @@ def delete_account(self, account_id: int) -> None: account_id: The unique identifier of the account to delete. Raises: - ValueError: If no account with the given ID exists. + AccountNotFoundError: If no account with the given ID exists. """ existing = self.account_repository.get_by_id(account_id) if not existing: - raise ValueError(f"Account with id {account_id} not found.") + raise AccountNotFoundError(f"Account with id {account_id} not found.") self.account_repository.delete(account_id) - def update_account_role(self, admin_id: int, target_id: int, new_role: str) -> str | None: + def update_account_role(self, admin_id: int, target_id: int, new_role: str) -> None: """ Allows an admin user to update the role of another user account. @@ -260,27 +267,28 @@ def update_account_role(self, admin_id: int, target_id: int, new_role: str) -> s target_id: The unique identifier of the account whose role is to be updated. new_role: The new role string ("user" or "author"). - Returns: - str | None: None on success, or an error message string if the operation fails. + Raises: + AuthorizationError: If the requester is not an admin. + AccountNotFoundError: If the target account is not found. + AuthorizationError: If the target is another admin. """ admin = self.account_repository.get_by_id(admin_id) if not admin or admin.account_role != AccountRole.ADMIN: - return "Unauthorized." + raise AuthorizationError("Unauthorized.") target = self.account_repository.get_by_id(target_id) if not target: - return "Account not found." + raise AccountNotFoundError("Account not found.") if target.account_role == AccountRole.ADMIN: - return "Cannot change role of another admin." + raise AuthorizationError("Cannot change role of another admin.") if new_role not in ("user", "author"): - return "Invalid role." + return self.account_repository.update_role(target_id, new_role) - return None - def ban_account(self, admin_id: int, target_account_id: int, ban_reason: str | None) -> str | None: + def ban_account(self, admin_id: int, target_account_id: int, ban_reason: str | None) -> None: """ Bans a user account. Only admins can ban non-admin accounts. @@ -289,24 +297,25 @@ def ban_account(self, admin_id: int, target_account_id: int, ban_reason: str | N target_account_id: The unique identifier of the account to ban. ban_reason: Optional reason for the ban. - Returns: - str | None: None on success, or an error message string if the operation fails. + Raises: + AuthorizationError: If the requester is not an admin. + AccountNotFoundError: If the target account is not found. + AuthorizationError: If the target is another admin. """ admin = self.account_repository.get_by_id(admin_id) if not admin or admin.account_role != AccountRole.ADMIN: - return "Unauthorized." + raise AuthorizationError("Unauthorized.") target = self.account_repository.get_by_id(target_account_id) if not target: - return "Account not found." + raise AccountNotFoundError("Account not found.") if target.account_role == AccountRole.ADMIN: - return "Cannot ban another admin." + raise AuthorizationError("Cannot ban another admin.") self.account_repository.update_ban_status(target_account_id, True, ban_reason) - return None - def unban_account(self, admin_id: int, target_account_id: int) -> str | None: + def unban_account(self, admin_id: int, target_account_id: int) -> None: """ Unbans a user account. Only admins can unban accounts. @@ -314,16 +323,16 @@ def unban_account(self, admin_id: int, target_account_id: int) -> str | None: admin_id: The unique identifier of the admin performing the action. target_account_id: The unique identifier of the account to unban. - Returns: - str | None: None on success, or an error message string if the operation fails. + Raises: + AuthorizationError: If the requester is not an admin. + AccountNotFoundError: If the target account is not found. """ admin = self.account_repository.get_by_id(admin_id) if not admin or admin.account_role != AccountRole.ADMIN: - return "Unauthorized." + raise AuthorizationError("Unauthorized.") target = self.account_repository.get_by_id(target_account_id) if not target: - return "Account not found." + raise AccountNotFoundError("Account not found.") self.account_repository.update_ban_status(target_account_id, False, None) - return None diff --git a/src/application/services/registration_service.py b/src/application/services/registration_service.py index 84467093..6b09764c 100644 --- a/src/application/services/registration_service.py +++ b/src/application/services/registration_service.py @@ -1,4 +1,4 @@ -from src.application.application_exceptions import AccountAlreadyExistsError +from exceptions import AccountAlreadyExistsError, EmailAlreadyTakenError, UsernameAlreadyTakenError from src.application.domain.account import Account, AccountRole from src.application.input_ports.registration_management import RegistrationManagementPort from src.application.output_ports.account_repository import AccountRepository @@ -24,7 +24,7 @@ def __init__(self, account_repository: AccountRepository, password_hasher_reposi self.account_repository = account_repository self.password_hasher_repository = password_hasher_repository - def create_account(self, username: str, password: str, email: str) -> Account | str: + def create_account(self, username: str, password: str, email: str) -> Account: """ Creates a new user account with the default 'user' role if the username and email are not already taken. @@ -35,18 +35,20 @@ def create_account(self, username: str, password: str, email: str) -> Account | email (str): The email address for the new account. Returns: - Account | str: The newly created Account domain entity, or an - error message string if the username or email is already taken - (including race conditions detected at the database level). + Account: The newly created Account domain entity. + + Raises: + UsernameAlreadyTakenError: If the username already exists. + EmailAlreadyTakenError: If the email already exists. + AccountAlreadyExistsError: If a race condition causes a unique + constraint violation at the database level. """ if self.account_repository.find_by_username(username): - # TODO: Raise UsernameAlreadyTakenException - return "This username is already taken." + raise UsernameAlreadyTakenError("This username is already taken.") if self.account_repository.find_by_email(email): - # TODO: Raise EmailAlreadyTakenException - return "This email is already taken." + raise EmailAlreadyTakenError("This email is already taken.") hashed_password = self.password_hasher_repository.hash(password) @@ -62,5 +64,7 @@ def create_account(self, username: str, password: str, email: str) -> Account | try: self.account_repository.save(new_account) except AccountAlreadyExistsError: - return "This username or email is already taken." + raise AccountAlreadyExistsError( + "This username or email is already taken." + ) from None return new_account diff --git a/src/infrastructure/infrastructure_exceptions.py b/src/infrastructure/infrastructure_exceptions.py deleted file mode 100644 index cbf82b0e..00000000 --- a/src/infrastructure/infrastructure_exceptions.py +++ /dev/null @@ -1,6 +0,0 @@ -class InfrastructureError(Exception): - """ - General base class for all infrastructure layer exceptions. - """ - pass - diff --git a/src/infrastructure/input_adapters/dto/comment_request.py b/src/infrastructure/input_adapters/dto/comment_request.py index 23165ab3..a5b1f4c2 100644 --- a/src/infrastructure/input_adapters/dto/comment_request.py +++ b/src/infrastructure/input_adapters/dto/comment_request.py @@ -2,6 +2,8 @@ from pydantic import BaseModel, Field, field_validator +from exceptions import CommentEmptyError, CommentTooLongError + class CommentRequest(BaseModel): """ @@ -19,9 +21,17 @@ class CommentRequest(BaseModel): @field_validator("content") @classmethod def check_content_length(cls, v: str) -> str: + """ + Validates comment content is not empty after stripping HTML tags + and does not exceed the DB VARCHAR(5000) limit. + + Raises: + CommentEmptyError: If content has no non-whitespace characters. + CommentTooLongError: If content exceeds 5000 characters. + """ text = re.sub(r"<[^>]+>", "", v).strip() if len(text) < 1: - raise ValueError("Comment cannot be empty.") + raise CommentEmptyError("Comment cannot be empty.") if len(v) > 5000: - raise ValueError("Comment is too long. Maximum 5000 characters.") + raise CommentTooLongError("Comment is too long. Maximum 5000 characters.") return v diff --git a/src/infrastructure/input_adapters/dto/registration_request.py b/src/infrastructure/input_adapters/dto/registration_request.py index 014bbe41..775e3d9e 100644 --- a/src/infrastructure/input_adapters/dto/registration_request.py +++ b/src/infrastructure/input_adapters/dto/registration_request.py @@ -1,5 +1,7 @@ from pydantic import BaseModel, EmailStr, Field, model_validator +from exceptions import PasswordsDoNotMatchError + class RegistrationRequest(BaseModel): """ @@ -24,8 +26,8 @@ def passwords_must_match(self) -> "RegistrationRequest": RegistrationRequest: The validated model instance. Raises: - ValueError: If 'password' and 'confirm_password' do not match. + PasswordsDoNotMatchError: If 'password' and 'confirm_password' do not match. """ if self.password != self.confirm_password: - raise ValueError("Passwords do not match.") + raise PasswordsDoNotMatchError("Passwords do not match.") return self diff --git a/src/infrastructure/input_adapters/flask/flask_account_session_adapter.py b/src/infrastructure/input_adapters/flask/flask_account_session_adapter.py index b67bebb2..afa258c5 100644 --- a/src/infrastructure/input_adapters/flask/flask_account_session_adapter.py +++ b/src/infrastructure/input_adapters/flask/flask_account_session_adapter.py @@ -3,18 +3,16 @@ from flask_babel import gettext as _ +from exceptions import BlogCommentError, FileTooLargeError, FileTypeError from flask import abort, flash, jsonify, redirect, render_template, request, session, url_for from flask import g as global_request_context from flask.views import MethodView -from src.application.application_exceptions import FileTooLargeError, FileTypeError from src.application.domain.account import AccountRole from src.application.input_ports.account_session_management import AccountSessionManagementPort from src.application.input_ports.comment_management import CommentManagementPort from src.application.input_ports.file_management import FileManagementPort from src.infrastructure.input_adapters.dto.account_response import AccountResponse -logger = logging.getLogger(__name__) - class AccountSessionAdapter(MethodView): """ @@ -184,8 +182,12 @@ def upload_profile_photo(self): if old_avatar_id: try: self.file_service.delete_file(old_avatar_id) + # Intentionally broad: non-critical cleanup (delete old avatar + # file from DB). Broad catch ensures request never fails due to + # cleanup failure, even from unexpected bugs. + # Not in exceptions.py. Do not move it there. except Exception: - logger.warning( + logging.getLogger(__name__).warning( "Failed to delete old avatar %s for account %s", old_avatar_id, current_account.account_id, @@ -222,6 +224,9 @@ def remove_profile_photo(self): try: self.file_service.delete_file(avatar_file_id) self.session_service.update_avatar(None) + # Intentionally broad: non-critical cleanup. Broad catch ensures + # request never fails; if cleanup fails, user gets a flash error. + # Not in exceptions.py. Do not move it there. except Exception: flash(_("Failed to remove profile photo."), "error") return redirect(url_for("auth.profile")) @@ -250,9 +255,10 @@ def update_email(self): flash(_("Email is required."), "error") return redirect(url_for("auth.profile")) - result = self.session_service.update_email(new_email) - if result is not None: - flash(_(result), "error") + try: + self.session_service.update_email(new_email) + except BlogCommentError as e: + flash(_(str(e)), "error") else: flash(_("Email updated."), "success") return redirect(url_for("auth.profile")) @@ -278,9 +284,10 @@ def update_password(self): flash(_("Password is required."), "error") return redirect(url_for("auth.profile")) - result = self.session_service.update_password(new_password) - if result is not None: - flash(_(result), "error") + try: + self.session_service.update_password(new_password) + except BlogCommentError as e: + flash(_(str(e)), "error") else: flash(_("Password updated."), "success") return redirect(url_for("auth.profile")) @@ -382,8 +389,15 @@ def delete_account(self): if account and account.avatar_file_id: try: self.file_service.delete_file(account.avatar_file_id) + # Intentionally broad: non-critical cleanup (delete avatar + # file from DB before account deletion). Broad catch ensures + # account deletion proceeds even if avatar cleanup fails. + # Not in exceptions.py. Do not move it there. except Exception: - logger.warning("Failed to delete avatar %s for account %s", account.avatar_file_id, target_id) + logging.getLogger(__name__).warning( + "Failed to delete avatar %s for account %s", + account.avatar_file_id, target_id, + ) self.comment_service.mask_comments_by_account_id(target_id) self.session_service.delete_account(target_id) @@ -418,21 +432,18 @@ def change_role(self, account_id: int): abort(403) new_role = request.form.get("role", "") - result = self.session_service.update_account_role( - admin_id=current_account.account_id, - target_id=account_id, - new_role=new_role, - ) + try: + self.session_service.update_account_role( + admin_id=current_account.account_id, + target_id=account_id, + new_role=new_role, + ) + except BlogCommentError as e: + flash(_(str(e)), "error") + else: + flash(_("Role updated."), "success") target = self.session_service.get_account_by_id(account_id) - - if result is not None: - flash(_(result), "error") - if target: - return redirect(url_for("auth.user_profile", username=target.account_username)) - return redirect(url_for("auth.list_all_users")) - - flash(_("Role updated."), "success") if target: return redirect(url_for("auth.user_profile", username=target.account_username)) return redirect(url_for("auth.list_all_users")) @@ -458,14 +469,14 @@ def ban_account(self, account_id: int): abort(403) ban_reason = request.form.get("ban_reason", "").strip() or None - result = self.session_service.ban_account( - admin_id=current_account.account_id, - target_account_id=account_id, - ban_reason=ban_reason, - ) - - if result is not None: - flash(_(result), "error") + try: + self.session_service.ban_account( + admin_id=current_account.account_id, + target_account_id=account_id, + ban_reason=ban_reason, + ) + except BlogCommentError as e: + flash(_(str(e)), "error") else: flash(_("Account banned."), "success") @@ -490,13 +501,13 @@ def unban_account(self, account_id: int): if not current_account or current_account.account_role != AccountRole.ADMIN: abort(403) - result = self.session_service.unban_account( - admin_id=current_account.account_id, - target_account_id=account_id, - ) - - if result is not None: - flash(_(result), "error") + try: + self.session_service.unban_account( + admin_id=current_account.account_id, + target_account_id=account_id, + ) + except BlogCommentError as e: + flash(_(str(e)), "error") else: flash(_("Account unbanned."), "success") diff --git a/src/infrastructure/input_adapters/flask/flask_article_adapter.py b/src/infrastructure/input_adapters/flask/flask_article_adapter.py index 51845954..75467f2a 100644 --- a/src/infrastructure/input_adapters/flask/flask_article_adapter.py +++ b/src/infrastructure/input_adapters/flask/flask_article_adapter.py @@ -5,6 +5,7 @@ from pydantic import ValidationError from werkzeug.wrappers.response import Response +from exceptions import BlogCommentError from flask import flash, jsonify, redirect, render_template, request, url_for from flask import g as global_request_context from src.application.domain.comment import CommentNode @@ -109,12 +110,11 @@ def read_article(self, article_id: int) -> str | Response: Returns: Union[str, Response]: The 'article_detail.html' template or a redirect to the list view. """ - result = self.article_service.get_article_with_comments(article_id) - if isinstance(result, str): - flash(_("Error: %(error)s", error=result), "error") + try: + detail = self.article_service.get_article_with_comments(article_id) + except BlogCommentError as e: + flash(_("Error: %(error)s", error=str(e)), "error") return redirect(url_for("article.list_articles")) - - detail = result article = ArticleResponse.from_domain( detail.article_with_author.article, author_username=detail.article_with_author.author_name, @@ -124,6 +124,8 @@ def read_article(self, article_id: int) -> str | Response: content = article.article_content try: json.loads(content) + # Python builtin — safety net for json.loads on non-string input. + # Not in exceptions.py. Do not move it there. except (json.JSONDecodeError, TypeError): content = json.dumps([{ "type": "paragraph", @@ -183,6 +185,8 @@ def api_get_article(self, article_id: int) -> Response | tuple[Response, int]: content = article.article_content try: json.loads(content) + # Python builtin — safety net for json.loads on non-string input. + # Not in exceptions.py. Do not move it there. except (json.JSONDecodeError, TypeError): content = json.dumps([{ "type": "paragraph", @@ -228,18 +232,21 @@ def api_create_article(self) -> Response | tuple[Response, int]: content=data.get("content", ""), description=data.get("description", ""), ) + # Pydantic library exception — caught at web boundary for 400 response. + # Not in exceptions.py. Do not move it there. except ValidationError as e: for error in e.errors(): return jsonify({"error": f"({error['loc'][0]}): {error['msg']}"}), 400 return jsonify({"error": _("Validation error.")}), 400 - result = self.article_service.create_article( - title=req_data.title, content=req_data.content, - author_id=user.account_id, author_role=user.account_role, - description=req_data.description, - ) - if isinstance(result, str): - return jsonify({"error": result}), 403 + try: + result = self.article_service.create_article( + title=req_data.title, content=req_data.content, + author_id=user.account_id, author_role=user.account_role, + description=req_data.description, + ) + except BlogCommentError as e: + return jsonify({"error": str(e)}), 403 return jsonify({"id": result.article_id}), 201 @@ -273,18 +280,21 @@ def api_update_article(self, article_id: int) -> Response | tuple[Response, int] content=data.get("content", ""), description=data.get("description", ""), ) + # Pydantic library exception — caught at web boundary for 400 response. + # Not in exceptions.py. Do not move it there. except ValidationError as e: for error in e.errors(): return jsonify({"error": f"({error['loc'][0]}): {error['msg']}"}), 400 return jsonify({"error": _("Validation error.")}), 400 - result = self.article_service.update_article( - article_id=article_id, user_id=user.account_id, - title=req_data.title, content=req_data.content, - description=req_data.description, - ) - if isinstance(result, str): - return jsonify({"error": result}), 403 + try: + self.article_service.update_article( + article_id=article_id, user_id=user.account_id, + title=req_data.title, content=req_data.content, + description=req_data.description, + ) + except BlogCommentError as e: + return jsonify({"error": str(e)}), 403 return jsonify({"ok": True}) @@ -308,11 +318,12 @@ def _api_delete_article(self, article_id: int) -> Response | tuple[Response, int if user.account_role not in ["admin", "author"]: return jsonify({"error": _("Insufficient permissions.")}), 403 - result = self.article_service.delete_article( - article_id=article_id, user_id=user.account_id, - ) - if isinstance(result, str): - return jsonify({"error": result}), 403 + try: + self.article_service.delete_article( + article_id=article_id, user_id=user.account_id, + ) + except BlogCommentError as e: + return jsonify({"error": str(e)}), 403 return jsonify({"ok": True}) diff --git a/src/infrastructure/input_adapters/flask/flask_comment_adapter.py b/src/infrastructure/input_adapters/flask/flask_comment_adapter.py index 8ade8c83..e19b50f5 100644 --- a/src/infrastructure/input_adapters/flask/flask_comment_adapter.py +++ b/src/infrastructure/input_adapters/flask/flask_comment_adapter.py @@ -4,6 +4,7 @@ from pydantic import ValidationError from werkzeug.wrappers.response import Response +from exceptions import BlogCommentError from flask import flash, redirect, request, url_for from flask import g as global_request_context from src.application.input_ports.comment_management import CommentManagementPort @@ -69,6 +70,8 @@ def create_comment(self, article_id: int) -> Response: try: req_data = CommentRequest(content=request.form.get("content", "")) + # Pydantic library exception — caught at web boundary for flash + redirect. + # Not in exceptions.py. Do not move it there. except ValidationError as e: for error in e.errors(): msg = error["msg"].removeprefix("Value error, ") @@ -80,14 +83,14 @@ def create_comment(self, article_id: int) -> Response: flash(_("You're posting too fast. Please wait %(remaining)ss before posting again.", remaining=remaining), "warning") return redirect(url_for("article.read_article", article_id=article_id)) - result = self.comment_service.create_comment( - article_id=article_id, - user_id=user.account_id, - content=req_data.content - ) - - if isinstance(result, str): - flash(_(result), "error") + try: + self.comment_service.create_comment( + article_id=article_id, + user_id=user.account_id, + content=req_data.content + ) + except BlogCommentError as e: + flash(_(str(e)), "error") else: flash(_("Comment added."), "success") @@ -114,6 +117,8 @@ def reply_to_comment(self, article_id: int, parent_comment_id: int) -> Response: try: req_data = CommentRequest(content=request.form.get("content", "")) + # Pydantic library exception — caught at web boundary for flash + redirect. + # Not in exceptions.py. Do not move it there. except ValidationError as e: for error in e.errors(): msg = error["msg"].removeprefix("Value error, ") @@ -125,14 +130,14 @@ def reply_to_comment(self, article_id: int, parent_comment_id: int) -> Response: flash(_("You're posting too fast. Please wait %(remaining)ss before posting again.", remaining=remaining), "warning") return redirect(url_for("article.read_article", article_id=article_id)) - result = self.comment_service.create_reply( - parent_comment_id=parent_comment_id, - user_id=user.account_id, - content=req_data.content - ) - - if isinstance(result, str): - flash(_(result), "error") + try: + self.comment_service.create_reply( + parent_comment_id=parent_comment_id, + user_id=user.account_id, + content=req_data.content + ) + except BlogCommentError as e: + flash(_(str(e)), "error") else: flash(_("Reply added."), "success") @@ -154,17 +159,15 @@ def delete_comment(self, article_id: int, comment_id: int) -> Response: flash(_("You must be signed in to delete comments."), "error") return redirect(url_for("auth.login")) - result = self.comment_service.delete_comment( - comment_id=comment_id, - user_id=user.account_id, - ) - - if isinstance(result, str): - flash(_(result), "error") - elif result is True: - flash(_("Comment deleted."), "success") + try: + self.comment_service.delete_comment( + comment_id=comment_id, + user_id=user.account_id, + ) + except BlogCommentError as e: + flash(_(str(e)), "error") else: - flash(_("Unauthorized or error."), "error") + flash(_("Comment deleted."), "success") return redirect(url_for("article.read_article", article_id=article_id)) @@ -186,14 +189,14 @@ def edit_comment(self, article_id: int, comment_id: int) -> Response: return redirect(url_for("auth.login")) content = request.form.get("content", "") - result = self.comment_service.edit_comment( - comment_id=comment_id, - user_id=user.account_id, - content=content, - ) - - if isinstance(result, str): - flash(_(result), "error") + try: + self.comment_service.edit_comment( + comment_id=comment_id, + user_id=user.account_id, + content=content, + ) + except BlogCommentError as e: + flash(_(str(e)), "error") else: flash(_("Comment updated."), "success") @@ -216,16 +219,14 @@ def hard_delete_comment(self, article_id: int, comment_id: int) -> Response: flash(_("You must be signed in to delete comments."), "error") return redirect(url_for("auth.login")) - result = self.comment_service.hard_delete_comment( - comment_id=comment_id, - user_id=user.account_id, - ) - - if isinstance(result, str): - flash(_(result), "error") - elif result is True: - flash(_("Comment permanently deleted."), "success") + try: + self.comment_service.hard_delete_comment( + comment_id=comment_id, + user_id=user.account_id, + ) + except BlogCommentError as e: + flash(_(str(e)), "error") else: - flash(_("Unauthorized or error."), "error") + flash(_("Comment permanently deleted."), "success") return redirect(url_for("article.read_article", article_id=article_id)) diff --git a/src/infrastructure/input_adapters/flask/flask_file_adapter.py b/src/infrastructure/input_adapters/flask/flask_file_adapter.py index 5e744afb..25a31ba3 100644 --- a/src/infrastructure/input_adapters/flask/flask_file_adapter.py +++ b/src/infrastructure/input_adapters/flask/flask_file_adapter.py @@ -2,8 +2,8 @@ from flask_babel import gettext as _ +from exceptions import FileTooLargeError, FileTypeError from flask import jsonify, request, send_file -from src.application.application_exceptions import FileTooLargeError, FileTypeError from src.application.input_ports.file_management import FileManagementPort from src.infrastructure.input_adapters.dto.file_upload_request import FileUploadRequest @@ -40,6 +40,8 @@ def upload_image(self): data=file_data, mime_type=uploaded_file.content_type or "application/octet-stream", ) + # Intentionally broad: catches Pydantic ValidationError or unexpected errors + # from file upload request parsing. Not in exceptions.py. Do not move it there. except Exception as e: return jsonify({"error": str(e)}), 400 diff --git a/src/infrastructure/input_adapters/flask/flask_login_adapter.py b/src/infrastructure/input_adapters/flask/flask_login_adapter.py index 5bef17cc..096ae364 100644 --- a/src/infrastructure/input_adapters/flask/flask_login_adapter.py +++ b/src/infrastructure/input_adapters/flask/flask_login_adapter.py @@ -1,6 +1,7 @@ from flask_babel import gettext as _ from pydantic import ValidationError +from exceptions import AccountBannedError, AuthenticationError from flask import flash, redirect, render_template, request, url_for from flask import g as global_request_context from flask.views import MethodView @@ -49,22 +50,24 @@ def authenticate(self): username=submitted_username, password=request.form.get("password", "") ) + # Pydantic library exception — caught at web boundary for flash + redirect. + # Not in exceptions.py. Do not move it there. except ValidationError as e: for error in e.errors(): location = str(error["loc"][0]) if error["loc"] else "Request" flash(_("Validation Error (%(location)s): %(message)s", location=location, message=error["msg"]), "error") return render_template("login.html", current_user=user, username=submitted_username) - result = self.login_service.authenticate_user( - username=login_data.username, - password=login_data.password - ) - - if not isinstance(result, str): + try: + self.login_service.authenticate_user( + username=login_data.username, + password=login_data.password + ) + except AccountBannedError: + flash(_("This account has been banned."), "error") + except AuthenticationError: + flash(_("Invalid username or password."), "error") + else: return redirect(url_for("article.list_articles")) - if result == "This account has been banned.": - flash(_(result), "error") - else: - flash(_("Invalid username or password."), "error") return render_template("login.html", current_user=user, username=login_data.username) diff --git a/src/infrastructure/input_adapters/flask/flask_registration_adapter.py b/src/infrastructure/input_adapters/flask/flask_registration_adapter.py index 7e457164..5354f10f 100644 --- a/src/infrastructure/input_adapters/flask/flask_registration_adapter.py +++ b/src/infrastructure/input_adapters/flask/flask_registration_adapter.py @@ -1,6 +1,7 @@ from flask_babel import gettext as _ from pydantic import ValidationError +from exceptions import BlogCommentError from flask import flash, redirect, render_template, request, url_for from flask import g as global_request_context from flask.views import MethodView @@ -52,20 +53,22 @@ def register(self): password=request.form.get("password", ""), confirm_password=request.form.get("confirm_password", "") ) + # Pydantic library exception — caught at web boundary for flash + redirect. + # Not in exceptions.py. Do not move it there. except ValidationError as e: for error in e.errors(): location = str(error["loc"][0]) if error["loc"] else "Request" flash(_("%(location)s: %(message)s", location=location, message=error["msg"]), "error") return render_template("registration.html", current_user=user, username=submitted_username, email=submitted_email) - result = self.registration_service.create_account( - username=reg_data.username, - password=reg_data.password, - email=reg_data.email - ) - - if isinstance(result, str): - flash(_(result), "error") + try: + self.registration_service.create_account( + username=reg_data.username, + password=reg_data.password, + email=reg_data.email + ) + except BlogCommentError as e: + flash(_(str(e)), "error") return render_template("registration.html", current_user=user, username=reg_data.username, email=reg_data.email) flash(_("Registration successful. Please sign in."), "success") diff --git a/src/infrastructure/output_adapters/in_memory/account_repository.py b/src/infrastructure/output_adapters/in_memory/account_repository.py index 9541590e..3fc387fe 100644 --- a/src/infrastructure/output_adapters/in_memory/account_repository.py +++ b/src/infrastructure/output_adapters/in_memory/account_repository.py @@ -1,5 +1,6 @@ from datetime import datetime +from exceptions import AccountNotFoundError from src.application.domain.account import Account, AccountRole from src.application.output_ports.account_repository import AccountRepository @@ -225,11 +226,11 @@ def update_ban_status(self, account_id: int, is_banned: bool, ban_reason: str | ban_reason: Optional reason for the ban, or None to clear. Raises: - ValueError: If no account with the given ID exists. + AccountNotFoundError: If no account with the given ID exists. """ account = self._accounts.get(account_id) if account is None: - raise ValueError(f"Account with id {account_id} not found.") + raise AccountNotFoundError(f"Account with id {account_id} not found.") account.is_banned = is_banned account.ban_reason = ban_reason diff --git a/src/infrastructure/output_adapters/security/argon2_password_hasher_adapter.py b/src/infrastructure/output_adapters/security/argon2_password_hasher_adapter.py index 216b08a8..7a209062 100644 --- a/src/infrastructure/output_adapters/security/argon2_password_hasher_adapter.py +++ b/src/infrastructure/output_adapters/security/argon2_password_hasher_adapter.py @@ -52,8 +52,12 @@ def verify(self, password: str, hashed_password: str) -> bool: """ try: return self._hasher.verify(hashed_password, password) + # Argon2 library exception — caught, returns False. + # Not in exceptions.py. Do not move it there. except VerifyMismatchError: return False + # Argon2 library exception — caught for legacy hash fallback. + # Not in exceptions.py. Do not move it there. except InvalidHashError: is_correct = secrets.compare_digest(password, hashed_password) self._hasher.hash("dummy_password_for_timing_consistency") @@ -71,5 +75,7 @@ def check_needs_rehash(self, hashed_password: str) -> bool: """ try: return self._hasher.check_needs_rehash(hashed_password) + # Argon2 library exception — caught for legacy hash fallback. + # Not in exceptions.py. Do not move it there. except InvalidHashError: return True diff --git a/src/infrastructure/output_adapters/sqlalchemy/sqlalchemy_account_adapter.py b/src/infrastructure/output_adapters/sqlalchemy/sqlalchemy_account_adapter.py index 2d6da6b8..d35db3b7 100644 --- a/src/infrastructure/output_adapters/sqlalchemy/sqlalchemy_account_adapter.py +++ b/src/infrastructure/output_adapters/sqlalchemy/sqlalchemy_account_adapter.py @@ -1,3 +1,4 @@ +import logging from typing import cast from psycopg2.errors import UniqueViolation @@ -5,19 +6,24 @@ from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session -from src.application.application_exceptions import AccountAlreadyExistsError +from exceptions import AccountAlreadyExistsError, AccountNotFoundError, DatabaseError from src.application.domain.account import Account from src.application.output_ports.account_repository import AccountRepository from src.infrastructure.output_adapters.dto.account_record import AccountRecord from src.infrastructure.output_adapters.sqlalchemy.models.sqlalchemy_account_model import AccountModel +from src.infrastructure.output_adapters.sqlalchemy.sqlalchemy_base_adapter import ( + SqlAlchemyBaseAdapter, +) -class SqlAlchemyAccountAdapter(AccountRepository): +class SqlAlchemyAccountAdapter(SqlAlchemyBaseAdapter, AccountRepository): """ SQLAlchemy-based implementation of the AccountRepository port. This adapter manages the persistence and retrieval of Account domain entities using SQLAlchemy ORM and the PostgreSQL database. + + All methods may raise DatabaseError on database failure. """ def __init__(self, session: Session): @@ -27,7 +33,7 @@ def __init__(self, session: Session): Args: session (Session): An active SQLAlchemy database session. """ - self._session = session + super().__init__(session) def _to_domain(self, model: AccountModel) -> Account: """ @@ -52,7 +58,7 @@ def find_by_username(self, username: str) -> Account | None: Returns: Account | None: The domain account if found, otherwise None. """ - model = self._session.query(AccountModel).filter_by(account_username=username).first() + model = self._db_query_first(AccountModel, account_username=username) if model is None: return None return self._to_domain(model) @@ -67,7 +73,7 @@ def get_by_id(self, account_id: int) -> Account | None: Returns: Account | None: The domain account if found, otherwise None. """ - model = self._session.get(AccountModel, account_id) + model = self._db_get(AccountModel, account_id) if model is None: return None return self._to_domain(model) @@ -85,8 +91,8 @@ def get_by_ids(self, account_ids: list[int]) -> list[Account]: if not account_ids: return [] - models = ( - self._session.query(AccountModel) + models = self._db_query_raw( + lambda: self._session.query(AccountModel) .filter(AccountModel.account_id.in_(account_ids)) .all() ) @@ -102,7 +108,7 @@ def find_by_email(self, email: str) -> Account | None: Returns: Account | None: The domain account if found, otherwise None. """ - model = self._session.query(AccountModel).filter_by(account_email=email).first() + model = self._db_query_first(AccountModel, account_email=email) if model is None: return None return self._to_domain(model) @@ -121,10 +127,9 @@ def save(self, account: Account) -> None: Raises: AccountAlreadyExistsError: If a unique constraint violation occurs on the username or email column. - RuntimeError: If an unexpected unique constraint violation occurs. """ if account.account_id and account.account_id > 0: - model = self._session.get(AccountModel, account.account_id) + model = self._db_get(AccountModel, account.account_id) if not model: model = AccountModel() else: @@ -134,11 +139,12 @@ def save(self, account: Account) -> None: model.account_password = account.account_password model.account_email = account.account_email model.account_role = account.account_role.value - self._session.add(model) + self._db_add(model) try: - self._session.commit() + self._db_commit() + # SQLAlchemy library exception — caught to translate to domain exception. + # Not in exceptions.py. Do not move it there. except IntegrityError as e: - self._session.rollback() constraint_name = cast(UniqueViolation, e.orig).diag.constraint_name if e.orig else None if constraint_name == "accounts_account_username_key": @@ -150,9 +156,8 @@ def save(self, account: Account) -> None: "This email is already taken." ) from None else: - raise RuntimeError( - f"Unexpected unique constraint violation: {constraint_name}" - ) from None + logging.getLogger(__name__).warning("Unexpected unique constraint violation: %s", constraint_name) + raise AccountAlreadyExistsError("Could not create account.") from None account.account_id = model.account_id def update_avatar(self, account_id: int, avatar_file_id: str | None) -> None: @@ -167,11 +172,11 @@ def update_avatar(self, account_id: int, avatar_file_id: str | None) -> None: account_id: The ID of the account to update. avatar_file_id: The new avatar file UUID, or None to remove. """ - model = self._session.get(AccountModel, account_id) + model = self._db_get(AccountModel, account_id) if model is None: return model.avatar_file_id = avatar_file_id - self._session.commit() + self._db_commit() def update_email(self, account_id: int, new_email: str) -> None: """ @@ -187,19 +192,22 @@ def update_email(self, account_id: int, new_email: str) -> None: Raises: AccountAlreadyExistsError: If the new email is already taken by another account. + DatabaseError: If an unexpected constraint violation or DB + error occurs. """ - model = self._session.get(AccountModel, account_id) + model = self._db_get(AccountModel, account_id) if model is None: return model.account_email = new_email try: - self._session.commit() + self._db_commit() + # SQLAlchemy library exception — caught to translate to domain exception. + # Not in exceptions.py. Do not move it there. except IntegrityError as e: - self._session.rollback() constraint_name = cast(UniqueViolation, e.orig).diag.constraint_name if e.orig else None if constraint_name == "accounts_account_email_key": raise AccountAlreadyExistsError("This email is already taken.") from None - raise + raise DatabaseError("Unexpected database constraint violation.") from e def update_password(self, account_id: int, new_hashed_password: str) -> None: """ @@ -211,11 +219,11 @@ def update_password(self, account_id: int, new_hashed_password: str) -> None: account_id: The ID of the account to update. new_hashed_password: The new Argon2 hash to store. """ - model = self._session.get(AccountModel, account_id) + model = self._db_get(AccountModel, account_id) if model is None: return model.account_password = new_hashed_password - self._session.commit() + self._db_commit() def update_ban_status(self, account_id: int, is_banned: bool, ban_reason: str | None) -> None: """ @@ -230,12 +238,12 @@ def update_ban_status(self, account_id: int, is_banned: bool, ban_reason: str | is_banned: True to ban, False to unban. ban_reason: Optional reason for the ban, or None to clear. """ - model = self._session.get(AccountModel, account_id) + model = self._db_get(AccountModel, account_id) if model is None: return model.is_banned = is_banned model.ban_reason = ban_reason - self._session.commit() + self._db_commit() def update_role(self, account_id: int, new_role: str) -> None: """ @@ -245,11 +253,11 @@ def update_role(self, account_id: int, new_role: str) -> None: account_id: The ID of the account to update. new_role: The new role string ("user" or "author"). """ - model = self._session.get(AccountModel, account_id) + model = self._db_get(AccountModel, account_id) if model is None: return model.account_role = new_role - self._session.commit() + self._db_commit() def get_all(self) -> list[Account]: """ @@ -258,7 +266,7 @@ def get_all(self) -> list[Account]: Returns: list[Account]: A list of all Account domain entities. """ - models = self._session.query(AccountModel).all() + models = self._db_query_all(AccountModel) return [self._to_domain(model) for model in models] def get_all_paginated(self, page: int = 1, per_page: int = 20) -> list[Account]: @@ -272,8 +280,8 @@ def get_all_paginated(self, page: int = 1, per_page: int = 20) -> list[Account]: Returns: list[Account]: A list of Account domain entities for the given page. """ - models = ( - self._session.query(AccountModel) + models = self._db_query_raw( + lambda: self._session.query(AccountModel) .order_by(AccountModel.account_created_at.desc()) .limit(per_page) .offset((page - 1) * per_page) @@ -288,7 +296,7 @@ def count_all(self) -> int: Returns: int: The total count of accounts. """ - return self._session.query(AccountModel).count() + return self._db_query_raw(lambda: self._session.query(AccountModel).count()) def search(self, query: str, page: int = 1, per_page: int = 20) -> list[Account]: """ @@ -305,8 +313,8 @@ def search(self, query: str, page: int = 1, per_page: int = 20) -> list[Account] for the given page. """ like = f"%{query}%" - models = ( - self._session.query(AccountModel) + models = self._db_query_raw( + lambda: self._session.query(AccountModel) .filter( or_( AccountModel.account_username.ilike(like), @@ -331,8 +339,8 @@ def count_search(self, query: str) -> int: int: The total count of matching accounts. """ like = f"%{query}%" - return ( - self._session.query(AccountModel) + return self._db_query_raw( + lambda: self._session.query(AccountModel) .filter( or_( AccountModel.account_username.ilike(like), @@ -353,10 +361,10 @@ def delete(self, account_id: int) -> None: account_id (int): The unique identifier of the account to delete. Raises: - ValueError: If no account with the given ID exists. + AccountNotFoundError: If no account with the given ID exists. """ - model = self._session.get(AccountModel, account_id) + model = self._db_get(AccountModel, account_id) if model is None: - raise ValueError(f"Account with id {account_id} not found.") - self._session.delete(model) - self._session.commit() + raise AccountNotFoundError(f"Account with id {account_id} not found.") + self._db_delete(model) + self._db_commit() diff --git a/src/infrastructure/output_adapters/sqlalchemy/sqlalchemy_article_adapter.py b/src/infrastructure/output_adapters/sqlalchemy/sqlalchemy_article_adapter.py index 492ed72e..a491afb7 100644 --- a/src/infrastructure/output_adapters/sqlalchemy/sqlalchemy_article_adapter.py +++ b/src/infrastructure/output_adapters/sqlalchemy/sqlalchemy_article_adapter.py @@ -6,14 +6,19 @@ from src.infrastructure.output_adapters.dto.article_record import ArticleRecord from src.infrastructure.output_adapters.sqlalchemy.models.sqlalchemy_account_model import AccountModel from src.infrastructure.output_adapters.sqlalchemy.models.sqlalchemy_article_model import ArticleModel +from src.infrastructure.output_adapters.sqlalchemy.sqlalchemy_base_adapter import ( + SqlAlchemyBaseAdapter, +) -class SqlAlchemyArticleAdapter(ArticleRepository): +class SqlAlchemyArticleAdapter(SqlAlchemyBaseAdapter, ArticleRepository): """ SQLAlchemy-based implementation of the ArticleRepository port. This adapter manages the persistence and retrieval of Article domain entities using SQLAlchemy ORM and the PostgreSQL database. + + All methods may raise DatabaseError on database failure. """ def __init__(self, session: Session): @@ -23,7 +28,7 @@ def __init__(self, session: Session): Args: session (Session): An active SQLAlchemy database session. """ - self._session = session + super().__init__(session) def _to_domain(self, model: ArticleModel) -> Article: """ @@ -45,8 +50,8 @@ def get_all_ordered_by_date_desc(self) -> list[Article]: Returns: list[Article]: A list of all Article domain entities. """ - models = ( - self._session.query(ArticleModel) + models = self._db_query_raw( + lambda: self._session.query(ArticleModel) .order_by(desc(ArticleModel.article_published_at)) .all() ) @@ -62,7 +67,7 @@ def get_by_id(self, article_id: int) -> Article | None: Returns: Article | None: The Article domain entity if found, None otherwise. """ - model = self._session.get(ArticleModel, article_id) + model = self._db_get(ArticleModel, article_id) if model is None: return None return self._to_domain(model) @@ -77,15 +82,17 @@ def save(self, article: Article) -> None: article (Article): The Article domain entity to persist. """ if article.article_id and article.article_id > 0: - self._session.query(ArticleModel).filter_by( - article_id=article.article_id, - ).update({ + self._db_query_raw( + lambda: self._session.query(ArticleModel).filter_by( + article_id=article.article_id, + ).update({ ArticleModel.article_title: article.article_title, ArticleModel.article_description: article.article_description, ArticleModel.article_content: article.article_content, ArticleModel.article_edited_at: article.article_edited_at, - }) - self._session.commit() + }) + ) + self._db_commit() return model = ArticleModel() @@ -94,8 +101,8 @@ def save(self, article: Article) -> None: model.article_description = article.article_description model.article_content = article.article_content model.article_edited_at = article.article_edited_at - self._session.add(model) - self._session.commit() + self._db_add(model) + self._db_commit() article.article_id = model.article_id def delete(self, article: Article) -> None: @@ -105,10 +112,12 @@ def delete(self, article: Article) -> None: Args: article (Article): The Article domain entity to delete. """ - self._session.query(ArticleModel).filter_by( - article_id=article.article_id, - ).delete() - self._session.commit() + self._db_query_raw( + lambda: self._session.query(ArticleModel).filter_by( + article_id=article.article_id, + ).delete() + ) + self._db_commit() def get_paginated(self, page: int, per_page: int) -> list[Article]: """ @@ -122,8 +131,8 @@ def get_paginated(self, page: int, per_page: int) -> list[Article]: list[Article]: A list of Article domain entities for the specified page. """ offset = (page - 1) * per_page - models = ( - self._session.query(ArticleModel) + models = self._db_query_raw( + lambda: self._session.query(ArticleModel) .order_by(desc(ArticleModel.article_published_at)) .offset(offset) .limit(per_page) @@ -139,7 +148,7 @@ def count_all(self) -> int: Returns: int: The total count of articles in the database. """ - return self._session.query(ArticleModel).count() + return self._db_query_raw(lambda: self._session.query(ArticleModel).count()) def search(self, query: str, page: int, per_page: int) -> list[Article]: """ @@ -158,8 +167,8 @@ def search(self, query: str, page: int, per_page: int) -> list[Article]: """ like = f"%{query}%" offset = (page - 1) * per_page - models = ( - self._session.query(ArticleModel) + models = self._db_query_raw( + lambda: self._session.query(ArticleModel) .outerjoin( AccountModel, ArticleModel.article_author_id == AccountModel.account_id, @@ -191,8 +200,8 @@ def count_search(self, query: str) -> int: The total number of articles matching the query. """ like = f"%{query}%" - return ( - self._session.query(ArticleModel) + return self._db_query_raw( + lambda: self._session.query(ArticleModel) .outerjoin( AccountModel, ArticleModel.article_author_id == AccountModel.account_id, diff --git a/src/infrastructure/output_adapters/sqlalchemy/sqlalchemy_base_adapter.py b/src/infrastructure/output_adapters/sqlalchemy/sqlalchemy_base_adapter.py new file mode 100644 index 00000000..dd8c9bd4 --- /dev/null +++ b/src/infrastructure/output_adapters/sqlalchemy/sqlalchemy_base_adapter.py @@ -0,0 +1,136 @@ +from sqlalchemy.exc import IntegrityError, SQLAlchemyError +from sqlalchemy.orm import Session + +from exceptions import DatabaseError + + +class SqlAlchemyBaseAdapter: + """Base for SQLAlchemy output adapters. + + Wraps common DB operations to translate SQLAlchemyError -> DatabaseError. + IntegrityError is re-raised for callers that handle unique constraints. + """ + + def __init__(self, session: Session): + """Initialize with an active SQLAlchemy database session. + + Args: + session: Active DB session. + """ + self._session = session + + def _db_get(self, model_class, pk): + """Get a record by primary key. + + Args: + model_class: SQLAlchemy model class. + pk: Primary key value. + + Raises: + DatabaseError: If a SQLAlchemy error occurs. + """ + try: + return self._session.get(model_class, pk) + # SQLAlchemy library exception — caught and translated to DatabaseError. + # Not in exceptions.py. Do not move it there. + except SQLAlchemyError as e: + raise DatabaseError("Database read failed.") from e + + def _db_add(self, model): + """Add a model instance to the session. + + Args: + model: SQLAlchemy model instance. + + Raises: + DatabaseError: If a SQLAlchemy error occurs. + """ + try: + self._session.add(model) + # SQLAlchemy library exception — caught and translated to DatabaseError. + # Not in exceptions.py. Do not move it there. + except SQLAlchemyError as e: + raise DatabaseError("Database insert failed.") from e + + def _db_delete(self, model): + """Delete a model instance from the session. + + Args: + model: SQLAlchemy model instance. + + Raises: + DatabaseError: If a SQLAlchemy error occurs. + """ + try: + self._session.delete(model) + # SQLAlchemy library exception — caught and translated to DatabaseError. + # Not in exceptions.py. Do not move it there. + except SQLAlchemyError as e: + raise DatabaseError("Database delete failed.") from e + + def _db_commit(self): + """Commit the current transaction. + + Handles rollback on error. IntegrityError is re-raised so callers + can handle unique constraint violations with domain exceptions. + + Raises: + IntegrityError: Re-raised for caller-specific handling. + DatabaseError: If a non-integrity SQLAlchemy error occurs. + """ + try: + self._session.commit() + # SQLAlchemy library exception — re-raised for constraint handling by callers. + # Not in exceptions.py. Do not move it there. + except IntegrityError: + self._session.rollback() + raise + # SQLAlchemy library exception — caught and translated to DatabaseError. + # Not in exceptions.py. Do not move it there. + except SQLAlchemyError as e: + self._session.rollback() + raise DatabaseError("Database commit failed.") from e + + def _db_query_first(self, model_class, **filters): + """Query.filter_by(**filters).first() with error wrapping. + + Args: + model_class: SQLAlchemy model class. + **filters: Column-value filter pairs. + + Raises: + DatabaseError: If a SQLAlchemy error occurs. + """ + return self._db_query_raw( + lambda: self._session.query(model_class).filter_by(**filters).first() + ) + + def _db_query_all(self, model_class, **filters): + """Query.filter_by(**filters).all() with error wrapping. + + Args: + model_class: SQLAlchemy model class. + **filters: Column-value filter pairs. + + Raises: + DatabaseError: If a SQLAlchemy error occurs. + """ + return self._db_query_raw( + lambda: self._session.query(model_class).filter_by(**filters).all() + ) + + def _db_query_raw(self, query_fn): + """Execute an arbitrary query function with error wrapping. + + Args: + query_fn: A callable that performs SQLAlchemy query operations. + + Raises: + DatabaseError: If a SQLAlchemy error occurs. + """ + try: + return query_fn() + # SQLAlchemy library exception — caught and translated to DatabaseError. + # Not in exceptions.py. Do not move it there. + except SQLAlchemyError as e: + raise DatabaseError("Database query failed.") from e diff --git a/src/infrastructure/output_adapters/sqlalchemy/sqlalchemy_comment_adapter.py b/src/infrastructure/output_adapters/sqlalchemy/sqlalchemy_comment_adapter.py index 12e3c4a7..984818ac 100644 --- a/src/infrastructure/output_adapters/sqlalchemy/sqlalchemy_comment_adapter.py +++ b/src/infrastructure/output_adapters/sqlalchemy/sqlalchemy_comment_adapter.py @@ -4,14 +4,19 @@ from src.application.output_ports.comment_repository import CommentRepository from src.infrastructure.output_adapters.dto.comment_record import CommentRecord from src.infrastructure.output_adapters.sqlalchemy.models.sqlalchemy_comment_model import CommentModel +from src.infrastructure.output_adapters.sqlalchemy.sqlalchemy_base_adapter import ( + SqlAlchemyBaseAdapter, +) -class SqlAlchemyCommentAdapter(CommentRepository): +class SqlAlchemyCommentAdapter(SqlAlchemyBaseAdapter, CommentRepository): """ SQLAlchemy-based implementation of the CommentRepository port. This adapter manages the persistence and retrieval of Comment domain entities using SQLAlchemy ORM and the database. + + All methods may raise DatabaseError on database failure. """ def __init__(self, session: Session): @@ -21,7 +26,7 @@ def __init__(self, session: Session): Args: session (Session): An active SQLAlchemy database session. """ - self._session = session + super().__init__(session) def _to_domain(self, model: CommentModel) -> Comment: """ @@ -46,9 +51,10 @@ def save(self, comment: Comment) -> None: comment (Comment): The domain Comment entity to persist. """ if comment.comment_id and comment.comment_id > 0: - self._session.query(CommentModel).filter_by( - comment_id=comment.comment_id, - ).update({ + self._db_query_raw( + lambda: self._session.query(CommentModel).filter_by( + comment_id=comment.comment_id, + ).update({ CommentModel.comment_article_id: comment.comment_article_id, CommentModel.comment_written_account_id: comment.comment_written_account_id, CommentModel.comment_reply_to: comment.comment_reply_to, @@ -56,12 +62,13 @@ def save(self, comment: Comment) -> None: CommentModel.is_deleted: comment.is_deleted, CommentModel.deleted_at: comment.deleted_at, CommentModel.edited_at: comment.edited_at, - }) - self._session.commit() + }) + ) + self._db_commit() return model = CommentModel() - self._session.add(model) + self._db_add(model) model.comment_article_id = comment.comment_article_id model.comment_written_account_id = comment.comment_written_account_id model.comment_reply_to = comment.comment_reply_to @@ -69,7 +76,7 @@ def save(self, comment: Comment) -> None: model.is_deleted = comment.is_deleted model.deleted_at = comment.deleted_at model.edited_at = comment.edited_at - self._session.commit() + self._db_commit() def get_by_id(self, comment_id: int) -> Comment | None: """ @@ -81,7 +88,7 @@ def get_by_id(self, comment_id: int) -> Comment | None: Returns: Comment | None: The Comment domain entity if found, None otherwise. """ - model = self._session.get(CommentModel, comment_id) + model = self._db_get(CommentModel, comment_id) if model is None: return None return self._to_domain(model) @@ -96,7 +103,7 @@ def get_all_by_article_id(self, article_id: int) -> list[Comment]: Returns: list[Comment]: A list of Comment domain entities. """ - models = self._session.query(CommentModel).filter_by(comment_article_id=article_id).all() + models = self._db_query_all(CommentModel, comment_article_id=article_id) return [self._to_domain(model) for model in models] def get_by_reply_to(self, comment_id: int) -> list[Comment]: @@ -109,9 +116,7 @@ def get_by_reply_to(self, comment_id: int) -> list[Comment]: Returns: list[Comment]: A list of direct child Comment domain entities. """ - models = self._session.query(CommentModel).filter_by( - comment_reply_to=comment_id, - ).all() + models = self._db_query_all(CommentModel, comment_reply_to=comment_id) return [self._to_domain(model) for model in models] def get_by_account_id(self, account_id: int) -> list[Comment]: @@ -124,9 +129,7 @@ def get_by_account_id(self, account_id: int) -> list[Comment]: Returns: list[Comment]: A list of Comment domain entities for this author. """ - models = self._session.query(CommentModel).filter_by( - comment_written_account_id=account_id, - ).all() + models = self._db_query_all(CommentModel, comment_written_account_id=account_id) return [self._to_domain(model) for model in models] def delete(self, comment_id: int) -> None: @@ -136,7 +139,9 @@ def delete(self, comment_id: int) -> None: Args: comment_id (int): The unique identifier of the comment to delete. """ - self._session.query(CommentModel).filter_by( - comment_id=comment_id, - ).delete() - self._session.commit() + self._db_query_raw( + lambda: self._session.query(CommentModel).filter_by( + comment_id=comment_id, + ).delete() + ) + self._db_commit() diff --git a/src/infrastructure/output_adapters/sqlalchemy/sqlalchemy_file_storage_adapter.py b/src/infrastructure/output_adapters/sqlalchemy/sqlalchemy_file_storage_adapter.py index 5c5ccc19..473c390d 100644 --- a/src/infrastructure/output_adapters/sqlalchemy/sqlalchemy_file_storage_adapter.py +++ b/src/infrastructure/output_adapters/sqlalchemy/sqlalchemy_file_storage_adapter.py @@ -6,14 +6,19 @@ from src.application.domain.file_record import FileRecord from src.application.output_ports.file_storage_repository import FileStorageRepository from src.infrastructure.output_adapters.sqlalchemy.models.sqlalchemy_uploaded_file_model import UploadedFileModel +from src.infrastructure.output_adapters.sqlalchemy.sqlalchemy_base_adapter import ( + SqlAlchemyBaseAdapter, +) -class SqlAlchemyFileStorageAdapter(FileStorageRepository): +class SqlAlchemyFileStorageAdapter(SqlAlchemyBaseAdapter, FileStorageRepository): """SQLAlchemy-based implementation of FileStorageRepository. Persists uploaded files as BYTEA in the uploaded_files table. Maps directly between UploadedFileModel (ORM) and FileRecord (domain). No DTO needed — FileRecord has no enums or complex conversions. + + All methods may raise DatabaseError on database failure. """ def __init__(self, session: Session): @@ -22,7 +27,7 @@ def __init__(self, session: Session): Args: session: Active DB session. """ - self._session = session + super().__init__(session) def save(self, file_record: FileRecord) -> FileRecord: """Persist a file record to the database. @@ -41,8 +46,8 @@ def save(self, file_record: FileRecord) -> FileRecord: file_data=file_record.data, created_at=file_record.created_at, ) - self._session.add(model) - self._session.commit() + self._db_add(model) + self._db_commit() return file_record def get(self, file_id: str) -> FileRecord | None: @@ -54,7 +59,7 @@ def get(self, file_id: str) -> FileRecord | None: Returns: FileRecord if found, None otherwise. """ - model = self._session.get(UploadedFileModel, file_id) + model = self._db_get(UploadedFileModel, file_id) if model is None: return None return FileRecord( @@ -74,8 +79,8 @@ def delete(self, file_id: str) -> None: Args: file_id: UUID string. """ - model = self._session.get(UploadedFileModel, file_id) + model = self._db_get(UploadedFileModel, file_id) if model is None: return - self._session.delete(model) - self._session.commit() + self._db_delete(model) + self._db_commit() diff --git a/tests/exceptions_tests.py b/tests/exceptions_tests.py deleted file mode 100644 index 8cc07435..00000000 --- a/tests/exceptions_tests.py +++ /dev/null @@ -1,5 +0,0 @@ -class ExceptionTest(Exception): - """ - Base class for all test-specific exceptions. - """ - pass diff --git a/tests/tests_infrastructure/tests_input_adapters/tests_flask/test_flask_account_session_adapter.py b/tests/tests_infrastructure/tests_input_adapters/tests_flask/test_flask_account_session_adapter.py index df5b675b..be4af806 100644 --- a/tests/tests_infrastructure/tests_input_adapters/tests_flask/test_flask_account_session_adapter.py +++ b/tests/tests_infrastructure/tests_input_adapters/tests_flask/test_flask_account_session_adapter.py @@ -444,7 +444,8 @@ def test_update_email_unauthenticated(self): def test_update_email_error(self): fake_user = create_test_account(account_id=1, account_email="old@test.com") self.mock_session_service.get_current_account.return_value = fake_user - self.mock_session_service.update_email.return_value = "This email is already taken." + from exceptions import EmailAlreadyTakenError + self.mock_session_service.update_email.side_effect = EmailAlreadyTakenError("This email is already taken.") response = self.client.post( "/profile/email", data={"email": "taken@test.com"}, @@ -534,7 +535,8 @@ def test_admin_change_role_nonexistent_target(self): admin = create_test_account(account_id=1, account_role=AccountRole.ADMIN) self.set_current_user(admin) self.mock_session_service.get_current_account.return_value = admin - self.mock_session_service.update_account_role.return_value = "Account not found." + from exceptions import AccountNotFoundError + self.mock_session_service.update_account_role.side_effect = AccountNotFoundError("Account not found.") self.mock_session_service.get_account_by_id.return_value = None response = self.client.post( "/admin/users/999/role", @@ -612,7 +614,8 @@ def test_admin_ban_another_admin_returns_302_with_flash(self): admin = create_test_account(account_id=1, account_role=AccountRole.ADMIN) self.set_current_user(admin) self.mock_session_service.get_current_account.return_value = admin - self.mock_session_service.ban_account.return_value = "Cannot ban another admin." + from exceptions import AuthorizationError + self.mock_session_service.ban_account.side_effect = AuthorizationError("Cannot ban another admin.") response = self.client.post("/admin/users/2/ban", data={"ban_reason": "Spam"}, follow_redirects=True) assert response.status_code == 200 assert b"Cannot ban another admin" in response.data @@ -622,7 +625,8 @@ def test_admin_ban_nonexistent_user_redirects_with_flash(self): admin = create_test_account(account_id=1, account_role=AccountRole.ADMIN) self.set_current_user(admin) self.mock_session_service.get_current_account.return_value = admin - self.mock_session_service.ban_account.return_value = "Account not found." + from exceptions import AccountNotFoundError + self.mock_session_service.ban_account.side_effect = AccountNotFoundError("Account not found.") response = self.client.post("/admin/users/999/ban", data={"ban_reason": "Spam"}, follow_redirects=True) assert response.status_code == 200 assert b"Account not found" in response.data diff --git a/tests/tests_infrastructure/tests_input_adapters/tests_flask/test_flask_article_adapter.py b/tests/tests_infrastructure/tests_input_adapters/tests_flask/test_flask_article_adapter.py index 42432ff5..0279a645 100644 --- a/tests/tests_infrastructure/tests_input_adapters/tests_flask/test_flask_article_adapter.py +++ b/tests/tests_infrastructure/tests_input_adapters/tests_flask/test_flask_article_adapter.py @@ -305,9 +305,10 @@ def _save_side_effect(article): def test_author_create_article_service_error(self): author = create_test_account(account_id=10, account_role=AccountRole.AUTHOR) self._prepare_user_context(author) + from exceptions import BlogCommentError mock_service = Mock(spec=ArticleService) self.adapter.article_service = mock_service - mock_service.create_article.return_value = "Service Error Message" + mock_service.create_article.side_effect = BlogCommentError("Service Error Message") response = self.client.post( "/api/articles", @@ -407,7 +408,8 @@ def test_update_article_validation_error(self): def test_delete_article_service_error(self): author = create_test_account(account_id=10, account_role=AccountRole.AUTHOR) self._prepare_user_context(author) - self.adapter.article_service.delete_article = Mock(return_value="Delete Error") + from exceptions import BlogCommentError + self.adapter.article_service.delete_article = Mock(side_effect=BlogCommentError("Delete Error")) response = self.client.delete("/api/articles/1") assert response.status_code == 403 assert response.get_json() == {"error": "Delete Error"} diff --git a/tests/tests_infrastructure/tests_input_adapters/tests_flask/test_flask_comment_adapter.py b/tests/tests_infrastructure/tests_input_adapters/tests_flask/test_flask_comment_adapter.py index f3dafbf3..22938192 100644 --- a/tests/tests_infrastructure/tests_input_adapters/tests_flask/test_flask_comment_adapter.py +++ b/tests/tests_infrastructure/tests_input_adapters/tests_flask/test_flask_comment_adapter.py @@ -83,7 +83,8 @@ def test_create_comment_validation_error(self): def test_create_comment_service_error_string(self): user = create_test_account(account_id=123) self.set_current_user(user) - self.mock_comment_service.create_comment.return_value = "Article not found" + from exceptions import ArticleNotFoundError + self.mock_comment_service.create_comment.side_effect = ArticleNotFoundError("Article not found") response = self.client.post("/articles/1/comments", data={"content": "Valid"}, follow_redirects=True) assert b"Article not found" in response.data assert b"alert-error" in response.data @@ -133,7 +134,8 @@ def test_reply_validation_error(self): def test_reply_service_error_string(self): user = create_test_account(account_id=123) self.set_current_user(user) - self.mock_comment_service.create_reply.return_value = "Parent not found" + from exceptions import CommentNotFoundError + self.mock_comment_service.create_reply.side_effect = CommentNotFoundError("Parent not found") response = self.client.post("/articles/1/comments/10/reply", data={"content": "Valid"}, follow_redirects=True) assert b"Parent not found" in response.data assert b"alert-error" in response.data @@ -176,17 +178,20 @@ def test_delete_comment_requires_login(self): def test_delete_comment_service_error_string(self): user = create_test_account(account_id=1, account_role=AccountRole.USER) self.set_current_user(user) - self.mock_comment_service.delete_comment.return_value = "Comment not found" + from exceptions import CommentNotFoundError + self.mock_comment_service.delete_comment.side_effect = CommentNotFoundError("Comment not found") response = self.client.post("/articles/1/comments/99/delete", follow_redirects=True) assert b"Comment not found" in response.data assert b"alert-error" in response.data - def test_delete_comment_unauthorized_none_return(self): + def test_delete_comment_unauthorized_raises(self): user = create_test_account(account_id=1, account_role=AccountRole.USER) self.set_current_user(user) - self.mock_comment_service.delete_comment.return_value = None + from exceptions import OwnershipError + self.mock_comment_service.delete_comment.side_effect = OwnershipError( + "Unauthorized: You can only delete your own comments." + ) response = self.client.post("/articles/1/comments/99/delete", follow_redirects=True) - assert b"Unauthorized or error" in response.data assert b"alert-error" in response.data @@ -242,7 +247,8 @@ def test_hard_delete_comment_not_authenticated(self): def test_hard_delete_comment_service_error_string(self): user = create_test_account(account_id=1, account_role=AccountRole.USER) self.set_current_user(user) - self.mock_comment_service.hard_delete_comment.return_value = "Comment not found" + from exceptions import CommentNotFoundError + self.mock_comment_service.hard_delete_comment.side_effect = CommentNotFoundError("Comment not found") response = self.client.post("/articles/1/comments/99/delete-permanent", follow_redirects=True) assert b"Comment not found" in response.data assert b"alert-error" in response.data diff --git a/tests/tests_infrastructure/tests_output_adapters/test_flask_session_adapter.py b/tests/tests_infrastructure/tests_output_adapters/test_flask_session_adapter.py index 5844496a..73c798cd 100644 --- a/tests/tests_infrastructure/tests_output_adapters/test_flask_session_adapter.py +++ b/tests/tests_infrastructure/tests_output_adapters/test_flask_session_adapter.py @@ -6,7 +6,6 @@ from src.application.domain.account import Account, AccountRole from src.application.output_ports.account_repository import AccountRepository from src.infrastructure.output_adapters.session.flask_session_adapter import FlaskSessionAdapter -from tests.exceptions_tests import ExceptionTest from tests.test_domain_factories import create_test_account @@ -110,7 +109,9 @@ def test_get_account_with_corrupted_structure_returns_none(self): def test_get_account_repository_timeout_resilience(self): from flask import session as flask_session - self.mock_repo.get_by_id.side_effect = ExceptionTest("DB Timeout") + # Intentionally NOT in exceptions.py: test-only. Exception sufficient. + # Do not move to exceptions.py. + self.mock_repo.get_by_id.side_effect = Exception("DB Timeout") with self.app.test_request_context(): flask_session[self.adapter._KEY_USER_ID] = 123 # TODO: Add try/except in FlaskSessionAdapter.get_account to return None if the repository fails. @@ -118,7 +119,7 @@ def test_get_account_repository_timeout_resilience(self): try: retrieved = self.adapter.get_account() assert retrieved is None - except ExceptionTest: + except Exception: # Fallback until the TODO is implemented pass diff --git a/tests/tests_infrastructure/tests_output_adapters/tests_sqlalchemy/test_sqlalchemy_account_adapter.py b/tests/tests_infrastructure/tests_output_adapters/tests_sqlalchemy/test_sqlalchemy_account_adapter.py index c5fcd56a..ad367efc 100644 --- a/tests/tests_infrastructure/tests_output_adapters/tests_sqlalchemy/test_sqlalchemy_account_adapter.py +++ b/tests/tests_infrastructure/tests_output_adapters/tests_sqlalchemy/test_sqlalchemy_account_adapter.py @@ -1,6 +1,6 @@ import pytest -from src.application.application_exceptions import AccountAlreadyExistsError +from exceptions import AccountAlreadyExistsError from src.application.domain.account import AccountRole from src.infrastructure.output_adapters.sqlalchemy.models.sqlalchemy_account_model import AccountModel from src.infrastructure.output_adapters.sqlalchemy.sqlalchemy_account_adapter import SqlAlchemyAccountAdapter diff --git a/tests/tests_services/test_article_service.py b/tests/tests_services/test_article_service.py index 0052b4f7..af3751f8 100644 --- a/tests/tests_services/test_article_service.py +++ b/tests/tests_services/test_article_service.py @@ -2,6 +2,14 @@ from datetime import UTC, datetime from unittest.mock import MagicMock +import pytest + +from exceptions import ( + AccountNotFoundError, + ArticleNotFoundError, + InsufficientPermissionsError, + OwnershipError, +) from src.application.domain.account import AccountRole from src.application.domain.article import Article from src.application.input_ports.file_management import FileManagementPort @@ -76,30 +84,30 @@ def test_create_article_unauthorized_role(self): self.mock_account_repo.get_by_id.return_value = fake_account - result = self.service.create_article( - title="Hacked Article", - content="Bad Content !", - author_id=fake_account.account_id, - author_role=fake_account.account_role, - ) + with pytest.raises(InsufficientPermissionsError, match="Insufficient permissions"): + self.service.create_article( + title="Hacked Article", + content="Bad Content !", + author_id=fake_account.account_id, + author_role=fake_account.account_role, + ) self.mock_account_repo.get_by_id.assert_called_once_with(fake_account.account_id) self.mock_article_repo.save.assert_not_called() - assert result == "Insufficient permissions." def test_create_article_account_not_found(self): self.mock_account_repo.get_by_id.return_value = None - result = self.service.create_article( - title="Ghost Article", - content="Content from beyond!", - author_id=999, - author_role=AccountRole.AUTHOR, - ) + with pytest.raises(AccountNotFoundError, match="not found"): + self.service.create_article( + title="Ghost Article", + content="Content from beyond!", + author_id=999, + author_role=AccountRole.AUTHOR, + ) self.mock_account_repo.get_by_id.assert_called_once_with(999) self.mock_article_repo.save.assert_not_called() - assert result == "Account not found." class TestGetArticles(ArticleServiceTestBase): @@ -235,48 +243,48 @@ def test_update_article_unauthorized(self): fake_account = create_test_account(account_id=99, account_role=AccountRole.AUTHOR) self.mock_account_repo.get_by_id.return_value = fake_account - result = self.service.update_article( - article_id=fake_article.article_id, - user_id=fake_account.account_id, - title="Hacked Title", - content="Hacked Content", - ) + with pytest.raises(OwnershipError, match="not the author"): + self.service.update_article( + article_id=fake_article.article_id, + user_id=fake_account.account_id, + title="Hacked Title", + content="Hacked Content", + ) self.mock_article_repo.get_by_id.assert_called_once_with(fake_article.article_id) self.mock_article_repo.save.assert_not_called() - assert result == "Unauthorized : You are not the author of this article." def test_update_article_insufficient_role(self): fake_account = create_test_account(account_id=1, account_role=AccountRole.USER) self.mock_account_repo.get_by_id.return_value = fake_account - result = self.service.update_article( - article_id=1, - user_id=fake_account.account_id, - title="Hacked Title", - content="Hacked Content", - ) + with pytest.raises(InsufficientPermissionsError, match="Insufficient permissions"): + self.service.update_article( + article_id=1, + user_id=fake_account.account_id, + title="Hacked Title", + content="Hacked Content", + ) self.mock_article_repo.get_by_id.assert_not_called() self.mock_account_repo.get_by_id.assert_called_once_with(fake_account.account_id) self.mock_article_repo.save.assert_not_called() - assert result == "Insufficient permissions." def test_update_article_not_found(self): self.mock_article_repo.get_by_id.return_value = None fake_account = create_test_account(account_role=AccountRole.AUTHOR) self.mock_account_repo.get_by_id.return_value = fake_account - result = self.service.update_article( - article_id=999, - user_id=fake_account.account_id, - title="New Title", - content="New Content", - ) + with pytest.raises(ArticleNotFoundError, match="not found"): + self.service.update_article( + article_id=999, + user_id=fake_account.account_id, + title="New Title", + content="New Content", + ) self.mock_article_repo.get_by_id.assert_called_once_with(999) self.mock_article_repo.save.assert_not_called() - assert result == "Article not found." class TestDeleteArticle(ArticleServiceTestBase): @@ -307,21 +315,25 @@ def test_delete_article_unauthorized_ownership(self): fake_author_other = create_test_account(account_id=99, account_role=AccountRole.AUTHOR) self.mock_article_repo.get_by_id.return_value = fake_article self.mock_account_repo.get_by_id.return_value = fake_author_other - result = self.service.delete_article(article_id=fake_article.article_id, user_id=fake_author_other.account_id) + + with pytest.raises(OwnershipError, match="Only authors or admins"): + self.service.delete_article(article_id=fake_article.article_id, user_id=fake_author_other.account_id) + self.mock_account_repo.get_by_id.assert_called_once_with(fake_author_other.account_id) self.mock_article_repo.get_by_id.assert_called_once_with(fake_article.article_id) self.mock_article_repo.delete.assert_not_called() - assert result == "Unauthorized : Only authors or admins can delete articles." def test_delete_article_not_found(self): fake_account = create_test_account(account_role=AccountRole.AUTHOR) self.mock_article_repo.get_by_id.return_value = None self.mock_account_repo.get_by_id.return_value = fake_account - result = self.service.delete_article(article_id=999, user_id=fake_account.account_id) + + with pytest.raises(ArticleNotFoundError, match="not found"): + self.service.delete_article(article_id=999, user_id=fake_account.account_id) + self.mock_account_repo.get_by_id.assert_called_once_with(fake_account.account_id) self.mock_article_repo.get_by_id.assert_called_once_with(999) self.mock_article_repo.delete.assert_not_called() - assert result == "Article not found." class TestGetAuthorName(ArticleServiceTestBase): @@ -355,7 +367,6 @@ def test_get_article_with_comments_success(self): self.mock_account_repo.get_by_ids.return_value = fake_accounts result = self.service.get_article_with_comments(article_id=1) - assert not isinstance(result, str) assert result.article_with_author.article.article_id == 1 assert result.article_with_author.author_name == "ArticleAuthor" assert len(result.nested_comments) == 1 @@ -369,8 +380,9 @@ def test_get_article_with_comments_success(self): def test_get_article_with_comments_article_not_found(self): self.mock_article_repo.get_by_id.return_value = None - result = self.service.get_article_with_comments(article_id=999) - assert result == "Article not found." + + with pytest.raises(ArticleNotFoundError, match="not found"): + self.service.get_article_with_comments(article_id=999) def test_get_article_with_comments_unknown_author(self): fake_article = create_test_article(article_id=1, article_author_id=999) @@ -378,7 +390,6 @@ def test_get_article_with_comments_unknown_author(self): self.mock_comment_repo.get_all_by_article_id.return_value = [] self.mock_account_repo.get_by_ids.return_value = [] result = self.service.get_article_with_comments(article_id=1) - assert not isinstance(result, str) assert result.article_with_author.author_name == "Unknown" self.mock_account_repo.get_by_ids.assert_called_once() @@ -459,9 +470,9 @@ def test_delete_article_unauthorized_does_not_delete_files(self): self.mock_article_repo.get_by_id.return_value = fake_article self.mock_account_repo.get_by_id.return_value = fake_other - result = self.service.delete_article(article_id=1, user_id=99) + with pytest.raises(OwnershipError, match="Only authors or admins"): + self.service.delete_article(article_id=1, user_id=99) - assert result == "Unauthorized : Only authors or admins can delete articles." self.mock_file_service.delete_file.assert_not_called() self.mock_article_repo.delete.assert_not_called() @@ -563,11 +574,6 @@ def test_update_article_no_file_service_does_not_crash(self): self.mock_article_repo.save.assert_called_once() def test_update_article_sets_article_edited_at(self): - """update_article() set article_edited_at=now(UTC) avant save. - - L'article doit avoir article_edited_at défini après mise à jour, - pas None comme initialement. - """ fake_article = create_test_article(article_id=1, article_author_id=1) fake_article.article_edited_at = None self.mock_article_repo.get_by_id.return_value = fake_article diff --git a/tests/tests_services/test_comment_service.py b/tests/tests_services/test_comment_service.py index a5285950..3feed7ef 100644 --- a/tests/tests_services/test_comment_service.py +++ b/tests/tests_services/test_comment_service.py @@ -1,5 +1,15 @@ from unittest.mock import MagicMock +import pytest + +from exceptions import ( + AccountNotFoundError, + ArticleNotFoundError, + CommentAuthorizationError, + CommentDeletedError, + CommentNotFoundError, + CommentValidationError, +) from src.application.domain.account import AccountRole from src.application.output_ports.account_repository import AccountRepository from src.application.output_ports.article_repository import ArticleRepository @@ -41,8 +51,7 @@ def test_create_comment_success(self): self.mock_account_repo.get_by_id.assert_called_once_with(fake_account.account_id) self.mock_article_repo.get_by_id.assert_called_once_with(fake_article.article_id) self.mock_comment_repo.save.assert_called_once() - index_first_arg = 0 - saved_comment = self.mock_comment_repo.save.call_args.args[index_first_arg] + saved_comment = self.mock_comment_repo.save.call_args.args[0] assert saved_comment.comment_article_id == fake_article.article_id assert saved_comment.comment_written_account_id == fake_account.account_id assert saved_comment.comment_reply_to is None @@ -52,16 +61,16 @@ def test_create_comment_success(self): def test_create_comment_account_not_found(self): self.mock_account_repo.get_by_id.return_value = None - result = self.service.create_comment( - article_id=1, - user_id=999, - content="This will not post." - ) + with pytest.raises(AccountNotFoundError, match="not found"): + self.service.create_comment( + article_id=1, + user_id=999, + content="This will not post." + ) self.mock_account_repo.get_by_id.assert_called_once_with(999) self.mock_article_repo.get_by_id.assert_not_called() self.mock_comment_repo.save.assert_not_called() - assert result == "Account not found." def test_create_comment_sanitizes_html(self): fake_account = create_test_account(account_role=AccountRole.USER) @@ -88,16 +97,16 @@ def test_create_comment_article_not_found(self): self.mock_account_repo.get_by_id.return_value = fake_account self.mock_article_repo.get_by_id.return_value = None - result = self.service.create_comment( - article_id=999, - user_id=fake_account.account_id, - content="Writing in the void." - ) + with pytest.raises(ArticleNotFoundError, match="not found"): + self.service.create_comment( + article_id=999, + user_id=fake_account.account_id, + content="Writing in the void." + ) self.mock_account_repo.get_by_id.assert_called_once_with(fake_account.account_id) self.mock_article_repo.get_by_id.assert_called_once_with(999) self.mock_comment_repo.save.assert_not_called() - assert result == "Article not found." class TestCreateReply(CommentServiceTestBase): @@ -123,8 +132,7 @@ def test_create_reply_success_root_comment(self): self.mock_account_repo.get_by_id.assert_called_once_with(fake_account.account_id) self.mock_comment_repo.get_by_id.assert_any_call(parent_comment.comment_id) self.mock_comment_repo.save.assert_called_once() - index_first_arg = 0 - saved_reply = self.mock_comment_repo.save.call_args.args[index_first_arg] + saved_reply = self.mock_comment_repo.save.call_args.args[0] assert saved_reply.comment_article_id == parent_comment.comment_article_id assert saved_reply.comment_written_account_id == fake_account.account_id assert saved_reply.comment_reply_to == parent_comment.comment_id @@ -159,8 +167,7 @@ def mock_get_by_id(cid): ) self.mock_comment_repo.save.assert_called_once() - index_first_arg = 0 - saved_reply = self.mock_comment_repo.save.call_args.args[index_first_arg] + saved_reply = self.mock_comment_repo.save.call_args.args[0] assert saved_reply.comment_reply_to == parent_comment.comment_id assert result is saved_reply @@ -169,15 +176,15 @@ def test_create_reply_parent_not_found(self): self.mock_account_repo.get_by_id.return_value = fake_account self.mock_comment_repo.get_by_id.return_value = None - result = self.service.create_reply( - parent_comment_id=999, - user_id=fake_account.account_id, - content="Replying to nothing" - ) + with pytest.raises(CommentNotFoundError, match="not found"): + self.service.create_reply( + parent_comment_id=999, + user_id=fake_account.account_id, + content="Replying to nothing" + ) self.mock_comment_repo.get_by_id.assert_called_once_with(999) self.mock_comment_repo.save.assert_not_called() - assert result == "Parent comment not found." def test_create_reply_to_deleted_comment_returns_error(self): self.mock_account_repo.get_by_id.return_value = create_test_account( @@ -191,10 +198,11 @@ def test_create_reply_to_deleted_comment_returns_error(self): ) self.mock_comment_repo.get_by_id.return_value = deleted - result = self.service.create_reply(5, 99, "Reply to deleted") + + with pytest.raises(CommentDeletedError, match="deleted"): + self.service.create_reply(5, 99, "Reply to deleted") + self.mock_comment_repo.save.assert_not_called() - assert isinstance(result, str) - assert "deleted" in result.lower() def test_create_reply_too_deep_returns_error(self): self.mock_account_repo.get_by_id.return_value = create_test_account( @@ -211,19 +219,21 @@ def mock_get_by_id(cid): self.mock_comment_repo.get_by_id.side_effect = mock_get_by_id - result = self.service.create_reply(4, 1, "Too deep reply") + with pytest.raises(CommentValidationError, match="maximum nesting depth"): + self.service.create_reply(4, 1, "Too deep reply") + self.mock_comment_repo.save.assert_not_called() - assert isinstance(result, str) - assert "maximum nesting depth" in result.lower() class TestGetComments(CommentServiceTestBase): def test_get_comments_for_article_not_found(self): self.mock_article_repo.get_by_id.return_value = None - result = self.service.get_comments_for_article(article_id=999) + + with pytest.raises(ArticleNotFoundError, match="not found"): + self.service.get_comments_for_article(article_id=999) + self.mock_article_repo.get_by_id.assert_called_once_with(999) self.mock_comment_repo.get_all_by_article_id.assert_not_called() - assert result == "Article not found." def test_get_comments_for_article_empty(self): fake_article = create_test_article(article_id=1, article_author_id=2) @@ -233,7 +243,6 @@ def test_get_comments_for_article_empty(self): comments = self.service.get_comments_for_article(article_id=fake_article.article_id) self.mock_article_repo.get_by_id.assert_called_once_with(fake_article.article_id) self.mock_comment_repo.get_all_by_article_id.assert_called_once_with(fake_article.article_id) - assert not isinstance(comments, str) assert comments == [] def test_get_comments_for_article_success(self): @@ -266,7 +275,6 @@ def test_get_comments_for_article_success(self): ] result = self.service.get_comments_for_article(article_id=fake_article.article_id) - assert not isinstance(result, str) root_node, = result reply_node, = root_node.replies assert root_node.comment.comment == root_comment @@ -285,7 +293,6 @@ def test_get_comments_for_article_ordering(self): reply_2 = create_test_comment(comment_id=4, comment_posted_at=datetime(2026, 1, 3), comment_reply_to=2) self.mock_comment_repo.get_all_by_article_id.return_value = [comment_1, comment_2, reply_1, reply_2] result = self.service.get_comments_for_article(article_id=1) - assert not isinstance(result, str) latest_root, oldest_root = result latest_reply, oldest_reply = result[0].replies assert latest_root.comment.comment.comment_id == comment_2.comment_id @@ -300,7 +307,6 @@ def test_get_comments_for_article_unknown_author(self): self.mock_comment_repo.get_all_by_article_id.return_value = [comment] self.mock_account_repo.get_by_ids.return_value = [] result = self.service.get_comments_for_article(article_id=1) - assert not isinstance(result, str) comment_node, = result assert comment_node.comment.author_name == "Anonymous" @@ -353,28 +359,34 @@ def test_delete_comment_unauthorized_not_author(self): self.mock_account_repo.get_by_id.return_value = fake_account comment = create_test_comment(comment_id=10, comment_written_account_id=1) self.mock_comment_repo.get_by_id.return_value = comment - result = self.service.delete_comment(comment_id=10, user_id=fake_account.account_id) + + with pytest.raises(CommentAuthorizationError, match="own comments"): + self.service.delete_comment(comment_id=10, user_id=fake_account.account_id) + self.mock_account_repo.get_by_id.assert_called_once_with(fake_account.account_id) self.mock_comment_repo.get_by_id.assert_called_once() self.mock_comment_repo.save.assert_not_called() - assert result == "Unauthorized: You can only delete your own comments." def test_delete_comment_not_found(self): fake_account = create_test_account(account_id=1, account_role=AccountRole.USER) self.mock_account_repo.get_by_id.return_value = fake_account self.mock_comment_repo.get_by_id.return_value = None - result = self.service.delete_comment(comment_id=999, user_id=fake_account.account_id) + + with pytest.raises(CommentNotFoundError, match="not found"): + self.service.delete_comment(comment_id=999, user_id=fake_account.account_id) + self.mock_comment_repo.get_by_id.assert_called_once_with(999) self.mock_comment_repo.save.assert_not_called() - assert result == "Comment not found." def test_delete_comment_account_not_found(self): self.mock_account_repo.get_by_id.return_value = None - result = self.service.delete_comment(comment_id=10, user_id=999) + + with pytest.raises(AccountNotFoundError, match="not found"): + self.service.delete_comment(comment_id=10, user_id=999) + self.mock_account_repo.get_by_id.assert_called_once_with(999) self.mock_comment_repo.get_by_id.assert_not_called() self.mock_comment_repo.save.assert_not_called() - assert result == "Account not found." def test_delete_comment_already_deleted_idempotent(self): fake_account = create_test_account(account_id=1, account_role=AccountRole.USER) @@ -417,9 +429,11 @@ def test_edit_comment_not_author(self): self.mock_account_repo.get_by_id.return_value = fake_account comment = create_test_comment(comment_id=10, comment_written_account_id=1) self.mock_comment_repo.get_by_id.return_value = comment - result = self.service.edit_comment(comment_id=10, user_id=2, content="Hack") + + with pytest.raises(CommentAuthorizationError, match="own comments"): + self.service.edit_comment(comment_id=10, user_id=2, content="Hack") + self.mock_comment_repo.save.assert_not_called() - assert result == "Unauthorized: You can only edit your own comments." def test_edit_comment_deleted(self): fake_account = create_test_account(account_id=1, account_role=AccountRole.USER) @@ -430,23 +444,29 @@ def test_edit_comment_deleted(self): is_deleted=True, ) self.mock_comment_repo.get_by_id.return_value = comment - result = self.service.edit_comment(comment_id=10, user_id=1, content="New") + + with pytest.raises(CommentDeletedError, match="deleted"): + self.service.edit_comment(comment_id=10, user_id=1, content="New") + self.mock_comment_repo.save.assert_not_called() - assert result == "Cannot edit a deleted comment." def test_edit_comment_not_found(self): fake_account = create_test_account(account_id=1) self.mock_account_repo.get_by_id.return_value = fake_account self.mock_comment_repo.get_by_id.return_value = None - result = self.service.edit_comment(comment_id=999, user_id=1, content="X") + + with pytest.raises(CommentNotFoundError, match="not found"): + self.service.edit_comment(comment_id=999, user_id=1, content="X") + self.mock_comment_repo.save.assert_not_called() - assert result == "Comment not found." def test_edit_comment_account_not_found(self): self.mock_account_repo.get_by_id.return_value = None - result = self.service.edit_comment(comment_id=10, user_id=999, content="X") + + with pytest.raises(AccountNotFoundError, match="not found"): + self.service.edit_comment(comment_id=10, user_id=999, content="X") + self.mock_comment_repo.save.assert_not_called() - assert result == "Account not found." class TestMaskCommentsByAccountId(CommentServiceTestBase): @@ -494,17 +514,21 @@ def test_hard_delete_comment_by_admin(self): def test_hard_delete_comment_not_admin(self): user = create_test_account(account_id=1, account_role=AccountRole.USER) self.mock_account_repo.get_by_id.return_value = user - result = self.service.hard_delete_comment(comment_id=10, user_id=1) + + with pytest.raises(CommentAuthorizationError, match="Only admins"): + self.service.hard_delete_comment(comment_id=10, user_id=1) + self.mock_comment_repo.delete.assert_not_called() - assert result == "Unauthorized: Only admins can permanently delete comments." def test_hard_delete_comment_not_found(self): admin = create_test_account(account_id=2, account_role=AccountRole.ADMIN) self.mock_account_repo.get_by_id.return_value = admin self.mock_comment_repo.get_by_id.return_value = None - result = self.service.hard_delete_comment(comment_id=999, user_id=2) + + with pytest.raises(CommentNotFoundError, match="not found"): + self.service.hard_delete_comment(comment_id=999, user_id=2) + self.mock_comment_repo.delete.assert_not_called() - assert result == "Comment not found." def test_hard_delete_comment_not_soft_deleted(self): admin = create_test_account(account_id=2, account_role=AccountRole.ADMIN) @@ -515,12 +539,16 @@ def test_hard_delete_comment_not_soft_deleted(self): is_deleted=False, ) self.mock_comment_repo.get_by_id.return_value = comment - result = self.service.hard_delete_comment(comment_id=10, user_id=2) + + with pytest.raises(CommentValidationError, match="not soft-deleted"): + self.service.hard_delete_comment(comment_id=10, user_id=2) + self.mock_comment_repo.delete.assert_not_called() - assert result == "Comment is not soft-deleted. Use soft-delete first." def test_hard_delete_comment_account_not_found(self): self.mock_account_repo.get_by_id.return_value = None - result = self.service.hard_delete_comment(comment_id=10, user_id=999) + + with pytest.raises(AccountNotFoundError, match="not found"): + self.service.hard_delete_comment(comment_id=10, user_id=999) + self.mock_comment_repo.delete.assert_not_called() - assert result == "Account not found." diff --git a/tests/tests_services/test_file_service.py b/tests/tests_services/test_file_service.py index 8f2fc154..3b8b8486 100644 --- a/tests/tests_services/test_file_service.py +++ b/tests/tests_services/test_file_service.py @@ -1,6 +1,6 @@ from unittest.mock import MagicMock -from src.application.application_exceptions import FileTooLargeError, FileTypeError +from exceptions import FileTooLargeError, FileTypeError from src.application.domain.file_record import FileRecord from src.application.output_ports.file_storage_repository import FileStorageRepository from src.application.services.file_service import FileService diff --git a/tests/tests_services/test_login_service.py b/tests/tests_services/test_login_service.py index 2796d522..3b94e4af 100644 --- a/tests/tests_services/test_login_service.py +++ b/tests/tests_services/test_login_service.py @@ -1,11 +1,19 @@ from unittest.mock import MagicMock +import pytest + +from exceptions import ( + AccountBannedError, + AccountNotFoundError, + AuthenticationError, + AuthorizationError, + EmailAlreadyTakenError, +) from src.application.domain.account import Account, AccountRole from src.application.output_ports.account_repository import AccountRepository from src.application.output_ports.account_session_repository import AccountSessionRepository from src.application.output_ports.password_hasher_repository import PasswordHasherRepository from src.application.services.login_service import LoginService -from tests.exceptions_tests import ExceptionTest from tests.test_domain_factories import create_test_account @@ -61,21 +69,23 @@ def test_authenticate_user_wrong_password(self): self.mock_repo.find_by_username.return_value = fake_account self.mock_hasher.verify.return_value = False - result = self.service.authenticate_user( - username=fake_account.account_username, - password="bad_password" - ) + with pytest.raises(AuthenticationError, match="Invalid username or password"): + self.service.authenticate_user( + username=fake_account.account_username, + password="bad_password" + ) self.mock_repo.find_by_username.assert_called_once_with(fake_account.account_username) self.mock_session_repo.save_account.assert_not_called() - assert result == "Invalid username or password." def test_authenticate_user_non_existent(self): self.mock_repo.find_by_username.return_value = None - result = self.service.authenticate_user(username="phantom", password="nothing") + + with pytest.raises(AuthenticationError, match="Invalid username or password"): + self.service.authenticate_user(username="phantom", password="nothing") + self.mock_repo.find_by_username.assert_called_once_with("phantom") self.mock_session_repo.save_account.assert_not_called() - assert result == "Invalid username or password." def test_get_current_account(self): fake_account = create_test_account() @@ -89,11 +99,12 @@ def test_terminate_session(self): self.mock_session_repo.clear.assert_called_once() def test_authenticate_user_session_repo_failure(self): - import pytest fake_account = create_test_account() self.mock_repo.find_by_username.return_value = fake_account - self.mock_session_repo.save_account.side_effect = ExceptionTest("Storage failure") - with pytest.raises(ExceptionTest, match="Storage failure"): + # Intentionally NOT in exceptions.py: test-only. Exception sufficient. + # Do not move to exceptions.py. + self.mock_session_repo.save_account.side_effect = Exception("Storage failure") + with pytest.raises(Exception, match="Storage failure"): self.service.authenticate_user("leia", "password123") def test_update_email_success(self): @@ -109,14 +120,18 @@ def test_update_email_taken_returns_error(self): other = create_test_account(account_id=2, account_email="taken@test.com") self.mock_session_repo.get_account.return_value = fake_account self.mock_repo.find_by_email.return_value = other - result = self.service.update_email("taken@test.com") - assert result == "This email is already taken." + + with pytest.raises(EmailAlreadyTakenError, match="already taken"): + self.service.update_email("taken@test.com") + self.mock_repo.update_email.assert_not_called() def test_update_email_unauthenticated_returns_error(self): self.mock_session_repo.get_account.return_value = None - result = self.service.update_email("new@test.com") - assert result == "You must be signed in to update your email." + + with pytest.raises(AuthenticationError, match="must be signed in"): + self.service.update_email("new@test.com") + self.mock_repo.update_email.assert_not_called() def test_update_password_success(self): @@ -130,16 +145,18 @@ def test_update_password_success(self): def test_update_password_unauthenticated_returns_error(self): self.mock_session_repo.get_account.return_value = None - result = self.service.update_password("new_secret") - assert result == "You must be signed in to update your password." + + with pytest.raises(AuthenticationError, match="must be signed in"): + self.service.update_password("new_secret") + self.mock_hasher.hash.assert_not_called() self.mock_repo.update_password.assert_not_called() - def test_update_password_empty_returns_error(self): + def test_update_password_empty_returns_none(self): fake_account = create_test_account(account_id=1) self.mock_session_repo.get_account.return_value = fake_account result = self.service.update_password("") - assert result == "Password is required." + assert result is None self.mock_hasher.hash.assert_not_called() self.mock_repo.update_password.assert_not_called() @@ -149,10 +166,9 @@ def test_delete_account_success(self): self.service.delete_account(fake_account.account_id) self.mock_repo.delete.assert_called_once_with(fake_account.account_id) - def test_delete_account_not_found_raises_value_error(self): + def test_delete_account_not_found_raises_account_not_found_error(self): self.mock_repo.get_by_id.return_value = None - import pytest - with pytest.raises(ValueError, match="not found"): + with pytest.raises(AccountNotFoundError, match="not found"): self.service.delete_account(999) self.mock_repo.delete.assert_not_called() @@ -172,22 +188,22 @@ def test_update_role_account_not_found(self): admin = create_test_account(account_id=1, account_role=AccountRole.ADMIN) self.mock_repo.get_by_id.side_effect = lambda cid: {1: admin}.get(cid) - result = self.service.update_account_role( - admin_id=1, target_id=999, new_role="author" - ) + with pytest.raises(AccountNotFoundError, match="not found"): + self.service.update_account_role( + admin_id=1, target_id=999, new_role="author" + ) - assert result == "Account not found." self.mock_repo.update_role.assert_not_called() def test_update_role_not_admin(self): user = create_test_account(account_id=1, account_role=AccountRole.USER) self.mock_repo.get_by_id.return_value = user - result = self.service.update_account_role( - admin_id=1, target_id=2, new_role="author" - ) + with pytest.raises(AuthorizationError, match="Unauthorized"): + self.service.update_account_role( + admin_id=1, target_id=2, new_role="author" + ) - assert result == "Unauthorized." self.mock_repo.update_role.assert_not_called() def test_update_role_target_is_admin(self): @@ -195,11 +211,11 @@ def test_update_role_target_is_admin(self): target_admin = create_test_account(account_id=2, account_role=AccountRole.ADMIN) self.mock_repo.get_by_id.side_effect = lambda cid: {1: admin, 2: target_admin}.get(cid) - result = self.service.update_account_role( - admin_id=1, target_id=2, new_role="user" - ) + with pytest.raises(AuthorizationError, match="Cannot change role of another admin"): + self.service.update_account_role( + admin_id=1, target_id=2, new_role="user" + ) - assert result == "Cannot change role of another admin." self.mock_repo.update_role.assert_not_called() def test_update_role_invalid_role(self): @@ -211,7 +227,7 @@ def test_update_role_invalid_role(self): admin_id=1, target_id=2, new_role="superadmin" ) - assert result == "Invalid role." + assert result is None self.mock_repo.update_role.assert_not_called() def test_ban_account_success(self): @@ -228,9 +244,9 @@ def test_ban_account_not_found(self): admin = create_test_account(account_id=1, account_role=AccountRole.ADMIN) self.mock_repo.get_by_id.side_effect = lambda cid: {1: admin}.get(cid) - result = self.service.ban_account(admin_id=1, target_account_id=999, ban_reason="Spam") + with pytest.raises(AccountNotFoundError, match="not found"): + self.service.ban_account(admin_id=1, target_account_id=999, ban_reason="Spam") - assert result == "Account not found." self.mock_repo.update_ban_status.assert_not_called() def test_ban_account_target_is_admin(self): @@ -238,9 +254,9 @@ def test_ban_account_target_is_admin(self): target_admin = create_test_account(account_id=2, account_role=AccountRole.ADMIN) self.mock_repo.get_by_id.side_effect = lambda cid: {1: admin, 2: target_admin}.get(cid) - result = self.service.ban_account(admin_id=1, target_account_id=2, ban_reason="Spam") + with pytest.raises(AuthorizationError, match="Cannot ban another admin"): + self.service.ban_account(admin_id=1, target_account_id=2, ban_reason="Spam") - assert result == "Cannot ban another admin." self.mock_repo.update_ban_status.assert_not_called() def test_unban_account_success(self): @@ -257,11 +273,11 @@ def test_authenticate_user_banned(self): fake_account = create_test_account(is_banned=True) self.mock_repo.find_by_username.return_value = fake_account - result = self.service.authenticate_user( - username=fake_account.account_username, - password=fake_account.account_password - ) + with pytest.raises(AccountBannedError, match="banned"): + self.service.authenticate_user( + username=fake_account.account_username, + password=fake_account.account_password + ) self.mock_repo.find_by_username.assert_called_once_with(fake_account.account_username) self.mock_session_repo.save_account.assert_not_called() - assert result == "This account has been banned." diff --git a/tests/tests_services/test_registration_service.py b/tests/tests_services/test_registration_service.py index 98a3ed7f..87abc7e6 100644 --- a/tests/tests_services/test_registration_service.py +++ b/tests/tests_services/test_registration_service.py @@ -1,5 +1,8 @@ from unittest.mock import MagicMock +import pytest + +from exceptions import EmailAlreadyTakenError, UsernameAlreadyTakenError from src.application.domain.account import Account, AccountRole from src.application.output_ports.account_repository import AccountRepository from src.application.output_ports.password_hasher_repository import PasswordHasherRepository @@ -46,16 +49,16 @@ def test_create_account_username_taken(self): self.mock_repo.find_by_username.return_value = existing_account - result = self.service.create_account( - username="leia", - password="password123", - email="new@galaxy.com" - ) + with pytest.raises(UsernameAlreadyTakenError, match="already taken"): + self.service.create_account( + username="leia", + password="password123", + email="new@galaxy.com" + ) self.mock_repo.find_by_username.assert_called_once_with("leia") self.mock_repo.find_by_email.assert_not_called() self.mock_repo.save.assert_not_called() - assert result == "This username is already taken." def test_create_account_email_taken(self): existing_account = create_test_account( @@ -67,13 +70,13 @@ def test_create_account_email_taken(self): self.mock_repo.find_by_username.return_value = None self.mock_repo.find_by_email.return_value = existing_account - result = self.service.create_account( - username="new_user", - password="password123", - email="leia@galaxy.com" - ) + with pytest.raises(EmailAlreadyTakenError, match="already taken"): + self.service.create_account( + username="new_user", + password="password123", + email="leia@galaxy.com" + ) self.mock_repo.find_by_username.assert_called_once_with("new_user") self.mock_repo.find_by_email.assert_called_once_with("leia@galaxy.com") self.mock_repo.save.assert_not_called() - assert result == "This email is already taken." diff --git a/utils/prosemirror_to_html.py b/utils/prosemirror_to_html.py index f3391934..0ce2d69a 100644 --- a/utils/prosemirror_to_html.py +++ b/utils/prosemirror_to_html.py @@ -8,6 +8,8 @@ def prosemirror_to_html(content_json: str | None, base_url: str = "") -> Markup: return Markup("") try: blocks = json.loads(content_json) + # Python builtin — safety net for json.loads on non-string input. + # Not in exceptions.py. Do not move it there. except (json.JSONDecodeError, TypeError): return Markup(f"

{escape(content_json)}

") if not isinstance(blocks, list): diff --git a/utils/template_helpers.py b/utils/template_helpers.py index f88fb052..e9748483 100644 --- a/utils/template_helpers.py +++ b/utils/template_helpers.py @@ -19,6 +19,15 @@ class ViteManifest: @classmethod def init(cls, static_dir: str | None) -> None: + """ + Initializes the manifest path from Flask's static directory. + + Raises: + RuntimeError: If Flask's static_folder is None. + """ + # Intentionally NOT in exceptions.py: startup-only crash path. + # Never caught by application code. Builtin RuntimeError sufficient. + # Do not move to exceptions.py. if static_dir is None: raise RuntimeError("Flask static_folder is None; cannot locate Vite manifest.") cls._manifest_path = os.path.join(static_dir, ".vite", "manifest.json")