Unit Test added

Author: MonsefRH

backend/models/scan_history.py

@@ -2,6 +2,8 @@
 import pytz
 from  utils.db import db 
 from sqlalchemy.dialects.postgresql import JSONB
+from models.user import User
+from models.selected_repo import SelectedRepo
 
 class ScanHistory(db.Model):
     __tablename__ = "scan_history"
@@ -17,5 +19,5 @@ class ScanHistory(db.Model):
     input_type = db.Column(db.String(50))
     scan_type = db.Column(db.String(50))
 
-    user = db.relationship("User", backref="scan_history")
-    repo = db.relationship("SelectedRepo", backref="scan_history")
+    user = db.relationship(User, backref="scan_history")
+    repo = db.relationship(SelectedRepo, backref="scan_history")

backend/models/user.py

@@ -9,6 +9,7 @@ class User(db.Model):
     email = db.Column(db.String(100), unique=True, nullable=False)
     password = db.Column(db.Text)
     role = db.Column(db.String(20), nullable=False, default="user")
+    email_notifications_enabled = db.Column(db.Boolean, default=True)
     created_at = db.Column(db.DateTime, default=datetime.now(pytz.UTC))
 
     __table_args__ = (

backend/schemas/admin_dto.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel
+from pydantic import BaseModel, field_validator
 from typing import List
 
 class UserStatsResponse(BaseModel):
@@ -9,7 +9,22 @@ class UserStatsResponse(BaseModel):
     created_at: str
     scan_count: int
 
+    @field_validator('scan_count')
+    @classmethod
+    def validate_scan_count(cls, v):
+        if v < 0:
+            raise ValueError('Scan count must be greater than 0')
+        return v
+
+
 class AdminStatsResponse(BaseModel):
     total_users: int
     total_scans: int
-    users: List[UserStatsResponse]
+    users: List[UserStatsResponse]
+
+    @field_validator('total_users', 'total_scans')
+    @classmethod
+    def validate_positive(cls, v):
+        if v < 0:
+            raise ValueError('Value must be greater than or equal to 0')
+        return v

backend/schemas/checkov_dto.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel
+from pydantic import BaseModel, Field, field_validator
 from typing import List, Optional
 
 class CheckovCheck(BaseModel):
@@ -28,15 +28,22 @@ class CheckovResult(BaseModel):
     message: Optional[str] = None
 
 class CheckovResponse(BaseModel):
-    scan_id: int
+    scan_id: int = Field(...,description="The ID of the checkov Response")
     results: CheckovResult
 
+    @field_validator("scan_id")
+    @classmethod
+    def validate(cls, v) -> int :
+        if v <= 0 :
+            raise ValueError("Checkov scan id must be greater than 0")
+        return v
+
 class CheckovContentRequest(BaseModel):
-    content: str
+    content: str = Field(...,min_length=1,description="The content of the checkov scanned file")
     framework: Optional[str] = "terraform"
 
 class CheckovRepoRequest(BaseModel):
-    repo_url: str
+    repo_url: str = Field(...,min_length=1, description="The URL of the  repository")
 
 class CheckovErrorResponse(BaseModel):
     error: str

backend/schemas/semgrep_dto.py

@@ -1,12 +1,12 @@
-from pydantic import BaseModel
-from typing import List, Dict, Any, Optional
+from pydantic import BaseModel, Field, field_validator
+from typing import List, Dict, Optional
 
 class SemgrepFinding(BaseModel):
     check_id: Optional[str]
     message: Optional[str]
     file_path: Optional[str]
     file_line_range: List[int]
-    severity: str
+    severity: str = Field(..., min_length=1)
     code: Optional[str]
     suggestion: Optional[str]
 
@@ -16,9 +16,16 @@ class SemgrepResult(BaseModel):
 
 class SemgrepResponse(BaseModel):
     scan_id: int
-    status: str
+    status: str  = Field(...,min_length=1)
     results: SemgrepResult
 
+    @field_validator('scan_id')
+    @classmethod
+    def validate(cls, v:int) -> int:
+        if v <= 0 :
+            raise ValueError("Scan ID must be greater than 0")
+        return v
+
 class SemgrepErrorResponse(BaseModel):
     error: str
     details: Optional[str] = None

backend/schemas/t5_dto.py

@@ -1,11 +1,11 @@
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 from typing import Optional
 
 class T5Request(BaseModel):
-    dockerfile: str
+    dockerfile: str = Field(...,min_length=1,description='Dockerfile')
 
 class T5Response(BaseModel):
-    scan_id: int
+    scan_id: int = Field(...,gt=0,description='scan id')
     correction: str
     explanation: str
 

backend/schemas/user_dto.py

@@ -1,26 +1,32 @@
-from pydantic import BaseModel
+from pydantic import BaseModel, Field, EmailStr, field_validator
 from typing import Optional
 
+
 class RegisterRequest(BaseModel):
-    name: str
-    email: str
-    password: str
+    name: str = Field(..., min_length=1, description="Full name, cannot be empty")
+    email: EmailStr = Field(..., description="Valid email address")
+    password: str = Field(..., min_length=5, description="Password at least 5 characters")
+    @field_validator('name')
+    @classmethod
+    def validate_name(cls, v: str) -> str:
+        if not v.strip():  # Extra check for whitespace-only
+            raise ValueError("Name cannot be empty or whitespace-only")
+        return v.strip()
 
 class VerifyCodeRequest(BaseModel):
-    email: str
-    code: str
+    email: EmailStr = Field(..., description="Valid email address")
+    code: str = Field(...,min_length=6,max_length=6, description="Valid code")
 
 class LoginRequest(BaseModel):
-    email: str
-    password: str
-
+    email: EmailStr = Field(..., description="Valid email address")
+    password: str = Field(..., min_length=5, description="Password at least 5 characters")
 class SetPasswordRequest(BaseModel):
-    password: str
+    password: str = Field(..., min_length=5, description="Password at least 5 characters")
 
 class UserResponse(BaseModel):
     id: int
     name: str
-    email: str
+    email: EmailStr = Field(..., description="Valid email address")
     role: str
 
 class RegisterResponse(BaseModel):

backend/services/checkov_service.py

@@ -9,7 +9,7 @@
 import json
 import logging
 import time
-from datetime import datetime
+from datetime import datetime, timezone
 
 import google.generativeai as genai
 from google.api_core import exceptions
@@ -449,7 +449,7 @@ def run_checkov_scan(user_id, input_type, content=None, file_path=None, repo_url
 
             result = run_checkov_on_single_file(temp_file_path)
             result.setdefault("meta", {})
-            result["meta"]["executed_at"] = datetime.utcnow().isoformat() + "Z"
+            result["meta"]["executed_at"] = datetime.now(timezone.utc).isoformat() + "Z"
 
             files_to_save = [(clean_path(temp_file_path), content)]
             scan_id = save_scan_history(user_id, result, input_type, files_to_save=files_to_save)
@@ -464,7 +464,7 @@ def run_checkov_scan(user_id, input_type, content=None, file_path=None, repo_url
 
             result = run_checkov_on_dir(file_path, is_file=True)
             result.setdefault("meta", {})
-            result["meta"]["executed_at"] = datetime.utcnow().isoformat() + "Z"
+            result["meta"]["executed_at"] = datetime.now(timezone.utc).isoformat() + "Z"
 
             files_to_save = [(clean_path(file_path), file_content)]
             scan_id = save_scan_history(user_id, result, input_type, files_to_save=files_to_save)
@@ -496,7 +496,7 @@ def run_checkov_scan(user_id, input_type, content=None, file_path=None, repo_url
 
             result = run_checkov_on_dir(temp_dir, is_file=False)
             result.setdefault("meta", {})
-            result["meta"]["executed_at"] = datetime.utcnow().isoformat() + "Z"
+            result["meta"]["executed_at"] = datetime.now(timezone.utc).isoformat() + "Z"
 
             scan_id = save_scan_history(user_id, result, input_type, files_to_save=files_found)
             return {"scan_id": scan_id, "results": result}
@@ -531,7 +531,7 @@ def remove_readonly(func, path, _):
 
             result = run_checkov_on_dir(temp_dir, is_file=False)
             result.setdefault("meta", {})
-            result["meta"]["executed_at"] = datetime.utcnow().isoformat() + "Z"
+            result["meta"]["executed_at"] = datetime.now(timezone.utc).isoformat() + "Z"
 
             scan_id = save_scan_history(user_id, result, input_type, repo_url=repo_url, files_to_save=files_found)
             return {"scan_id": scan_id, "results": result}

backend/services/t5_service.py

@@ -1,7 +1,7 @@
 import os
 import logging
 import torch
-from datetime import datetime
+from datetime import datetime ,UTC
 from transformers import AutoTokenizer, AutoModelForSeq2SeqLM
 import google.generativeai as genai
 from utils.db import db
@@ -87,7 +87,7 @@ def _send_finish_email_t5(to_email: str, user_name: str, executed_at: str):
             <p>Thank you for using <strong>SafeOps</strong>!</p>
           </div>
           <div class="footer">
-            &copy; {datetime.utcnow().year} SafeOps — Automated Security Platform
+            &copy; {datetime.now(UTC).year} SafeOps — Automated Security Platform
           </div>
         </div>
       </body>
@@ -148,7 +148,7 @@ def correct_dockerfile(user_id, dockerfile):
             explanation = f"[Erreur Gemini] Impossible de générer les explications: {str(e)}"
 
         # Save in DB
-        executed_at = datetime.utcnow().isoformat() + "Z"
+        executed_at = datetime.now(UTC).isoformat() + "Z"
         result = {
             "status": "success",
             "correction": fixed_code,

backend/services/user_service.py

@@ -5,6 +5,7 @@
 import logging
 from flask_bcrypt import Bcrypt
 from flask_jwt_extended import create_access_token, create_refresh_token
+from sqlalchemy.exc import SQLAlchemyError
 from utils.db import db
 from models.user import User
 from models.pending_user import PendingUser
@@ -48,18 +49,15 @@ def register_user(name, email, password):
 
     name = name.strip()
     hashed_password = bcrypt.generate_password_hash(password).decode("utf-8")
-
+    # Move duplicate checks OUTSIDE try to avoid catching
+    if User.query.filter((User.email == email) | (User.name == name)).first():
+        logger.error(f"User with email {email} or name {name} already exists")
+        raise ValueError("A user with this email or name already exists")
+
+    if PendingUser.query.filter_by(email=email).first():
+        logger.error(f"Verification request pending for email {email}")
+        raise ValueError("A verification request for this email is already pending")
     try:
-        # Check if email or name already exists in users
-        if User.query.filter((User.email == email) | (User.name == name)).first():
-            logger.error(f"User with email {email} or name {name} already exists")
-            raise ValueError("A user with this email or name already exists")
-
-        # Check if email is pending
-        if PendingUser.query.filter_by(email=email).first():
-            logger.error(f"Verification request pending for email {email}")
-            raise ValueError("A verification request for this email is already pending")
-
         # Store in pending_users
         verification_code = generate_verification_code()
         expires_at = datetime.now(timezone.utc) + CODE_EXPIRATION
@@ -86,31 +84,30 @@ def register_user(name, email, password):
     except Exception as e:
         logger.error(f"Registration failed for email {email}: {str(e)}")
         db.session.rollback()
-        raise RuntimeError("Registration failed")
+        raise
 
 def verify_code(email, code):
     """Verify the code and create a user account."""
     if not email or not code:
         logger.error("Missing email or verification code")
         raise ValueError("Email and verification code are required")
 
-    try:
-        pending_user = PendingUser.query.filter_by(email=email).first()
-        if not pending_user:
-            logger.error(f"No pending registration for email {email}")
-            raise ValueError("No pending registration found for this email")
-
-        expires_at_aware = pytz.utc.localize(pending_user.expires_at) if pending_user.expires_at.tzinfo is None else pending_user.expires_at
-        if datetime.now(timezone.utc) > expires_at_aware:
-            db.session.delete(pending_user)
-            db.session.commit()
-            logger.error(f"Verification code expired for email {email}")
-            raise ValueError("Verification code has expired")
-
-        if code != pending_user.verification_code:
-            logger.error(f"Invalid verification code for email {email}")
-            raise ValueError("Invalid verification code")
+    pending_user = PendingUser.query.filter_by(email=email).first()
+    if not pending_user:
+        logger.error(f"No pending registration for email {email}")
+        raise ValueError("No pending registration found for this email")
+    if code != pending_user.verification_code:
+        logger.error(f"Invalid verification code for email {email}")
+        raise ValueError("Invalid verification code")
+    expires_at_aware = pytz.utc.localize(
+        pending_user.expires_at) if pending_user.expires_at.tzinfo is None else pending_user.expires_at
+    if datetime.now(timezone.utc) > expires_at_aware:
+        db.session.delete(pending_user)
+        db.session.commit()
+        logger.error(f"Verification code expired for email {email}")
+        raise ValueError("Verification code has expired")
 
+    try:
         # Move to users
         user = User(
             name=pending_user.name,
@@ -189,7 +186,7 @@ def login_user(ip, email, password):
             logger.error(f"Invalid credentials for email {email}")
             raise ValueError("Invalid credentials")
 
-    except Exception as e:
+    except (SQLAlchemyError, OSError) as e:
         logger.error(f"Login failed for email {email}: {str(e)}")
         db.session.rollback()
         raise RuntimeError("Internal server error")
@@ -205,12 +202,11 @@ def set_password(user_id, password):
         raise ValueError("Password must be at least 5 characters long")
 
     hashed_password = bcrypt.generate_password_hash(password).decode("utf-8")
-
+    user = User.query.get(user_id)
+    if not user:
+        logger.error(f"User not found: {user_id}")
+        raise ValueError("User not found")
     try:
-        user = User.query.get(user_id)
-        if not user:
-            logger.error(f"User not found: {user_id}")
-            raise ValueError("User not found")
         user.password = hashed_password
         db.session.commit()
         logger.info(f"Password set successfully for user_id {user_id}")