unit test completed

Author: MonsefRH

backend/schemas/dashboard_dto.py

@@ -1,9 +1,18 @@
-from pydantic import BaseModel
+from pydantic import BaseModel, field_validator, ValidationError
+
 
 class DashboardStatsResponse(BaseModel):
     policies: int
     alerts: int
     security_score: int
 
+    @field_validator('policies','alerts','security_score')
+    @classmethod
+    def ensure_not_negative(cls, v):
+        if v < 0:
+            raise ValueError(f'Policies value {v} is negative')
+        return v
+
+
 class DashboardErrorResponse(BaseModel):
     error: str

backend/schemas/full_scan_dto.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 from typing import Dict, Optional, Any
 
 class T5Correction(BaseModel):
@@ -7,13 +7,13 @@ class T5Correction(BaseModel):
     error: Optional[str] = None
 
 class FullScanResponse(BaseModel):
-    checkov: Dict[str, Any]  # Includes scan_id and results or error
+    checkov: Dict[str, Any]
     t5: Dict[str, T5Correction]
-    checkov_corrected: Dict[str, Any]  # Includes scan_id and results or error
-    semgrep: Dict[str, Any]  # SemgrepScanResponse or error
+    checkov_corrected: Dict[str, Any]
+    semgrep: Dict[str, Any]
 
 class FullScanRequest(BaseModel):
-    repo_url: str
+    repo_url: str=Field(...,min_length=20,description="The full repository URL")
 
 class FullScanErrorResponse(BaseModel):
     error: str

backend/schemas/history_dto.py

@@ -1,18 +1,24 @@
-from pydantic import BaseModel
+from pydantic import BaseModel, field_validator
 from typing import Optional, Dict, Any
 
 class ScanHistoryResponse(BaseModel):
     id: int
-    repo_id: Optional[int]
+    repo_id: Optional[int] = None
     scan_result: Dict[str, Any]
-    repo_url: Optional[str]
-    item_name: Optional[str]
-    status: Optional[str]
-    score: Optional[int]
-    compliant: Optional[bool]
+    repo_url: Optional[str] = None
+    item_name: Optional[str] = None
+    status: Optional[str] = None
+    score: Optional[int] = None
+    compliant: Optional[bool] = None
     created_at: str
-    input_type: Optional[str]
-    scan_type: Optional[str]
+    input_type: Optional[str] = None
+    scan_type: Optional[str] = None
+
+    @field_validator('id',mode='before')
+    def validate_id(cls, v):
+        if v < 0:
+            raise ValueError("Id must be positive")
+        return v
 
 class HistoryErrorResponse(BaseModel):
     error: str

backend/services/full_scan_service.py

@@ -136,8 +136,6 @@ def run_full_scan(user_id, repo_url, jwt_token):
 
             # Step 4: Semgrep scan
             try:
-                # ⚠️ on utilise la version de run_semgrep fournie par ton service.
-                # Elle gère déjà la sauvegarde + notifs + email CSV (selon ta version actuelle).
                 semgrep_result = run_semgrep(
                     user_id=user_id,
                     input_type="repo",

backend/services/github_service.py

@@ -121,12 +121,12 @@ def github_callback(code):
 
 def get_github_repos(user_id):
     """Fetch GitHub repositories for a user."""
+    github_user = GithubUser.query.filter_by(user_id=user_id).first()
+    if not github_user:
+        logger.error(f"No GitHub account linked for user_id {user_id}")
+        raise ValueError("Compte GitHub non lié")
+    access_token = github_user.access_token
     try:
-        github_user = GithubUser.query.filter_by(user_id=user_id).first()
-        if not github_user:
-            logger.error(f"No GitHub account linked for user_id {user_id}")
-            raise ValueError("Compte GitHub non lié")
-        access_token = github_user.access_token
 
         headers = {"Authorization": f"Bearer {access_token}"}
         response = requests.get("https://api.github.com/user/repos", headers=headers)
@@ -234,11 +234,12 @@ def save_selected_repos(user_id, selected_repos):
 
 def get_repo_configs(user_id):
     """Fetch and store configuration files from selected repositories."""
+    github_user = GithubUser.query.filter_by(user_id=user_id).first()
+    if not github_user:
+        logger.error(f"No GitHub account linked for user_id {user_id}")
+        raise ValueError("Compte GitHub non lié")
     try:
-        github_user = GithubUser.query.filter_by(user_id=user_id).first()
-        if not github_user:
-            logger.error(f"No GitHub account linked for user_id {user_id}")
-            raise ValueError("Compte GitHub non lié")
+
         access_token = github_user.access_token
 
         repos = SelectedRepo.query.filter_by(user_id=user_id).all()

backend/services/notification_service.py

@@ -9,7 +9,6 @@
 from utils.db import db
 from models.user import User
 from models.notification import Notification
-from models.user_preference import UserPreference
 
 # On réutilise tes identifiants SMTP déjà définis dans services/email_template.py
 from services.email_template import EMAIL_ADDRESS, EMAIL_PASSWORD

backend/services/report_service.py

@@ -2,7 +2,7 @@
 import os
 import csv
 import tempfile
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import List, Dict, Any, Tuple
 import smtplib
 from email.message import EmailMessage
@@ -138,7 +138,7 @@ def generate_csv_for_scan(scan_id: int) -> Tuple[str, str]:
         })
 
     repo_part = (scan.repo_url or "local").rstrip("/").split("/")[-1]
