mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 04:51:51 +00:00
feat(jira): add get_metadata (#8630)
This commit is contained in:
+2
-16
@@ -4,23 +4,9 @@ All notable changes to the **Prowler SDK** are documented in this file.
|
||||
## [v5.12.0] (Prowler UNRELEASED)
|
||||
|
||||
### Added
|
||||
|
||||
### Changed
|
||||
|
||||
### Fixed
|
||||
|
||||
---
|
||||
|
||||
## [v5.11.1] (Prowler UNRELEASED)
|
||||
|
||||
### Fixed
|
||||
|
||||
---
|
||||
|
||||
## [v5.12.0] (Prowler UNRELEASED)
|
||||
|
||||
### Added
|
||||
- Get Jira Project's metadata [(#8630)](https://github.com/prowler-cloud/prowler/pull/8630)
|
||||
- Add more fields for the Jira ticket and handle custom fields errors [(#8601)](https://github.com/prowler-cloud/prowler/pull/8601)
|
||||
- Get Jira projects from test_connection [(#8634)](https://github.com/prowler-cloud/prowler/pull/8634)
|
||||
|
||||
### Changed
|
||||
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import base64
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Dict
|
||||
|
||||
@@ -35,6 +36,17 @@ from prowler.lib.outputs.jira.exceptions.exceptions import (
|
||||
from prowler.providers.common.models import Connection
|
||||
|
||||
|
||||
@dataclass
|
||||
class JiraConnection(Connection):
|
||||
"""
|
||||
Represents a Jira connection object.
|
||||
Attributes:
|
||||
projects (dict): Dictionary of projects in Jira.
|
||||
"""
|
||||
|
||||
projects: dict = None
|
||||
|
||||
|
||||
class Jira:
|
||||
"""
|
||||
Jira class to interact with the Jira API
|
||||
@@ -463,7 +475,7 @@ class Jira:
|
||||
api_token: str = None,
|
||||
domain: str = None,
|
||||
raise_on_exception: bool = True,
|
||||
) -> Connection:
|
||||
) -> JiraConnection:
|
||||
"""Test the connection to Jira
|
||||
|
||||
Args:
|
||||
@@ -476,7 +488,7 @@ class Jira:
|
||||
- raise_on_exception: Whether to raise an exception or not
|
||||
|
||||
Returns:
|
||||
- Connection: The connection object
|
||||
- JiraConnection: The connection object
|
||||
|
||||
Raises:
|
||||
- JiraGetCloudIDNoResourcesError: No resources were found in Jira when getting the cloud id
|
||||
@@ -484,6 +496,8 @@ class Jira:
|
||||
- JiraGetCloudIDError: Failed to get the cloud ID from Jira
|
||||
- JiraAuthenticationError: Failed to authenticate
|
||||
- JiraTestConnectionError: Failed to test the connection
|
||||
- JiraNoProjectsError: No projects found in Jira
|
||||
- JiraGetProjectsResponseError: Failed to get projects from Jira, response code did not match 200
|
||||
"""
|
||||
try:
|
||||
jira = Jira(
|
||||
@@ -494,60 +508,51 @@ class Jira:
|
||||
api_token=api_token,
|
||||
domain=domain,
|
||||
)
|
||||
access_token = jira.get_access_token()
|
||||
projects = jira.get_projects()
|
||||
|
||||
if not access_token:
|
||||
return ValueError("Failed to get access token")
|
||||
|
||||
if jira.using_basic_auth:
|
||||
headers = {"Authorization": f"Basic {access_token}"}
|
||||
else:
|
||||
headers = {"Authorization": f"Bearer {access_token}"}
|
||||
|
||||
response = requests.get(
|
||||
f"https://api.atlassian.com/ex/jira/{jira.cloud_id}/rest/api/3/myself",
|
||||
headers=headers,
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
return Connection(is_connected=True)
|
||||
else:
|
||||
return Connection(is_connected=False, error=response.json())
|
||||
except JiraGetCloudIDNoResourcesError as no_resources_error:
|
||||
return JiraConnection(is_connected=True, projects=projects)
|
||||
except JiraNoProjectsError as no_projects_error:
|
||||
logger.error(
|
||||
f"{no_resources_error.__class__.__name__}[{no_resources_error.__traceback__.tb_lineno}]: {no_resources_error}"
|
||||
f"{no_projects_error.__class__.__name__}[{no_projects_error.__traceback__.tb_lineno}]: {no_projects_error}"
|
||||
)
|
||||
if raise_on_exception:
|
||||
raise no_resources_error
|
||||
return Connection(error=no_resources_error)
|
||||
raise no_projects_error
|
||||
return JiraConnection(error=no_projects_error)
|
||||
except JiraGetCloudIDResponseError as response_error:
|
||||
logger.error(
|
||||
f"{response_error.__class__.__name__}[{response_error.__traceback__.tb_lineno}]: {response_error}"
|
||||
)
|
||||
if raise_on_exception:
|
||||
raise response_error
|
||||
return Connection(error=response_error)
|
||||
return JiraConnection(error=response_error)
|
||||
except JiraGetCloudIDError as cloud_id_error:
|
||||
logger.error(
|
||||
f"{cloud_id_error.__class__.__name__}[{cloud_id_error.__traceback__.tb_lineno}]: {cloud_id_error}"
|
||||
)
|
||||
if raise_on_exception:
|
||||
raise cloud_id_error
|
||||
return Connection(error=cloud_id_error)
|
||||
return JiraConnection(error=cloud_id_error)
|
||||
except JiraAuthenticationError as auth_error:
|
||||
logger.error(
|
||||
f"{auth_error.__class__.__name__}[{auth_error.__traceback__.tb_lineno}]: {auth_error}"
|
||||
)
|
||||
if raise_on_exception:
|
||||
raise auth_error
|
||||
return Connection(error=auth_error)
|
||||
return JiraConnection(error=auth_error)
|
||||
except JiraBasicAuthError as basic_auth_error:
|
||||
logger.error(
|
||||
f"{basic_auth_error.__class__.__name__}[{basic_auth_error.__traceback__.tb_lineno}]: {basic_auth_error}"
|
||||
)
|
||||
if raise_on_exception:
|
||||
raise basic_auth_error
|
||||
return Connection(error=basic_auth_error)
|
||||
return JiraConnection(error=basic_auth_error)
|
||||
except JiraGetProjectsResponseError as projects_response_error:
|
||||
logger.error(
|
||||
f"{projects_response_error.__class__.__name__}[{projects_response_error.__traceback__.tb_lineno}]: {projects_response_error}"
|
||||
)
|
||||
if raise_on_exception:
|
||||
raise projects_response_error
|
||||
return JiraConnection(error=projects_response_error)
|
||||
except Exception as error:
|
||||
logger.error(f"Failed to test connection: {error}")
|
||||
if raise_on_exception:
|
||||
@@ -555,7 +560,7 @@ class Jira:
|
||||
message="Failed to test connection on the Jira integration",
|
||||
file=os.path.basename(__file__),
|
||||
)
|
||||
return Connection(is_connected=False, error=error)
|
||||
return JiraConnection(is_connected=False, error=error)
|
||||
|
||||
def get_projects(self) -> Dict[str, str]:
|
||||
"""Get the projects from Jira
|
||||
@@ -682,6 +687,92 @@ class Jira:
|
||||
file=os.path.basename(__file__),
|
||||
)
|
||||
|
||||
def get_metadata(self) -> dict:
|
||||
"""Get the metadata from Jira
|
||||
|
||||
Returns:
|
||||
- dict: The projects and issue types from Jira as a dictionary, the projects format is {"KEY": {"name": "NAME", "issue_types": ["ISSUE_TYPE_1", "ISSUE_TYPE_2"]}}
|
||||
"""
|
||||
try:
|
||||
access_token = self.get_access_token()
|
||||
|
||||
if not access_token:
|
||||
return ValueError("Failed to get access token")
|
||||
|
||||
if self._using_basic_auth:
|
||||
headers = {"Authorization": f"Basic {access_token}"}
|
||||
else:
|
||||
headers = {"Authorization": f"Bearer {access_token}"}
|
||||
|
||||
response = requests.get(
|
||||
f"https://api.atlassian.com/ex/jira/{self.cloud_id}/rest/api/3/project",
|
||||
headers=headers,
|
||||
)
|
||||
if response.status_code == 200:
|
||||
projects_data = {}
|
||||
projects_list = response.json()
|
||||
if not projects_list:
|
||||
logger.error("No projects found")
|
||||
raise JiraNoProjectsError(
|
||||
message="No projects found in Jira",
|
||||
file=os.path.basename(__file__),
|
||||
)
|
||||
else:
|
||||
for project in projects_list:
|
||||
project_response = requests.get(
|
||||
f"https://api.atlassian.com/ex/jira/{self.cloud_id}/rest/api/3/issue/createmeta?projectKeys={project['key']}&expand=projects.issuetypes.fields",
|
||||
headers=headers,
|
||||
)
|
||||
if project_response.status_code == 200:
|
||||
project_metadata = project_response.json()
|
||||
if len(project_metadata["projects"]) == 0:
|
||||
logger.warning(
|
||||
f"No project metadata found for project {project['key']}, setting empty issue types"
|
||||
)
|
||||
issue_types = []
|
||||
else:
|
||||
issue_types = [
|
||||
issue_type["name"]
|
||||
for issue_type in project_metadata["projects"][0][
|
||||
"issuetypes"
|
||||
]
|
||||
]
|
||||
else:
|
||||
raise JiraGetAvailableIssueTypesResponseError(
|
||||
message="Failed to get available issue types from Jira",
|
||||
file=os.path.basename(__file__),
|
||||
)
|
||||
projects_data[project["key"]] = {
|
||||
"name": project["name"],
|
||||
"issue_types": issue_types,
|
||||
}
|
||||
return projects_data
|
||||
else:
|
||||
logger.error(
|
||||
f"Failed to get projects: {response.status_code} - {response.text}"
|
||||
)
|
||||
raise JiraGetProjectsResponseError(
|
||||
message="Failed to get projects from Jira",
|
||||
file=os.path.basename(__file__),
|
||||
)
|
||||
except JiraNoProjectsError as no_projects_error:
|
||||
raise no_projects_error
|
||||
except JiraGetAvailableIssueTypesResponseError as issue_types_error:
|
||||
raise JiraGetProjectsError(
|
||||
message=f"Failed to get projects and issue types from Jira: {issue_types_error}",
|
||||
file=os.path.basename(__file__),
|
||||
)
|
||||
except JiraRefreshTokenError as refresh_error:
|
||||
raise refresh_error
|
||||
except JiraRefreshTokenResponseError as response_error:
|
||||
raise response_error
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to get projects: {e}")
|
||||
raise JiraGetProjectsError(
|
||||
message="Failed to get projects from Jira",
|
||||
file=os.path.basename(__file__),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def get_color_from_status(status: str) -> str:
|
||||
"""Get the color from the status
|
||||
|
||||
@@ -13,9 +13,11 @@ from prowler.lib.outputs.jira.exceptions.exceptions import (
|
||||
JiraGetAvailableIssueTypesError,
|
||||
JiraGetCloudIDError,
|
||||
JiraGetProjectsError,
|
||||
JiraGetProjectsResponseError,
|
||||
JiraNoProjectsError,
|
||||
JiraRefreshTokenError,
|
||||
JiraRequiredCustomFieldsError,
|
||||
JiraTestConnectionError,
|
||||
)
|
||||
from prowler.lib.outputs.jira.jira import Jira
|
||||
|
||||
@@ -293,25 +295,19 @@ class TestJiraIntegration:
|
||||
with pytest.raises(JiraRefreshTokenError):
|
||||
self.jira_integration.refresh_access_token()
|
||||
|
||||
@patch.object(Jira, "get_access_token", return_value="valid_access_token")
|
||||
@patch.object(Jira, "get_cloud_id", return_value="test_cloud_id")
|
||||
@patch.object(Jira, "get_auth", return_value=None)
|
||||
@patch("prowler.lib.outputs.jira.jira.requests.get")
|
||||
def test_test_connection_successful(
|
||||
self, mock_get, mock_get_cloud_id, mock_get_auth, mock_get_access_token
|
||||
):
|
||||
"""Test that a successful connection returns an active Connection object."""
|
||||
@patch.object(
|
||||
Jira,
|
||||
"get_projects",
|
||||
return_value={"PROJ1": "Project One", "PROJ2": "Project Two"},
|
||||
)
|
||||
def test_test_connection_successful(self, mock_get_projects, mock_get_auth):
|
||||
"""Test that a successful connection returns an active Connection object with projects."""
|
||||
# To disable vulture
|
||||
mock_get_cloud_id = mock_get_cloud_id
|
||||
mock_get_projects = mock_get_projects
|
||||
mock_get_auth = mock_get_auth
|
||||
mock_get_access_token = mock_get_access_token
|
||||
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = {"id": "test_user_id"}
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
connection = self.jira_integration.test_connection(
|
||||
connection = Jira.test_connection(
|
||||
redirect_uri=self.redirect_uri,
|
||||
client_id=self.client_id,
|
||||
client_secret=self.client_secret,
|
||||
@@ -319,26 +315,23 @@ class TestJiraIntegration:
|
||||
|
||||
assert connection.is_connected
|
||||
assert connection.error is None
|
||||
assert connection.projects == {"PROJ1": "Project One", "PROJ2": "Project Two"}
|
||||
|
||||
@patch.object(Jira, "get_access_token", return_value="valid_access_token")
|
||||
@patch.object(Jira, "get_cloud_id", return_value="test_cloud_id")
|
||||
@patch.object(Jira, "get_basic_auth", return_value=None)
|
||||
@patch("prowler.lib.outputs.jira.jira.requests.get")
|
||||
@patch.object(
|
||||
Jira,
|
||||
"get_projects",
|
||||
return_value={"PROJ1": "Project One", "PROJ2": "Project Two"},
|
||||
)
|
||||
def test_test_connection_successful_basic_auth(
|
||||
self, mock_get, mock_get_cloud_id, mock_get_auth, mock_get_access_token
|
||||
self, mock_get_projects, mock_get_basic_auth
|
||||
):
|
||||
"""Test that a successful connection returns an active Connection object."""
|
||||
"""Test that a successful connection returns an active Connection object with projects."""
|
||||
# To disable vulture
|
||||
mock_get_cloud_id = mock_get_cloud_id
|
||||
mock_get_auth = mock_get_auth
|
||||
mock_get_access_token = mock_get_access_token
|
||||
mock_get_projects = mock_get_projects
|
||||
mock_get_basic_auth = mock_get_basic_auth
|
||||
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = {"id": "test_user_id"}
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
connection = self.jira_integration_basic_auth.test_connection(
|
||||
connection = Jira.test_connection(
|
||||
user_mail=self.user_mail,
|
||||
api_token=self.api_token,
|
||||
domain=self.domain,
|
||||
@@ -346,19 +339,20 @@ class TestJiraIntegration:
|
||||
|
||||
assert connection.is_connected
|
||||
assert connection.error is None
|
||||
assert connection.projects == {"PROJ1": "Project One", "PROJ2": "Project Two"}
|
||||
|
||||
@patch.object(
|
||||
Jira,
|
||||
"get_access_token",
|
||||
"get_auth",
|
||||
side_effect=JiraAuthenticationError("Failed to authenticate with Jira"),
|
||||
)
|
||||
def test_test_connection_failed(self, mock_get_access_token):
|
||||
def test_test_connection_failed(self, mock_get_auth):
|
||||
"""Test that a failed connection raises JiraAuthenticationError."""
|
||||
# To disable vulture
|
||||
mock_get_access_token = mock_get_access_token
|
||||
mock_get_auth = mock_get_auth
|
||||
|
||||
with pytest.raises(JiraAuthenticationError):
|
||||
self.jira_integration.test_connection(
|
||||
Jira.test_connection(
|
||||
redirect_uri=self.redirect_uri,
|
||||
client_id=self.client_id,
|
||||
client_secret=self.client_secret,
|
||||
@@ -366,21 +360,120 @@ class TestJiraIntegration:
|
||||
|
||||
@patch.object(
|
||||
Jira,
|
||||
"get_cloud_id",
|
||||
"get_basic_auth",
|
||||
side_effect=JiraBasicAuthError("Failed to authenticate with Jira"),
|
||||
)
|
||||
def test_test_connection_failed_basic_auth(self, mock_get_access_token):
|
||||
"""Test that a failed connection raises JiraAuthenticationError."""
|
||||
def test_test_connection_failed_basic_auth(self, mock_get_basic_auth):
|
||||
"""Test that a failed connection raises JiraBasicAuthError."""
|
||||
# To disable vulture
|
||||
mock_get_access_token = mock_get_access_token
|
||||
mock_get_basic_auth = mock_get_basic_auth
|
||||
|
||||
with pytest.raises(JiraBasicAuthError):
|
||||
self.jira_integration_basic_auth.test_connection(
|
||||
Jira.test_connection(
|
||||
user_mail=self.user_mail,
|
||||
api_token=self.api_token,
|
||||
domain=self.domain,
|
||||
)
|
||||
|
||||
@patch.object(Jira, "get_auth", return_value=None)
|
||||
@patch.object(
|
||||
Jira, "get_projects", side_effect=JiraNoProjectsError("No projects found")
|
||||
)
|
||||
def test_test_connection_no_projects_found(self, mock_get_projects, mock_get_auth):
|
||||
"""Test that test_connection raises JiraNoProjectsError when no projects are found."""
|
||||
# To disable vulture
|
||||
mock_get_projects = mock_get_projects
|
||||
mock_get_auth = mock_get_auth
|
||||
|
||||
with pytest.raises(JiraNoProjectsError):
|
||||
Jira.test_connection(
|
||||
redirect_uri=self.redirect_uri,
|
||||
client_id=self.client_id,
|
||||
client_secret=self.client_secret,
|
||||
)
|
||||
|
||||
@patch.object(Jira, "get_auth", return_value=None)
|
||||
@patch.object(
|
||||
Jira,
|
||||
"get_projects",
|
||||
side_effect=JiraGetProjectsResponseError("Projects request failed"),
|
||||
)
|
||||
def test_test_connection_projects_request_error(
|
||||
self, mock_get_projects, mock_get_auth
|
||||
):
|
||||
"""Test that test_connection raises JiraGetProjectsResponseError when projects request fails."""
|
||||
# To disable vulture
|
||||
mock_get_projects = mock_get_projects
|
||||
mock_get_auth = mock_get_auth
|
||||
|
||||
with pytest.raises(JiraGetProjectsResponseError):
|
||||
Jira.test_connection(
|
||||
redirect_uri=self.redirect_uri,
|
||||
client_id=self.client_id,
|
||||
client_secret=self.client_secret,
|
||||
)
|
||||
|
||||
@patch.object(Jira, "get_auth", return_value=None)
|
||||
@patch.object(
|
||||
Jira, "get_projects", side_effect=JiraNoProjectsError("No projects found")
|
||||
)
|
||||
def test_test_connection_no_projects_found_no_exception(
|
||||
self, mock_get_projects, mock_get_auth
|
||||
):
|
||||
"""Test that test_connection returns error connection object when no projects found and raise_on_exception=False."""
|
||||
# To disable vulture
|
||||
mock_get_projects = mock_get_projects
|
||||
mock_get_auth = mock_get_auth
|
||||
|
||||
connection = Jira.test_connection(
|
||||
redirect_uri=self.redirect_uri,
|
||||
client_id=self.client_id,
|
||||
client_secret=self.client_secret,
|
||||
raise_on_exception=False,
|
||||
)
|
||||
|
||||
assert not connection.is_connected
|
||||
assert isinstance(connection.error, JiraNoProjectsError)
|
||||
|
||||
@patch.object(Jira, "get_auth", return_value=None)
|
||||
@patch.object(
|
||||
Jira,
|
||||
"get_projects",
|
||||
side_effect=JiraGetProjectsResponseError("Projects request failed"),
|
||||
)
|
||||
def test_test_connection_projects_request_error_no_exception(
|
||||
self, mock_get_projects, mock_get_auth
|
||||
):
|
||||
"""Test that test_connection returns error connection object when projects request fails and raise_on_exception=False."""
|
||||
# To disable vulture
|
||||
mock_get_projects = mock_get_projects
|
||||
mock_get_auth = mock_get_auth
|
||||
|
||||
connection = Jira.test_connection(
|
||||
redirect_uri=self.redirect_uri,
|
||||
client_id=self.client_id,
|
||||
client_secret=self.client_secret,
|
||||
raise_on_exception=False,
|
||||
)
|
||||
|
||||
assert not connection.is_connected
|
||||
assert isinstance(connection.error, JiraGetProjectsResponseError)
|
||||
|
||||
@patch.object(Jira, "get_auth", return_value=None)
|
||||
@patch.object(Jira, "get_projects", side_effect=Exception("Unexpected error"))
|
||||
def test_test_connection_unexpected_error(self, mock_get_projects, mock_get_auth):
|
||||
"""Test that test_connection raises JiraTestConnectionError on unexpected exceptions."""
|
||||
# To disable vulture
|
||||
mock_get_projects = mock_get_projects
|
||||
mock_get_auth = mock_get_auth
|
||||
|
||||
with pytest.raises(JiraTestConnectionError):
|
||||
Jira.test_connection(
|
||||
redirect_uri=self.redirect_uri,
|
||||
client_id=self.client_id,
|
||||
client_secret=self.client_secret,
|
||||
)
|
||||
|
||||
@patch.object(Jira, "get_access_token", return_value="valid_access_token")
|
||||
@patch.object(
|
||||
Jira, "cloud_id", new_callable=PropertyMock, return_value="test_cloud_id"
|
||||
@@ -903,3 +996,270 @@ class TestJiraIntegration:
|
||||
def test_get_severity_color(self, severity, expected_color):
|
||||
"""Test that get_severity_color returns the correct color for a severity."""
|
||||
assert self.jira_integration.get_severity_color(severity) == expected_color
|
||||
|
||||
@patch.object(Jira, "get_access_token", return_value="valid_access_token")
|
||||
@patch.object(
|
||||
Jira, "cloud_id", new_callable=PropertyMock, return_value="test_cloud_id"
|
||||
)
|
||||
@patch("prowler.lib.outputs.jira.jira.requests.get")
|
||||
def test_get_projects_and_issue_types_successful(
|
||||
self, mock_get, mock_cloud_id, mock_get_access_token
|
||||
):
|
||||
"""Test successful retrieval of metadata associated to projects from Jira."""
|
||||
# To disable vulture
|
||||
mock_cloud_id = mock_cloud_id
|
||||
mock_get_access_token = mock_get_access_token
|
||||
|
||||
# Mock the projects response
|
||||
mock_projects_response = MagicMock()
|
||||
mock_projects_response.status_code = 200
|
||||
mock_projects_response.json.return_value = [
|
||||
{"key": "PROJ1", "name": "Project One"},
|
||||
{"key": "PROJ2", "name": "Project Two"},
|
||||
]
|
||||
|
||||
# Mock the issue types responses
|
||||
mock_issue_types_response_1 = MagicMock()
|
||||
mock_issue_types_response_1.status_code = 200
|
||||
mock_issue_types_response_1.json.return_value = {
|
||||
"projects": [
|
||||
{
|
||||
"issuetypes": [
|
||||
{"name": "Bug", "id": "1"},
|
||||
{"name": "Task", "id": "2"},
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
mock_issue_types_response_2 = MagicMock()
|
||||
mock_issue_types_response_2.status_code = 200
|
||||
mock_issue_types_response_2.json.return_value = {
|
||||
"projects": [
|
||||
{
|
||||
"issuetypes": [
|
||||
{"name": "Story", "id": "3"},
|
||||
{"name": "Epic", "id": "4"},
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
# Configure side_effect to return different responses for different calls
|
||||
mock_get.side_effect = [
|
||||
mock_projects_response,
|
||||
mock_issue_types_response_1,
|
||||
mock_issue_types_response_2,
|
||||
]
|
||||
|
||||
jira_metadata = self.jira_integration.get_metadata()
|
||||
|
||||
expected_result = {
|
||||
"PROJ1": {
|
||||
"name": "Project One",
|
||||
"issue_types": ["Bug", "Task"],
|
||||
},
|
||||
"PROJ2": {
|
||||
"name": "Project Two",
|
||||
"issue_types": ["Story", "Epic"],
|
||||
},
|
||||
}
|
||||
|
||||
assert jira_metadata == expected_result
|
||||
|
||||
# Verify the correct number of calls were made
|
||||
assert mock_get.call_count == 3
|
||||
|
||||
# Verify the URLs called
|
||||
calls = mock_get.call_args_list
|
||||
assert (
|
||||
calls[0][0][0]
|
||||
== "https://api.atlassian.com/ex/jira/test_cloud_id/rest/api/3/project"
|
||||
)
|
||||
assert "projectKeys=PROJ1" in calls[1][0][0]
|
||||
assert "projectKeys=PROJ2" in calls[2][0][0]
|
||||
|
||||
@patch.object(Jira, "get_access_token", return_value="valid_access_token")
|
||||
@patch.object(
|
||||
Jira, "cloud_id", new_callable=PropertyMock, return_value="test_cloud_id"
|
||||
)
|
||||
@patch("prowler.lib.outputs.jira.jira.requests.get")
|
||||
def test_get_projects_and_issue_types_no_projects_found(
|
||||
self, mock_get, mock_cloud_id, mock_get_access_token
|
||||
):
|
||||
"""Test that get_metadata raises JiraNoProjectsError when no projects are found."""
|
||||
# To disable vulture
|
||||
mock_cloud_id = mock_cloud_id
|
||||
mock_get_access_token = mock_get_access_token
|
||||
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.json.return_value = []
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
with pytest.raises(JiraNoProjectsError):
|
||||
self.jira_integration.get_metadata()
|
||||
|
||||
@patch.object(Jira, "get_access_token", return_value="valid_access_token")
|
||||
@patch.object(
|
||||
Jira, "cloud_id", new_callable=PropertyMock, return_value="test_cloud_id"
|
||||
)
|
||||
@patch("prowler.lib.outputs.jira.jira.requests.get")
|
||||
def test_get_projects_and_issue_types_projects_response_error(
|
||||
self, mock_get, mock_cloud_id, mock_get_access_token
|
||||
):
|
||||
"""Test that get_metadata raises JiraGetProjectsError when projects request fails."""
|
||||
# To disable vulture
|
||||
mock_cloud_id = mock_cloud_id
|
||||
mock_get_access_token = mock_get_access_token
|
||||
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 404
|
||||
mock_response.text = "Not Found"
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
with pytest.raises(JiraGetProjectsError):
|
||||
self.jira_integration.get_metadata()
|
||||
|
||||
@patch.object(Jira, "get_access_token", return_value="valid_access_token")
|
||||
@patch.object(
|
||||
Jira, "cloud_id", new_callable=PropertyMock, return_value="test_cloud_id"
|
||||
)
|
||||
@patch("prowler.lib.outputs.jira.jira.requests.get")
|
||||
def test_get_projects_and_issue_types_issue_types_response_error(
|
||||
self, mock_get, mock_cloud_id, mock_get_access_token
|
||||
):
|
||||
"""Test that get_metadata raises JiraGetProjectsError when issue types request fails."""
|
||||
# To disable vulture
|
||||
mock_cloud_id = mock_cloud_id
|
||||
mock_get_access_token = mock_get_access_token
|
||||
|
||||
# Mock successful projects response
|
||||
mock_projects_response = MagicMock()
|
||||
mock_projects_response.status_code = 200
|
||||
mock_projects_response.json.return_value = [
|
||||
{"key": "PROJ1", "name": "Project One"}
|
||||
]
|
||||
|
||||
# Mock failed issue types response
|
||||
mock_issue_types_response = MagicMock()
|
||||
mock_issue_types_response.status_code = 404
|
||||
|
||||
mock_get.side_effect = [mock_projects_response, mock_issue_types_response]
|
||||
|
||||
with pytest.raises(JiraGetProjectsError):
|
||||
self.jira_integration.get_metadata()
|
||||
|
||||
@patch.object(Jira, "get_access_token", return_value="valid_access_token")
|
||||
@patch.object(
|
||||
Jira, "cloud_id", new_callable=PropertyMock, return_value="test_cloud_id"
|
||||
)
|
||||
@patch("prowler.lib.outputs.jira.jira.requests.get")
|
||||
def test_get_projects_and_issue_types_no_project_metadata(
|
||||
self, mock_get, mock_cloud_id, mock_get_access_token
|
||||
):
|
||||
"""Test that get_metadata returns empty issue_types when project metadata is empty."""
|
||||
# To disable vulture
|
||||
mock_cloud_id = mock_cloud_id
|
||||
mock_get_access_token = mock_get_access_token
|
||||
|
||||
# Mock successful projects response
|
||||
mock_projects_response = MagicMock()
|
||||
mock_projects_response.status_code = 200
|
||||
mock_projects_response.json.return_value = [
|
||||
{"key": "PROJ1", "name": "Project One"}
|
||||
]
|
||||
|
||||
# Mock issue types response with empty projects list
|
||||
mock_issue_types_response = MagicMock()
|
||||
mock_issue_types_response.status_code = 200
|
||||
mock_issue_types_response.json.return_value = {"projects": []}
|
||||
|
||||
mock_get.side_effect = [mock_projects_response, mock_issue_types_response]
|
||||
|
||||
projects_and_issue_types = self.jira_integration.get_metadata()
|
||||
|
||||
expected_result = {
|
||||
"PROJ1": {
|
||||
"name": "Project One",
|
||||
"issue_types": [],
|
||||
}
|
||||
}
|
||||
|
||||
assert projects_and_issue_types == expected_result
|
||||
|
||||
@patch.object(
|
||||
Jira,
|
||||
"get_access_token",
|
||||
side_effect=JiraRefreshTokenError("Failed to refresh the access token"),
|
||||
)
|
||||
def test_get_projects_and_issue_types_refresh_token_error(
|
||||
self, mock_get_access_token
|
||||
):
|
||||
"""Test that get_metadata raises JiraRefreshTokenError when refreshing the token fails."""
|
||||
# To disable vulture
|
||||
mock_get_access_token = mock_get_access_token
|
||||
|
||||
with pytest.raises(JiraRefreshTokenError):
|
||||
self.jira_integration.get_metadata()
|
||||
|
||||
@patch.object(Jira, "get_access_token", return_value="valid_access_token")
|
||||
@patch.object(
|
||||
Jira, "cloud_id", new_callable=PropertyMock, return_value="test_cloud_id"
|
||||
)
|
||||
@patch("prowler.lib.outputs.jira.jira.requests.get")
|
||||
def test_get_projects_and_issue_types_mixed_scenarios(
|
||||
self, mock_get, mock_cloud_id, mock_get_access_token
|
||||
):
|
||||
"""Test get_metadata with mixed success and empty metadata scenarios."""
|
||||
# To disable vulture
|
||||
mock_cloud_id = mock_cloud_id
|
||||
mock_get_access_token = mock_get_access_token
|
||||
|
||||
# Mock projects response with two projects
|
||||
mock_projects_response = MagicMock()
|
||||
mock_projects_response.status_code = 200
|
||||
mock_projects_response.json.return_value = [
|
||||
{"key": "PROJ1", "name": "Project One"},
|
||||
{"key": "PROJ2", "name": "Project Two"},
|
||||
]
|
||||
|
||||
# Mock successful issue types response for first project
|
||||
mock_issue_types_response_1 = MagicMock()
|
||||
mock_issue_types_response_1.status_code = 200
|
||||
mock_issue_types_response_1.json.return_value = {
|
||||
"projects": [
|
||||
{
|
||||
"issuetypes": [
|
||||
{"name": "Bug", "id": "1"},
|
||||
{"name": "Task", "id": "2"},
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
# Mock empty issue types response for second project
|
||||
mock_issue_types_response_2 = MagicMock()
|
||||
mock_issue_types_response_2.status_code = 200
|
||||
mock_issue_types_response_2.json.return_value = {"projects": []}
|
||||
|
||||
mock_get.side_effect = [
|
||||
mock_projects_response,
|
||||
mock_issue_types_response_1,
|
||||
mock_issue_types_response_2,
|
||||
]
|
||||
|
||||
projects_and_issue_types = self.jira_integration.get_metadata()
|
||||
|
||||
expected_result = {
|
||||
"PROJ1": {
|
||||
"name": "Project One",
|
||||
"issue_types": ["Bug", "Task"],
|
||||
},
|
||||
"PROJ2": {
|
||||
"name": "Project Two",
|
||||
"issue_types": [],
|
||||
},
|
||||
}
|
||||
|
||||
assert projects_and_issue_types == expected_result
|
||||
|
||||
Reference in New Issue
Block a user