mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-05-06 08:47:18 +00:00
fix(sdk): add autouse mock_aws fixture and leak detector to prevent AWS test leaks (#10605)
This commit is contained in:
@@ -209,11 +209,11 @@ jobs:
|
||||
echo "AWS service_paths='${STEPS_AWS_SERVICES_OUTPUTS_SERVICE_PATHS}'"
|
||||
|
||||
if [ "${STEPS_AWS_SERVICES_OUTPUTS_RUN_ALL}" = "true" ]; then
|
||||
poetry run pytest -p no:randomly -n auto --cov=./prowler/providers/aws --cov-report=xml:aws_coverage.xml tests/providers/aws
|
||||
poetry run pytest -n auto --cov=./prowler/providers/aws --cov-report=xml:aws_coverage.xml tests/providers/aws
|
||||
elif [ -z "${STEPS_AWS_SERVICES_OUTPUTS_SERVICE_PATHS}" ]; then
|
||||
echo "No AWS service paths detected; skipping AWS tests."
|
||||
else
|
||||
poetry run pytest -p no:randomly -n auto --cov=./prowler/providers/aws --cov-report=xml:aws_coverage.xml ${STEPS_AWS_SERVICES_OUTPUTS_SERVICE_PATHS}
|
||||
poetry run pytest -n auto --cov=./prowler/providers/aws --cov-report=xml:aws_coverage.xml ${STEPS_AWS_SERVICES_OUTPUTS_SERVICE_PATHS}
|
||||
fi
|
||||
env:
|
||||
STEPS_AWS_SERVICES_OUTPUTS_RUN_ALL: ${{ steps.aws-services.outputs.run_all }}
|
||||
|
||||
@@ -14,6 +14,10 @@ All notable changes to the **Prowler SDK** are documented in this file.
|
||||
- Azure Network Watcher flow log checks now require workspace-backed Traffic Analytics for `network_flow_log_captured_sent` and align metadata with VNet-compatible flow log guidance [(#10645)](https://github.com/prowler-cloud/prowler/pull/10645)
|
||||
- Azure compliance entries for legacy Network Watcher flow log controls now use retirement-aware guidance and point new deployments to VNet flow logs
|
||||
|
||||
### 🐞 Fixed
|
||||
|
||||
- AWS SDK test isolation: autouse `mock_aws` fixture and leak detector in `conftest.py` to prevent tests from hitting real AWS endpoints, with idempotent organization setup for tests calling `set_mocked_aws_provider` multiple times [(#10605)](https://github.com/prowler-cloud/prowler/pull/10605)
|
||||
|
||||
---
|
||||
|
||||
## [5.25.1] (Prowler v5.25.1)
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
import pytest
|
||||
from unittest.mock import patch
|
||||
|
||||
from moto import mock_aws
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _mock_aws_globally():
|
||||
"""Activate moto's mock_aws for every test under tests/providers/aws/.
|
||||
|
||||
This prevents any test from accidentally hitting real AWS endpoints,
|
||||
even if it forgets to add @mock_aws on the method. Tests that never
|
||||
call boto3 are unaffected (mock_aws is a no-op in that case).
|
||||
"""
|
||||
with mock_aws():
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _detect_aws_leaks():
|
||||
"""Fail the test if any HTTP request reaches a real AWS endpoint."""
|
||||
calls = []
|
||||
original_send = None
|
||||
|
||||
try:
|
||||
from botocore.httpsession import URLLib3Session
|
||||
|
||||
original_send = URLLib3Session.send
|
||||
except ImportError:
|
||||
yield
|
||||
return
|
||||
|
||||
def tracking_send(self, request):
|
||||
url = getattr(request, "url", str(request))
|
||||
if ".amazonaws.com" in url:
|
||||
calls.append(url)
|
||||
return original_send(self, request)
|
||||
|
||||
with patch.object(URLLib3Session, "send", tracking_send):
|
||||
yield
|
||||
|
||||
if calls:
|
||||
pytest.fail(
|
||||
f"Test leaked {len(calls)} real AWS call(s):\n"
|
||||
+ "\n".join(f" - {url}" for url in calls[:5])
|
||||
)
|
||||
@@ -116,6 +116,12 @@ def set_mocked_aws_provider(
|
||||
status: list[str] = [],
|
||||
create_default_organization: bool = True,
|
||||
) -> AwsProvider:
|
||||
if audited_regions is None:
|
||||
raise ValueError(
|
||||
"audited_regions is None, which means all 36 regions will be used. "
|
||||
"Pass an explicit list of regions instead."
|
||||
)
|
||||
|
||||
if create_default_organization:
|
||||
# Create default AWS Organization
|
||||
create_default_aws_organization()
|
||||
@@ -191,7 +197,13 @@ def create_default_aws_organization():
|
||||
mockdomain = "moto-example.org"
|
||||
mockemail = "@".join([mockname, mockdomain])
|
||||
|
||||
_ = organizations_client.create_organization(FeatureSet="ALL")["Organization"]["Id"]
|
||||
try:
|
||||
_ = organizations_client.create_organization(FeatureSet="ALL")["Organization"][
|
||||
"Id"
|
||||
]
|
||||
except organizations_client.exceptions.AlreadyInOrganizationException:
|
||||
return
|
||||
|
||||
account_id = organizations_client.create_account(
|
||||
AccountName=mockname, Email=mockemail
|
||||
)["CreateAccountStatus"]["AccountId"]
|
||||
|
||||
Reference in New Issue
Block a user