Merge pull request #44 from google/ads-offline-conversion-improvements

Ads offline conversion improvements

megalista_dataflow/uploaders/google_ads/conversions/google_ads_offline_conversions_uploader.py

@@ -15,7 +15,7 @@
 import logging
 
 import apache_beam as beam
-from models.execution import Batch, Execution
+from models.execution import Batch, Execution, AccountConfig, Destination
 from uploaders import utils
 from uploaders.google_ads import ADS_API_VERSION
 
@@ -28,7 +28,6 @@ def __init__(self, oauth_credentials, developer_token):
     super().__init__()
     self.oauth_credentials = oauth_credentials
     self.developer_token = developer_token
-    self.active = self.developer_token is not None
 
   def _get_ads_service(self, customer_id: str):
     return utils.get_ads_service('GoogleAdsService', ADS_API_VERSION,
@@ -45,10 +44,19 @@ def _get_oc_service(self, customer_id):
   def start_bundle(self):
     pass
 
+  def _get_customer_id(self, account_config:AccountConfig, destination:Destination) -> str:
+    """
+      If the customer_id is present on the destination, returns it, otherwise defaults to the account_config info.
+    """
+    if len(destination.destination_metadata) >= 2 and len(destination.destination_metadata[1]) > 0:
+      return destination.destination_metadata[1].replace('-', '')
+    return account_config.google_ads_account_id.replace('-', '')
+
+
   @staticmethod
   def _assert_conversion_name_is_present(execution: Execution):
     destination = execution.destination.destination_metadata
-    if len(destination) != 1:
+    if len(destination) is 0:
       raise ValueError('Missing destination information. Found {}'.format(
           len(destination)))
 
@@ -59,14 +67,10 @@ def _assert_conversion_name_is_present(execution: Execution):
   @utils.safe_process(
       logger=logging.getLogger('megalista.GoogleAdsOfflineUploader'))
   def process(self, batch: Batch, **kwargs):
-    if not self.active:
-      logging.getLogger().warning(
-          'Skipping upload, parameters not configured.')
-      return
     execution = batch.execution
     self._assert_conversion_name_is_present(execution)
 
-    customer_id = execution.account_config.google_ads_account_id.replace('-', '')
+    customer_id = self._get_customer_id(execution.account_config, execution.destination)
     oc_service = self._get_oc_service(customer_id)
 
     resource_name = self._get_resource_name(customer_id, execution.destination.destination_metadata[0])
@@ -88,21 +92,19 @@ def _do_upload(oc_service, conversion_resource_name, customer_id, rows):
 
     upload_data = {
       'customer_id': customer_id,
-      'partial_failure': False,
+      'partial_failure': True,
       'validate_only': False,
       'conversions': conversions
     }
 
-
     response = oc_service.upload_click_conversions(request=upload_data)
     utils.print_partial_error_messages(_DEFAULT_LOGGER, 'uploading offline conversions', response)
 
   def _get_resource_name(self, customer_id: str, name: str):
-      resource_name = None
       service = self._get_ads_service(customer_id)
       query = f"SELECT conversion_action.resource_name FROM conversion_action WHERE conversion_action.name = '{name}'"
       response_query = service.search_stream(customer_id=customer_id, query=query)
       for batch in response_query:
         for row in batch.results:
-          resource_name = row.conversion_action.resource_name
-      return resource_name
+          return row.conversion_action.resource_name
+      raise Exception(f'Conversion "{name}" could not be found on account {customer_id}')

megalista_dataflow/uploaders/google_ads/conversions/google_ads_offline_conversions_uploader_test.py

@@ -11,9 +11,11 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+from unittest.mock import MagicMock
 
 from apache_beam.options.value_provider import StaticValueProvider
 import pytest
+
 from uploaders.google_ads.conversions.google_ads_offline_conversions_uploader import GoogleAdsOfflineUploaderDoFn
 from models.execution import AccountConfig
 from models.execution import Destination
@@ -24,7 +26,7 @@
 from models.execution import Batch
 from models.oauth_credentials import OAuthCredentials
 
-_account_config = AccountConfig('account_id', False, 'ga_account_id', '', '')
+_account_config = AccountConfig('123-45567-890', False, 'ga_account_id', '', '')
 
 
 @pytest.fixture
@@ -44,19 +46,22 @@ def test_get_service(mocker, uploader):
   assert uploader._get_oc_service(mocker.ANY) is not None
 
 
-def test_not_active(mocker, caplog):
-  credential_id = StaticValueProvider(str, 'id')
-  secret = StaticValueProvider(str, 'secret')
-  access = StaticValueProvider(str, 'access')
-  refresh = StaticValueProvider(str, 'refresh')
-  credentials = OAuthCredentials(credential_id, secret, access, refresh)
-  uploader_dofn = GoogleAdsOfflineUploaderDoFn(credentials, None)
-  mocker.patch.object(uploader_dofn, '_get_oc_service')
-  uploader_dofn.process(Batch(None, []))
-  uploader_dofn._get_oc_service.assert_not_called()
-  assert 'Skipping upload, parameters not configured.' in caplog.text
+def arrange_conversion_resource_name_api_call(mocker, uploader, conversion_resource_name):
+  mocker.patch.object(uploader, '_get_ads_service')
+
+  resource_name_result = MagicMock()
+  resource_name_result.conversion_action.resource_name = conversion_resource_name
+
+  resource_name_batch_response = MagicMock()
+  resource_name_batch_response.results = [resource_name_result]
+
+  uploader._get_ads_service.return_value.search_stream.return_value = [resource_name_batch_response]
 
 def test_conversion_upload(mocker, uploader):
+  # arrange
+  conversion_resource_name = 'user_list_resouce'
+  arrange_conversion_resource_name_api_call(mocker, uploader, conversion_resource_name)
+
   mocker.patch.object(uploader, '_get_oc_service')
   conversion_name = 'user_list'
   destination = Destination(
@@ -79,21 +84,84 @@ def test_conversion_upload(mocker, uploader):
           'amount': '234',
           'gclid': '567'
       }])
