feat(templates): add pagination with total counts to catalog and campaign picker
Aegis CI / lint-and-test (push) Has been cancelled
Snyk Security Scan / Python vulnerabilities (backend) (push) Has been cancelled
Snyk Security Scan / npm vulnerabilities (frontend) (push) Has been cancelled
Snyk Security Scan / Docker image vulnerabilities (backend) (push) Has been cancelled
Aegis CI / lint-and-test (push) Has been cancelled
Snyk Security Scan / Python vulnerabilities (backend) (push) Has been cancelled
Snyk Security Scan / npm vulnerabilities (frontend) (push) Has been cancelled
Snyk Security Scan / Docker image vulnerabilities (backend) (push) Has been cancelled
Test Catalog's pagination showed only 'Page N' with no way to know how many pages existed, and the campaign 'Add Test' modal's template picker had no pagination at all — with 2800+ templates in the catalog, its flat 100-row fetch made most templates unreachable unless a search happened to match. Adds GET /test-templates/count (open to any authenticated user, mirrors the list endpoint's filters) and wires real 'Page N of M' + total-available counts into both.
This commit is contained in:
@@ -58,6 +58,7 @@ from app.services.audit_service import log_action
|
||||
# Import from app.services.test_template_service
|
||||
from app.services.test_template_service import (
|
||||
bulk_activate,
|
||||
count_templates,
|
||||
get_template_or_raise,
|
||||
get_template_stats,
|
||||
list_templates,
|
||||
@@ -156,6 +157,35 @@ def _list_templates_handler(
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GET /test-templates/count — total matching a filter, for pagination
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@router.get("/count")
|
||||
def _count_templates_handler(
|
||||
source: Optional[str] = Query(None),
|
||||
platform: Optional[str] = Query(None),
|
||||
severity: Optional[str] = Query(None),
|
||||
mitre_technique_id: Optional[str] = Query(None),
|
||||
search: Optional[str] = Query(None),
|
||||
is_active: Optional[bool] = Query(None),
|
||||
db: Session = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user),
|
||||
) -> dict:
|
||||
"""Return the total count of templates matching the same filters as the
|
||||
list endpoint, so the catalog UI can show a true page count."""
|
||||
return {"total": count_templates(
|
||||
db,
|
||||
source=source,
|
||||
platform=platform,
|
||||
severity=severity,
|
||||
mitre_technique_id=mitre_technique_id,
|
||||
search=search,
|
||||
is_active=is_active,
|
||||
)}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GET /test-templates/stats — catalog statistics (admin)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -23,6 +23,42 @@ from app.models.test import Test
|
||||
from app.utils import escape_like
|
||||
|
||||
|
||||
def _build_template_query(
|
||||
db: Session,
|
||||
*,
|
||||
source: str | None = None,
|
||||
platform: str | None = None,
|
||||
severity: str | None = None,
|
||||
mitre_technique_id: str | None = None,
|
||||
search: str | None = None,
|
||||
is_active: bool | None = None,
|
||||
):
|
||||
"""Build the filtered TestTemplate query shared by list_templates() and
|
||||
count_templates() — a single source of truth so a paginated page of
|
||||
results and the total count it's paginated against can never drift
|
||||
apart (same pattern used for tests in test_crud_service.py)."""
|
||||
query = db.query(TestTemplate)
|
||||
if is_active is not None:
|
||||
query = query.filter(TestTemplate.is_active == is_active)
|
||||
if source:
|
||||
query = query.filter(TestTemplate.source == source)
|
||||
if platform:
|
||||
query = query.filter(TestTemplate.platform.ilike(f"%{escape_like(platform)}%"))
|
||||
if severity:
|
||||
query = query.filter(TestTemplate.severity == severity)
|
||||
if mitre_technique_id:
|
||||
query = query.filter(TestTemplate.mitre_technique_id == mitre_technique_id)
|
||||
if search:
|
||||
pattern = f"%{escape_like(search)}%"
|
||||
query = query.filter(
|
||||
or_(
|
||||
TestTemplate.name.ilike(pattern),
|
||||
TestTemplate.description.ilike(pattern),
|
||||
)
|
||||
)
|
||||
return query
|
||||
|
||||
|
||||
# Define function list_templates
|
||||
def list_templates(
|
||||
# Entry: db
|
||||
@@ -46,51 +82,16 @@ def list_templates(
|
||||
limit: int = 50,
|
||||
) -> list:
|
||||
"""Return paginated, filterable list of test templates."""
|
||||
# Assign query = db.query(TestTemplate)
|
||||
query = db.query(TestTemplate)
|
||||
# Check: is_active is not None
|
||||
if is_active is not None:
|
||||
# Assign query = query.filter(TestTemplate.is_active == is_active)
|
||||
query = query.filter(TestTemplate.is_active == is_active)
|
||||
query = _build_template_query(
|
||||
db, source=source, platform=platform, severity=severity,
|
||||
mitre_technique_id=mitre_technique_id, search=search, is_active=is_active,
|
||||
)
|
||||
|
||||
# Check: source
|
||||
if source:
|
||||
# Assign query = query.filter(TestTemplate.source == source)
|
||||
query = query.filter(TestTemplate.source == source)
|
||||
# Check: platform
|
||||
if platform:
|
||||
# Assign query = query.filter(TestTemplate.platform.ilike(f"%{escape_like(platform)}...
|
||||
query = query.filter(TestTemplate.platform.ilike(f"%{escape_like(platform)}%"))
|
||||
# Check: severity
|
||||
if severity:
|
||||
# Assign query = query.filter(TestTemplate.severity == severity)
|
||||
query = query.filter(TestTemplate.severity == severity)
|
||||
# Check: mitre_technique_id
|
||||
if mitre_technique_id:
|
||||
# Assign query = query.filter(TestTemplate.mitre_technique_id == mitre_technique_id)
|
||||
query = query.filter(TestTemplate.mitre_technique_id == mitre_technique_id)
|
||||
# Check: search
|
||||
if search:
|
||||
# Assign pattern = f"%{escape_like(search)}%"
|
||||
pattern = f"%{escape_like(search)}%"
|
||||
# Assign query = query.filter(
|
||||
query = query.filter(
|
||||
or_(
|
||||
TestTemplate.name.ilike(pattern),
|
||||
TestTemplate.description.ilike(pattern),
|
||||
)
|
||||
)
|
||||
|
||||
# Assign templates = (
|
||||
templates = (
|
||||
query
|
||||
# Chain .order_by() call
|
||||
.order_by(TestTemplate.mitre_technique_id, TestTemplate.name)
|
||||
# Chain .offset() call
|
||||
.offset(offset)
|
||||
# Chain .limit() call
|
||||
.limit(limit)
|
||||
# Chain .all() call
|
||||
.all()
|
||||
)
|
||||
|
||||
@@ -108,10 +109,29 @@ def list_templates(
|
||||
for t in templates:
|
||||
t.existing_test_count = counts.get(t.mitre_technique_id, 0)
|
||||
|
||||
# Return templates
|
||||
return templates
|
||||
|
||||
|
||||
def count_templates(
|
||||
db: Session,
|
||||
*,
|
||||
source: str | None = None,
|
||||
platform: str | None = None,
|
||||
severity: str | None = None,
|
||||
mitre_technique_id: str | None = None,
|
||||
search: str | None = None,
|
||||
is_active: bool | None = None,
|
||||
) -> int:
|
||||
"""Return the total count of templates matching the same filters as
|
||||
list_templates() — lets the catalog UI show a true page count even
|
||||
though the list endpoint caps how many rows it returns per request."""
|
||||
query = _build_template_query(
|
||||
db, source=source, platform=platform, severity=severity,
|
||||
mitre_technique_id=mitre_technique_id, search=search, is_active=is_active,
|
||||
)
|
||||
return query.count()
|
||||
|
||||
|
||||
# Define function get_template_stats
|
||||
def get_template_stats(db: Session) -> dict:
|
||||
"""Return catalog statistics: totals by source, platform, active/inactive."""
|
||||
|
||||
@@ -182,11 +182,14 @@ def test_soft_delete():
|
||||
|
||||
|
||||
def test_search_filter():
|
||||
from app.routers.test_templates import list_templates
|
||||
source = inspect.getsource(list_templates)
|
||||
from app.services.test_template_service import _build_template_query, list_templates
|
||||
# The actual filter-building (including search) now lives in the shared
|
||||
# _build_template_query() helper, reused by both list and count.
|
||||
source = inspect.getsource(_build_template_query)
|
||||
assert "search" in source
|
||||
assert "name" in source and "description" in source
|
||||
assert "ilike" in source
|
||||
assert "_build_template_query" in inspect.getsource(list_templates)
|
||||
print(" [PASS] Search filter searches in name and description")
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
"""GET /test-templates/count — total matching the same filters as the list
|
||||
endpoint, so the catalog UI can compute a true page count."""
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def technique(api, auth_headers):
|
||||
resp = api(
|
||||
"post", "/api/v1/techniques", auth_headers,
|
||||
json={"mitre_id": "T1059.300", "name": "Count Endpoint Technique"},
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
return resp.json()["id"]
|
||||
|
||||
|
||||
def _create_template(api, headers, mitre_technique_id, name):
|
||||
resp = api(
|
||||
"post", "/api/v1/test-templates", headers,
|
||||
json={"mitre_technique_id": mitre_technique_id, "name": name},
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
return resp.json()["id"]
|
||||
|
||||
|
||||
def test_count_matches_number_of_matching_templates(
|
||||
client, db, api, auth_headers, red_lead_headers, technique,
|
||||
):
|
||||
_create_template(api, red_lead_headers, "T1059.300", "Count test template A")
|
||||
_create_template(api, red_lead_headers, "T1059.300", "Count test template B")
|
||||
|
||||
resp = api(
|
||||
"get", "/api/v1/test-templates/count", auth_headers,
|
||||
params={"mitre_technique_id": "T1059.300"},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["total"] == 2
|
||||
|
||||
|
||||
def test_count_respects_search_filter(client, db, api, auth_headers, red_lead_headers, technique):
|
||||
_create_template(api, red_lead_headers, "T1059.300", "Unique Zebra Template")
|
||||
_create_template(api, red_lead_headers, "T1059.300", "Something else entirely")
|
||||
|
||||
resp = api("get", "/api/v1/test-templates/count", auth_headers, params={"search": "Zebra"})
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["total"] == 1
|
||||
|
||||
|
||||
def test_count_matches_list_length_when_under_page_size(
|
||||
client, db, api, auth_headers, red_lead_headers, technique,
|
||||
):
|
||||
_create_template(api, red_lead_headers, "T1059.300", "Parity check template")
|
||||
|
||||
listed = api(
|
||||
"get", "/api/v1/test-templates", auth_headers,
|
||||
params={"mitre_technique_id": "T1059.300"},
|
||||
)
|
||||
counted = api(
|
||||
"get", "/api/v1/test-templates/count", auth_headers,
|
||||
params={"mitre_technique_id": "T1059.300"},
|
||||
)
|
||||
assert len(listed.json()) == counted.json()["total"]
|
||||
|
||||
|
||||
def test_count_accessible_to_any_authenticated_role(client, db, api, red_tech_headers):
|
||||
"""Pagination is a basic UX need, not a lead-only privilege like /stats."""
|
||||
resp = api("get", "/api/v1/test-templates/count", red_tech_headers)
|
||||
assert resp.status_code == 200, resp.text
|
||||
@@ -147,8 +147,11 @@ def test_list_templates_with_filters():
|
||||
assert "search" in source, "List must accept search filter"
|
||||
assert "mitre_technique_id" in source, "List must accept mitre_technique_id filter"
|
||||
|
||||
# Verify ilike is used for search
|
||||
assert "ilike" in source, "Search should use ilike for case-insensitive matching"
|
||||
# Verify ilike is used for search — lives in the shared
|
||||
# _build_template_query() helper now, reused by list and count.
|
||||
from app.services.test_template_service import _build_template_query
|
||||
assert "ilike" in inspect.getsource(_build_template_query), \
|
||||
"Search should use ilike for case-insensitive matching"
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
|
||||
Reference in New Issue
Block a user