0002601b50
Aegis CI / lint-and-test (push) Has been cancelled
Snyk Security Scan / Python vulnerabilities (backend) (push) Has been cancelled
Snyk Security Scan / npm vulnerabilities (frontend) (push) Has been cancelled
Snyk Security Scan / Docker image vulnerabilities (backend) (push) Has been cancelled
Test Catalog's pagination showed only 'Page N' with no way to know how many pages existed, and the campaign 'Add Test' modal's template picker had no pagination at all — with 2800+ templates in the catalog, its flat 100-row fetch made most templates unreachable unless a search happened to match. Adds GET /test-templates/count (open to any authenticated user, mirrors the list endpoint's filters) and wires real 'Page N of M' + total-available counts into both.
212 lines
8.1 KiB
Python
212 lines
8.1 KiB
Python
"""Validation tests for T-111: TestTemplates CRUD Router.
|
|
|
|
Tests the router structure, endpoint presence, and filter logic.
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import uuid
|
|
from unittest.mock import MagicMock
|
|
from types import ModuleType
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stubs
|
|
# ---------------------------------------------------------------------------
|
|
|
|
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:
|
|
pydantic_settings_mock = ModuleType("pydantic_settings")
|
|
class _BaseSettings:
|
|
def __init__(self, **kwargs): pass
|
|
def __init_subclass__(cls, **kwargs): super().__init_subclass__(**kwargs)
|
|
pydantic_settings_mock.BaseSettings = _BaseSettings
|
|
sys.modules["pydantic_settings"] = pydantic_settings_mock
|
|
|
|
if "app.config" not in sys.modules:
|
|
config_mod = 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
|
|
config_mod.settings = _FakeSettings()
|
|
sys.modules["app.config"] = config_mod
|
|
|
|
if "app.database" not in sys.modules:
|
|
db_mod = ModuleType("app.database")
|
|
db_mod.Base = type("Base", (), {"metadata": MagicMock()})
|
|
db_mod.get_db = MagicMock()
|
|
sys.modules["app.database"] = db_mod
|
|
|
|
for mod_name in [
|
|
"taxii2client", "taxii2client.v20",
|
|
"jose", "boto3", "botocore", "botocore.exceptions",
|
|
"apscheduler", "apscheduler.schedulers",
|
|
"apscheduler.schedulers.background",
|
|
"apscheduler.triggers", "apscheduler.triggers.cron",
|
|
]:
|
|
if mod_name not in sys.modules:
|
|
m = ModuleType(mod_name)
|
|
if mod_name == "taxii2client.v20": m.Server = MagicMock
|
|
elif mod_name == "jose": m.JWTError = Exception; m.jwt = MagicMock()
|
|
elif mod_name == "boto3": m.client = MagicMock()
|
|
elif mod_name == "botocore.exceptions": m.ClientError = Exception
|
|
elif mod_name == "apscheduler.schedulers.background": m.BackgroundScheduler = MagicMock
|
|
elif mod_name == "apscheduler.triggers.cron": m.CronTrigger = MagicMock
|
|
sys.modules[mod_name] = m
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Imports
|
|
# ---------------------------------------------------------------------------
|
|
|
|
from app.routers.test_templates import router
|
|
import inspect
|
|
|
|
|
|
def _get_route_paths():
|
|
routes = {}
|
|
for route in router.routes:
|
|
path = getattr(route, "path", "")
|
|
methods = getattr(route, "methods", set())
|
|
for method in methods:
|
|
routes[f"{method} {path}"] = route
|
|
return routes
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 1. GET /test-templates returns paginated list
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_list_endpoint_exists():
|
|
routes = _get_route_paths()
|
|
found = any("GET" in k and (k.endswith(" ") or k == "GET " or k == "GET /")
|
|
for k in routes) or any("GET" in k and "{template_id}" not in k and "by-technique" not in k for k in routes)
|
|
assert found, f"GET /test-templates not found. Routes: {list(routes.keys())}"
|
|
print(" [PASS] GET /test-templates returns paginated list")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 2. GET /test-templates?source=atomic_red_team filters by source
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_list_has_source_filter():
|
|
from app.routers.test_templates import list_templates
|
|
source = inspect.getsource(list_templates)
|
|
assert "source" in source and "filter" in source.lower()
|
|
print(" [PASS] GET /test-templates?source=atomic_red_team filters by source")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 3. GET /test-templates?platform=windows filters by platform
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_list_has_platform_filter():
|
|
from app.routers.test_templates import list_templates
|
|
source = inspect.getsource(list_templates)
|
|
assert "platform" in source and "filter" in source.lower()
|
|
print(" [PASS] GET /test-templates?platform=windows filters by platform")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 4. GET /test-templates/by-technique/T1059.001 returns technique templates
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_by_technique_endpoint():
|
|
routes = _get_route_paths()
|
|
found = any("by-technique" in k and "GET" in k for k in routes)
|
|
assert found, f"GET /test-templates/by-technique/{{mitre_id}} not found. Routes: {list(routes.keys())}"
|
|
print(" [PASS] GET /test-templates/by-technique/{mitre_id} endpoint exists")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 5. POST /test-templates only accessible by admin
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_create_admin_only():
|
|
from app.routers.test_templates import create_template
|
|
source = inspect.getsource(create_template)
|
|
assert "require_any_role" in source or "require_role" in source
|
|
print(" [PASS] POST /test-templates only accessible by admin")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 6. DELETE /test-templates/{id} does soft delete (is_active=False)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_soft_delete():
|
|
from app.routers.test_templates import delete_template
|
|
source = inspect.getsource(delete_template)
|
|
assert "is_active" in source and "False" in source
|
|
print(" [PASS] DELETE /test-templates/{id} does soft delete (is_active=False)")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 7. Search filter looks in name and description
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_search_filter():
|
|
from app.services.test_template_service import _build_template_query, list_templates
|
|
# The actual filter-building (including search) now lives in the shared
|
|
# _build_template_query() helper, reused by both list and count.
|
|
source = inspect.getsource(_build_template_query)
|
|
assert "search" in source
|
|
assert "name" in source and "description" in source
|
|
assert "ilike" in source
|
|
assert "_build_template_query" in inspect.getsource(list_templates)
|
|
print(" [PASS] Search filter searches in name and description")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Run all
|
|
# ---------------------------------------------------------------------------
|
|
|
|
if __name__ == "__main__":
|
|
print("T-111 Validation: TestTemplates CRUD Router")
|
|
print("=" * 50)
|
|
test_list_endpoint_exists()
|
|
test_list_has_source_filter()
|
|
test_list_has_platform_filter()
|
|
test_by_technique_endpoint()
|
|
test_create_admin_only()
|
|
test_soft_delete()
|
|
test_search_filter()
|
|
print("=" * 50)
|
|
print("ALL T-111 validations PASSED!")
|