Kitos
7af6be10be
T-106: Create test_workflow_service.py with state-machine transitions for the complete test lifecycle (draft -> red_executing -> blue_evaluating -> in_review -> validated/rejected), dual validation by Red/Blue leads, and reopen capability with field cleanup. T-107: Update status_service.py to use detection_result from Blue Team instead of legacy result field, and differentiate between partial progress (some validated) vs all-in-progress states. T-108: Create atomic_import_service.py that downloads the Atomic Red Team repo as a ZIP (avoiding API rate limits), parses all atomics YAML files, and creates idempotent TestTemplate records mapped to MITRE techniques. Includes validation tests for all three tasks (19 checks total).
55 lines
2.3 KiB
Python
55 lines
2.3 KiB
Python
"""add_test_templates_table
|
|
|
|
Revision ID: b004templates
|
|
Revises: b003dualvalid
|
|
Create Date: 2026-02-09 10:03:00.000000
|
|
|
|
"""
|
|
from typing import Sequence, Union
|
|
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
|
|
|
|
# revision identifiers, used by Alembic.
|
|
revision: str = 'b004templates'
|
|
down_revision: Union[str, Sequence[str], None] = 'b003dualvalid'
|
|
branch_labels: Union[str, Sequence[str], None] = None
|
|
depends_on: Union[str, Sequence[str], None] = None
|
|
|
|
|
|
def upgrade() -> None:
|
|
"""Create the test_templates table with indexes."""
|
|
op.create_table(
|
|
'test_templates',
|
|
sa.Column('id', sa.UUID(), nullable=False, default=sa.text('gen_random_uuid()')),
|
|
sa.Column('mitre_technique_id', sa.String(), nullable=False),
|
|
sa.Column('name', sa.String(), nullable=False),
|
|
sa.Column('description', sa.Text(), nullable=True),
|
|
sa.Column('source', sa.String(), nullable=False),
|
|
sa.Column('source_url', sa.String(), nullable=True),
|
|
sa.Column('attack_procedure', sa.Text(), nullable=True),
|
|
sa.Column('expected_detection', sa.Text(), nullable=True),
|
|
sa.Column('platform', sa.String(), nullable=True),
|
|
sa.Column('tool_suggested', sa.String(), nullable=True),
|
|
sa.Column('severity', sa.String(), nullable=True),
|
|
sa.Column('atomic_test_id', sa.String(), nullable=True),
|
|
sa.Column('is_active', sa.Boolean(), nullable=True, server_default=sa.text('true')),
|
|
sa.Column('created_at', sa.DateTime(), nullable=True),
|
|
sa.PrimaryKeyConstraint('id'),
|
|
)
|
|
|
|
op.create_index('ix_test_templates_mitre_technique_id', 'test_templates', ['mitre_technique_id'])
|
|
op.create_index('ix_test_templates_source', 'test_templates', ['source'])
|
|
op.create_index('ix_test_templates_platform', 'test_templates', ['platform'])
|
|
op.create_index('ix_test_templates_severity', 'test_templates', ['severity'])
|
|
|
|
|
|
def downgrade() -> None:
|
|
"""Drop the test_templates table and its indexes."""
|
|
op.drop_index('ix_test_templates_severity', table_name='test_templates')
|
|
op.drop_index('ix_test_templates_platform', table_name='test_templates')
|
|
op.drop_index('ix_test_templates_source', table_name='test_templates')
|
|
op.drop_index('ix_test_templates_mitre_technique_id', table_name='test_templates')
|
|
op.drop_table('test_templates')
|