ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-10-09 21:14:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(integrations): allow sending findings to Jira from the API (#8645)

Browse Source
This commit is contained in:
Víctor Fernández Poyatos
2025-09-05 14:28:34 +02:00
committed by GitHub
parent f87b2089fb
commit 43fe9c6860
11 changed files with 702 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/backend/api/filters.py
+20
View File
@@ -860,3 +860,23 @@ class ProcessorFilter(FilterSet):
field_name="processor_type",
lookup_expr="in",
)
class IntegrationJiraFindingsFilter(FilterSet):
# To be expanded as needed
finding_id = UUIDFilter(field_name="id", lookup_expr="exact")
finding_id__in = UUIDInFilter(field_name="id", lookup_expr="in")
class Meta:
model = Finding
fields = {}
def filter_queryset(self, queryset):
# Validate that there is at least one filter provided
if not self.data:
raise ValidationError(
{
"findings": "No finding filters provided. At least one filter is required."
}
)
return super().filter_queryset(queryset)
api/src/backend/api/specs/v1.yaml
+98
View File
@@ -2846,6 +2846,71 @@ paths:
schema:
$ref: '#/components/schemas/IntegrationCreateResponse'
description: ''
/api/v1/integrations/{integration_pk}/jira/dispatches:
post:
operationId: integrations_jira_dispatches_create
description: Send a set of filtered findings to the given integration. At least
one finding filter must be provided.
summary: Send findings to a Jira integration
parameters:
- in: query
name: filter[finding_id]
schema:
type: string
format: uuid
- in: query
name: filter[finding_id__in]
schema:
type: array
items:
type: string
format: uuid
description: Multiple values may be separated by commas.
explode: false
style: form
- in: path
name: integration_pk
schema:
type: string
required: true
tags:
- Integration
requestBody:
content:
application/vnd.api+json:
schema:
$ref: '#/components/schemas/IntegrationJiraDispatchRequest'
application/x-www-form-urlencoded:
schema:
$ref: '#/components/schemas/IntegrationJiraDispatchRequest'
multipart/form-data:
schema:
$ref: '#/components/schemas/IntegrationJiraDispatchRequest'
required: true
security:
- jwtAuth: []
responses:
'202':
content:
application/vnd.api+json:
schema:
$ref: '#/components/schemas/TaskResponse'
examples:
Task queued:
summary: Task queued
value:
data:
type: tasks
id: 497f6eca-6276-4993-bfeb-53cbbbba6f08
attributes:
inserted_at: '2019-08-24T14:15:22Z'
completed_at: '2019-08-24T14:15:22Z'
name: string
state: available
result: null
task_args: null
metadata: null
description: ''
/api/v1/integrations/{id}:
get:
operationId: integrations_retrieve
@@ -9444,6 +9509,39 @@ components:
$ref: '#/components/schemas/IntegrationCreate'
required:
- data
IntegrationJiraDispatchRequest:
type: object
properties:
data:
type: object
required:
- type
additionalProperties: false
properties:
type:
type: string
description: The [type](https://jsonapi.org/format/#document-resource-object-identification)
member is used to describe resource objects that share common attributes
and relationships.
enum:
- integrations-jira-dispatches
attributes:
type: object
properties:
project_key:
type: string
minLength: 1
issue_type:
enum:
- Task
type: string
description: '* `Task` - Task'
x-spec-enum-id: b527b0cec62087c1
required:
- project_key
- issue_type
required:
- data
IntegrationResponse:
type: object
properties:
api/src/backend/api/tests/test_utils.py
+2 -2
View File
@@ -617,7 +617,7 @@ class TestProwlerIntegrationConnectionTest:
"domain": "example.atlassian.net",
}
integration.configuration = {
"issue_types": ["Bug", "Task"], # Existing configuration
"issue_types": ["Task"], # Existing configuration
"projects": {"OLD_PROJ": "Old Project"}, # Will be overwritten
}
@@ -647,7 +647,7 @@ class TestProwlerIntegrationConnectionTest:
}
# Verify other configuration fields were preserved
assert integration.configuration["issue_types"] == ["Bug", "Task"]
assert integration.configuration["issue_types"] == ["Task"]
# Verify integration.save() was called
integration.save.assert_called_once()
api/src/backend/api/utils.py
+3 -12
View File
@@ -10,7 +10,7 @@ from api.db_utils import rls_transaction
from api.exceptions import InvitationTokenExpiredException
from api.models import Integration, Invitation, Processor, Provider, Resource
from api.v1.serializers import FindingMetadataSerializer
from prowler.lib.outputs.jira.jira import Jira, JiraBasicAuthError, JiraNoProjectsError
from prowler.lib.outputs.jira.jira import Jira, JiraBasicAuthError
from prowler.providers.aws.aws_provider import AwsProvider
from prowler.providers.aws.lib.s3.s3 import S3
from prowler.providers.aws.lib.security_hub.security_hub import SecurityHub
@@ -353,20 +353,11 @@ def initialize_prowler_integration(integration: Integration) -> Jira:
# TODO Refactor other integrations to use this function
if integration.integration_type == Integration.IntegrationChoices.JIRA:
try:
return Jira(
**integration.credentials, domain=integration.configuration["domain"]
)
return Jira(**integration.credentials)
except JiraBasicAuthError as jira_auth_error:
with rls_transaction(str(integration.tenant_id)):
integration.configuration["projects"] = {}
integration.connected = False
integration.connection_last_checked_at = datetime.now(tz=timezone.utc)
integration.save()
raise jira_auth_error
def get_jira_integration_metadata(jira_integration: Integration) -> dict:
prowler_jira = initialize_prowler_integration(jira_integration)
try:
return prowler_jira.get_jira_metadata()
except JiraNoProjectsError:
return {}
api/src/backend/api/v1/serializer_utils/integrations.py
+1 -1
View File
@@ -70,7 +70,7 @@ class SecurityHubConfigSerializer(BaseValidateSerializer):
class JiraConfigSerializer(BaseValidateSerializer):
domain = serializers.CharField(read_only=True)
issue_types = serializers.ListField(
read_only=True, child=serializers.CharField(), default=["Task", "Bug"]
read_only=True, child=serializers.CharField(), default=["Task"]
)
projects = serializers.DictField(read_only=True)
api/src/backend/api/v1/serializers.py
+39 -1
View File
@@ -2051,7 +2051,7 @@ class BaseWriteIntegrationSerializer(BaseWriteSerializer):
)
config_serializer = JiraConfigSerializer
# Create non-editable configuration for JIRA integration
default_jira_issue_types = ["Task", "Bug"]
default_jira_issue_types = ["Task"]
configuration.update(
{
"projects": {},
@@ -2270,6 +2270,44 @@ class IntegrationUpdateSerializer(BaseWriteIntegrationSerializer):
return representation
class IntegrationJiraDispatchSerializer(serializers.Serializer):
"""
Serializer for dispatching findings to JIRA integration.
"""
project_key = serializers.CharField(required=True)
issue_type = serializers.ChoiceField(required=True, choices=["Task"])
class JSONAPIMeta:
resource_name = "integrations-jira-dispatches"
def validate(self, attrs):
validated_attrs = super().validate(attrs)
integration_instance = Integration.objects.get(
id=self.context.get("integration_id")
)
if integration_instance.integration_type != Integration.IntegrationChoices.JIRA:
raise ValidationError(
{"integration_type": "The given integration is not a JIRA integration"}
)
if not integration_instance.enabled:
raise ValidationError(
{"integration": "The given integration is not enabled"}
)
project_key = attrs.get("project_key")
if project_key not in integration_instance.configuration.get("projects", {}):
raise ValidationError(
{
"project_key": "The given project key is not available for this JIRA integration. Refresh the "
"connection if this is an error."
}
)
return validated_attrs
# Processors
api/src/backend/api/v1/urls.py
+9
View File
@@ -12,6 +12,7 @@ from api.v1.views import (
FindingViewSet,
GithubSocialLoginView,
GoogleSocialLoginView,
IntegrationJiraViewSet,
IntegrationViewSet,
InvitationAcceptViewSet,
InvitationViewSet,
@@ -73,6 +74,13 @@ tenants_router.register(
users_router = routers.NestedSimpleRouter(router, r"users", lookup="user")
users_router.register(r"memberships", MembershipViewSet, basename="user-membership")
integrations_router = routers.NestedSimpleRouter(
router, r"integrations", lookup="integration"
)
integrations_router.register(
r"jira", IntegrationJiraViewSet, basename="integration-jira"
)
urlpatterns = [
path("tokens", CustomTokenObtainView.as_view(), name="token-obtain"),
path("tokens/refresh", CustomTokenRefreshView.as_view(), name="token-refresh"),
@@ -162,6 +170,7 @@ urlpatterns = [
path("", include(router.urls)),
path("", include(tenants_router.urls)),
path("", include(users_router.urls)),
path("", include(integrations_router.urls)),
path("schema", SchemaView.as_view(), name="schema"),
path("docs", SpectacularRedocView.as_view(url_name="schema"), name="docs"),
]
api/src/backend/api/v1/views.py
+84
View File
@@ -62,6 +62,7 @@ from tasks.tasks import (
check_provider_connection_task,
delete_provider_task,
delete_tenant_task,
jira_integration_task,
perform_scan_task,
)
@@ -75,8 +76,10 @@ from api.db_utils import rls_transaction
from api.exceptions import TaskFailedException
from api.filters import (
ComplianceOverviewFilter,
CustomDjangoFilterBackend,
FindingFilter,
IntegrationFilter,
IntegrationJiraFindingsFilter,
InvitationFilter,
LatestFindingFilter,
LatestResourceFilter,
@@ -143,6 +146,7 @@ from api.v1.serializers import (
FindingMetadataSerializer,
FindingSerializer,
IntegrationCreateSerializer,
IntegrationJiraDispatchSerializer,
IntegrationSerializer,
IntegrationUpdateSerializer,
InvitationAcceptSerializer,
@@ -3902,6 +3906,86 @@ class IntegrationViewSet(BaseRLSViewSet):
)
@extend_schema_view(
dispatches=extend_schema(
tags=["Integration"],
summary="Send findings to a Jira integration",
description="Send a set of filtered findings to the given integration. At least one finding filter must be "
"provided.",
responses={202: OpenApiResponse(response=TaskSerializer)},
filters=True,
)
)
class IntegrationJiraViewSet(BaseRLSViewSet):
queryset = Finding.all_objects.all()
serializer_class = IntegrationJiraDispatchSerializer
http_method_names = ["post"]
filter_backends = [CustomDjangoFilterBackend]
filterset_class = IntegrationJiraFindingsFilter
# RBAC required permissions
required_permissions = [Permissions.MANAGE_INTEGRATIONS]
@extend_schema(exclude=True)
def create(self, request, *args, **kwargs):
raise MethodNotAllowed(method="POST")
def get_queryset(self):
tenant_id = self.request.tenant_id
user_roles = get_role(self.request.user)
if user_roles.unlimited_visibility:
# User has unlimited visibility, return all findings
queryset = Finding.all_objects.filter(tenant_id=tenant_id)
else:
# User lacks permission, filter findings based on provider groups associated with the role
queryset = Finding.all_objects.filter(
scan__provider__in=get_providers(user_roles)
)
return queryset
@action(detail=False, methods=["post"], url_name="dispatches")
def dispatches(self, request, integration_pk=None):
get_object_or_404(Integration, pk=integration_pk)
serializer = self.get_serializer(
data=request.data, context={"integration_id": integration_pk}
)
serializer.is_valid(raise_exception=True)
if self.filter_queryset(self.get_queryset()).count() == 0:
raise ValidationError(
{"findings": "No findings match the provided filters"}
)
finding_ids = [
str(finding_id)
for finding_id in self.filter_queryset(self.get_queryset()).values_list(
"id", flat=True
)
]
project_key = serializer.validated_data["project_key"]
issue_type = serializer.validated_data["issue_type"]
with transaction.atomic():
task = jira_integration_task.delay(
tenant_id=self.request.tenant_id,
integration_id=integration_pk,
project_key=project_key,
issue_type=issue_type,
finding_ids=finding_ids,
)
prowler_task = Task.objects.get(id=task.id)
serializer = TaskSerializer(prowler_task)
return Response(
data=serializer.data,
status=status.HTTP_202_ACCEPTED,
headers={
"Content-Location": reverse(
"task-detail", kwargs={"pk": prowler_task.id}
)
},
)
@extend_schema_view(
list=extend_schema(
tags=["Lighthouse AI"],
api/src/backend/tasks/jobs/integrations.py
+76 -1
View File
@@ -7,7 +7,7 @@ from tasks.utils import batched
from api.db_utils import rls_transaction
from api.models import Finding, Integration, Provider
from api.utils import initialize_prowler_provider
from api.utils import initialize_prowler_integration, initialize_prowler_provider
from prowler.lib.outputs.asff.asff import ASFF
from prowler.lib.outputs.compliance.generic.generic import GenericCompliance
from prowler.lib.outputs.csv.csv import CSV
@@ -429,3 +429,78 @@ def upload_security_hub_integration(
f"Security Hub integrations failed for provider {provider_id}: {str(e)}"
)
return False
def send_findings_to_jira(
tenant_id: str,
integration_id: str,
project_key: str,
issue_type: str,
finding_ids: list[str],
):
with rls_transaction(tenant_id):
integration = Integration.objects.get(id=integration_id)
jira_integration = initialize_prowler_integration(integration)
num_tickets_created = 0
for finding_id in finding_ids:
with rls_transaction(tenant_id):
finding_instance = (
Finding.all_objects.select_related("scan__provider")
.prefetch_related("resources")
.get(id=finding_id)
)
# Extract resource information
resource = (
finding_instance.resources.first()
if finding_instance.resources.exists()
else None
)
resource_uid = resource.uid if resource else ""
resource_name = resource.name if resource else ""
resource_tags = {}
if resource and hasattr(resource, "tags"):
resource_tags = resource.get_tags(tenant_id)
# Get region
region = resource.region if resource and resource.region else ""
# Extract remediation information from check_metadata
check_metadata = finding_instance.check_metadata
remediation = check_metadata.get("remediation", {})
recommendation = remediation.get("recommendation", {})
remediation_code = remediation.get("code", {})
# Send the individual finding to Jira
result = jira_integration.send_finding(
check_id=finding_instance.check_id,
check_title=check_metadata.get("checktitle", ""),
severity=finding_instance.severity,
status=finding_instance.status,
status_extended=finding_instance.status_extended or "",
provider=finding_instance.scan.provider.provider,
region=region,
resource_uid=resource_uid,
resource_name=resource_name,
risk=check_metadata.get("risk", ""),
recommendation_text=recommendation.get("text", ""),
recommendation_url=recommendation.get("url", ""),
remediation_code_native_iac=remediation_code.get("nativeiac", ""),
remediation_code_terraform=remediation_code.get("terraform", ""),
remediation_code_cli=remediation_code.get("cli", ""),
remediation_code_other=remediation_code.get("other", ""),
resource_tags=resource_tags,
compliance=finding_instance.compliance or {},
project_key=project_key,
issue_type=issue_type,
)
if result:
num_tickets_created += 1
else:
logger.error(f"Failed to send finding {finding_id} to Jira")
return {
"created_count": num_tickets_created,
"failed_count": len(finding_ids) - num_tickets_created,
}
api/src/backend/tasks/tasks.py
+18
View File
@@ -22,6 +22,7 @@ from tasks.jobs.export import (
_upload_to_s3,
)
from tasks.jobs.integrations import (
send_findings_to_jira,
upload_s3_integration,
upload_security_hub_integration,
)
@@ -595,3 +596,20 @@ def security_hub_integration_task(
scan_id (str): The scan identifier
"""
return upload_security_hub_integration(tenant_id, provider_id, scan_id)
@shared_task(
base=RLSTask,
name="integration-jira",
queue="integrations",
)
def jira_integration_task(
tenant_id: str,
integration_id: str,
project_key: str,
issue_type: str,
finding_ids: list[str],
):
return send_findings_to_jira(
tenant_id, integration_id, project_key, issue_type, finding_ids
)
api/src/backend/tasks/tests/test_integrations.py
+352
View File
@@ -4,6 +4,7 @@ import pytest
from tasks.jobs.integrations import (
get_s3_client_from_integration,
get_security_hub_client_from_integration,
send_findings_to_jira,
upload_s3_integration,
upload_security_hub_integration,
)
@@ -1557,3 +1558,354 @@ class TestSecurityHubIntegrationUploads:
mock_security_hub.batch_send_to_security_hub.assert_called_once()
mock_security_hub.archive_previous_findings.assert_called_once()
@pytest.mark.django_db
class TestJiraIntegration:
@patch("tasks.jobs.integrations.rls_transaction")
@patch("tasks.jobs.integrations.Finding")
@patch("tasks.jobs.integrations.Integration")
@patch("tasks.jobs.integrations.initialize_prowler_integration")
def test_send_findings_to_jira_success(
self,
mock_initialize_integration,
mock_integration_model,
mock_finding_model,
mock_rls_transaction,
):
"""Test successful sending of findings to Jira using send_finding method"""
tenant_id = "tenant-123"
integration_id = "integration-456"
project_key = "PROJ"
issue_type = "Task"
finding_ids = ["finding-1", "finding-2"]
# Mock RLS transaction
mock_rls_transaction.return_value.__enter__ = MagicMock()
mock_rls_transaction.return_value.__exit__ = MagicMock()
# Mock integration
integration = MagicMock()
mock_integration_model.objects.get.return_value = integration
# Mock Jira integration
mock_jira_integration = MagicMock()
mock_jira_integration.send_finding.side_effect = [True, True] # Both succeed
mock_initialize_integration.return_value = mock_jira_integration
# Mock findings with resources
resource1 = MagicMock()
resource1.uid = "resource-uid-1"
resource1.name = "resource-name-1"
resource1.region = "us-east-1"
resource1.get_tags.return_value = {"env": "prod", "team": "security"}
resource2 = MagicMock()
resource2.uid = "resource-uid-2"
resource2.name = "resource-name-2"
resource2.region = "eu-west-1"
resource2.get_tags.return_value = {"env": "dev"}
finding1 = MagicMock()
finding1.id = "finding-1"
finding1.check_id = "check_001"
finding1.severity = "high"
finding1.status = "FAIL"
finding1.status_extended = "Resource is not compliant"
finding1.resource_regions = ["us-east-1"]
finding1.compliance = {"cis": ["1.1", "1.2"]}
finding1.resources.exists.return_value = True
finding1.resources.first.return_value = resource1
finding1.scan.provider.provider = "aws"
finding1.check_metadata = {
"checktitle": "Check Title 1",
"risk": "High risk finding",
"remediation": {
"recommendation": {
"text": "Fix this issue",
"url": "https://docs.example.com/fix",
},
"code": {
"nativeiac": "native code",
"terraform": "terraform code",
"cli": "aws cli command",
"other": "",
},
},
}
finding2 = MagicMock()
finding2.id = "finding-2"
finding2.check_id = "check_002"
finding2.severity = "medium"
finding2.status = "PASS"
finding2.status_extended = None
finding2.resource_regions = []
finding2.compliance = {}
finding2.resources.exists.return_value = True
finding2.resources.first.return_value = resource2
finding2.scan.provider.provider = "azure"
finding2.check_metadata = {
"checktitle": "Check Title 2",
"risk": "Medium risk",
"remediation": {
"recommendation": {"text": "Consider fixing", "url": ""},
"code": {},
},
}
mock_finding_model.all_objects.select_related.return_value.prefetch_related.return_value.get.side_effect = [
finding1,
finding2,
]
# Call the function
result = send_findings_to_jira(
tenant_id, integration_id, project_key, issue_type, finding_ids
)
# Assertions
assert result == {"created_count": 2, "failed_count": 0}
# Verify Jira integration was initialized
mock_initialize_integration.assert_called_once_with(integration)
# Verify send_finding was called twice with correct parameters
assert mock_jira_integration.send_finding.call_count == 2
# Verify first call
first_call = mock_jira_integration.send_finding.call_args_list[0]
assert first_call.kwargs["check_id"] == "check_001"
assert first_call.kwargs["check_title"] == "Check Title 1"
assert first_call.kwargs["severity"] == "high"
assert first_call.kwargs["status"] == "FAIL"
assert first_call.kwargs["resource_uid"] == "resource-uid-1"
assert first_call.kwargs["resource_name"] == "resource-name-1"
assert first_call.kwargs["region"] == "us-east-1"
assert first_call.kwargs["provider"] == "aws"
assert first_call.kwargs["project_key"] == project_key
assert first_call.kwargs["issue_type"] == issue_type
# Verify second call
second_call = mock_jira_integration.send_finding.call_args_list[1]
assert second_call.kwargs["check_id"] == "check_002"
assert second_call.kwargs["severity"] == "medium"
assert second_call.kwargs["status"] == "PASS"
@patch("tasks.jobs.integrations.rls_transaction")
@patch("tasks.jobs.integrations.Finding")
@patch("tasks.jobs.integrations.Integration")
@patch("tasks.jobs.integrations.initialize_prowler_integration")
@patch("tasks.jobs.integrations.logger")
def test_send_findings_to_jira_partial_failure(
self,
mock_logger,
mock_initialize_integration,
mock_integration_model,
mock_finding_model,
mock_rls_transaction,
):
"""Test partial failure when sending findings to Jira"""
tenant_id = "tenant-123"
integration_id = "integration-456"
project_key = "PROJ"
issue_type = "Task"
finding_ids = ["finding-1", "finding-2", "finding-3"]
# Mock RLS transaction
mock_rls_transaction.return_value.__enter__ = MagicMock()
mock_rls_transaction.return_value.__exit__ = MagicMock()
# Mock integration
integration = MagicMock()
mock_integration_model.objects.get.return_value = integration
# Mock Jira integration with mixed results
mock_jira_integration = MagicMock()
mock_jira_integration.send_finding.side_effect = [
True,
False,
True,
] # Second fails
mock_initialize_integration.return_value = mock_jira_integration
# Mock findings (simplified for this test)
findings = []
for i in range(3):
finding = MagicMock()
finding.id = f"finding-{i + 1}"
finding.check_id = f"check_{i + 1:03d}"
finding.severity = "low"
finding.status = "FAIL"
finding.status_extended = ""
finding.resource_regions = []
finding.compliance = {}
finding.resources.exists.return_value = False
finding.resources.first.return_value = None
finding.scan.provider.provider = "aws"
finding.check_metadata = {
"checktitle": f"Check {i + 1}",
"risk": "Low risk",
"remediation": {"recommendation": {}, "code": {}},
}
findings.append(finding)
mock_finding_model.all_objects.select_related.return_value.prefetch_related.return_value.get.side_effect = findings
# Call the function
result = send_findings_to_jira(
tenant_id, integration_id, project_key, issue_type, finding_ids
)
# Assertions
assert result == {"created_count": 2, "failed_count": 1}
# Verify error was logged for the failed finding
mock_logger.error.assert_called_with("Failed to send finding finding-2 to Jira")
@patch("tasks.jobs.integrations.rls_transaction")
@patch("tasks.jobs.integrations.Finding")
@patch("tasks.jobs.integrations.Integration")
@patch("tasks.jobs.integrations.initialize_prowler_integration")
def test_send_findings_to_jira_no_resources(
self,
mock_initialize_integration,
mock_integration_model,
mock_finding_model,
mock_rls_transaction,
):
"""Test sending findings to Jira when finding has no resources"""
tenant_id = "tenant-123"
integration_id = "integration-456"
project_key = "PROJ"
issue_type = "Task"
finding_ids = ["finding-1"]
# Mock RLS transaction
mock_rls_transaction.return_value.__enter__ = MagicMock()
mock_rls_transaction.return_value.__exit__ = MagicMock()
# Mock integration
integration = MagicMock()
mock_integration_model.objects.get.return_value = integration
# Mock Jira integration
mock_jira_integration = MagicMock()
mock_jira_integration.send_finding.return_value = True
mock_initialize_integration.return_value = mock_jira_integration
# Mock finding without resources
finding = MagicMock()
finding.id = "finding-1"
finding.check_id = "check_001"
finding.severity = "critical"
finding.status = "FAIL"
finding.status_extended = "Critical issue found"
finding.resource_regions = None
finding.compliance = {"pci": ["3.1"]}
finding.resources.exists.return_value = False
finding.resources.first.return_value = None
finding.scan.provider.provider = "gcp"
finding.check_metadata = {
"checktitle": "Critical Check",
"risk": "Very high risk",
"remediation": {
"recommendation": {
"text": "Immediate action required",
"url": "https://example.com/critical",
},
"code": {
"nativeiac": "",
"terraform": "terraform fix",
"cli": "",
"other": "manual fix",
},
},
}
mock_finding_model.all_objects.select_related.return_value.prefetch_related.return_value.get.return_value = finding
# Call the function
result = send_findings_to_jira(
tenant_id, integration_id, project_key, issue_type, finding_ids
)
# Assertions
assert result == {"created_count": 1, "failed_count": 0}
# Verify send_finding was called with empty resource fields
call_kwargs = mock_jira_integration.send_finding.call_args.kwargs
assert call_kwargs["resource_uid"] == ""
assert call_kwargs["resource_name"] == ""
assert call_kwargs["resource_tags"] == {}
assert call_kwargs["region"] == ""
@patch("tasks.jobs.integrations.rls_transaction")
@patch("tasks.jobs.integrations.Finding")
@patch("tasks.jobs.integrations.Integration")
@patch("tasks.jobs.integrations.initialize_prowler_integration")
def test_send_findings_to_jira_with_empty_check_metadata(
self,
mock_initialize_integration,
mock_integration_model,
mock_finding_model,
mock_rls_transaction,
):
"""Test sending findings to Jira when check_metadata is empty or missing fields"""
tenant_id = "tenant-123"
integration_id = "integration-456"
project_key = "PROJ"
issue_type = "Task"
finding_ids = ["finding-1"]
# Mock RLS transaction
mock_rls_transaction.return_value.__enter__ = MagicMock()
mock_rls_transaction.return_value.__exit__ = MagicMock()
# Mock integration
integration = MagicMock()
mock_integration_model.objects.get.return_value = integration
# Mock Jira integration
mock_jira_integration = MagicMock()
mock_jira_integration.send_finding.return_value = True
mock_initialize_integration.return_value = mock_jira_integration
# Mock finding with minimal/empty check_metadata
finding = MagicMock()
finding.id = "finding-1"
finding.check_id = "check_001"
finding.severity = "low"
finding.status = "PASS"
finding.status_extended = None
finding.resource_regions = []
finding.compliance = None
finding.resources.exists.return_value = False
finding.resources.first.return_value = None
finding.scan.provider.provider = "kubernetes"
finding.check_metadata = {} # Empty metadata
mock_finding_model.all_objects.select_related.return_value.prefetch_related.return_value.get.return_value = finding
# Call the function
result = send_findings_to_jira(
tenant_id, integration_id, project_key, issue_type, finding_ids
)
# Assertions
assert result == {"created_count": 1, "failed_count": 0}
# Verify send_finding was called with default/empty values
call_kwargs = mock_jira_integration.send_finding.call_args.kwargs
assert call_kwargs["check_title"] == ""
assert call_kwargs["risk"] == ""
assert call_kwargs["recommendation_text"] == ""
assert call_kwargs["recommendation_url"] == ""
assert call_kwargs["remediation_code_native_iac"] == ""
assert call_kwargs["remediation_code_terraform"] == ""
assert call_kwargs["remediation_code_cli"] == ""
assert call_kwargs["remediation_code_other"] == ""
assert call_kwargs["compliance"] == {}