+
+  # act
   uploader.process(batch)
 
-  uploader._get_oc_service.return_value.upload_click_conversions.assert_any_call(request = {
-    'customer_id': 'account_id',
-    'partial_failure': False,
+  #assert
+  uploader._get_ads_service.return_value.search_stream.assert_called_once_with(
+    customer_id='12345567890',
+    query=f"SELECT conversion_action.resource_name FROM conversion_action WHERE conversion_action.name = '{conversion_name}'"
+  )
+
+  uploader._get_oc_service.return_value.upload_click_conversions.assert_called_once_with(request={
+    'customer_id': '12345567890',
+    'partial_failure': True,
     'validate_only': False,
     'conversions': [{
-      'conversion_action': None,
+      'conversion_action': conversion_resource_name,
       'conversion_date_time': time1_result,
       'conversion_value': 123,
       'gclid': '456'
     }, {
-      'conversion_action': None,
+      'conversion_action': conversion_resource_name,
       'conversion_date_time': time2_result,
       'conversion_value': 234,
       'gclid': '567'
     }]
   })
+
+
+def test_upload_with_ads_account_override(mocker, uploader):
+  # arrange
+  conversion_resource_name = 'user_list_resouce'
+  arrange_conversion_resource_name_api_call(mocker, uploader, conversion_resource_name)
+
+  mocker.patch.object(uploader, '_get_oc_service')
+  conversion_name = 'user_list'
+  destination = Destination(
+    'dest1', DestinationType.ADS_OFFLINE_CONVERSION, ['user_list', '987-7654-123'])
+  source = Source('orig1', SourceType.BIG_QUERY, ['dt1', 'buyers'])
+  execution = Execution(_account_config, source, destination)
+
+  time1 = '2020-04-09T14:13:55.0005'
+  time1_result = '2020-04-09 14:13:55-03:00'
+
+  time2 = '2020-04-09T13:13:55.0005'
+  time2_result = '2020-04-09 13:13:55-03:00'
+
+  batch = Batch(execution, [{
+    'time': time1,
+    'amount': '123',
+    'gclid': '456'
+  }, {
+    'time': time2,
+    'amount': '234',
+    'gclid': '567'
+  }])
+
+  # act
+  uploader.process(batch)
+
+  # assert
+  uploader._get_ads_service.return_value.search_stream.assert_called_once_with(
+    customer_id='9877654123',
+    query=f"SELECT conversion_action.resource_name FROM conversion_action WHERE conversion_action.name = '{conversion_name}'"
+  )
+
+  uploader._get_oc_service.return_value.upload_click_conversions.assert_called_once_with(request={
+    'customer_id': '9877654123',
+    'partial_failure': True,
+    'validate_only': False,
+    'conversions': [{
+      'conversion_action': conversion_resource_name,
+      'conversion_date_time': time1_result,
+      'conversion_value': 123,
+      'gclid': '456'
+    }, {
+      'conversion_action': conversion_resource_name,
+      'conversion_date_time': time2_result,
+      'conversion_value': 234,
+      'gclid': '567'
+    }]
+  })