T-109: Rewrite tests router with full Red/Blue workflow endpoints - list with filters, create from template, Red/Blue team updates with state guards, start-execution, submit-red, submit-blue, validate-red, validate-blue, reopen, and timeline. All using workflow service from Phase 11. T-110: Rewrite evidence router with Red/Blue separation - upload with team field, list with team filter, delete with state-based permissions. Red Team edits in draft/red_executing, Blue Team in blue_evaluating, admin bypasses all. T-111: Create test_templates router with full CRUD - paginated list with source/platform/severity/search filters, by-technique lookup, admin-only create/update, and soft delete. Registered in main.py. T-112: Add POST /system/import-atomic-tests endpoint to system router - admin-only trigger for Atomic Red Team import with error handling and statistics response. Includes validation tests for all four tasks (35 checks total).
186 lines
7.0 KiB
Python
186 lines
7.0 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"
|
|
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_role("admin")' 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.routers.test_templates import list_templates
|
|
source = inspect.getsource(list_templates)
|
|
assert "search" in source
|
|
assert "name" in source and "description" in source
|
|
assert "ilike" in source
|
|
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!")
|