✨ Source Facebook Marketing: remove testing deleted records from integration test, move it to mock server tests (#37095)

Author: darynaishchenko

airbyte-integrations/connectors/source-facebook-marketing/integration_tests/test_streams.py

@@ -51,5 +51,17 @@ def with_end_date(self, end_date: datetime) -> ConfigBuilder:
         self._config["end_date"] = end_date.strftime(DATE_TIME_FORMAT)
         return self
 
+    def with_ad_statuses(self, statuses: List[str]) -> ConfigBuilder:
+        self._config["ad_statuses"] = statuses
+        return self
+
+    def with_campaign_statuses(self, statuses: List[str]) -> ConfigBuilder:
+        self._config["campaign_statuses"] = statuses
+        return self
+
+    def with_ad_set_statuses(self, statuses: List[str]) -> ConfigBuilder:
+        self._config["adset_statuses"] = statuses
+        return self
+
     def build(self) -> MutableMapping[str, Any]:
         return self._config

airbyte-integrations/connectors/source-facebook-marketing/unit_tests/integration/config.py

@@ -5,7 +5,7 @@
 
 from __future__ import annotations
 
-from typing import Any, List, Mapping, Optional, Union
+from typing import Any, Dict, List, Mapping, Optional, Union
 
 from airbyte_cdk.test.mock_http.request import HttpRequest
 
@@ -16,7 +16,30 @@ def get_account_request(account_id: Optional[str] = ACCOUNT_ID) -> RequestBuilde
     return RequestBuilder.get_account_endpoint(access_token=ACCESS_TOKEN, account_id=account_id)
 
 
+def get_ads_request(account_id: Optional[str] = ACCOUNT_ID) -> RequestBuilder:
+    return RequestBuilder.get_ad_endpoint(access_token=ACCESS_TOKEN, account_id=account_id)
+
+
+def get_campaigns_request(account_id: Optional[str] = ACCOUNT_ID) -> RequestBuilder:
+    return RequestBuilder.get_campaign_endpoint(access_token=ACCESS_TOKEN, account_id=account_id)
+
+
+def get_ad_sets_request(account_id: Optional[str] = ACCOUNT_ID) -> RequestBuilder:
+    return RequestBuilder.get_ad_sets_endpoint(access_token=ACCESS_TOKEN, account_id=account_id)
+
+
 class RequestBuilder:
+    @classmethod
+    def get_ad_endpoint(cls, access_token: str, account_id: str) -> RequestBuilder:
+        return cls(access_token=access_token, resource="ads").with_account_id(account_id)
+
+    @classmethod
+    def get_campaign_endpoint(cls, access_token: str, account_id: str) -> RequestBuilder:
+        return cls(access_token=access_token, resource="campaigns").with_account_id(account_id)
+
+    @classmethod
+    def get_ad_sets_endpoint(cls, access_token: str, account_id: str) -> RequestBuilder:
+        return cls(access_token=access_token, resource="adsets").with_account_id(account_id)
 
     @classmethod
     def get_account_endpoint(cls, access_token: str, account_id: str) -> RequestBuilder:
@@ -68,6 +91,10 @@ def with_body(self, body: Union[str, bytes, Mapping[str, Any]]) -> RequestBuilde
         self._body = body
         return self
 
+    def with_filtering(self, filters: List[Dict[str, Any]]):
+        self._query_params["filtering"] = self._get_formatted_filters(filters)
+        return self
+
     def build(self) -> HttpRequest:
         return HttpRequest(
             url=f"https://graph.facebook.com/v19.0/{self._account_sub_path()}{self._resource}",
@@ -81,3 +108,29 @@ def _account_sub_path(self) -> str:
     @staticmethod
     def _get_formatted_fields(fields: List[str]) -> str:
         return ",".join(fields)
+
+    @staticmethod
+    def _get_formatted_filters(filters: List[Dict[str, Any]]) -> str:
+        """
+        Used to create an acceptable by fb query param from list of dict filters in string format
+        From:
+        [{"field": "ad.effective_status", "operator": "IN", "value": ["ACTIVE", "ARCHIVED"]}, {"field": "ad.updated_time", "operator": "GREATER_THAN", "value": 1672531200}]
+        To:
+        '[{"field":"ad.effective_status","operator":"IN","value":["ACTIVE","ARCHIVED"]},' '{"field":"ad.updated_time","operator":"GREATER_THAN","value":1672531200}]'
+        """
+        field_filter = []
+        field_filters = []
+        for f in filters:
+            for key, value in f.items():
+                if isinstance(value, list):
+                    value = ",".join([f'"{s}"' for s in value])
+                    field_filter.append(f'"{key}":[{value}]')
+                elif isinstance(value, int):
+                    field_filter.append(f'"{key}":{value}')
+                else:
+                    field_filter.append(f'"{key}":"{value}"')
+            field_filters.append("{" + f'{",".join(field_filter)}' + "}")
+            field_filter = []
+
+        field_filters_str = f'[{",".join(field_filters)}]'
+        return field_filters_str