29acdf3e01
- Implement test script for new SAG module endpoints BE-003 (Tag State Management) and BE-004 (Bulk Operations). - Create test cases for creating, updating, and bulk operations on cases and tags. - Add a test for module deactivation to ensure data integrity is maintained. - Include setup and teardown for tests to clear database state before and after each test.
275 lines
9.7 KiB
Python
275 lines
9.7 KiB
Python
import logging
|
|
from fastapi import APIRouter, HTTPException, Query, Request, Form
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from pathlib import Path
|
|
from datetime import date
|
|
from app.core.database import execute_query
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter()
|
|
|
|
# Setup template directory - must be root "app" to allow extending shared/frontend/base.html
|
|
templates = Jinja2Templates(directory="app")
|
|
|
|
|
|
def build_location_tree(items: list) -> list:
|
|
"""Helper to build recursive location tree"""
|
|
nodes = {}
|
|
roots = []
|
|
|
|
for loc in items or []:
|
|
if not isinstance(loc, dict):
|
|
continue
|
|
loc_id = loc.get("id")
|
|
if loc_id is None:
|
|
continue
|
|
# Ensure we have the fields we need
|
|
nodes[loc_id] = {
|
|
"id": loc_id,
|
|
"name": loc.get("name"),
|
|
"location_type": loc.get("location_type"),
|
|
"parent_location_id": loc.get("parent_location_id"),
|
|
"children": []
|
|
}
|
|
|
|
for node in nodes.values():
|
|
parent_id = node.get("parent_location_id")
|
|
if parent_id and parent_id in nodes:
|
|
nodes[parent_id]["children"].append(node)
|
|
else:
|
|
roots.append(node)
|
|
|
|
def sort_nodes(node_list: list) -> None:
|
|
node_list.sort(key=lambda n: (n.get("name") or "").lower())
|
|
for n in node_list:
|
|
if n.get("children"):
|
|
sort_nodes(n["children"])
|
|
|
|
sort_nodes(roots)
|
|
return roots
|
|
|
|
|
|
@router.get("/hardware", response_class=HTMLResponse)
|
|
async def hardware_list(
|
|
request: Request,
|
|
status: str = Query(None),
|
|
asset_type: str = Query(None),
|
|
customer_id: int = Query(None),
|
|
q: str = Query(None)
|
|
):
|
|
"""Display list of all hardware."""
|
|
query = "SELECT * FROM hardware_assets WHERE deleted_at IS NULL"
|
|
params = []
|
|
|
|
if status:
|
|
query += " AND status = %s"
|
|
params.append(status)
|
|
if asset_type:
|
|
query += " AND asset_type = %s"
|
|
params.append(asset_type)
|
|
if customer_id:
|
|
query += " AND current_owner_customer_id = %s"
|
|
params.append(customer_id)
|
|
if q:
|
|
query += " AND (serial_number ILIKE %s OR model ILIKE %s OR brand ILIKE %s)"
|
|
search_param = f"%{q}%"
|
|
params.extend([search_param, search_param, search_param])
|
|
|
|
query += " ORDER BY created_at DESC"
|
|
hardware = execute_query(query, tuple(params))
|
|
|
|
# Get customer names for display
|
|
if hardware:
|
|
customer_ids = [h['current_owner_customer_id'] for h in hardware if h.get('current_owner_customer_id')]
|
|
if customer_ids:
|
|
customer_query = "SELECT id, navn FROM customers WHERE id = ANY(%s)"
|
|
customers = execute_query(customer_query, (customer_ids,))
|
|
customer_map = {c['id']: c['navn'] for c in customers} if customers else {}
|
|
|
|
# Add customer names to hardware
|
|
for h in hardware:
|
|
if h.get('current_owner_customer_id'):
|
|
h['customer_name'] = customer_map.get(h['current_owner_customer_id'], 'Unknown')
|
|
|
|
return templates.TemplateResponse("modules/hardware/templates/index.html", {
|
|
"request": request,
|
|
"hardware": hardware,
|
|
"current_status": status,
|
|
"current_asset_type": asset_type,
|
|
"search_query": q
|
|
})
|
|
|
|
|
|
@router.get("/hardware/new", response_class=HTMLResponse)
|
|
async def create_hardware_form(request: Request):
|
|
"""Display create hardware form."""
|
|
# Get customers for dropdown
|
|
customers = execute_query("SELECT id, navn FROM customers WHERE deleted_at IS NULL ORDER BY navn")
|
|
|
|
return templates.TemplateResponse("modules/hardware/templates/create.html", {
|
|
"request": request,
|
|
"customers": customers or []
|
|
})
|
|
|
|
|
|
@router.get("/hardware/{hardware_id}", response_class=HTMLResponse)
|
|
async def hardware_detail(request: Request, hardware_id: int):
|
|
"""Display hardware details."""
|
|
# Get hardware
|
|
query = "SELECT * FROM hardware_assets WHERE id = %s AND deleted_at IS NULL"
|
|
result = execute_query(query, (hardware_id,))
|
|
if not result:
|
|
raise HTTPException(status_code=404, detail="Hardware not found")
|
|
|
|
hardware = result[0]
|
|
|
|
# Get customer name if applicable
|
|
if hardware.get('current_owner_customer_id'):
|
|
customer_query = "SELECT navn FROM customers WHERE id = %s"
|
|
customer_result = execute_query(customer_query, (hardware['current_owner_customer_id'],))
|
|
if customer_result:
|
|
hardware['customer_name'] = customer_result[0]['navn']
|
|
|
|
# Get ownership history
|
|
ownership_query = """
|
|
SELECT * FROM hardware_ownership_history
|
|
WHERE hardware_id = %s AND deleted_at IS NULL
|
|
ORDER BY start_date DESC
|
|
"""
|
|
ownership = execute_query(ownership_query, (hardware_id,))
|
|
|
|
# Get customer names for ownership history
|
|
if ownership:
|
|
customer_ids = [o['owner_customer_id'] for o in ownership if o.get('owner_customer_id')]
|
|
if customer_ids:
|
|
customer_query = "SELECT id, navn FROM customers WHERE id = ANY(%s)"
|
|
customers = execute_query(customer_query, (customer_ids,))
|
|
customer_map = {c['id']: c['navn'] for c in customers} if customers else {}
|
|
|
|
for o in ownership:
|
|
if o.get('owner_customer_id'):
|
|
o['customer_name'] = customer_map.get(o['owner_customer_id'], 'Unknown')
|
|
|
|
# Get location history
|
|
location_query = """
|
|
SELECT * FROM hardware_location_history
|
|
WHERE hardware_id = %s AND deleted_at IS NULL
|
|
ORDER BY start_date DESC
|
|
"""
|
|
locations = execute_query(location_query, (hardware_id,))
|
|
|
|
# Get attachments
|
|
attachment_query = """
|
|
SELECT * FROM hardware_attachments
|
|
WHERE hardware_id = %s AND deleted_at IS NULL
|
|
ORDER BY uploaded_at DESC
|
|
"""
|
|
attachments = execute_query(attachment_query, (hardware_id,))
|
|
|
|
# Get related cases
|
|
case_query = """
|
|
SELECT hcr.*, s.titel, s.status, s.customer_id
|
|
FROM hardware_case_relations hcr
|
|
LEFT JOIN sag_sager s ON hcr.case_id = s.id
|
|
WHERE hcr.hardware_id = %s AND hcr.deleted_at IS NULL AND s.deleted_at IS NULL
|
|
ORDER BY hcr.created_at DESC
|
|
"""
|
|
cases = execute_query(case_query, (hardware_id,))
|
|
|
|
# Get tags
|
|
tag_query = """
|
|
SELECT * FROM hardware_tags
|
|
WHERE hardware_id = %s AND deleted_at IS NULL
|
|
ORDER BY created_at DESC
|
|
"""
|
|
tags = execute_query(tag_query, (hardware_id,))
|
|
|
|
# Get all active locations for the tree (including parent_id for structure)
|
|
all_locations_query = """
|
|
SELECT id, name, location_type, parent_location_id
|
|
FROM locations_locations
|
|
WHERE deleted_at IS NULL
|
|
ORDER BY name
|
|
"""
|
|
all_locations_flat = execute_query(all_locations_query)
|
|
location_tree = build_location_tree(all_locations_flat)
|
|
|
|
return templates.TemplateResponse("modules/hardware/templates/detail.html", {
|
|
"request": request,
|
|
"hardware": hardware,
|
|
"ownership": ownership or [],
|
|
"locations": locations or [],
|
|
"attachments": attachments or [],
|
|
"cases": cases or [],
|
|
"tags": tags or [],
|
|
"location_tree": location_tree or []
|
|
})
|
|
|
|
|
|
@router.get("/hardware/{hardware_id}/edit", response_class=HTMLResponse)
|
|
async def edit_hardware_form(request: Request, hardware_id: int):
|
|
"""Display edit hardware form."""
|
|
# Get hardware
|
|
query = "SELECT * FROM hardware_assets WHERE id = %s AND deleted_at IS NULL"
|
|
result = execute_query(query, (hardware_id,))
|
|
if not result:
|
|
raise HTTPException(status_code=404, detail="Hardware not found")
|
|
|
|
hardware = result[0]
|
|
|
|
# Get customers for dropdown
|
|
customers = execute_query("SELECT id, navn FROM customers WHERE deleted_at IS NULL ORDER BY navn")
|
|
|
|
return templates.TemplateResponse("modules/hardware/templates/edit.html", {
|
|
"request": request,
|
|
"hardware": hardware,
|
|
"customers": customers or []
|
|
})
|
|
|
|
|
|
@router.post("/hardware/{hardware_id}/location")
|
|
async def update_hardware_location(
|
|
request: Request,
|
|
hardware_id: int,
|
|
location_id: int = Form(...),
|
|
notes: str = Form(None)
|
|
):
|
|
"""Update hardware location."""
|
|
# Verify hardware exists
|
|
check_query = "SELECT id FROM hardware_assets WHERE id = %s AND deleted_at IS NULL"
|
|
if not execute_query(check_query, (hardware_id,)):
|
|
raise HTTPException(status_code=404, detail="Hardware not found")
|
|
|
|
# Verify location exists
|
|
loc_check = "SELECT name FROM locations_locations WHERE id = %s"
|
|
loc_result = execute_query(loc_check, (location_id,))
|
|
if not loc_result:
|
|
raise HTTPException(status_code=404, detail="Location not found")
|
|
location_name = loc_result[0]['name']
|
|
|
|
# 1. Close current location history
|
|
close_history_query = """
|
|
UPDATE hardware_location_history
|
|
SET end_date = %s
|
|
WHERE hardware_id = %s AND end_date IS NULL
|
|
"""
|
|
execute_query(close_history_query, (date.today(), hardware_id))
|
|
|
|
# 2. Add new location history
|
|
add_history_query = """
|
|
INSERT INTO hardware_location_history (hardware_id, location_id, location_name, start_date, notes)
|
|
VALUES (%s, %s, %s, %s, %s)
|
|
"""
|
|
execute_query(add_history_query, (hardware_id, location_id, location_name, date.today(), notes))
|
|
|
|
# 3. Update current location on asset
|
|
update_asset_query = """
|
|
UPDATE hardware_assets
|
|
SET current_location_id = %s, updated_at = NOW()
|
|
WHERE id = %s
|
|
"""
|
|
execute_query(update_asset_query, (location_id, hardware_id))
|
|
|
|
return RedirectResponse(url=f"/hardware/{hardware_id}", status_code=303)
|