-    ts = datetime.utcnow().strftime("%Y%m%d-%H%M%S")
+    ts = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
     filename = f"safeops_report_{scan.scan_type}_{repo_part}_{scan.id}_{ts}.csv"
     file_path = os.path.join(tempfile.gettempdir(), filename)
 
@@ -225,7 +225,7 @@ def send_csv_report_email(
             <p>Thank you for using <strong>SafeOps+</strong>!</p>
           </div>
           <div class="footer">
-            &copy; {datetime.utcnow().year} SafeOps+ — Automated Security Platform
+            &copy; {datetime.now(timezone.utc).year} SafeOps+ — Automated Security Platform
           </div>
         </div>
       </body>

backend/services/semgrep_service.py

@@ -1,4 +1,3 @@
-# services/semgrep_service.py
 from pathlib import Path
 import os
 import tempfile
@@ -19,9 +18,7 @@
 # Email + CSV
 from services.report_service import generate_csv_for_scan, send_csv_report_email
 
-# -------------------------------------------------------------------
-# Logging / Config
-# -------------------------------------------------------------------
+
 logging.basicConfig(
     level=logging.DEBUG,
     format='%(asctime)s %(levelname)s:%(name)s: %(message)s',
@@ -36,9 +33,6 @@
 genai.configure(api_key=GEMINI_API_KEY)
 
 
-# -------------------------------------------------------------------
-# Helpers
-# -------------------------------------------------------------------
 def clean_path(full_path: str) -> str:
     """Remove temp-dir prefixes from file paths for nicer display."""
     try:
@@ -73,9 +67,7 @@ def get_gemini_suggestion(finding: dict, max_retries: int = 3, delay: int = 2) -
     return "Review this finding and apply best practices."
 
 
-# -------------------------------------------------------------------
-# Semgrep execution
-# -------------------------------------------------------------------
+
 def run_semgrep(target_path: str) -> dict:
     """Run Semgrep and return normalized results (failed list + summary)."""
     cmd = ["semgrep", "scan", target_path, "--config=auto", "--json"]
@@ -133,9 +125,7 @@ def run_semgrep(target_path: str) -> dict:
         raise RuntimeError(f"Semgrep failed: {str(e)}")
 
 
-# -------------------------------------------------------------------
-# Persistence + Email (one HTML email with CSV; English; no emojis; no scan id)
-# -------------------------------------------------------------------
+
 def save_scan_history(user_id, result, input_type, repo_url=None):
     """Persist Semgrep results and send one HTML email (with CSV) — English, no emojis, no scan id in subject."""
     try:
@@ -179,10 +169,10 @@ def save_scan_history(user_id, result, input_type, repo_url=None):
                 send_csv_report_email(
                     to_email=user.email,
                     subject=subject,
-                    body_text=body,      # HTML-friendly; wrapped in blue template
+                    body_text=body,
                     csv_path=csv_path,
                     csv_filename=csv_filename,
-                    user_name=user.name,  # greet by name
+                    user_name=user.name,
                 )
         except Exception as e:
             logger.warning(f"Combined email (finish + CSV) failed for Semgrep scan_id {scan_id}: {e}")
@@ -194,10 +184,6 @@ def save_scan_history(user_id, result, input_type, repo_url=None):
         db.session.rollback()
         raise RuntimeError(f"Failed to save scan: {str(e)}")
 
-
-# -------------------------------------------------------------------
-# Entry point used by route
-# -------------------------------------------------------------------
 def validate_semgrep(user_id, input_type, file=None, repo_url=None, content=None, extension="py"):
     """
     Validate inputs and run Semgrep. Returns {"scan_id": id, **result}.

tests/conftest.py

@@ -11,6 +11,13 @@
 from fixtures.admin_fixtures import *
 from fixtures.checkov_fixtures import *
 from fixtures.semgrep_fixtures import *
+from fixtures.dashboard_fixtures import *
+from fixtures.fullscan_fixtures import *
+from fixtures.history_fixtures import *
+from fixtures.notification_fixtures import *
+from fixtures.report_fixtures import *
+from fixtures.github_fixtures import *
+from fixtures.google_fixtures import *