feat(jira): pause/hold status mapping, RT/BT dates, TTP, attack_success 3-value field, hours

This commit is contained in:
kitos
2026-07-06 16:19:39 +02:00
parent 19c4866103
commit 022c1c756a
13 changed files with 433 additions and 22 deletions
@@ -0,0 +1,48 @@
"""Convert tests.attack_success from Boolean to a 3-value enum.
Revision ID: b058
Revises: b057
Create Date: 2026-07-06
"""
from alembic import op
import sqlalchemy as sa
revision = "b058"
down_revision = "b057"
branch_labels = None
depends_on = None
_ENUM_NAME = "attacksuccessresult"
def upgrade() -> None:
attack_success_enum = sa.Enum(
"successful", "not_successful", "partially_successful", name=_ENUM_NAME
)
attack_success_enum.create(op.get_bind(), checkfirst=True)
op.add_column("tests", sa.Column("attack_success_new", attack_success_enum, nullable=True))
op.execute(
"UPDATE tests SET attack_success_new = CASE "
"WHEN attack_success = true THEN 'successful' "
"WHEN attack_success = false THEN 'not_successful' "
"ELSE NULL END"
)
op.drop_column("tests", "attack_success")
op.alter_column("tests", "attack_success_new", new_column_name="attack_success")
def downgrade() -> None:
op.add_column("tests", sa.Column("attack_success_bool", sa.Boolean(), nullable=True))
op.execute(
"UPDATE tests SET attack_success_bool = CASE "
"WHEN attack_success = 'successful' THEN true "
"WHEN attack_success = 'not_successful' THEN false "
"ELSE NULL END"
)
op.drop_column("tests", "attack_success")
op.alter_column("tests", "attack_success_bool", new_column_name="attack_success")
sa.Enum(name=_ENUM_NAME).drop(op.get_bind(), checkfirst=True)
+8
View File
@@ -81,6 +81,14 @@ class ContainmentResult(str, enum.Enum):
not_contained = "not_contained" not_contained = "not_contained"
class AttackSuccessResult(str, enum.Enum):
"""Outcome of a red-team attack from a success perspective."""
successful = "successful"
not_successful = "not_successful"
partially_successful = "partially_successful"
# Define class DataClassification # Define class DataClassification
class DataClassification(str, enum.Enum): class DataClassification(str, enum.Enum):
"""Data sensitivity classification levels for compliance and retention policies.""" """Data sensitivity classification levels for compliance and retention policies."""
+1
View File
@@ -7,6 +7,7 @@ working with ``from app.models.enums import ...``.
# Import # noqa: F401 from app.domain.enums # Import # noqa: F401 from app.domain.enums
from app.domain.enums import ( # noqa: F401 from app.domain.enums import ( # noqa: F401
AttackSuccessResult,
ContainmentResult, ContainmentResult,
DataClassification, DataClassification,
TeamSide, TeamSide,
+2 -2
View File
@@ -27,7 +27,7 @@ from sqlalchemy.orm import relationship
from app.database import Base from app.database import Base
# Import ContainmentResult, TestResult, TestState from app.models.enums # Import ContainmentResult, TestResult, TestState from app.models.enums
from app.models.enums import ContainmentResult, TestResult, TestState from app.models.enums import AttackSuccessResult, ContainmentResult, TestResult, TestState
# Define class Test # Define class Test
@@ -69,7 +69,7 @@ class Test(Base):
# ── Red Team fields ───────────────────────────────────────────── # ── Red Team fields ─────────────────────────────────────────────
red_summary = Column(Text, nullable=True) red_summary = Column(Text, nullable=True)
# Assign attack_success = Column(Boolean, nullable=True) # Assign attack_success = Column(Boolean, nullable=True)
attack_success = Column(Boolean, nullable=True) attack_success = Column(Enum(AttackSuccessResult, name="attacksuccessresult"), nullable=True)
execution_start_time = Column(DateTime, nullable=True) execution_start_time = Column(DateTime, nullable=True)
execution_end_time = Column(DateTime, nullable=True) execution_end_time = Column(DateTime, nullable=True)
# Assign red_validated_by = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=True) # Assign red_validated_by = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=True)
+2 -2
View File
@@ -49,7 +49,7 @@ from app.domain.unit_of_work import UnitOfWork
# Import limiter from app.limiter # Import limiter from app.limiter
from app.limiter import limiter from app.limiter import limiter
from app.models.enums import TestState, TestResult, TeamSide from app.models.enums import AttackSuccessResult, TestState, TestResult, TeamSide
from app.models.evidence import Evidence from app.models.evidence import Evidence
from app.storage import upload_file from app.storage import upload_file
from app.models.technique import Technique from app.models.technique import Technique
@@ -1687,7 +1687,7 @@ class RTEvidenceEntry(BaseModel):
class RTTechniqueEntry(BaseModel): class RTTechniqueEntry(BaseModel):
mitre_id: str mitre_id: str
result: str # "detected" | "not_detected" | "partially_detected" result: str # "detected" | "not_detected" | "partially_detected"
attack_success: bool = True attack_success: AttackSuccessResult = AttackSuccessResult.successful
platform: Optional[str] = None platform: Optional[str] = None
notes: Optional[str] = None notes: Optional[str] = None
evidence: list[RTEvidenceEntry] # REQUIRED — at least one image per technique evidence: list[RTEvidenceEntry] # REQUIRED — at least one image per technique
+3 -3
View File
@@ -9,7 +9,7 @@ from datetime import datetime
from pydantic import BaseModel, ConfigDict, model_validator from pydantic import BaseModel, ConfigDict, model_validator
# Import DataClassification from app.domain.enums # Import DataClassification from app.domain.enums
from app.domain.enums import ContainmentResult, DataClassification from app.domain.enums import AttackSuccessResult, ContainmentResult, DataClassification
from app.models.enums import TestResult, TestState from app.models.enums import TestResult, TestState
from app.schemas.evidence import EvidenceOut from app.schemas.evidence import EvidenceOut
@@ -79,7 +79,7 @@ class TestRedUpdate(BaseModel):
# Assign tool_used = None # Assign tool_used = None
tool_used: str | None = None tool_used: str | None = None
# Assign attack_success = None # Assign attack_success = None
attack_success: bool | None = None attack_success: AttackSuccessResult | None = None
# Assign red_summary = None # Assign red_summary = None
red_summary: str | None = None red_summary: str | None = None
execution_start_time: datetime | None = None execution_start_time: datetime | None = None
@@ -229,7 +229,7 @@ class TestOut(BaseModel):
# Red Team fields # Red Team fields
red_summary: str | None = None red_summary: str | None = None
# Assign attack_success = None # Assign attack_success = None
attack_success: bool | None = None attack_success: AttackSuccessResult | None = None
execution_start_time: datetime | None = None execution_start_time: datetime | None = None
execution_end_time: datetime | None = None execution_end_time: datetime | None = None
# Assign red_validated_by = None # Assign red_validated_by = None
+8 -3
View File
@@ -35,7 +35,7 @@ from app.database import SessionLocal
from app.models.audit import AuditLog from app.models.audit import AuditLog
# Import TeamSide, TechniqueStatus, TestResult, TestState from app.models.enums # Import TeamSide, TechniqueStatus, TestResult, TestState from app.models.enums
from app.models.enums import TeamSide, TechniqueStatus, TestResult, TestState from app.models.enums import AttackSuccessResult, TeamSide, TechniqueStatus, TestResult, TestState
# Import Evidence from app.models.evidence # Import Evidence from app.models.evidence
from app.models.evidence import Evidence from app.models.evidence import Evidence
@@ -354,8 +354,13 @@ def _seed_tests(db: Session, users: list[User], techniques: list[Technique], cou
if state in (TestState.blue_evaluating, TestState.in_review, TestState.validated, TestState.rejected): if state in (TestState.blue_evaluating, TestState.in_review, TestState.validated, TestState.rejected):
# Assign test.red_summary = f"Attack executed successfully using {test.tool_used}." # Assign test.red_summary = f"Attack executed successfully using {test.tool_used}."
test.red_summary = f"Attack executed successfully using {test.tool_used}." test.red_summary = f"Attack executed successfully using {test.tool_used}."
# Assign test.attack_success = random.choice([True, True, True, False]) test.attack_success = random.choice([
test.attack_success = random.choice([True, True, True, False]) AttackSuccessResult.successful,
AttackSuccessResult.successful,
AttackSuccessResult.successful,
AttackSuccessResult.partially_successful,
AttackSuccessResult.not_successful,
])
# Check: state in (TestState.in_review, TestState.validated, TestState.rejec... # Check: state in (TestState.in_review, TestState.validated, TestState.rejec...
if state in (TestState.in_review, TestState.validated, TestState.rejected): if state in (TestState.in_review, TestState.validated, TestState.rejected):
+1 -1
View File
@@ -105,7 +105,7 @@ def get_tests_analytics(
# Literal argument value # Literal argument value
"tool_used": t.tool_used, "tool_used": t.tool_used,
# Literal argument value # Literal argument value
"attack_success": t.attack_success, "attack_success": t.attack_success.value if t.attack_success else None,
# Literal argument value # Literal argument value
"remediation_status": t.remediation_status, "remediation_status": t.remediation_status,
} }
@@ -35,7 +35,7 @@ from typing import Any
import requests import requests
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from app.models.enums import TestState, TestResult from app.models.enums import AttackSuccessResult, TestState, TestResult
from app.models.evaluation_import import EvaluationImport from app.models.evaluation_import import EvaluationImport
from app.models.technique import Technique from app.models.technique import Technique
from app.models.test import Test from app.models.test import Test
@@ -625,7 +625,7 @@ def import_evaluation_round(
procedure_text=procedure_text, procedure_text=procedure_text,
created_by=current_user.id, created_by=current_user.id,
state=TestState.in_review, state=TestState.in_review,
attack_success=True, attack_success=AttackSuccessResult.successful,
red_summary=red_summary, red_summary=red_summary,
red_validation_status="approved", red_validation_status="approved",
red_validated_by=current_user.id, red_validated_by=current_user.id,
@@ -305,7 +305,10 @@ def build_test_results_report(
# Literal argument value # Literal argument value
"platform": t.platform, "platform": t.platform,
# Literal argument value # Literal argument value
"attack_success": t.attack_success, "attack_success": (
t.attack_success.value if t.attack_success and hasattr(t.attack_success, "value")
else str(t.attack_success) if t.attack_success else None
),
# Literal argument value # Literal argument value
"detection_result": ( "detection_result": (
t.detection_result.value if t.detection_result and hasattr(t.detection_result, "value") t.detection_result.value if t.detection_result and hasattr(t.detection_result, "value")
+179 -8
View File
@@ -65,6 +65,33 @@ from app.models.technique import Technique
from app.models.test import Test from app.models.test import Test
from app.models.user import User from app.models.user import User
# ---------------------------------------------------------------------------
# Custom field IDs (project-specific — see System Settings → Jira Configuration)
# ---------------------------------------------------------------------------
JIRA_FIELD_RT_START_DATE = "customfield_11803"
JIRA_FIELD_RT_END_DATE = "customfield_11804"
JIRA_FIELD_BT_START_DATE = "customfield_11805"
JIRA_FIELD_BT_END_DATE = "customfield_11806"
JIRA_FIELD_TTP = "customfield_11807"
JIRA_FIELD_SEVERITY = "customfield_10098"
JIRA_FIELD_ATTACK_SUCCESS = "customfield_11815"
JIRA_FIELD_ATTACK_DETECTED = "customfield_11809"
JIRA_FIELD_TIME_TO_DETECT = "customfield_11810"
JIRA_FIELD_TIME_TO_CONTAIN = "customfield_11811"
_ATTACK_SUCCESS_TO_JIRA = {
"successful": "Yes",
"not_successful": "No",
"partially_successful": "Partial",
}
_DETECTION_TO_JIRA = {
"detected": "Yes",
"not_detected": "No",
"partially_detected": "Partial",
}
# Assign logger = logging.getLogger(__name__) # Assign logger = logging.getLogger(__name__)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -388,7 +415,7 @@ def _build_state_comment(
lines += [ lines += [
"Red Team has finished execution and submitted evidence for Blue Team evaluation.", "Red Team has finished execution and submitted evidence for Blue Team evaluation.",
"", "",
f"*Attack Success:* {test.attack_success if test.attack_success is not None else 'N/A'}", f"*Attack Success:* {_enum_value(test.attack_success) or 'N/A'}",
] ]
if test.red_summary: if test.red_summary:
lines += ["", "h4. Red Team Summary", test.red_summary] lines += ["", "h4. Red Team Summary", test.red_summary]
@@ -627,6 +654,7 @@ def auto_create_test_issue(
issue_type = settings.JIRA_ISSUE_TYPE_TEST # always Task issue_type = settings.JIRA_ISSUE_TYPE_TEST # always Task
poc = test.procedure_text or "N/A" poc = test.procedure_text or "N/A"
severity = _technique_severity(technique).capitalize()
fields: dict = { fields: dict = {
"project": {"key": project_key}, "project": {"key": project_key},
"summary": f"[Aegis] {mitre_id}{test.name}", "summary": f"[Aegis] {mitre_id}{test.name}",
@@ -635,6 +663,8 @@ def auto_create_test_issue(
"labels": ["aegis", "security-test", mitre_id.replace(".", "-")], "labels": ["aegis", "security-test", mitre_id.replace(".", "-")],
# customfield_10309 = Proof of Concept field (required by team's Jira config) # customfield_10309 = Proof of Concept field (required by team's Jira config)
"customfield_10309": f"{{code}}{poc}{{code}}", "customfield_10309": f"{{code}}{poc}{{code}}",
JIRA_FIELD_TTP: mitre_id,
JIRA_FIELD_SEVERITY: severity,
} }
# Inherit campaign start date if available, otherwise use today # Inherit campaign start date if available, otherwise use today
@@ -803,20 +833,67 @@ def push_hold_event(
f"h3. ▶ Test Resumed\n\n" f"h3. ▶ Test Resumed\n\n"
f"*Resumed by:* {actor.username}\n" f"*Resumed by:* {actor.username}\n"
f"*At:* {ts}\n\n" f"*At:* {ts}\n\n"
f"_The test has been taken off hold and is active again._" f"_The test has been unblocked and is active again._"
) )
try: try:
jira.set_issue_status(link.jira_issue_key, "In Progress") jira.set_issue_status(link.jira_issue_key, "In Progress")
except Exception as exc_t: except Exception as exc_t:
logger.warning("Could not transition %s off hold: %s", link.jira_issue_key, exc_t) logger.warning("Could not transition %s off blocked: %s", link.jira_issue_key, exc_t)
else: else:
comment = ( comment = (
f"h3. Test Placed On Hold\n\n" f"h3. 🚫 Test Blocked\n\n"
f"*Placed on hold by:* {actor.username}\n" f"*Blocked by:* {actor.username}\n"
f"*At:* {ts}\n" f"*At:* {ts}\n"
f"*Reason:* {reason or 'No reason provided'}\n\n" f"*Reason:* {reason or 'No reason provided'}\n\n"
f"_This test has been paused. No action required until it is resumed._" f"_This test is blocked. No action required until it is resumed._"
) )
try:
jira.set_issue_status(link.jira_issue_key, "Blocked")
except Exception as exc_t:
logger.warning("Could not transition %s to Blocked: %s", link.jira_issue_key, exc_t)
jira.issue_add_comment(link.jira_issue_key, comment)
link.last_synced_at = datetime.utcnow()
db.flush()
logger.info("Posted hold event to Jira %s (resuming=%s)", link.jira_issue_key, resuming)
except Exception as exc:
logger.warning("Failed to push hold event for test %s: %s", test.id, exc, exc_info=True)
def push_pause_event(db: Session, test, actor, *, resuming: bool = False) -> None:
"""Post a timer pause/resume comment to Jira and toggle 'On Hold' / 'In Progress'.
Distinct from :func:`push_hold_event` — a short timer pause (the
operator stepping away briefly) maps to "On Hold", while an explicit
hold (something external blocking progress) maps to "Blocked".
Non-fatal — any Jira error is logged and swallowed.
"""
if not has_admin_jira_configured(db):
return
link = (
db.query(JiraLink)
.filter(
JiraLink.entity_type == JiraLinkEntityType.test,
JiraLink.entity_id == test.id,
)
.first()
)
if not link:
return
try:
jira = get_admin_jira_client(db)
ts = datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC")
if resuming:
comment = f"h3. ▶ Timer Resumed\n\n*Resumed by:* {actor.username}\n*At:* {ts}"
try:
jira.set_issue_status(link.jira_issue_key, "In Progress")
except Exception as exc_t:
logger.warning("Could not transition %s off On Hold: %s", link.jira_issue_key, exc_t)
else:
comment = f"h3. ⏸ Timer Paused\n\n*Paused by:* {actor.username}\n*At:* {ts}"
try: try:
jira.set_issue_status(link.jira_issue_key, "On Hold") jira.set_issue_status(link.jira_issue_key, "On Hold")
except Exception as exc_t: except Exception as exc_t:
@@ -825,9 +902,103 @@ def push_hold_event(
jira.issue_add_comment(link.jira_issue_key, comment) jira.issue_add_comment(link.jira_issue_key, comment)
link.last_synced_at = datetime.utcnow() link.last_synced_at = datetime.utcnow()
db.flush() db.flush()
logger.info("Posted hold event to Jira %s (resuming=%s)", link.jira_issue_key, resuming) logger.info("Posted pause event to Jira %s (resuming=%s)", link.jira_issue_key, resuming)
except Exception as exc: except Exception as exc:
logger.warning("Failed to push hold event for test %s: %s", test.id, exc, exc_info=True) logger.warning("Failed to push pause event for test %s: %s", test.id, exc, exc_info=True)
# ---------------------------------------------------------------------------
# RT/BT date + result custom-field sync
# ---------------------------------------------------------------------------
def _enum_value(v) -> Optional[str]:
"""Return the plain string value of an enum member, or the value itself."""
if v is None:
return None
return v.value if hasattr(v, "value") else str(v)
def _compute_hours(start: Optional[datetime], end: Optional[datetime]) -> Optional[float]:
"""Return the elapsed time between two timestamps in hours, or None if unavailable."""
if not start or not end:
return None
delta_hours = (end - start).total_seconds() / 3600
return round(delta_hours, 2) if delta_hours >= 0 else None
def _update_test_fields(db: Session, test: Test, fields: dict) -> None:
"""Best-effort update of custom fields on the Jira issue linked to *test*.
Non-fatal — any Jira error is logged and swallowed so it never blocks
the test workflow.
"""
if not fields or not has_admin_jira_configured(db):
return
link = (
db.query(JiraLink)
.filter(
JiraLink.entity_type == JiraLinkEntityType.test,
JiraLink.entity_id == test.id,
)
.first()
)
if not link:
return
try:
jira = get_admin_jira_client(db)
jira.update_issue_field(link.jira_issue_key, fields=fields)
link.last_synced_at = datetime.utcnow()
db.flush()
logger.info("Updated Jira fields %s on %s", list(fields.keys()), link.jira_issue_key)
except Exception as exc:
logger.warning(
"Failed to update Jira fields for test %s: %s", test.id, exc, exc_info=True
)
def push_rt_started(db: Session, test: Test) -> None:
"""Set the RT Start Date field — called when the operator hits Start Execution."""
_update_test_fields(db, test, {
JIRA_FIELD_RT_START_DATE: datetime.utcnow().strftime("%Y-%m-%d"),
})
def push_rt_submitted(db: Session, test: Test) -> None:
"""Set the RT End Date + Attack Success fields — called when Red submits for review."""
fields: dict = {JIRA_FIELD_RT_END_DATE: datetime.utcnow().strftime("%Y-%m-%d")}
attack_success = _enum_value(test.attack_success)
if attack_success in _ATTACK_SUCCESS_TO_JIRA:
fields[JIRA_FIELD_ATTACK_SUCCESS] = _ATTACK_SUCCESS_TO_JIRA[attack_success]
_update_test_fields(db, test, fields)
def push_bt_started(db: Session, test: Test) -> None:
"""Set the BT Start Date field — called when Blue picks up the test to evaluate."""
_update_test_fields(db, test, {
JIRA_FIELD_BT_START_DATE: datetime.utcnow().strftime("%Y-%m-%d"),
})
def push_bt_submitted(db: Session, test: Test) -> None:
"""Set BT End Date, Attack Detected, and time-to-detect/contain — Blue submits for review."""
fields: dict = {JIRA_FIELD_BT_END_DATE: datetime.utcnow().strftime("%Y-%m-%d")}
detection_result = _enum_value(test.detection_result)
if detection_result in _DETECTION_TO_JIRA:
fields[JIRA_FIELD_ATTACK_DETECTED] = _DETECTION_TO_JIRA[detection_result]
time_to_detect = _compute_hours(test.execution_end_time, test.detection_time)
if time_to_detect is not None:
fields[JIRA_FIELD_TIME_TO_DETECT] = time_to_detect
time_to_contain = _compute_hours(test.detection_time, test.containment_time)
if time_to_contain is not None:
fields[JIRA_FIELD_TIME_TO_CONTAIN] = time_to_contain
_update_test_fields(db, test, fields)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -223,6 +223,12 @@ def start_execution(db: Session, test: Test, user: User) -> Test:
except Exception as e: except Exception as e:
logger.warning("Jira push failed for test %s: %s", test.id, e, exc_info=True) logger.warning("Jira push failed for test %s: %s", test.id, e, exc_info=True)
try:
from app.services.jira_service import push_rt_started
push_rt_started(db, test)
except Exception as e:
logger.warning("Jira RT start date push failed for test %s: %s", test.id, e, exc_info=True)
return test return test
@@ -314,6 +320,12 @@ def submit_red_evidence(db: Session, test: Test, user: User) -> Test:
except Exception as e: except Exception as e:
logger.warning("Jira push failed for test %s: %s", test.id, e, exc_info=True) logger.warning("Jira push failed for test %s: %s", test.id, e, exc_info=True)
try:
from app.services.jira_service import push_rt_submitted
push_rt_submitted(db, test)
except Exception as e:
logger.warning("Jira RT end date push failed for test %s: %s", test.id, e, exc_info=True)
return test return test
@@ -410,6 +422,12 @@ def start_blue_work(db: Session, test: Test, user: User) -> Test:
except Exception as e: except Exception as e:
logger.warning("Notification failed for test %s: %s", test.id, e, exc_info=True) logger.warning("Notification failed for test %s: %s", test.id, e, exc_info=True)
try:
from app.services.jira_service import push_bt_started
push_bt_started(db, test)
except Exception as e:
logger.warning("Jira BT start date push failed for test %s: %s", test.id, e, exc_info=True)
return test return test
@@ -494,6 +512,12 @@ def submit_blue_evidence(db: Session, test: Test, user: User) -> Test:
except Exception as e: except Exception as e:
logger.warning("Jira push failed for test %s: %s", test.id, e, exc_info=True) logger.warning("Jira push failed for test %s: %s", test.id, e, exc_info=True)
try:
from app.services.jira_service import push_bt_submitted
push_bt_submitted(db, test)
except Exception as e:
logger.warning("Jira BT end date push failed for test %s: %s", test.id, e, exc_info=True)
return test return test
@@ -637,6 +661,13 @@ def pause_timer(db: Session, test: Test, user: User) -> Test:
# Keyword argument: details # Keyword argument: details
details={"state": test.state.value}, details={"state": test.state.value},
) )
try:
from app.services.jira_service import push_pause_event
push_pause_event(db, test, user, resuming=False)
except Exception as e:
logger.warning("Jira pause push failed for test %s: %s", test.id, e, exc_info=True)
# Return test # Return test
return test return test
@@ -692,6 +723,13 @@ def resume_timer(db: Session, test: Test, user: User) -> Test:
# Keyword argument: details # Keyword argument: details
details={"paused_seconds": paused_seconds, "state": test.state.value}, details={"paused_seconds": paused_seconds, "state": test.state.value},
) )
try:
from app.services.jira_service import push_pause_event
push_pause_event(db, test, user, resuming=True)
except Exception as e:
logger.warning("Jira resume push failed for test %s: %s", test.id, e, exc_info=True)
# Return test # Return test
return test return test
+137
View File
@@ -37,6 +37,143 @@ def test_sync_aegis_to_jira_adds_comment(mock_get_client, db):
assert link.last_synced_at is not None assert link.last_synced_at is not None
def _make_link(entity_id, issue_key="SEC-100"):
return JiraLink(
entity_type=JiraLinkEntityType.test,
entity_id=entity_id,
jira_issue_key=issue_key,
)
def _make_test(**overrides):
from app.models.test import Test
t = MagicMock(spec=Test)
t.id = uuid4()
t.name = "Test"
t.attack_success = None
t.detection_result = None
t.execution_end_time = None
t.detection_time = None
t.containment_time = None
for k, v in overrides.items():
setattr(t, k, v)
return t
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_admin_jira_client")
def test_push_pause_event_sets_on_hold(mock_get_client, mock_configured, db):
mock_jira = MagicMock()
mock_get_client.return_value = mock_jira
test = _make_test()
link = _make_link(test.id)
db.add(link)
db.commit()
jira_service.push_pause_event(db, test, MagicMock(username="alice"), resuming=False)
mock_jira.set_issue_status.assert_called_once_with(link.jira_issue_key, "On Hold")
mock_jira.issue_add_comment.assert_called_once()
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_admin_jira_client")
def test_push_pause_event_resume_sets_in_progress(mock_get_client, mock_configured, db):
mock_jira = MagicMock()
mock_get_client.return_value = mock_jira
test = _make_test()
link = _make_link(test.id)
db.add(link)
db.commit()
jira_service.push_pause_event(db, test, MagicMock(username="alice"), resuming=True)
mock_jira.set_issue_status.assert_called_once_with(link.jira_issue_key, "In Progress")
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_admin_jira_client")
def test_push_hold_event_sets_blocked(mock_get_client, mock_configured, db):
mock_jira = MagicMock()
mock_get_client.return_value = mock_jira
test = _make_test()
link = _make_link(test.id)
db.add(link)
db.commit()
jira_service.push_hold_event(db, test, MagicMock(username="alice"), resuming=False, reason="waiting on approval")
mock_jira.set_issue_status.assert_called_once_with(link.jira_issue_key, "Blocked")
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_admin_jira_client")
def test_push_rt_started_sets_date_field(mock_get_client, mock_configured, db):
mock_jira = MagicMock()
mock_get_client.return_value = mock_jira
test = _make_test()
link = _make_link(test.id)
db.add(link)
db.commit()
jira_service.push_rt_started(db, test)
mock_jira.update_issue_field.assert_called_once()
args, kwargs = mock_jira.update_issue_field.call_args
assert args[0] == link.jira_issue_key
assert jira_service.JIRA_FIELD_RT_START_DATE in kwargs["fields"]
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_admin_jira_client")
def test_push_rt_submitted_includes_attack_success(mock_get_client, mock_configured, db):
mock_jira = MagicMock()
mock_get_client.return_value = mock_jira
test = _make_test(attack_success=MagicMock(value="successful"))
link = _make_link(test.id)
db.add(link)
db.commit()
jira_service.push_rt_submitted(db, test)
fields = mock_jira.update_issue_field.call_args[1]["fields"]
assert fields[jira_service.JIRA_FIELD_RT_END_DATE]
assert fields[jira_service.JIRA_FIELD_ATTACK_SUCCESS] == "Yes"
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_admin_jira_client")
def test_push_bt_submitted_includes_detection_and_hours(mock_get_client, mock_configured, db):
mock_jira = MagicMock()
mock_get_client.return_value = mock_jira
exec_end = datetime(2026, 1, 1, 10, 0, 0)
detect = datetime(2026, 1, 1, 12, 0, 0)
contain = datetime(2026, 1, 1, 13, 30, 0)
test = _make_test(
detection_result=MagicMock(value="detected"),
execution_end_time=exec_end,
detection_time=detect,
containment_time=contain,
)
link = _make_link(test.id)
db.add(link)
db.commit()
jira_service.push_bt_submitted(db, test)
fields = mock_jira.update_issue_field.call_args[1]["fields"]
assert fields[jira_service.JIRA_FIELD_ATTACK_DETECTED] == "Yes"
assert fields[jira_service.JIRA_FIELD_TIME_TO_DETECT] == 2.0
assert fields[jira_service.JIRA_FIELD_TIME_TO_CONTAIN] == 1.5
def test_update_test_fields_noop_when_not_configured(db):
test = _make_test()
# has_admin_jira_configured defaults to False without mocking — should
# silently no-op rather than raise.
jira_service._update_test_fields(db, test, {"customfield_1": "x"})
@patch("app.services.jira_service.get_jira_client") @patch("app.services.jira_service.get_jira_client")
def test_search_jira_issues_maps_fields(mock_get_client): def test_search_jira_issues_maps_fields(mock_get_client):
mock_jira = MagicMock() mock_jira = MagicMock()