feat(tests): richer server-side filters + true total count for GET /tests
GET /tests caps limit at 200, so any page relying on len(results) for a "total" count (e.g. Validated Tests) silently under-reports once there are more matches than the page size — the 200 shown was a page cap, not the real total. - Adds GET /tests/count, sharing its query-building with list_tests() via a new _build_test_query() so the two can't drift apart. - Adds technique_search (free-text on MITRE ID or name), attack_success, detection_result, and validated_from/validated_to filters, so callers aren't limited to state/technique_id/platform/creator.
This commit is contained in:
@@ -10,8 +10,8 @@ from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
# Import Session, joinedload from sqlalchemy.orm
|
||||
from sqlalchemy import or_
|
||||
from sqlalchemy.orm import Session, joinedload
|
||||
from sqlalchemy import func, or_
|
||||
from sqlalchemy.orm import Query, Session, joinedload
|
||||
|
||||
# Import from app.domain.errors
|
||||
from app.domain.errors import (
|
||||
@@ -49,29 +49,27 @@ def determine_initial_classification(technique: Technique | None) -> str:
|
||||
return DataClassification.confidential.value
|
||||
|
||||
|
||||
# Define function list_tests
|
||||
def list_tests(
|
||||
# Entry: db
|
||||
def _build_test_query(
|
||||
db: Session,
|
||||
*,
|
||||
# Entry: state
|
||||
state: str | None = None,
|
||||
# Entry: technique_id
|
||||
technique_id: uuid.UUID | None = None,
|
||||
# Entry: platform
|
||||
technique_search: str | None = None,
|
||||
platform: str | None = None,
|
||||
# Entry: created_by
|
||||
created_by: uuid.UUID | None = None,
|
||||
# Entry: pending_validation_side
|
||||
pending_validation_side: str | None = None,
|
||||
# Entry: reviewer_id — "my reviews" queue for the red_review/blue_review gate stage
|
||||
reviewer_id: uuid.UUID | None = None,
|
||||
not_in_any_campaign: bool = False,
|
||||
offset: int = 0,
|
||||
# Entry: limit
|
||||
limit: int = 50,
|
||||
) -> list[Test]:
|
||||
"""Return a paginated list of tests with optional filters."""
|
||||
attack_success: str | None = None,
|
||||
detection_result: str | None = None,
|
||||
validated_from: datetime | None = None,
|
||||
validated_to: datetime | None = None,
|
||||
) -> Query:
|
||||
"""Build the filtered Test query shared by list_tests() and count_tests().
|
||||
|
||||
Kept as a single source of truth so the displayed page of results and
|
||||
the total count it's paginated against can never drift apart.
|
||||
"""
|
||||
query = db.query(Test).options(joinedload(Test.technique))
|
||||
|
||||
# Check: state
|
||||
@@ -82,6 +80,16 @@ def list_tests(
|
||||
if technique_id:
|
||||
# Assign query = query.filter(Test.technique_id == technique_id)
|
||||
query = query.filter(Test.technique_id == technique_id)
|
||||
# Free-text technique filter (MITRE ID or name) — doesn't require the
|
||||
# caller to already know the technique's UUID like technique_id does.
|
||||
if technique_search:
|
||||
pattern = f"%{escape_like(technique_search)}%"
|
||||
query = query.join(Technique, Test.technique_id == Technique.id).filter(
|
||||
or_(
|
||||
Technique.mitre_id.ilike(pattern),
|
||||
Technique.name.ilike(pattern),
|
||||
)
|
||||
)
|
||||
# Check: platform
|
||||
if platform:
|
||||
# Assign query = query.filter(Test.platform.ilike(f"%{escape_like(platform)}%"))
|
||||
@@ -125,10 +133,102 @@ def list_tests(
|
||||
linked = db.query(CampaignTest.test_id).distinct().subquery()
|
||||
query = query.filter(~Test.id.in_(linked))
|
||||
|
||||
# Return query.order_by(Test.created_at.desc()).offset(offset).limit(limit)....
|
||||
if attack_success:
|
||||
query = query.filter(Test.attack_success == attack_success)
|
||||
if detection_result:
|
||||
query = query.filter(Test.detection_result == detection_result)
|
||||
|
||||
# Validated-date range: whichever lead validated last (mirrors the
|
||||
# "Validated" column shown on the Validated Tests page).
|
||||
if validated_from or validated_to:
|
||||
validated_at = func.coalesce(Test.blue_validated_at, Test.red_validated_at)
|
||||
if validated_from:
|
||||
query = query.filter(validated_at >= validated_from)
|
||||
if validated_to:
|
||||
query = query.filter(validated_at <= validated_to)
|
||||
|
||||
return query
|
||||
|
||||
|
||||
# Define function list_tests
|
||||
def list_tests(
|
||||
db: Session,
|
||||
*,
|
||||
state: str | None = None,
|
||||
technique_id: uuid.UUID | None = None,
|
||||
technique_search: str | None = None,
|
||||
platform: str | None = None,
|
||||
created_by: uuid.UUID | None = None,
|
||||
pending_validation_side: str | None = None,
|
||||
# Entry: reviewer_id — "my reviews" queue for the red_review/blue_review gate stage
|
||||
reviewer_id: uuid.UUID | None = None,
|
||||
not_in_any_campaign: bool = False,
|
||||
attack_success: str | None = None,
|
||||
detection_result: str | None = None,
|
||||
validated_from: datetime | None = None,
|
||||
validated_to: datetime | None = None,
|
||||
offset: int = 0,
|
||||
# Entry: limit
|
||||
limit: int = 50,
|
||||
) -> list[Test]:
|
||||
"""Return a paginated list of tests with optional filters."""
|
||||
query = _build_test_query(
|
||||
db,
|
||||
state=state,
|
||||
technique_id=technique_id,
|
||||
technique_search=technique_search,
|
||||
platform=platform,
|
||||
created_by=created_by,
|
||||
pending_validation_side=pending_validation_side,
|
||||
reviewer_id=reviewer_id,
|
||||
not_in_any_campaign=not_in_any_campaign,
|
||||
attack_success=attack_success,
|
||||
detection_result=detection_result,
|
||||
validated_from=validated_from,
|
||||
validated_to=validated_to,
|
||||
)
|
||||
return query.order_by(Test.created_at.desc()).offset(offset).limit(limit).all()
|
||||
|
||||
|
||||
def count_tests(
|
||||
db: Session,
|
||||
*,
|
||||
state: str | None = None,
|
||||
technique_id: uuid.UUID | None = None,
|
||||
technique_search: str | None = None,
|
||||
platform: str | None = None,
|
||||
created_by: uuid.UUID | None = None,
|
||||
pending_validation_side: str | None = None,
|
||||
reviewer_id: uuid.UUID | None = None,
|
||||
not_in_any_campaign: bool = False,
|
||||
attack_success: str | None = None,
|
||||
detection_result: str | None = None,
|
||||
validated_from: datetime | None = None,
|
||||
validated_to: datetime | None = None,
|
||||
) -> int:
|
||||
"""Return the total count of tests matching the same filters as list_tests().
|
||||
|
||||
Lets the frontend show a true total even when the page size caps the
|
||||
number of rows actually returned (e.g. GET /tests's limit<=200).
|
||||
"""
|
||||
query = _build_test_query(
|
||||
db,
|
||||
state=state,
|
||||
technique_id=technique_id,
|
||||
technique_search=technique_search,
|
||||
platform=platform,
|
||||
created_by=created_by,
|
||||
pending_validation_side=pending_validation_side,
|
||||
reviewer_id=reviewer_id,
|
||||
not_in_any_campaign=not_in_any_campaign,
|
||||
attack_success=attack_success,
|
||||
detection_result=detection_result,
|
||||
validated_from=validated_from,
|
||||
validated_to=validated_to,
|
||||
)
|
||||
return query.count()
|
||||
|
||||
|
||||
# Define function create_test
|
||||
def create_test(
|
||||
# Entry: db
|
||||
|
||||
Reference in New Issue
Block a user