Kitos
bfce1a8a0e
Some checks failed
Aegis CI / lint-and-test (push) Has been cancelled
- Add UnitOfWork context manager in domain/unit_of_work.py with commit/rollback/flush API and auto-rollback on exception - Remove all db.commit() from test_workflow_service (8 calls), notification_service (4 calls), status_service (1 call) - Services now only stage changes via db.add/db.flush; caller owns the transaction boundary - Update routers/tests.py: wrap 9 workflow endpoints in UnitOfWork context managers - Update routers/notifications.py: wrap mark_as_read and mark_all_as_read in UnitOfWork
48 lines
1.9 KiB
Python
48 lines
1.9 KiB
Python
"""Service for recalculating the global status of a Technique
|
|
based on the state and result of its associated tests.
|
|
|
|
V2 rules account for dual Red/Blue validation and use
|
|
``detection_result`` (filled by Blue Team) instead of the legacy
|
|
``result`` field.
|
|
|
|
This function mutates the technique but does **not** commit.
|
|
The caller is responsible for committing the session.
|
|
"""
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.enums import TechniqueStatus, TestState
|
|
from app.models.technique import Technique
|
|
|
|
|
|
def recalculate_technique_status(db: Session, technique: Technique) -> None:
|
|
"""Recompute ``technique.status_global`` from its tests and commit.
|
|
|
|
Rules (v2)
|
|
----------
|
|
1. No tests → ``not_evaluated``
|
|
2. All tests ``validated`` → look at detection results:
|
|
- All ``detected`` → ``validated``
|
|
- Any ``partially_detected`` → ``partial``
|
|
- Otherwise → ``not_covered``
|
|
3. Some tests ``validated``, others still in progress → ``partial``
|
|
4. All tests in intermediate states (no validated) → ``in_progress``
|
|
"""
|
|
tests = technique.tests
|
|
|
|
if not tests:
|
|
technique.status_global = TechniqueStatus.not_evaluated
|
|
elif all(t.state == TestState.validated for t in tests):
|
|
# All validated — inspect detection results
|
|
results = [t.detection_result for t in tests if t.detection_result]
|
|
if results and all(str(r) == "detected" or r == "detected" for r in results):
|
|
technique.status_global = TechniqueStatus.validated
|
|
elif any(str(r) == "partially_detected" or r == "partially_detected" for r in results):
|
|
technique.status_global = TechniqueStatus.partial
|
|
else:
|
|
technique.status_global = TechniqueStatus.not_covered
|
|
elif any(t.state == TestState.validated for t in tests):
|
|
technique.status_global = TechniqueStatus.partial
|
|
else:
|
|
technique.status_global = TechniqueStatus.in_progress
|