d81a8f41b4
- Implemented a comprehensive end-to-end testing script for the Sag module's HTTP API, covering various functionalities including case creation, updates, and file uploads. - Introduced a safe HTML sanitizer utility to ensure safe rendering of HTML content in the BMC Hub UI. - Added database migrations for new features including WAN connection marking for wall outlets, permanent audit trails for supplier invoices, and dedicated permissions for the Sag module. - Created a migration center for manual subscription and invoice migrations with relevant tables and indexes. - Added tests for migration center functionalities, ensuring stability and correctness of the new features.
119 lines
4.4 KiB
Python
119 lines
4.4 KiB
Python
import logging
|
|
from fastapi import APIRouter, HTTPException, Request, Depends
|
|
from typing import Optional
|
|
|
|
from app.core.database import execute_query
|
|
from app.models.schemas import Solution, SolutionCreate, SolutionUpdate
|
|
from app.core.auth_dependencies import require_any_permission
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter()
|
|
case_edit_access = require_any_permission("cases.edit", "tickets.edit")
|
|
|
|
@router.get("/sag/{sag_id}/solution", response_model=Optional[Solution])
|
|
async def get_solution(sag_id: int):
|
|
"""Get the solution associated with a case."""
|
|
try:
|
|
query = "SELECT * FROM sag_solutions WHERE sag_id = %s"
|
|
result = execute_query(query, (sag_id,))
|
|
if not result:
|
|
return None
|
|
return result[0]
|
|
except Exception as e:
|
|
logger.error("❌ Error getting solution for case %s: %s", sag_id, e)
|
|
raise HTTPException(status_code=500, detail="Failed to get solution")
|
|
|
|
@router.post(
|
|
"/sag/{sag_id}/solution",
|
|
response_model=Solution,
|
|
dependencies=[Depends(case_edit_access)],
|
|
)
|
|
async def create_solution(sag_id: int, solution: SolutionCreate, request: Request):
|
|
"""Create a solution for a case."""
|
|
try:
|
|
# Check if case exists
|
|
case_check = execute_query("SELECT id FROM sag_sager WHERE id = %s", (sag_id,))
|
|
if not case_check:
|
|
raise HTTPException(status_code=404, detail="Case not found")
|
|
|
|
# Check if solution already exists
|
|
check = execute_query("SELECT id FROM sag_solutions WHERE sag_id = %s", (sag_id,))
|
|
if check:
|
|
raise HTTPException(status_code=400, detail="Solution already exists for this case")
|
|
|
|
query = """
|
|
INSERT INTO sag_solutions
|
|
(sag_id, title, description, solution_type, result, created_by_user_id)
|
|
VALUES (%s, %s, %s, %s, %s, %s)
|
|
RETURNING *
|
|
"""
|
|
params = (
|
|
sag_id,
|
|
solution.title,
|
|
solution.description,
|
|
solution.solution_type,
|
|
solution.result,
|
|
getattr(request.state, "user_id", None)
|
|
)
|
|
|
|
result = execute_query(query, params)
|
|
if result:
|
|
logger.info("✅ Solution created for case: %s", sag_id)
|
|
return result[0]
|
|
raise HTTPException(status_code=500, detail="Failed to create solution")
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error("❌ Error creating solution: %s", e)
|
|
raise HTTPException(status_code=500, detail="Failed to create solution")
|
|
|
|
@router.patch(
|
|
"/sag/{sag_id}/solution",
|
|
response_model=Solution,
|
|
dependencies=[Depends(case_edit_access)],
|
|
)
|
|
async def update_solution(sag_id: int, updates: SolutionUpdate):
|
|
"""Update a solution."""
|
|
try:
|
|
# Check if solution exists
|
|
check = execute_query("SELECT id FROM sag_solutions WHERE sag_id = %s", (sag_id,))
|
|
if not check:
|
|
raise HTTPException(status_code=404, detail="Solution not found")
|
|
|
|
# Build dynamic update query
|
|
set_clauses = []
|
|
params = []
|
|
|
|
# Helper to check and add params
|
|
if updates.title is not None:
|
|
set_clauses.append("title = %s")
|
|
params.append(updates.title)
|
|
if updates.description is not None:
|
|
set_clauses.append("description = %s")
|
|
params.append(updates.description)
|
|
if updates.solution_type is not None:
|
|
set_clauses.append("solution_type = %s")
|
|
params.append(updates.solution_type)
|
|
if updates.result is not None:
|
|
set_clauses.append("result = %s")
|
|
params.append(updates.result)
|
|
|
|
if not set_clauses:
|
|
raise HTTPException(status_code=400, detail="No fields to update")
|
|
|
|
set_clauses.append("updated_at = NOW()")
|
|
|
|
params.append(sag_id)
|
|
query = f"UPDATE sag_solutions SET {', '.join(set_clauses)} WHERE sag_id = %s RETURNING *"
|
|
|
|
result = execute_query(query, tuple(params))
|
|
if result:
|
|
logger.info("✅ Solution updated for case: %s", sag_id)
|
|
return result[0]
|
|
raise HTTPException(status_code=500, detail="Failed to update solution")
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error("❌ Error updating solution: %s", e)
|
|
raise HTTPException(status_code=500, detail="Failed to update solution")
|