From 226869d3083e32eec92fa77abcbad5f523741eef Mon Sep 17 00:00:00 2001 From: kitos Date: Thu, 9 Jul 2026 12:45:05 +0200 Subject: [PATCH] feat(tests): richer server-side filters + true total count for GET /tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- backend/app/routers/tests.py | 78 +++++++++---- backend/app/services/test_crud_service.py | 134 +++++++++++++++++++--- backend/tests/test_test_list_filters.py | 85 ++++++++++++++ 3 files changed, 259 insertions(+), 38 deletions(-) create mode 100644 backend/tests/test_test_list_filters.py diff --git a/backend/app/routers/tests.py b/backend/app/routers/tests.py index b1f5af3..9a76697 100644 --- a/backend/app/routers/tests.py +++ b/backend/app/routers/tests.py @@ -118,6 +118,11 @@ from app.services.test_crud_service import ( list_tests as crud_list_tests, ) +# Import from app.services.test_crud_service +from app.services.test_crud_service import ( + count_tests as crud_count_tests, +) + # Import from app.services.test_crud_service from app.services.test_crud_service import ( update_test as crud_update_test, @@ -203,13 +208,14 @@ def _mask_for_team_blindness(test_out: TestOut, *, viewer_role: str) -> TestOut: # --------------------------------------------------------------------------- -@router.get("", response_model=list[TestOut]) -# Define function list_tests -def list_tests( +def _test_list_filter_params( # Entry: state state: Optional[str] = Query(None, description="Filter by test state"), # Entry: technique_id technique_id: Optional[uuid.UUID] = Query(None, description="Filter by technique"), + technique_search: Optional[str] = Query( + None, description="Free-text filter on technique MITRE ID or name" + ), # Entry: platform platform: Optional[str] = Query(None, description="Filter by platform"), # Entry: created_by @@ -225,6 +231,52 @@ def list_tests( not_in_any_campaign: bool = Query( False, description="Only return tests not linked to any campaign" ), + attack_success: Optional[str] = Query(None, description="Filter by attack success outcome"), + detection_result: Optional[str] = Query(None, description="Filter by detection outcome"), + validated_from: Optional[datetime] = Query( + None, description="Only tests validated on/after this date" + ), + validated_to: Optional[datetime] = Query( + None, description="Only tests validated on/before this date" + ), +) -> dict: + """Shared filter params for GET /tests and GET /tests/count, so the two + endpoints can never silently drift out of sync with each other.""" + return { + "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, + } + + +@router.get("/count") +def count_tests( + filters: dict = Depends(_test_list_filter_params), + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user), +) -> dict: + """Return the true total of tests matching the given filters. + + GET /tests caps `limit` at 200, so pages that show "N results" need + this to report the real total rather than len(page) — which silently + under-reports whenever there are more matches than the page size. + """ + return {"total": crud_count_tests(db, **filters)} + + +@router.get("", response_model=list[TestOut]) +# Define function list_tests +def list_tests( + filters: dict = Depends(_test_list_filter_params), offset: int = Query(0, ge=0), # Entry: limit limit: int = Query(50, ge=1, le=200), @@ -236,12 +288,7 @@ def list_tests( """Return a paginated list of tests, optionally filtered by state, technique, platform or creator. Args: - state (Optional[str]): Filter by test state (e.g. ``draft``, ``validated``). - technique_id (Optional[uuid.UUID]): Filter tests belonging to a specific technique. - platform (Optional[str]): Filter by target platform (e.g. ``windows``, ``linux``). - created_by (Optional[uuid.UUID]): Filter by the UUID of the creator. - pending_validation_side (Optional[str]): Filter ``in_review`` tests pending validation - on ``'red'`` or ``'blue'`` side. + filters (dict): Shared filter params — see ``_test_list_filter_params``. offset (int): Number of records to skip for pagination. limit (int): Maximum number of records to return. db (Session): SQLAlchemy database session. @@ -253,18 +300,7 @@ def list_tests( # Return crud_list_tests( return crud_list_tests( db, - # Keyword argument: state - state=state, - # Keyword argument: technique_id - technique_id=technique_id, - # Keyword argument: platform - platform=platform, - # Keyword argument: created_by - created_by=created_by, - # Keyword argument: pending_validation_side - pending_validation_side=pending_validation_side, - reviewer_id=reviewer_id, - not_in_any_campaign=not_in_any_campaign, + **filters, offset=offset, # Keyword argument: limit limit=limit, diff --git a/backend/app/services/test_crud_service.py b/backend/app/services/test_crud_service.py index db239c7..6ab51bc 100644 --- a/backend/app/services/test_crud_service.py +++ b/backend/app/services/test_crud_service.py @@ -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 diff --git a/backend/tests/test_test_list_filters.py b/backend/tests/test_test_list_filters.py new file mode 100644 index 0000000..c0a0647 --- /dev/null +++ b/backend/tests/test_test_list_filters.py @@ -0,0 +1,85 @@ +"""Tests for the expanded GET /tests filters and the new GET /tests/count. + +Block: Validated Tests page couldn't show a true total once there were +more than `limit` (max 200) matching tests, and had no server-side way to +filter by technique/platform/date/outcome beyond a client-side name search. +count_tests() shares its query-building with list_tests() (_build_test_query) +so the two can't silently drift out of sync. +""" + +from datetime import datetime, timedelta + +from app.models.enums import AttackSuccessResult, TestResult, TestState +from app.models.technique import Technique +from app.models.test import Test +from app.services.test_crud_service import count_tests, list_tests + + +def _seed(db, owner_id, *, mitre_id, name, technique_name=None, state=TestState.validated, **overrides): + tech = Technique(mitre_id=mitre_id, name=technique_name or f"Technique {mitre_id}", tactic="execution", platforms=["windows"]) + db.add(tech) + db.flush() + test = Test(technique_id=tech.id, name=name, state=state, created_by=owner_id, **overrides) + db.add(test) + db.commit() + db.refresh(test) + return test + + +def test_count_tests_matches_list_tests_total_beyond_page_size(db, red_lead_user): + for i in range(5): + _seed(db, red_lead_user.id, mitre_id=f"T900{i}", name=f"Validated {i}") + + total = count_tests(db, state="validated") + page = list_tests(db, state="validated", limit=2) + + assert total == 5 + assert len(page) == 2 + + +def test_technique_search_matches_mitre_id_or_name(db, red_lead_user): + _seed(db, red_lead_user.id, mitre_id="T1059.001", name="PowerShell test", technique_name="Command Line Interface") + _seed(db, red_lead_user.id, mitre_id="T1055", name="Process Injection test", technique_name="Process Injection") + + by_id = list_tests(db, technique_search="T1059") + by_name = list_tests(db, technique_search="injection") + + assert [t.name for t in by_id] == ["PowerShell test"] + assert [t.name for t in by_name] == ["Process Injection test"] + + +def test_attack_success_and_detection_result_filters(db, red_lead_user): + _seed( + db, red_lead_user.id, mitre_id="T9010", name="Successful and detected", + attack_success=AttackSuccessResult.successful, detection_result=TestResult.detected, + ) + _seed( + db, red_lead_user.id, mitre_id="T9011", name="Successful and not detected", + attack_success=AttackSuccessResult.successful, detection_result=TestResult.not_detected, + ) + + results = list_tests(db, attack_success="successful", detection_result="detected") + + assert [t.name for t in results] == ["Successful and detected"] + + +def test_validated_date_range_filter(db, red_lead_user): + old = _seed(db, red_lead_user.id, mitre_id="T9020", name="Old validation") + old.blue_validated_at = datetime.utcnow() - timedelta(days=30) + recent = _seed(db, red_lead_user.id, mitre_id="T9021", name="Recent validation") + recent.blue_validated_at = datetime.utcnow() - timedelta(days=1) + db.commit() + + results = list_tests(db, validated_from=datetime.utcnow() - timedelta(days=7)) + + assert [t.name for t in results] == ["Recent validation"] + + +def test_count_tests_endpoint_returns_true_total(api, db, red_lead_user, auth_headers): + for i in range(3): + _seed(db, red_lead_user.id, mitre_id=f"T903{i}", name=f"Counted {i}") + + resp = api("get", "/api/v1/tests/count?state=validated", auth_headers) + + assert resp.status_code == 200 + assert resp.json()["total"] == 3