test(jira): add hourly sync job tests [FASE-1.7]
Some checks failed
Aegis CI / lint-and-test (push) Has been cancelled
Some checks failed
Aegis CI / lint-and-test (push) Has been cancelled
Verify skip when disabled, per-link sync invocation, and continued batch on single-link failures.
This commit is contained in:
@@ -53,7 +53,7 @@ def create_link(
|
||||
audit_service.log_action(
|
||||
db,
|
||||
user_id=user.id,
|
||||
action="jira_link_created",
|
||||
action="JIRA_LINK_CREATED",
|
||||
entity_type="jira_link",
|
||||
entity_id=str(link.id),
|
||||
details={
|
||||
|
||||
68
backend/tests/test_jira_router.py
Normal file
68
backend/tests/test_jira_router.py
Normal file
@@ -0,0 +1,68 @@
|
||||
"""Jira router API tests (FASE-1.3)."""
|
||||
|
||||
import uuid
|
||||
|
||||
from app.models.audit import AuditLog
|
||||
from app.models.jira_link import JiraLink, JiraLinkEntityType
|
||||
|
||||
|
||||
def test_create_link_rejects_invalid_issue_key(client, admin_user, auth_headers):
|
||||
response = client.post(
|
||||
"/api/v1/jira/links",
|
||||
json={
|
||||
"entity_type": "test",
|
||||
"entity_id": str(uuid.uuid4()),
|
||||
"jira_issue_key": "invalid-key",
|
||||
},
|
||||
headers=auth_headers,
|
||||
)
|
||||
# Pydantic validation (422) or global validation handler (400)
|
||||
assert response.status_code in (400, 422)
|
||||
|
||||
|
||||
def test_create_link_returns_201(client, admin_user, auth_headers, db):
|
||||
entity_id = uuid.uuid4()
|
||||
response = client.post(
|
||||
"/api/v1/jira/links",
|
||||
json={
|
||||
"entity_type": "test",
|
||||
"entity_id": str(entity_id),
|
||||
"jira_issue_key": "SEC-1234",
|
||||
},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == 201
|
||||
data = response.json()
|
||||
assert data["jira_issue_key"] == "SEC-1234"
|
||||
assert data["entity_id"] == str(entity_id)
|
||||
|
||||
audit = (
|
||||
db.query(AuditLog)
|
||||
.filter(AuditLog.action == "JIRA_LINK_CREATED")
|
||||
.order_by(AuditLog.timestamp.desc())
|
||||
.first()
|
||||
)
|
||||
assert audit is not None
|
||||
assert audit.details.get("jira_issue_key") == "SEC-1234"
|
||||
|
||||
|
||||
def test_list_links_filters_by_entity(client, admin_user, auth_headers, db):
|
||||
entity_id = uuid.uuid4()
|
||||
link = JiraLink(
|
||||
entity_type=JiraLinkEntityType.test,
|
||||
entity_id=entity_id,
|
||||
jira_issue_key="SEC-555",
|
||||
created_by=admin_user.id,
|
||||
)
|
||||
db.add(link)
|
||||
db.commit()
|
||||
|
||||
response = client.get(
|
||||
"/api/v1/jira/links",
|
||||
params={"entity_type": "test", "entity_id": str(entity_id)},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
rows = response.json()
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["jira_issue_key"] == "SEC-555"
|
||||
64
backend/tests/test_worklog_service.py
Normal file
64
backend/tests/test_worklog_service.py
Normal file
@@ -0,0 +1,64 @@
|
||||
"""Worklog service tests — integrity hash (FASE-1.5)."""
|
||||
|
||||
from datetime import datetime
|
||||
from uuid import uuid4
|
||||
|
||||
from app.services import worklog_service
|
||||
|
||||
|
||||
def test_create_worklog_sets_integrity_hash(db, admin_user):
|
||||
entity_id = uuid4()
|
||||
started = datetime(2026, 5, 18, 9, 0, 0)
|
||||
wl = worklog_service.create_worklog(
|
||||
db,
|
||||
entity_type="test",
|
||||
entity_id=entity_id,
|
||||
user_id=admin_user.id,
|
||||
activity_type="red_team",
|
||||
started_at=started,
|
||||
duration_seconds=3600,
|
||||
description="Manual entry",
|
||||
)
|
||||
db.commit()
|
||||
db.refresh(wl)
|
||||
|
||||
assert wl.integrity_hash
|
||||
assert len(wl.integrity_hash) == 64
|
||||
assert worklog_service.verify_worklog_integrity(wl) is True
|
||||
|
||||
|
||||
def test_verify_worklog_integrity_detects_tampering(db, admin_user):
|
||||
wl = worklog_service.create_worklog(
|
||||
db,
|
||||
entity_type="test",
|
||||
entity_id=uuid4(),
|
||||
user_id=admin_user.id,
|
||||
activity_type="blue_validation",
|
||||
started_at=datetime(2026, 5, 18, 10, 0, 0),
|
||||
duration_seconds=1800,
|
||||
)
|
||||
db.commit()
|
||||
db.refresh(wl)
|
||||
|
||||
wl.duration_seconds = 9999
|
||||
assert worklog_service.verify_worklog_integrity(wl) is False
|
||||
|
||||
|
||||
def test_list_worklogs_filters_by_entity(db, admin_user):
|
||||
eid = uuid4()
|
||||
worklog_service.create_worklog(
|
||||
db,
|
||||
entity_type="campaign",
|
||||
entity_id=eid,
|
||||
user_id=admin_user.id,
|
||||
activity_type="reporting",
|
||||
started_at=datetime(2026, 5, 18, 11, 0, 0),
|
||||
duration_seconds=600,
|
||||
)
|
||||
db.commit()
|
||||
|
||||
rows = worklog_service.list_worklogs(
|
||||
db, entity_type="campaign", entity_id=eid,
|
||||
)
|
||||
assert len(rows) == 1
|
||||
assert rows[0].entity_id == eid
|
||||
Reference in New Issue
Block a user