[Integration][GCP] | Optionally convert resource fields to camelCase …

…for non cloud asset apis (#1118)

# Description

**What** - Resolves an issue with Google Cloud Pub/Sub integration where
key names were returned in `snake_case` instead of camelCase.

**Why** - The integration is returning these on snake case which have
[affected data model.](https://getport.atlassian.net/browse/PORT-11122)

**How** - Set the `preserving_proto_field_name` flag to False within the
to_dict method. This change ensures that all field names are converted
to `camelCase` by default, bypassing the need to manually handle key
renaming.

## Type of change

Please leave one option from the following and delete the rest:

- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] New Integration (non-breaking change which adds a new integration)
- [ ] Breaking change (fix or feature that would cause existing
functionality to not work as expected)
- [ ] Non-breaking change (fix of existing functionality that will not
change current behavior)
- [ ] Documentation (added/updated documentation)

<h4> All tests should be run against the port production
environment(using a testing org). </h4>

### Core testing checklist

- [x] Integration able to create all default resources from scratch
- [x] Resync finishes successfully
- [x] Resync able to create entities
- [x] Resync able to update entities
- [x] Resync able to detect and delete entities
- [x] Scheduled resync able to abort existing resync and start a new one
- [x] Tested with at least 2 integrations from scratch
- [ ] Tested with Kafka and Polling event listeners
- [ ] Tested deletion of entities that don't pass the selector


### Integration testing checklist

- [x] Integration able to create all default resources from scratch
- [x] Resync able to create entities
- [x] Resync able to update entities
- [x] Resync able to detect and delete entities
- [x] Resync finishes successfully
- [ ] If new resource kind is added or updated in the integration, add
example raw data, mapping and expected result to the `examples` folder
in the integration directory.
- [ ] If resource kind is updated, run the integration with the example
data and check if the expected result is achieved
- [ ] If new resource kind is added or updated, validate that
live-events for that resource are working as expected
- [ ] Docs PR link [here](#)

### Preflight checklist

- [ ] Handled rate limiting
- [ ] Handled pagination
- [ ] Implemented the code in async
- [ ] Support Multi account

## Screenshots

Include screenshots from your environment showing how the resources of
the integration will look.

## API Documentation

Provide links to the API documentation used for this integration.

---------

Co-authored-by: PagesCoffy <[email protected]>
Co-authored-by: Michael Kofi Armah <[email protected]>
Co-authored-by: Matan <[email protected]>

integrations/gcp/CHANGELOG.md

@@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 <!-- towncrier release notes start -->
 
+## 0.1.76 (2024-12-09)
+
+
+### Improvements
+
+- Added `preserveApiResponseCaseStyle` selector to optionally convert resource fields to and from `snake_case` and `camelCase` for non-cloud asset APIs.
+
+
 ## 0.1.75 (2024-12-04)
 
 

integrations/gcp/gcp_core/overrides.py

@@ -17,7 +17,25 @@ class GCPCloudResourceConfig(ResourceConfig):
     selector: GCPCloudResourceSelector
 
 
+class GCPResourceSelector(Selector):
+    preserve_api_response_case_style: bool | None = Field(
+        default=None,
+        alias="preserveApiResponseCaseStyle",
+        description=(
+            "Controls whether to preserve the Google Cloud API's original field format instead of using protobuf's default snake case. "
+            "When False (default): Uses protobuf's default snake_case format (existing behavior). "
+            "When True: Preserves the specific API's original format (e.g., camelCase for PubSub). "
+            "If not set, defaults to False to maintain existing behavior (snake_case for all APIs)."
+            "Note that this setting does not affect resources fetched from the cloud asset API"
+        ),
+    )
+
+
+class GCPResourceConfig(ResourceConfig):
+    selector: GCPResourceSelector
+
+
 class GCPPortAppConfig(PortAppConfig):
-    resources: list[GCPCloudResourceConfig | ResourceConfig] = Field(
-        default_factory=list
+    resources: list[GCPCloudResourceConfig | GCPResourceConfig | ResourceConfig] = (
+        Field(default_factory=list)
     )

integrations/gcp/gcp_core/utils.py

@@ -5,6 +5,7 @@
 from collections.abc import MutableSequence
 from typing import Any, TypedDict, Tuple
 
+from gcp_core.errors import ResourceNotFoundError
 from loguru import logger
 import proto  # type: ignore
 from port_ocean.context.event import event
@@ -42,14 +43,56 @@ class AssetData(TypedDict):
 
 
 def parse_latest_resource_from_asset(asset_data: AssetData) -> dict[Any, Any]:
-    max_versioned_resource_data = max(
-        asset_data["versioned_resources"], key=lambda x: x["version"]
+    """
+    Parse the latest version of a resource from asset data.
+
+    Attempts to find the versioned resources using either snake_case or camelCase key,
+    as the input format depends on how the asset data was originally serialized.
+
+    Args:
+        asset_data: Asset data containing versioned resources
+
+    Returns:
+        dict: The most recent version of the resource
+
+    Raises:
+        ResourceNotFoundError: If neither versioned_resources nor versionedResources is found
+    """
+    # Try both key formats since we don't control the input format
+    versioned_resources = asset_data.get("versioned_resources") or asset_data.get(
+        "versionedResources"
     )
+    if not isinstance(versioned_resources, list):
+        raise ResourceNotFoundError(
+            "Could not find versioned resources under either 'versioned_resources' or 'versionedResources'. "
+            "Please ensure the asset data contains a list of versioned resources in the expected format."
+        )
+
+    # Ensure each item in the list is a VersionedResource
+    versioned_resources = typing.cast(list[VersionedResource], versioned_resources)
+
+    max_versioned_resource_data = max(versioned_resources, key=lambda x: x["version"])
     return max_versioned_resource_data["resource"]
 
 
+def should_use_snake_case() -> bool:
+    """
+    Determines whether to use snake_case for field names based on preserve_api_response_case_style config.
+
+    Returns:
+        bool: True to use snake_case, False to preserve API's original case style
+    """
+    selector = get_current_resource_config().selector
+    preserve_api_case = getattr(selector, "preserve_api_response_case_style", False)
+    return not preserve_api_case
+
+
 def parse_protobuf_message(message: proto.Message) -> dict[str, Any]:
-    return proto.Message.to_dict(message)
+    """
+    Parse protobuf message to dict, controlling field name case style.
+    """
+    use_snake_case = should_use_snake_case()
+    return proto.Message.to_dict(message, preserving_proto_field_name=use_snake_case)
 
 
 def parse_protobuf_messages(

integrations/gcp/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "gcp"
-version = "0.1.75"
+version = "0.1.76"
 description = "A GCP ocean integration"
 authors = ["Matan Geva <[email protected]>"]
 

integrations/gcp/tests/gcp_core/search/test_resource_searches.py

@@ -1,5 +1,8 @@
-from typing import Any
-from unittest.mock import AsyncMock, patch
+from typing import Any, Generator
+from unittest.mock import AsyncMock, patch, MagicMock
+import pytest
+from port_ocean.context.event import event_context
+from port_ocean.context.ocean import initialize_port_ocean_context
 
 from port_ocean.core.ocean_types import ASYNC_GENERATOR_RESYNC_TYPE
 from google.pubsub_v1.types import pubsub
@@ -13,10 +16,25 @@ async def mock_subscription_pages(
     yield [{"name": "subscription_3"}, {"name": "subscription_4"}]  # Second page
 
 
-@patch(
-    "port_ocean.context.ocean.PortOceanContext.integration_config",
-    return_value={"search_all_resources_per_minute_quota": 100},
-)
+@pytest.fixture(autouse=True)
+def mock_ocean_context() -> None:
+    """Fixture to initialize the PortOcean context."""
+    mock_app = MagicMock()
+    mock_app.config.integration.config = {"search_all_resources_per_minute_quota": 100}
+    initialize_port_ocean_context(mock_app)
+
+
+@pytest.fixture
+def integration_config_mock() -> Generator[Any, Any, Any]:
+    """Fixture to mock integration configuration."""
+    with patch(
+        "port_ocean.context.ocean.PortOceanContext.integration_config",
+        new_callable=MagicMock,
+    ) as mock:
+        yield mock
+
+
+@pytest.mark.asyncio
 @patch("gcp_core.search.paginated_query.paginated_query", new=mock_subscription_pages)
 @patch("google.pubsub_v1.services.subscriber.SubscriberAsyncClient", new=AsyncMock)
 async def test_list_all_subscriptions_per_project(integration_config_mock: Any) -> None:
@@ -41,12 +59,10 @@ async def test_list_all_subscriptions_per_project(integration_config_mock: Any)
     assert actual_subscriptions == expected_subscriptions
 
 
-@patch(
-    "port_ocean.context.ocean.PortOceanContext.integration_config",
-    return_value={"search_all_resources_per_minute_quota": 100},
-)
+@pytest.mark.asyncio
+@patch("gcp_core.utils.get_current_resource_config")
 async def test_get_single_subscription(
-    integration_config: Any, monkeypatch: Any
+    get_current_resource_config_mock: MagicMock, monkeypatch: Any
 ) -> None:
     # Arrange
     subscriber_async_client_mock = AsyncMock
@@ -59,6 +75,11 @@ async def test_get_single_subscription(
         {"name": "subscription_name"}
     )
 
+    # Mock the resource config
+    mock_resource_config = MagicMock()
+    mock_resource_config.selector = MagicMock(preserve_api_response_case_style=False)
+    get_current_resource_config_mock.return_value = mock_resource_config
+
     from gcp_core.search.resource_searches import get_single_subscription
 
     expected_subscription = {
@@ -75,23 +96,22 @@ async def test_get_single_subscription(
     }
     mock_project = "project_name"
 
-    # Act
-    actual_subscription = await get_single_subscription(
-        mock_project, "subscription_name"
-    )
+    # Act within event context
+    async with event_context("test_event"):
+        actual_subscription = await get_single_subscription(
+            mock_project, "subscription_name"
+        )
 
     # Assert
     assert actual_subscription == expected_subscription
 
 
-@patch(
-    "port_ocean.context.ocean.PortOceanContext.integration_config",
-    return_value={"search_all_resources_per_minute_quota": 100},
-)
-async def test_feed_to_resource(integration_config: Any, monkeypatch: Any) -> None:
+@pytest.mark.asyncio
+@patch("gcp_core.utils.get_current_resource_config")
+async def test_feed_to_resource(
+    get_current_resource_config_mock: MagicMock, monkeypatch: Any
+) -> None:
     # Arrange
-
-    ## Mock project client
     projects_async_client_mock = AsyncMock
     monkeypatch.setattr(
         "google.cloud.resourcemanager_v3.ProjectsAsyncClient",
@@ -102,7 +122,6 @@ async def test_feed_to_resource(integration_config: Any, monkeypatch: Any) -> No
         {"name": "project_name"}
     )
 
-    ## Mock publisher client
     publisher_async_client_mock = AsyncMock
     monkeypatch.setattr(
         "google.pubsub_v1.services.publisher.PublisherAsyncClient",
@@ -113,6 +132,11 @@ async def test_feed_to_resource(integration_config: Any, monkeypatch: Any) -> No
         {"name": "topic_name"}
     )
 
+    # Mock the resource config
+    mock_resource_config = MagicMock()
+    mock_resource_config.selector = MagicMock(preserve_api_response_case_style=False)
+    get_current_resource_config_mock.return_value = mock_resource_config
+
     from gcp_core.search.resource_searches import feed_event_to_resource
 
     mock_asset_name = "projects/project_name/topics/topic_name"
@@ -144,13 +168,105 @@ async def test_feed_to_resource(integration_config: Any, monkeypatch: Any) -> No
         "state": 0,
     }
 
-    # Act
-    actual_resource = await feed_event_to_resource(
-        asset_type=mock_asset_type,
-        asset_name=mock_asset_name,
-        project_id=mock_asset_project_name,
-        asset_data=mock_asset_data,
-    )
+    # Act within event context
+    async with event_context("test_event"):
+        actual_resource = await feed_event_to_resource(
+            asset_type=mock_asset_type,
+            asset_name=mock_asset_name,
+            project_id=mock_asset_project_name,
+            asset_data=mock_asset_data,
+        )
 
     # Assert
     assert actual_resource == expected_resource
+
+
+@pytest.mark.asyncio
+@patch("gcp_core.utils.get_current_resource_config")
+async def test_preserve_case_style_combined(
+    get_current_resource_config_mock: MagicMock, monkeypatch: Any
+) -> None:
+    # Arrange
+    subscriber_async_client_mock = AsyncMock
+    monkeypatch.setattr(
+        "google.pubsub_v1.services.subscriber.SubscriberAsyncClient",
+        subscriber_async_client_mock,
+    )
+    subscriber_async_client_mock.get_subscription = AsyncMock()
+
+    # Mock for preserve_case_style = True
+    subscriber_async_client_mock.get_subscription.return_value = pubsub.Subscription(
+        {
+            "name": "subscription_name",
+            "topic": "projects/project_name/topics/topic_name",
+            "ack_deadline_seconds": 0,
+            "retain_acked_messages": False,
+            "labels": {},
+            "enable_message_ordering": False,
+            "filter": "",
+            "detached": False,
+            "enable_exactly_once_delivery": False,
+            "state": 0,
+        }
+    )
+
+    # Mock the resource config with preserve_api_response_case_style set to True
+    mock_resource_config_true = MagicMock()
+    mock_resource_config_true.selector = MagicMock(
+        preserve_api_response_case_style=True
+    )
+    get_current_resource_config_mock.return_value = mock_resource_config_true
+
+    from gcp_core.search.resource_searches import get_single_subscription
+
+    expected_subscription_true = {
+        "ackDeadlineSeconds": 0,
+        "detached": False,
+        "enableExactlyOnceDelivery": False,
+        "enableMessageOrdering": False,
+        "filter": "",
+        "labels": {},
+        "name": "subscription_name",
+        "retainAckedMessages": False,
+        "state": 0,
+        "topic": "projects/project_name/topics/topic_name",
+    }
+    mock_project = "project_name"
+
+    # Act within event context for preserve_case_style = True
+    async with event_context("test_event"):
+        actual_subscription_true = await get_single_subscription(
+            mock_project, "subscription_name"
+        )
+
+    # Assert for preserve_case_style = True
+    assert actual_subscription_true == expected_subscription_true
+
+    # Mock for preserve_case_style = False
+    mock_resource_config_false = MagicMock()
+    mock_resource_config_false.selector = MagicMock(
+        preserve_api_response_case_style=False
+    )
+    get_current_resource_config_mock.return_value = mock_resource_config_false
+
+    expected_subscription_false = {
+        "ack_deadline_seconds": 0,
+        "detached": False,
+        "enable_exactly_once_delivery": False,
+        "enable_message_ordering": False,
+        "filter": "",
+        "labels": {},
+        "name": "subscription_name",
+        "retain_acked_messages": False,
+        "state": 0,
+        "topic": "projects/project_name/topics/topic_name",
+    }
+
+    # Act within event context for preserve_case_style = False
+    async with event_context("test_event"):
+        actual_subscription_false = await get_single_subscription(
+            mock_project, "subscription_name"
+        )
+
+    # Assert for preserve_case_style = False
+    assert actual_subscription_false == expected_subscription_false