Kitos
9e204b78ec
- Add test_test_entity.py with 46 pure unit tests covering the full domain entity - Fix _FakeSettings in 11 test files (REPORT_TEMPLATES_DIR, JIRA, TEMPO) - Fix stale db.commit assertions to db.flush after UoW refactor - Add missing mock fields for TestEntity.from_orm compatibility - Make database.py skip pool args for SQLite in test environment - Disable slowapi rate limiter in test client fixture - Inject test engine into app.database to fix threading errors - Update role assertions to match current require_any_role policy - Mark 6 legacy V1 endpoint tests as xfail (replaced by V2 workflow)
433 lines
16 KiB
Python
433 lines
16 KiB
Python
"""T-127: Tests de métricas actualizadas.
|
|
|
|
Tests for the V2 metrics endpoints (pipeline, team-activity, validation-rate)
|
|
and for the technique status recalculation logic with the new test states.
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import uuid
|
|
import inspect
|
|
from unittest.mock import MagicMock, patch, PropertyMock
|
|
from types import ModuleType
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stub heavy dependencies before importing app modules
|
|
# ---------------------------------------------------------------------------
|
|
|
|
backend_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
|
if backend_dir not in sys.path:
|
|
sys.path.insert(0, backend_dir)
|
|
|
|
if "pydantic_settings" not in sys.modules:
|
|
_ps = ModuleType("pydantic_settings")
|
|
class _BaseSettings:
|
|
def __init__(self, **kwargs): pass
|
|
def __init_subclass__(cls, **kwargs): super().__init_subclass__(**kwargs)
|
|
_ps.BaseSettings = _BaseSettings
|
|
sys.modules["pydantic_settings"] = _ps
|
|
|
|
if "app.config" not in sys.modules:
|
|
_cfg = ModuleType("app.config")
|
|
class _FakeSettings:
|
|
DATABASE_URL = "sqlite:///:memory:"
|
|
SECRET_KEY = "test"
|
|
ALGORITHM = "HS256"
|
|
ACCESS_TOKEN_EXPIRE_MINUTES = 60
|
|
MINIO_ENDPOINT = "localhost:9000"
|
|
MINIO_ACCESS_KEY = "test"
|
|
MINIO_SECRET_KEY = "test"
|
|
MINIO_BUCKET = "test"
|
|
REPORT_TEMPLATES_DIR = "app/templates/reports"
|
|
REPORT_OUTPUT_DIR = "/tmp/aegis_reports"
|
|
COMPANY_NAME = "Test Org"
|
|
COMPANY_LOGO_PATH = "app/templates/reports/assets/logo.png"
|
|
JIRA_ENABLED = False
|
|
JIRA_URL = ""
|
|
JIRA_USERNAME = ""
|
|
JIRA_API_TOKEN = ""
|
|
JIRA_IS_CLOUD = True
|
|
JIRA_DEFAULT_PROJECT = ""
|
|
JIRA_ISSUE_TYPE_TEST = "Task"
|
|
JIRA_ISSUE_TYPE_CAMPAIGN = "Epic"
|
|
TEMPO_ENABLED = False
|
|
TEMPO_API_TOKEN = ""
|
|
TEMPO_DEFAULT_WORK_TYPE = "Red Team"
|
|
NVD_API_KEY = ""
|
|
STALE_THRESHOLD_DAYS = 365
|
|
CORS_ORIGINS = "http://localhost:3000"
|
|
SCORING_WEIGHT_TESTS = 40
|
|
SCORING_WEIGHT_DETECTION_RULES = 20
|
|
SCORING_WEIGHT_D3FEND = 15
|
|
SCORING_WEIGHT_FRESHNESS = 15
|
|
SCORING_WEIGHT_PLATFORM_DIVERSITY = 10
|
|
_cfg.settings = _FakeSettings()
|
|
sys.modules["app.config"] = _cfg
|
|
|
|
if "app.database" not in sys.modules:
|
|
_db = ModuleType("app.database")
|
|
_db.Base = type("Base", (), {"metadata": MagicMock()})
|
|
_db.get_db = MagicMock()
|
|
sys.modules["app.database"] = _db
|
|
|
|
for _mod in [
|
|
"taxii2client", "taxii2client.v20",
|
|
"jose", "boto3", "botocore", "botocore.exceptions",
|
|
"apscheduler", "apscheduler.schedulers",
|
|
"apscheduler.schedulers.background",
|
|
"apscheduler.triggers", "apscheduler.triggers.cron",
|
|
]:
|
|
if _mod not in sys.modules:
|
|
m = ModuleType(_mod)
|
|
if _mod == "taxii2client.v20": m.Server = MagicMock
|
|
elif _mod == "jose": m.JWTError = Exception; m.jwt = MagicMock()
|
|
elif _mod == "boto3": m.client = MagicMock()
|
|
elif _mod == "botocore.exceptions": m.ClientError = Exception
|
|
elif _mod == "apscheduler.schedulers.background": m.BackgroundScheduler = MagicMock
|
|
elif _mod == "apscheduler.triggers.cron": m.CronTrigger = MagicMock
|
|
sys.modules[_mod] = m
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Imports
|
|
# ---------------------------------------------------------------------------
|
|
|
|
from app.models.enums import TestState, TestResult, TechniqueStatus
|
|
from app.services.status_service import recalculate_technique_status
|
|
from app.routers.metrics import router as metrics_router
|
|
|
|
|
|
def _get_route_paths():
|
|
routes = {}
|
|
for route in metrics_router.routes:
|
|
path = getattr(route, "path", "")
|
|
methods = getattr(route, "methods", set())
|
|
for method in methods:
|
|
routes[f"{method} {path}"] = route
|
|
return routes
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers for technique status recalculation tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_test(state: TestState, detection_result=None) -> MagicMock:
|
|
t = MagicMock()
|
|
t.id = uuid.uuid4()
|
|
t.state = state
|
|
t.detection_result = detection_result
|
|
t.red_validation_status = None
|
|
t.blue_validation_status = None
|
|
return t
|
|
|
|
|
|
def _make_technique(tests=None) -> MagicMock:
|
|
tech = MagicMock()
|
|
tech.id = uuid.uuid4()
|
|
tech.tests = tests or []
|
|
tech.status_global = TechniqueStatus.not_evaluated
|
|
return tech
|
|
|
|
|
|
def _make_db() -> MagicMock:
|
|
return MagicMock()
|
|
|
|
|
|
# ===========================================================================
|
|
# 1. test_pipeline_metrics — endpoint exists and queries TestState
|
|
# ===========================================================================
|
|
|
|
|
|
def test_pipeline_metrics_endpoint_exists():
|
|
"""GET /metrics/test-pipeline endpoint exists."""
|
|
routes = _get_route_paths()
|
|
found = any("test-pipeline" in k and "GET" in k for k in routes)
|
|
assert found, f"GET /metrics/test-pipeline not found. Routes: {list(routes.keys())}"
|
|
|
|
|
|
def test_pipeline_metrics_queries_all_states():
|
|
"""Pipeline endpoint groups by all test states."""
|
|
from app.routers.metrics import test_pipeline
|
|
source = inspect.getsource(test_pipeline)
|
|
|
|
assert "Test.state" in source, "Must query Test.state"
|
|
assert "group_by" in source, "Must group by state"
|
|
assert "TestPipelineCounts" in source, "Must return TestPipelineCounts schema"
|
|
|
|
|
|
# ===========================================================================
|
|
# 2. test_team_activity_metrics — endpoint exists and calculates correctly
|
|
# ===========================================================================
|
|
|
|
|
|
def test_team_activity_endpoint_exists():
|
|
"""GET /metrics/team-activity endpoint exists."""
|
|
routes = _get_route_paths()
|
|
found = any("team-activity" in k and "GET" in k for k in routes)
|
|
assert found, f"GET /metrics/team-activity not found. Routes: {list(routes.keys())}"
|
|
|
|
|
|
def test_team_activity_calculates_both_teams():
|
|
"""Team activity endpoint returns data for both Red and Blue teams."""
|
|
from app.routers.metrics import team_activity
|
|
source = inspect.getsource(team_activity)
|
|
|
|
assert "Red Team" in source or "red" in source.lower(), "Must include Red Team data"
|
|
assert "Blue Team" in source or "blue" in source.lower(), "Must include Blue Team data"
|
|
assert "tests_completed" in source, "Must calculate completed tests"
|
|
assert "tests_pending" in source, "Must calculate pending tests"
|
|
|
|
|
|
def test_team_activity_red_pending_states():
|
|
"""Red Team pending includes draft and red_executing."""
|
|
from app.routers.metrics import team_activity
|
|
source = inspect.getsource(team_activity)
|
|
|
|
assert "draft" in source, "Red pending must include draft"
|
|
assert "red_executing" in source, "Red pending must include red_executing"
|
|
|
|
|
|
def test_team_activity_blue_pending_states():
|
|
"""Blue Team pending includes blue_evaluating."""
|
|
from app.routers.metrics import team_activity
|
|
source = inspect.getsource(team_activity)
|
|
|
|
assert "blue_evaluating" in source, "Blue pending must include blue_evaluating"
|
|
|
|
|
|
# ===========================================================================
|
|
# 3. test_technique_status_recalculation_with_new_states
|
|
# ===========================================================================
|
|
|
|
|
|
def test_technique_no_tests_is_not_evaluated():
|
|
"""Technique with no tests -> not_evaluated."""
|
|
tech = _make_technique(tests=[])
|
|
db = _make_db()
|
|
|
|
recalculate_technique_status(db, tech)
|
|
assert tech.status_global == TechniqueStatus.not_evaluated
|
|
|
|
|
|
def test_technique_all_validated_detected():
|
|
"""All tests validated with detected -> technique validated."""
|
|
tests = [
|
|
_make_test(TestState.validated, detection_result="detected"),
|
|
_make_test(TestState.validated, detection_result="detected"),
|
|
]
|
|
tech = _make_technique(tests=tests)
|
|
db = _make_db()
|
|
|
|
recalculate_technique_status(db, tech)
|
|
assert tech.status_global == TechniqueStatus.validated
|
|
|
|
|
|
def test_technique_all_validated_partially_detected():
|
|
"""All tests validated with partially_detected -> technique partial."""
|
|
tests = [
|
|
_make_test(TestState.validated, detection_result="detected"),
|
|
_make_test(TestState.validated, detection_result="partially_detected"),
|
|
]
|
|
tech = _make_technique(tests=tests)
|
|
db = _make_db()
|
|
|
|
recalculate_technique_status(db, tech)
|
|
assert tech.status_global == TechniqueStatus.partial
|
|
|
|
|
|
def test_technique_all_validated_not_detected():
|
|
"""All tests validated with not_detected -> technique not_covered."""
|
|
tests = [
|
|
_make_test(TestState.validated, detection_result="not_detected"),
|
|
]
|
|
tech = _make_technique(tests=tests)
|
|
db = _make_db()
|
|
|
|
recalculate_technique_status(db, tech)
|
|
assert tech.status_global == TechniqueStatus.not_covered
|
|
|
|
|
|
def test_technique_mixed_validated_and_in_progress():
|
|
"""Some validated, some still in pipeline -> technique partial."""
|
|
tests = [
|
|
_make_test(TestState.validated, detection_result="detected"),
|
|
_make_test(TestState.red_executing),
|
|
]
|
|
tech = _make_technique(tests=tests)
|
|
db = _make_db()
|
|
|
|
recalculate_technique_status(db, tech)
|
|
assert tech.status_global == TechniqueStatus.partial
|
|
|
|
|
|
def test_technique_all_in_progress():
|
|
"""All tests in intermediate states (no validated) -> technique in_progress."""
|
|
tests = [
|
|
_make_test(TestState.draft),
|
|
_make_test(TestState.red_executing),
|
|
_make_test(TestState.blue_evaluating),
|
|
]
|
|
tech = _make_technique(tests=tests)
|
|
db = _make_db()
|
|
|
|
recalculate_technique_status(db, tech)
|
|
assert tech.status_global == TechniqueStatus.in_progress
|
|
|
|
|
|
def test_technique_with_in_review_tests():
|
|
"""Tests in in_review are still in-progress (not yet validated)."""
|
|
tests = [
|
|
_make_test(TestState.in_review),
|
|
]
|
|
tech = _make_technique(tests=tests)
|
|
db = _make_db()
|
|
|
|
recalculate_technique_status(db, tech)
|
|
assert tech.status_global == TechniqueStatus.in_progress
|
|
|
|
|
|
def test_technique_with_rejected_tests():
|
|
"""Rejected tests count as in-progress (need rework)."""
|
|
tests = [
|
|
_make_test(TestState.rejected),
|
|
]
|
|
tech = _make_technique(tests=tests)
|
|
db = _make_db()
|
|
|
|
recalculate_technique_status(db, tech)
|
|
assert tech.status_global == TechniqueStatus.in_progress
|
|
|
|
|
|
# ===========================================================================
|
|
# 4. test_coverage_with_dual_validation
|
|
# ===========================================================================
|
|
|
|
|
|
def test_coverage_correct_after_dual_validation():
|
|
"""After dual validation (both approved), technique status is correct."""
|
|
# A test that completed the full pipeline with detection
|
|
test = _make_test(TestState.validated, detection_result="detected")
|
|
test.red_validation_status = "approved"
|
|
test.blue_validation_status = "approved"
|
|
|
|
tech = _make_technique(tests=[test])
|
|
db = _make_db()
|
|
|
|
recalculate_technique_status(db, tech)
|
|
assert tech.status_global == TechniqueStatus.validated
|
|
|
|
|
|
def test_coverage_partial_when_one_detected_one_partial():
|
|
"""Mixed detection results after dual validation -> partial coverage."""
|
|
test1 = _make_test(TestState.validated, detection_result="detected")
|
|
test1.red_validation_status = "approved"
|
|
test1.blue_validation_status = "approved"
|
|
|
|
test2 = _make_test(TestState.validated, detection_result="partially_detected")
|
|
test2.red_validation_status = "approved"
|
|
test2.blue_validation_status = "approved"
|
|
|
|
tech = _make_technique(tests=[test1, test2])
|
|
db = _make_db()
|
|
|
|
recalculate_technique_status(db, tech)
|
|
assert tech.status_global == TechniqueStatus.partial
|
|
|
|
|
|
# ===========================================================================
|
|
# 5. test_validation_rate_endpoint — approval/rejection rates
|
|
# ===========================================================================
|
|
|
|
|
|
def test_validation_rate_endpoint_exists():
|
|
"""GET /metrics/validation-rate endpoint exists."""
|
|
routes = _get_route_paths()
|
|
found = any("validation-rate" in k and "GET" in k for k in routes)
|
|
assert found, f"GET /metrics/validation-rate not found. Routes: {list(routes.keys())}"
|
|
|
|
|
|
def test_validation_rate_queries_both_roles():
|
|
"""Validation rate endpoint returns data for both red_lead and blue_lead."""
|
|
from app.routers.metrics import validation_rate
|
|
source = inspect.getsource(validation_rate)
|
|
|
|
assert "red_validation_status" in source, "Must query red_validation_status"
|
|
assert "blue_validation_status" in source, "Must query blue_validation_status"
|
|
assert "approved" in source, "Must count approved validations"
|
|
assert "rejected" in source, "Must count rejected validations"
|
|
assert "approval_rate" in source, "Must calculate approval_rate"
|
|
|
|
|
|
# ===========================================================================
|
|
# 6. test_recent_tests_endpoint — latest 10 tests
|
|
# ===========================================================================
|
|
|
|
|
|
def test_recent_tests_endpoint_exists():
|
|
"""GET /metrics/recent-tests endpoint exists."""
|
|
routes = _get_route_paths()
|
|
found = any("recent-tests" in k and "GET" in k for k in routes)
|
|
assert found, f"GET /metrics/recent-tests not found. Routes: {list(routes.keys())}"
|
|
|
|
|
|
def test_recent_tests_limits_to_10():
|
|
"""Recent tests endpoint limits to 10 results."""
|
|
from app.routers.metrics import recent_tests
|
|
source = inspect.getsource(recent_tests)
|
|
|
|
assert "limit(10)" in source or ".limit(10)" in source, \
|
|
"Must limit to 10 recent tests"
|
|
assert "created_at" in source, "Must order by created_at"
|
|
|
|
|
|
# ===========================================================================
|
|
# 7. test_original_endpoints_still_work
|
|
# ===========================================================================
|
|
|
|
|
|
def test_summary_endpoint_exists():
|
|
"""GET /metrics/summary (original) endpoint still exists."""
|
|
routes = _get_route_paths()
|
|
found = any("summary" in k and "GET" in k for k in routes)
|
|
assert found, f"GET /metrics/summary not found. Routes: {list(routes.keys())}"
|
|
|
|
|
|
def test_by_tactic_endpoint_exists():
|
|
"""GET /metrics/by-tactic (original) endpoint still exists."""
|
|
routes = _get_route_paths()
|
|
found = any("by-tactic" in k and "GET" in k for k in routes)
|
|
assert found, f"GET /metrics/by-tactic not found. Routes: {list(routes.keys())}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Run all
|
|
# ---------------------------------------------------------------------------
|
|
|
|
if __name__ == "__main__":
|
|
print("T-127 Validation: Metrics V2 Tests")
|
|
print("=" * 55)
|
|
test_pipeline_metrics_endpoint_exists()
|
|
test_pipeline_metrics_queries_all_states()
|
|
test_team_activity_endpoint_exists()
|
|
test_team_activity_calculates_both_teams()
|
|
test_team_activity_red_pending_states()
|
|
test_team_activity_blue_pending_states()
|
|
test_technique_no_tests_is_not_evaluated()
|
|
test_technique_all_validated_detected()
|
|
test_technique_all_validated_partially_detected()
|
|
test_technique_all_validated_not_detected()
|
|
test_technique_mixed_validated_and_in_progress()
|
|
test_technique_all_in_progress()
|
|
test_technique_with_in_review_tests()
|
|
test_technique_with_rejected_tests()
|
|
test_coverage_correct_after_dual_validation()
|
|
test_coverage_partial_when_one_detected_one_partial()
|
|
test_validation_rate_endpoint_exists()
|
|
test_validation_rate_queries_both_roles()
|
|
test_recent_tests_endpoint_exists()
|
|
test_recent_tests_limits_to_10()
|
|
test_summary_endpoint_exists()
|
|
test_by_tactic_endpoint_exists()
|
|
print("=" * 55)
|
|
print("ALL T-127 validations PASSED!")
|