mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 21:11:53 +00:00
chore: ruff format
This commit is contained in:
@@ -1055,9 +1055,9 @@ class TestProviderViewSet:
|
||||
|
||||
included_data = response.json()["included"]
|
||||
for expected_type in expected_resources:
|
||||
assert any(d.get("type") == expected_type for d in included_data), (
|
||||
f"Expected type '{expected_type}' not found in included data"
|
||||
)
|
||||
assert any(
|
||||
d.get("type") == expected_type for d in included_data
|
||||
), f"Expected type '{expected_type}' not found in included data"
|
||||
|
||||
def test_providers_retrieve(self, authenticated_client, providers_fixture):
|
||||
provider1, *_ = providers_fixture
|
||||
@@ -4919,13 +4919,13 @@ class TestAttackPathsScanViewSet:
|
||||
content_type=API_JSON_CONTENT_TYPE,
|
||||
)
|
||||
if i < 10:
|
||||
assert response.status_code == status.HTTP_200_OK, (
|
||||
f"Request {i + 1} should succeed with 200 OK, got {response.status_code}"
|
||||
)
|
||||
assert (
|
||||
response.status_code == status.HTTP_200_OK
|
||||
), f"Request {i + 1} should succeed with 200 OK, got {response.status_code}"
|
||||
else:
|
||||
assert response.status_code == status.HTTP_429_TOO_MANY_REQUESTS, (
|
||||
f"Request {i + 1} should be throttled"
|
||||
)
|
||||
assert (
|
||||
response.status_code == status.HTTP_429_TOO_MANY_REQUESTS
|
||||
), f"Request {i + 1} should be throttled"
|
||||
|
||||
# -- Timeout simulation -------------------------------------------------------
|
||||
|
||||
@@ -5128,9 +5128,9 @@ class TestResourceViewSet:
|
||||
|
||||
included_data = response.json()["included"]
|
||||
for expected_type in expected_resources:
|
||||
assert any(d.get("type") == expected_type for d in included_data), (
|
||||
f"Expected type '{expected_type}' not found in included data"
|
||||
)
|
||||
assert any(
|
||||
d.get("type") == expected_type for d in included_data
|
||||
), f"Expected type '{expected_type}' not found in included data"
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"filter_name, filter_value, expected_count",
|
||||
@@ -5679,9 +5679,9 @@ class TestResourceViewSet:
|
||||
(e for e in errors if e["source"]["parameter"] == expected_invalid_param),
|
||||
None,
|
||||
)
|
||||
assert error is not None, (
|
||||
f"Expected error for parameter '{expected_invalid_param}'"
|
||||
)
|
||||
assert (
|
||||
error is not None
|
||||
), f"Expected error for parameter '{expected_invalid_param}'"
|
||||
assert error["code"] == "invalid"
|
||||
assert error["status"] == "400" # Must be string per JSON:API spec
|
||||
assert expected_invalid_param in error["detail"]
|
||||
@@ -6213,16 +6213,16 @@ class TestResourceViewSet:
|
||||
# Test with completely malformed token
|
||||
client.credentials(HTTP_AUTHORIZATION="Bearer not.a.valid.jwt.token")
|
||||
response = client.get(reverse("resource-events", kwargs={"pk": resource.id}))
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED, (
|
||||
f"Expected 401 for malformed token but got {response.status_code}"
|
||||
)
|
||||
assert (
|
||||
response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
), f"Expected 401 for malformed token but got {response.status_code}"
|
||||
|
||||
# Test with empty bearer token
|
||||
client.credentials(HTTP_AUTHORIZATION="Bearer ")
|
||||
response = client.get(reverse("resource-events", kwargs={"pk": resource.id}))
|
||||
assert response.status_code == status.HTTP_401_UNAUTHORIZED, (
|
||||
f"Expected 401 for empty bearer token but got {response.status_code}"
|
||||
)
|
||||
assert (
|
||||
response.status_code == status.HTTP_401_UNAUTHORIZED
|
||||
), f"Expected 401 for empty bearer token but got {response.status_code}"
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@@ -6283,9 +6283,9 @@ class TestFindingViewSet:
|
||||
|
||||
included_data = response.json()["included"]
|
||||
for expected_type in expected_resources:
|
||||
assert any(d.get("type") == expected_type for d in included_data), (
|
||||
f"Expected type '{expected_type}' not found in included data"
|
||||
)
|
||||
assert any(
|
||||
d.get("type") == expected_type for d in included_data
|
||||
), f"Expected type '{expected_type}' not found in included data"
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"filter_name, filter_value, expected_count",
|
||||
@@ -6824,9 +6824,9 @@ class TestJWTFields:
|
||||
reverse("token-obtain"), data, format="json"
|
||||
)
|
||||
|
||||
assert response.status_code == status.HTTP_200_OK, (
|
||||
f"Unexpected status code: {response.status_code}"
|
||||
)
|
||||
assert (
|
||||
response.status_code == status.HTTP_200_OK
|
||||
), f"Unexpected status code: {response.status_code}"
|
||||
|
||||
access_token = response.data["attributes"]["access"]
|
||||
payload = jwt.decode(access_token, options={"verify_signature": False})
|
||||
@@ -6840,23 +6840,23 @@ class TestJWTFields:
|
||||
# Verify expected fields
|
||||
for field in expected_fields:
|
||||
assert field in payload, f"The field '{field}' is not in the JWT"
|
||||
assert payload[field] == expected_fields[field], (
|
||||
f"The value of '{field}' does not match"
|
||||
)
|
||||
assert (
|
||||
payload[field] == expected_fields[field]
|
||||
), f"The value of '{field}' does not match"
|
||||
|
||||
# Verify time fields are integers
|
||||
for time_field in ["exp", "iat", "nbf"]:
|
||||
assert time_field in payload, f"The field '{time_field}' is not in the JWT"
|
||||
assert isinstance(payload[time_field], int), (
|
||||
f"The field '{time_field}' is not an integer"
|
||||
)
|
||||
assert isinstance(
|
||||
payload[time_field], int
|
||||
), f"The field '{time_field}' is not an integer"
|
||||
|
||||
# Verify identification fields are non-empty strings
|
||||
for id_field in ["jti", "sub", "tenant_id"]:
|
||||
assert id_field in payload, f"The field '{id_field}' is not in the JWT"
|
||||
assert isinstance(payload[id_field], str) and payload[id_field], (
|
||||
f"The field '{id_field}' is not a valid string"
|
||||
)
|
||||
assert (
|
||||
isinstance(payload[id_field], str) and payload[id_field]
|
||||
), f"The field '{id_field}' is not a valid string"
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@@ -10795,9 +10795,9 @@ class TestIntegrationViewSet:
|
||||
|
||||
included_data = response.json()["included"]
|
||||
for expected_type in expected_resources:
|
||||
assert any(d.get("type") == expected_type for d in included_data), (
|
||||
f"Expected type '{expected_type}' not found in included data"
|
||||
)
|
||||
assert any(
|
||||
d.get("type") == expected_type for d in included_data
|
||||
), f"Expected type '{expected_type}' not found in included data"
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"integration_type, configuration, credentials",
|
||||
@@ -12234,9 +12234,9 @@ class TestLighthouseConfigViewSet:
|
||||
)
|
||||
# Check that API key is masked with asterisks only
|
||||
masked_api_key = data["attributes"]["api_key"]
|
||||
assert all(c == "*" for c in masked_api_key), (
|
||||
"API key should contain only asterisks"
|
||||
)
|
||||
assert all(
|
||||
c == "*" for c in masked_api_key
|
||||
), "API key should contain only asterisks"
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"field_name, invalid_value",
|
||||
@@ -15783,12 +15783,12 @@ class TestFindingGroupViewSet:
|
||||
assert response_p1.status_code == status.HTTP_200_OK
|
||||
p1_check_ids = {item["id"] for item in response_p1.json()["data"]}
|
||||
# Provider1 has scan1 with 4 checks
|
||||
assert len(p1_check_ids) == 4, (
|
||||
f"Provider1 should have 4 checks, got {len(p1_check_ids)}"
|
||||
)
|
||||
assert "cloudtrail_enabled" not in p1_check_ids, (
|
||||
"cloudtrail_enabled should NOT be in provider1"
|
||||
)
|
||||
assert (
|
||||
len(p1_check_ids) == 4
|
||||
), f"Provider1 should have 4 checks, got {len(p1_check_ids)}"
|
||||
assert (
|
||||
"cloudtrail_enabled" not in p1_check_ids
|
||||
), "cloudtrail_enabled should NOT be in provider1"
|
||||
|
||||
# Get finding groups for provider2 only
|
||||
response_p2 = authenticated_client.get(
|
||||
@@ -15798,12 +15798,12 @@ class TestFindingGroupViewSet:
|
||||
assert response_p2.status_code == status.HTTP_200_OK
|
||||
p2_check_ids = {item["id"] for item in response_p2.json()["data"]}
|
||||
# Provider2 has scan2 with 1 check
|
||||
assert len(p2_check_ids) == 1, (
|
||||
f"Provider2 should have 1 check, got {len(p2_check_ids)}"
|
||||
)
|
||||
assert "cloudtrail_enabled" in p2_check_ids, (
|
||||
"cloudtrail_enabled should be in provider2"
|
||||
)
|
||||
assert (
|
||||
len(p2_check_ids) == 1
|
||||
), f"Provider2 should have 1 check, got {len(p2_check_ids)}"
|
||||
assert (
|
||||
"cloudtrail_enabled" in p2_check_ids
|
||||
), "cloudtrail_enabled should be in provider2"
|
||||
|
||||
# Test provider_type filter actually filters data
|
||||
def test_finding_groups_provider_type_filter_actually_filters(
|
||||
@@ -15826,9 +15826,9 @@ class TestFindingGroupViewSet:
|
||||
{"filter[inserted_at]": TODAY, "filter[provider_type]": "gcp"},
|
||||
)
|
||||
assert response_gcp.status_code == status.HTTP_200_OK
|
||||
assert len(response_gcp.json()["data"]) == 0, (
|
||||
"GCP filter should return 0 results"
|
||||
)
|
||||
assert (
|
||||
len(response_gcp.json()["data"]) == 0
|
||||
), "GCP filter should return 0 results"
|
||||
|
||||
def test_finding_groups_pagination(
|
||||
self, authenticated_client, finding_groups_fixture
|
||||
|
||||
Reference in New Issue
Block a user