mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-08-21 13:20:57 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8fcb8e53b7 | ||
|
|
84e9a91186 | ||
|
|
c1eceb9b5a | ||
|
|
082915cad9 | ||
|
|
ec6878c824 |
@@ -0,0 +1 @@
|
||||
`functiongraph_function_vpc_configured` check for Huawei Cloud provider: FunctionGraph functions are configured within a VPC
|
||||
@@ -384,6 +384,21 @@ class HuaweiCloudSession:
|
||||
.build()
|
||||
)
|
||||
|
||||
elif service == "functiongraph":
|
||||
from huaweicloudsdkfunctiongraph.v2 import FunctionGraphClient
|
||||
from huaweicloudsdkfunctiongraph.v2.region.functiongraph_region import (
|
||||
FunctionGraphRegion,
|
||||
)
|
||||
|
||||
client_region = region or self._region
|
||||
return (
|
||||
FunctionGraphClient.new_builder()
|
||||
.with_credentials(self._get_basic_credentials(client_region))
|
||||
.with_http_config(self._http_config())
|
||||
.with_region(_aligned_region(FunctionGraphRegion, client_region))
|
||||
.build()
|
||||
)
|
||||
|
||||
else:
|
||||
raise HuaweiCloudServiceError(
|
||||
message=f"Huawei Cloud service '{service}' is not supported"
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
from prowler.providers.common.provider import Provider
|
||||
from prowler.providers.huaweicloud.services.functiongraph.functiongraph_service import (
|
||||
FunctionGraph,
|
||||
)
|
||||
|
||||
functiongraph_client = FunctionGraph(Provider.get_global_provider())
|
||||
+36
@@ -0,0 +1,36 @@
|
||||
{
|
||||
"Provider": "huaweicloud",
|
||||
"CheckID": "functiongraph_function_vpc_configured",
|
||||
"CheckTitle": "FunctionGraph functions are configured within a VPC",
|
||||
"CheckType": [],
|
||||
"ServiceName": "functiongraph",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "FunctionGraphFunction",
|
||||
"ResourceGroup": "serverless",
|
||||
"Description": "Ensure FunctionGraph functions are attached to a VPC so workloads that require private connectivity can use VPC network controls.",
|
||||
"Risk": "Functions without a VPC attachment cannot privately access VPC resources such as databases or internal services. They may depend on public service endpoints, reducing network isolation and increasing exposure to public-network paths.",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://support.huaweicloud.com/intl/en-us/api-functiongraph/functiongraph_06_0111.html"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "In the FunctionGraph console, open the function, edit its network configuration, select the required VPC, subnet, and security groups, and save the configuration.",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Attach functions that require private resource access to an appropriate VPC, subnet, and least-privilege security groups.",
|
||||
"Url": "https://hub.prowler.com/check/functiongraph_function_vpc_configured"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"trust-boundaries"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": ""
|
||||
}
|
||||
+29
@@ -0,0 +1,29 @@
|
||||
from prowler.lib.check.models import Check, CheckReportHuaweiCloud
|
||||
from prowler.providers.huaweicloud.services.functiongraph.functiongraph_client import (
|
||||
functiongraph_client,
|
||||
)
|
||||
|
||||
|
||||
class functiongraph_function_vpc_configured(Check):
|
||||
"""Check if FunctionGraph functions are configured within a VPC."""
|
||||
|
||||
def execute(self) -> list[CheckReportHuaweiCloud]:
|
||||
findings = []
|
||||
for function in functiongraph_client.functions:
|
||||
report = CheckReportHuaweiCloud(
|
||||
metadata=self.metadata(),
|
||||
resource=function,
|
||||
)
|
||||
|
||||
if function.vpc_id:
|
||||
report.status = "PASS"
|
||||
report.status_extended = f"Function '{function.name}' is configured within VPC '{function.vpc_id}'."
|
||||
else:
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
f"Function '{function.name}' is not configured within a VPC."
|
||||
)
|
||||
|
||||
findings.append(report)
|
||||
|
||||
return findings
|
||||
@@ -0,0 +1,118 @@
|
||||
from typing import List, Optional
|
||||
|
||||
from prowler.lib.logger import logger
|
||||
from prowler.lib.scan_filters.scan_filters import is_resource_filtered
|
||||
from prowler.providers.huaweicloud.lib.service.service import HuaweiCloudService
|
||||
from prowler.providers.huaweicloud.models import HuaweiCloudBaseModel
|
||||
|
||||
|
||||
class FunctionGraph(HuaweiCloudService):
|
||||
"""
|
||||
FunctionGraph service class for Huawei Cloud.
|
||||
|
||||
This class provides methods to interact with Huawei Cloud FunctionGraph service
|
||||
to retrieve serverless functions and their security configuration.
|
||||
"""
|
||||
|
||||
def __init__(self, provider):
|
||||
"""Initialize FunctionGraph and discover functions in enabled regions.
|
||||
|
||||
Args:
|
||||
provider: The Huawei Cloud provider used to create regional clients.
|
||||
"""
|
||||
super().__init__(__class__.__name__, provider)
|
||||
|
||||
self.functions: List[FunctionGraphFunction] = []
|
||||
|
||||
if getattr(self.session, "is_mock", False):
|
||||
self._load_mock_data()
|
||||
return
|
||||
|
||||
self.__threading_call__(
|
||||
self._list_functions, iterator=self.regional_clients.items()
|
||||
)
|
||||
|
||||
def _load_mock_data(self):
|
||||
"""Load mock data for testing."""
|
||||
region = "la-south-2"
|
||||
self.functions = [
|
||||
FunctionGraphFunction(
|
||||
id="fg-mock-001",
|
||||
name="function-secure",
|
||||
arn="urn:fss:la-south-2:project:function:default:function-secure:latest",
|
||||
runtime="Python3.9",
|
||||
timeout=30,
|
||||
memory_size=128,
|
||||
vpc_id="vpc-12345",
|
||||
region=region,
|
||||
),
|
||||
FunctionGraphFunction(
|
||||
id="fg-mock-002",
|
||||
name="function-insecure",
|
||||
arn="urn:fss:la-south-2:project:function:default:function-insecure:latest",
|
||||
runtime="Python3.9",
|
||||
timeout=30,
|
||||
memory_size=128,
|
||||
vpc_id=None,
|
||||
region=region,
|
||||
),
|
||||
]
|
||||
|
||||
def _list_functions(self, regional_client_item):
|
||||
"""List every FunctionGraph function in one region."""
|
||||
region, regional_client = regional_client_item
|
||||
logger.info(f"FunctionGraph - Listing Functions in {region}...")
|
||||
discovered_functions = []
|
||||
marker = None
|
||||
|
||||
try:
|
||||
from huaweicloudsdkfunctiongraph.v2 import ListFunctionsRequest
|
||||
|
||||
while True:
|
||||
request = ListFunctionsRequest(marker=marker, maxitems="400")
|
||||
response = self._call_with_retries(
|
||||
regional_client.list_functions, request
|
||||
)
|
||||
|
||||
for function in getattr(response, "functions", None) or []:
|
||||
resource_id = getattr(function, "resource_id", None) or ""
|
||||
if self.audit_resources and not is_resource_filtered(
|
||||
resource_id, self.audit_resources
|
||||
):
|
||||
continue
|
||||
discovered_functions.append(
|
||||
FunctionGraphFunction(
|
||||
id=resource_id,
|
||||
name=getattr(function, "func_name", None) or resource_id,
|
||||
arn=getattr(function, "func_urn", None) or "",
|
||||
runtime=getattr(function, "runtime", None) or "",
|
||||
timeout=getattr(function, "timeout", None) or 0,
|
||||
memory_size=getattr(function, "memory_size", None) or 0,
|
||||
vpc_id=getattr(function, "func_vpc_id", None),
|
||||
region=region,
|
||||
)
|
||||
)
|
||||
|
||||
next_marker = getattr(response, "next_marker", None)
|
||||
if not next_marker or str(next_marker) == marker:
|
||||
break
|
||||
marker = str(next_marker)
|
||||
|
||||
self.functions.extend(discovered_functions)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
|
||||
class FunctionGraphFunction(HuaweiCloudBaseModel):
|
||||
"""FunctionGraph function model."""
|
||||
|
||||
id: str
|
||||
name: str = ""
|
||||
arn: str = ""
|
||||
runtime: str = ""
|
||||
timeout: int = 0
|
||||
memory_size: int = 0
|
||||
vpc_id: Optional[str] = None
|
||||
region: str = ""
|
||||
@@ -123,6 +123,7 @@ dependencies = [
|
||||
"huaweicloudsdkecs==3.1.204",
|
||||
"huaweicloudsdkelb==3.1.204",
|
||||
"huaweicloudsdkevs==3.1.204",
|
||||
"huaweicloudsdkfunctiongraph==3.1.204",
|
||||
"huaweicloudsdkiam==3.1.204",
|
||||
"huaweicloudsdkkms==3.1.204",
|
||||
"huaweicloudsdkobs==3.1.204",
|
||||
@@ -260,6 +261,7 @@ constraint-dependencies = [
|
||||
"huaweicloudsdkecs==3.1.204",
|
||||
"huaweicloudsdkelb==3.1.204",
|
||||
"huaweicloudsdkevs==3.1.204",
|
||||
"huaweicloudsdkfunctiongraph==3.1.204",
|
||||
"huaweicloudsdkiam==3.1.204",
|
||||
"huaweicloudsdkkms==3.1.204",
|
||||
"huaweicloudsdkobs==3.1.204",
|
||||
|
||||
@@ -50,6 +50,31 @@ class TestHuaweiCloudProviderSetupSession:
|
||||
session = HuaweicloudProvider.setup_session()
|
||||
assert session.get_credentials().ak == ACCESS_KEY
|
||||
|
||||
def test_creates_functiongraph_client_for_europe(self):
|
||||
session = HuaweiCloudSession(
|
||||
HuaweiCloudCredentials(ak=ACCESS_KEY, sk=SECRET_KEY)
|
||||
)
|
||||
functiongraph_client = mock.MagicMock()
|
||||
builder = mock.MagicMock()
|
||||
builder.with_credentials.return_value.with_http_config.return_value.with_region.return_value.build.return_value = (
|
||||
functiongraph_client
|
||||
)
|
||||
|
||||
with mock.patch(
|
||||
"huaweicloudsdkfunctiongraph.v2.FunctionGraphClient.new_builder",
|
||||
return_value=builder,
|
||||
):
|
||||
client = session.client("functiongraph", "eu-west-101")
|
||||
|
||||
assert client is functiongraph_client
|
||||
region = builder.with_credentials.return_value.with_http_config.return_value.with_region.call_args.args[
|
||||
0
|
||||
]
|
||||
assert region.id == "eu-west-101"
|
||||
assert region.endpoints == [
|
||||
"https://functiongraph.eu-west-101.myhuaweicloud.eu"
|
||||
]
|
||||
|
||||
|
||||
class TestHuaweiCloudProviderValidateCredentials:
|
||||
def test_resolves_caller_identity_from_iam(self):
|
||||
|
||||
+184
@@ -0,0 +1,184 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.lib.check.models import CheckMetadata
|
||||
from tests.providers.huaweicloud.huaweicloud_fixtures import (
|
||||
set_mocked_huaweicloud_provider,
|
||||
)
|
||||
|
||||
|
||||
class Test_functiongraph_function_vpc_configured:
|
||||
def test_metadata_describes_vpc_attachment_accurately(self):
|
||||
metadata = CheckMetadata.parse_file(
|
||||
"prowler/providers/huaweicloud/services/functiongraph/"
|
||||
"functiongraph_function_vpc_configured/"
|
||||
"functiongraph_function_vpc_configured.metadata.json"
|
||||
)
|
||||
|
||||
assert metadata.ResourceIdTemplate == ""
|
||||
assert "direct internet access" not in metadata.Risk
|
||||
assert metadata.AdditionalURLs == [
|
||||
"https://support.huaweicloud.com/intl/en-us/api-functiongraph/functiongraph_06_0111.html"
|
||||
]
|
||||
assert metadata.Remediation.Code.Other
|
||||
|
||||
def test_functiongraph_vpc_configured_pass(self):
|
||||
functiongraph_client = mock.MagicMock()
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_huaweicloud_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.huaweicloud.services.functiongraph.functiongraph_function_vpc_configured.functiongraph_function_vpc_configured.functiongraph_client",
|
||||
new=functiongraph_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.huaweicloud.services.functiongraph.functiongraph_function_vpc_configured.functiongraph_function_vpc_configured import (
|
||||
functiongraph_function_vpc_configured,
|
||||
)
|
||||
from prowler.providers.huaweicloud.services.functiongraph.functiongraph_service import (
|
||||
FunctionGraphFunction,
|
||||
)
|
||||
|
||||
functiongraph_client.functions = [
|
||||
FunctionGraphFunction(
|
||||
id="fg-001",
|
||||
name="function-secure",
|
||||
arn="urn:fss:la-south-2:project:function:default:function-secure:latest",
|
||||
runtime="Python3.9",
|
||||
timeout=30,
|
||||
memory_size=128,
|
||||
vpc_id="vpc-12345",
|
||||
region="la-south-2",
|
||||
),
|
||||
]
|
||||
functiongraph_client.audited_account = "123456789012"
|
||||
|
||||
check = functiongraph_function_vpc_configured()
|
||||
results = check.execute()
|
||||
|
||||
assert len(results) == 1
|
||||
assert results[0].status == "PASS"
|
||||
assert results[0].resource_id == "fg-001"
|
||||
assert results[0].resource_name == "function-secure"
|
||||
assert results[0].resource_arn.endswith("function-secure:latest")
|
||||
assert "configured within VPC" in results[0].status_extended
|
||||
|
||||
def test_functiongraph_vpc_configured_fail(self):
|
||||
functiongraph_client = mock.MagicMock()
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_huaweicloud_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.huaweicloud.services.functiongraph.functiongraph_function_vpc_configured.functiongraph_function_vpc_configured.functiongraph_client",
|
||||
new=functiongraph_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.huaweicloud.services.functiongraph.functiongraph_function_vpc_configured.functiongraph_function_vpc_configured import (
|
||||
functiongraph_function_vpc_configured,
|
||||
)
|
||||
from prowler.providers.huaweicloud.services.functiongraph.functiongraph_service import (
|
||||
FunctionGraphFunction,
|
||||
)
|
||||
|
||||
functiongraph_client.functions = [
|
||||
FunctionGraphFunction(
|
||||
id="fg-002",
|
||||
name="function-insecure",
|
||||
arn="urn:fss:la-south-2:project:function:default:function-insecure:latest",
|
||||
runtime="Python3.9",
|
||||
timeout=30,
|
||||
memory_size=128,
|
||||
vpc_id=None,
|
||||
region="la-south-2",
|
||||
),
|
||||
]
|
||||
functiongraph_client.audited_account = "123456789012"
|
||||
|
||||
check = functiongraph_function_vpc_configured()
|
||||
results = check.execute()
|
||||
|
||||
assert len(results) == 1
|
||||
assert results[0].status == "FAIL"
|
||||
assert results[0].resource_id == "fg-002"
|
||||
assert "not configured within a VPC" in results[0].status_extended
|
||||
|
||||
def test_functiongraph_vpc_configured_mixed(self):
|
||||
functiongraph_client = mock.MagicMock()
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_huaweicloud_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.huaweicloud.services.functiongraph.functiongraph_function_vpc_configured.functiongraph_function_vpc_configured.functiongraph_client",
|
||||
new=functiongraph_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.huaweicloud.services.functiongraph.functiongraph_function_vpc_configured.functiongraph_function_vpc_configured import (
|
||||
functiongraph_function_vpc_configured,
|
||||
)
|
||||
from prowler.providers.huaweicloud.services.functiongraph.functiongraph_service import (
|
||||
FunctionGraphFunction,
|
||||
)
|
||||
|
||||
functiongraph_client.functions = [
|
||||
FunctionGraphFunction(
|
||||
id="fg-001",
|
||||
name="function-secure",
|
||||
arn="urn:fss:la-south-2:project:function:default:function-secure:latest",
|
||||
runtime="Python3.9",
|
||||
timeout=30,
|
||||
memory_size=128,
|
||||
vpc_id="vpc-12345",
|
||||
region="la-south-2",
|
||||
),
|
||||
FunctionGraphFunction(
|
||||
id="fg-002",
|
||||
name="function-insecure",
|
||||
arn="urn:fss:la-south-2:project:function:default:function-insecure:latest",
|
||||
runtime="Python3.9",
|
||||
timeout=30,
|
||||
memory_size=128,
|
||||
vpc_id=None,
|
||||
region="la-south-2",
|
||||
),
|
||||
]
|
||||
functiongraph_client.audited_account = "123456789012"
|
||||
|
||||
check = functiongraph_function_vpc_configured()
|
||||
results = check.execute()
|
||||
|
||||
assert len(results) == 2
|
||||
assert results[0].status == "PASS"
|
||||
assert results[1].status == "FAIL"
|
||||
|
||||
def test_functiongraph_vpc_configured_empty(self):
|
||||
functiongraph_client = mock.MagicMock()
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_huaweicloud_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.huaweicloud.services.functiongraph.functiongraph_function_vpc_configured.functiongraph_function_vpc_configured.functiongraph_client",
|
||||
new=functiongraph_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.huaweicloud.services.functiongraph.functiongraph_function_vpc_configured.functiongraph_function_vpc_configured import (
|
||||
functiongraph_function_vpc_configured,
|
||||
)
|
||||
|
||||
functiongraph_client.functions = []
|
||||
functiongraph_client.audited_account = "123456789012"
|
||||
|
||||
check = functiongraph_function_vpc_configured()
|
||||
results = check.execute()
|
||||
|
||||
assert len(results) == 0
|
||||
+139
@@ -0,0 +1,139 @@
|
||||
from types import SimpleNamespace
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.huaweicloud.services.functiongraph.functiongraph_service import (
|
||||
FunctionGraph,
|
||||
)
|
||||
from tests.providers.huaweicloud.huaweicloud_fixtures import (
|
||||
set_mocked_huaweicloud_provider,
|
||||
)
|
||||
|
||||
|
||||
class TestFunctionGraphService:
|
||||
def test_real_session_lists_functions(self):
|
||||
function = SimpleNamespace(
|
||||
resource_id="resource-1",
|
||||
func_urn="urn:fss:eu-west-101:project-1:function:default:first:latest",
|
||||
func_name="first",
|
||||
runtime="Python3.9",
|
||||
timeout=30,
|
||||
memory_size=128,
|
||||
func_vpc_id=None,
|
||||
)
|
||||
regional_client = SimpleNamespace(list_functions=mock.MagicMock())
|
||||
regional_client.list_functions.return_value = SimpleNamespace(
|
||||
functions=[function], next_marker=None
|
||||
)
|
||||
provider = set_mocked_huaweicloud_provider()
|
||||
provider.session = SimpleNamespace(client=mock.MagicMock())
|
||||
provider.generate_regional_clients = mock.MagicMock(
|
||||
return_value={"eu-west-101": regional_client}
|
||||
)
|
||||
provider.get_default_region = mock.MagicMock(return_value="eu-west-101")
|
||||
|
||||
service = FunctionGraph(provider)
|
||||
|
||||
assert len(service.functions) == 1
|
||||
assert service.functions[0].region == "eu-west-101"
|
||||
regional_client.list_functions.assert_called_once()
|
||||
|
||||
def test_lists_every_page_and_maps_sdk_fields(self):
|
||||
first_function = SimpleNamespace(
|
||||
resource_id="resource-1",
|
||||
func_urn="urn:fss:eu-west-101:project-1:function:default:first:latest",
|
||||
func_name="first",
|
||||
runtime=None,
|
||||
timeout=None,
|
||||
memory_size=None,
|
||||
func_vpc_id=None,
|
||||
)
|
||||
second_function = SimpleNamespace(
|
||||
resource_id="resource-2",
|
||||
func_urn="urn:fss:eu-west-101:project-1:function:default:second:latest",
|
||||
func_name="second",
|
||||
runtime="Python3.9",
|
||||
timeout=30,
|
||||
memory_size=128,
|
||||
func_vpc_id="vpc-1",
|
||||
)
|
||||
regional_client = mock.MagicMock(region="eu-west-101")
|
||||
regional_client.list_functions.side_effect = [
|
||||
SimpleNamespace(functions=[first_function], next_marker=400),
|
||||
SimpleNamespace(functions=[second_function], next_marker=None),
|
||||
]
|
||||
service = FunctionGraph.__new__(FunctionGraph)
|
||||
service.functions = []
|
||||
service.audit_resources = []
|
||||
service._call_with_retries = mock.MagicMock(
|
||||
side_effect=regional_client.list_functions.side_effect
|
||||
)
|
||||
|
||||
service._list_functions(("eu-west-101", regional_client))
|
||||
|
||||
assert [function.id for function in service.functions] == [
|
||||
"resource-1",
|
||||
"resource-2",
|
||||
]
|
||||
assert service.functions[0].arn == first_function.func_urn
|
||||
assert service.functions[0].runtime == ""
|
||||
assert service.functions[0].timeout == 0
|
||||
assert service.functions[0].memory_size == 0
|
||||
assert service.functions[0].vpc_id is None
|
||||
assert service.functions[1].vpc_id == "vpc-1"
|
||||
assert service._call_with_retries.call_count == 2
|
||||
first_request = service._call_with_retries.call_args_list[0].args[1]
|
||||
second_request = service._call_with_retries.call_args_list[1].args[1]
|
||||
assert first_request.maxitems == "400"
|
||||
assert first_request.marker is None
|
||||
assert second_request.marker == "400"
|
||||
|
||||
def test_api_error_does_not_fabricate_functions(self):
|
||||
function = SimpleNamespace(
|
||||
resource_id="resource-1",
|
||||
func_urn="urn:fss:eu-west-101:project-1:function:default:first:latest",
|
||||
func_name="first",
|
||||
runtime="Python3.9",
|
||||
timeout=30,
|
||||
memory_size=128,
|
||||
func_vpc_id=None,
|
||||
)
|
||||
regional_client = mock.MagicMock(region="eu-west-101")
|
||||
service = FunctionGraph.__new__(FunctionGraph)
|
||||
service.functions = []
|
||||
service.audit_resources = []
|
||||
service._call_with_retries = mock.MagicMock(
|
||||
side_effect=[
|
||||
SimpleNamespace(functions=[function], next_marker=400),
|
||||
Exception("boom"),
|
||||
]
|
||||
)
|
||||
|
||||
service._list_functions(("eu-west-101", regional_client))
|
||||
|
||||
assert service.functions == []
|
||||
|
||||
def test_repeated_marker_stops_pagination(self):
|
||||
function = SimpleNamespace(
|
||||
resource_id="resource-1",
|
||||
func_urn="urn:fss:eu-west-101:project-1:function:default:first:latest",
|
||||
func_name="first",
|
||||
runtime="Python3.9",
|
||||
timeout=30,
|
||||
memory_size=128,
|
||||
func_vpc_id=None,
|
||||
)
|
||||
regional_client = mock.MagicMock(region="eu-west-101")
|
||||
service = FunctionGraph.__new__(FunctionGraph)
|
||||
service.functions = []
|
||||
service.audit_resources = []
|
||||
service._call_with_retries = mock.MagicMock(
|
||||
side_effect=[
|
||||
SimpleNamespace(functions=[function], next_marker=400),
|
||||
SimpleNamespace(functions=[], next_marker=400),
|
||||
]
|
||||
)
|
||||
|
||||
service._list_functions(("eu-west-101", regional_client))
|
||||
|
||||
assert [item.id for item in service.functions] == ["resource-1"]
|
||||
assert service._call_with_retries.call_count == 2
|
||||
@@ -100,6 +100,7 @@ constraints = [
|
||||
{ name = "huaweicloudsdkecs", specifier = "==3.1.204" },
|
||||
{ name = "huaweicloudsdkelb", specifier = "==3.1.204" },
|
||||
{ name = "huaweicloudsdkevs", specifier = "==3.1.204" },
|
||||
{ name = "huaweicloudsdkfunctiongraph", specifier = "==3.1.204" },
|
||||
{ name = "huaweicloudsdkiam", specifier = "==3.1.204" },
|
||||
{ name = "huaweicloudsdkkms", specifier = "==3.1.204" },
|
||||
{ name = "huaweicloudsdkobs", specifier = "==3.1.204" },
|
||||
@@ -2310,6 +2311,17 @@ wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/f7/cf/531dc55fd9d0f3bbd3eef24c7e4d78c6a1ba8eb80fa506e3574d72bcc98a/huaweicloudsdkevs-3.1.204-py3-none-any.whl", hash = "sha256:9118ac4c576e54aa7eaa926949e2b6824c5f038a2274b51d9a304d37fc0d7e2f", size = 251404, upload-time = "2026-07-09T09:02:50.05Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "huaweicloudsdkfunctiongraph"
|
||||
version = "3.1.204"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
dependencies = [
|
||||
{ name = "huaweicloudsdkcore" },
|
||||
]
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/1a/10/19e2f8a0f15aa6c3c1b2f67eb58e5d06aac0b3eef527283624acfd946b0a/huaweicloudsdkfunctiongraph-3.1.204-py3-none-any.whl", hash = "sha256:0999e57f56b157817c9d8d68940279feff6c1170693e2f41eef21a1fd12a7de9", size = 556873, upload-time = "2026-07-09T09:02:52.402Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "huaweicloudsdkiam"
|
||||
version = "3.1.204"
|
||||
@@ -3753,6 +3765,7 @@ dependencies = [
|
||||
{ name = "huaweicloudsdkecs" },
|
||||
{ name = "huaweicloudsdkelb" },
|
||||
{ name = "huaweicloudsdkevs" },
|
||||
{ name = "huaweicloudsdkfunctiongraph" },
|
||||
{ name = "huaweicloudsdkiam" },
|
||||
{ name = "huaweicloudsdkkms" },
|
||||
{ name = "huaweicloudsdkobs" },
|
||||
@@ -3873,6 +3886,7 @@ requires-dist = [
|
||||
{ name = "huaweicloudsdkecs", specifier = "==3.1.204" },
|
||||
{ name = "huaweicloudsdkelb", specifier = "==3.1.204" },
|
||||
{ name = "huaweicloudsdkevs", specifier = "==3.1.204" },
|
||||
{ name = "huaweicloudsdkfunctiongraph", specifier = "==3.1.204" },
|
||||
{ name = "huaweicloudsdkiam", specifier = "==3.1.204" },
|
||||
{ name = "huaweicloudsdkkms", specifier = "==3.1.204" },
|
||||
{ name = "huaweicloudsdkobs", specifier = "==3.1.204" },
|
||||
|
||||
Reference in New Issue
Block a user