refactor(heatmap): extract business logic to dedicated service
Some checks failed
Aegis CI / lint-and-test (push) Has been cancelled

Move layer dispatch, entity-not-found checks, and validation from router to heatmap_service. Router now only validates requests, calls service, and formats responses (no HTTPException, no business logic). Service raises EntityNotFoundError/BusinessRuleViolation instead of returning None. Add build_navigator_export() for centralized dispatch. 29 new tests (253 total, 0 failures).
This commit is contained in:
2026-02-18 16:09:51 +01:00
parent 1338d52cd0
commit e651ef8a8c
3 changed files with 243 additions and 53 deletions

View File

@@ -15,6 +15,7 @@ from typing import Optional
from sqlalchemy import func, or_
from sqlalchemy.orm import Session
from app.domain.errors import BusinessRuleViolation, EntityNotFoundError
from app.models.campaign import Campaign, CampaignTest
from app.models.detection_rule import DetectionRule
from app.models.defensive_technique import DefensiveTechniqueMapping
@@ -206,14 +207,14 @@ def build_threat_actor_layer(
platforms: str | None = None,
tactics: str | None = None,
min_score: int = 0,
) -> dict | None:
) -> dict:
"""Threat actor layer -- techniques used by an actor with coverage colour.
Returns ``None`` if the actor does not exist.
Raises :class:`EntityNotFoundError` if the actor does not exist.
"""
actor = db.query(ThreatActor).filter(ThreatActor.id == actor_id).first()
if not actor:
return None
raise EntityNotFoundError("ThreatActor", actor_id)
layer = _build_layer_skeleton(
f"Threat Actor: {actor.name}",
@@ -364,14 +365,14 @@ def build_campaign_layer(
platforms: str | None = None,
tactics: str | None = None,
min_score: int = 0,
) -> dict | None:
) -> dict:
"""Campaign layer -- techniques in a campaign, coloured by test state.
Returns ``None`` if the campaign does not exist.
Raises :class:`EntityNotFoundError` if the campaign does not exist.
"""
campaign = db.query(Campaign).filter(Campaign.id == campaign_id).first()
if not campaign:
return None
raise EntityNotFoundError("Campaign", campaign_id)
layer = _build_layer_skeleton(
f"Campaign: {campaign.name}",
@@ -450,3 +451,49 @@ def build_campaign_layer(
})
return layer
# ── Layer dispatch (for Navigator export) ────────────────────────────
_LAYER_BUILDERS = {
"coverage": lambda db, **kw: build_coverage_layer(db, **kw),
"detection-rules": lambda db, **kw: build_detection_rules_layer(db, **kw),
}
_LAYER_BUILDERS_WITH_ID = {
"threat-actor": lambda db, lid, **kw: build_threat_actor_layer(db, lid, **kw),
"campaign": lambda db, lid, **kw: build_campaign_layer(db, lid, **kw),
}
SUPPORTED_LAYER_TYPES = set(_LAYER_BUILDERS) | set(_LAYER_BUILDERS_WITH_ID)
def build_navigator_export(
db: Session,
layer_type: str,
*,
layer_id: str | None = None,
platforms: str | None = None,
tactics: str | None = None,
min_score: int = 0,
) -> dict:
"""Build a heatmap layer dict by type name.
Raises :class:`BusinessRuleViolation` for unknown layer types or
missing ``layer_id``. Raises :class:`EntityNotFoundError` when
an entity-bound layer (threat-actor, campaign) references a
non-existent record.
"""
kwargs = dict(platforms=platforms, tactics=tactics, min_score=min_score)
if layer_type in _LAYER_BUILDERS:
return _LAYER_BUILDERS[layer_type](db, **kwargs)
if layer_type in _LAYER_BUILDERS_WITH_ID:
if not layer_id:
raise BusinessRuleViolation(
f"layer_id is required for '{layer_type}' layer"
)
return _LAYER_BUILDERS_WITH_ID[layer_type](db, layer_id, **kwargs)
raise BusinessRuleViolation(f"Unknown layer type: {layer_type}")