feat: Add test suite for examples/misc

This commit introduces a new test suite for the Python scripts located in the examples/misc directory.

The following scripts now have corresponding unit tests:
- add_ad_group_image_asset.py
- campaign_report_to_csv.py
- set_custom_client_timeouts.py
- upload_image_asset.py

The tests are located in the examples/misc/tests subdirectory. They utilize unittest.mock to mock Google Ads API interactions, ensuring that the scripts' logic is verified without making actual API calls. Key aspects tested include:
- Correct service calls and parameter construction.
- Proper handling of API responses and exceptions (e.g., DeadlineExceeded).
- File I/O operations (e.g., CSV generation).
- Correct usage of client configurations like timeouts and retries.

An __init__.py file has been added to the tests directory to enable test discovery.

Author: google-labs-jules[bot]

examples/misc/tests/.gitkeep

@@ -0,0 +1,108 @@
+import unittest
+from unittest.mock import patch, MagicMock
+import sys
+import os
+
+# Add the project root directory to sys.path to allow robust import of the script under test.
+# This assumes the test file is located at 'examples/misc/tests/test_add_ad_group_image_asset.py'
+# and the script to test is at 'examples/misc/add_ad_group_image_asset.py'.
+# The project root would then be '../..' from this file's directory.
+_PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", ".."))
+if _PROJECT_ROOT not in sys.path:
+    sys.path.insert(0, _PROJECT_ROOT)
+
+from examples.misc import add_ad_group_image_asset
+
+class TestAddAdGroupImageAsset(unittest.TestCase):
+
+    @patch('examples.misc.add_ad_group_image_asset.GoogleAdsClient.load_from_storage')
+    def test_main_function(self, mock_load_from_storage):
+        # Mock the GoogleAdsClient instance
+        mock_ads_client = MagicMock()
+        mock_load_from_storage.return_value = mock_ads_client
+
+        # Mock the AdGroupAssetService
+        mock_ad_group_asset_service = mock_ads_client.get_service.return_value
+
+        # Define test parameters and expected values
+        customer_id = "1234567890"
+        ad_group_id = "111222333"
+        asset_id = "987654321"
+        
+        expected_asset_resource_name = f"customers/{customer_id}/assets/{asset_id}"
+        expected_ad_group_resource_name = f"customers/{customer_id}/adGroups/{ad_group_id}"
+        mock_created_resource_name = f"customers/{customer_id}/adGroupAssets/{ad_group_id}~{asset_id}"
+
+        # Configure mock path helper methods
+        mock_ad_group_asset_service.asset_path.return_value = expected_asset_resource_name
+        mock_ad_group_asset_service.ad_group_path.return_value = expected_ad_group_resource_name
+
+        # Configure mock for client.get_type("AdGroupAssetOperation")
+        # This mock will be used by the script to create the operation object
+        mock_operation = MagicMock()
+        mock_ads_client.get_type.return_value = mock_operation
+        
+        # Configure mock for AssetFieldTypeEnum
+        # The script accesses client.enums.AssetFieldTypeEnum.AD_IMAGE
+        mock_ads_client.enums.AssetFieldTypeEnum.AD_IMAGE = "AD_IMAGE_ENUM_VALUE_FOR_TEST"
+
+        # Configure mock for the mutate_ad_group_assets method response
+        mock_mutate_response = MagicMock()
+        mock_mutate_response.results = [MagicMock(resource_name=mock_created_resource_name)]
+        mock_ad_group_asset_service.mutate_ad_group_assets.return_value = mock_mutate_response
+
+        # Call the main function of the script, capturing print output
+        with patch('builtins.print') as mock_print:
+            add_ad_group_image_asset.main(
+                mock_ads_client,
+                customer_id,
+                ad_group_id,
+                asset_id
+            )
+
+        # --- Assertions ---
+
+        # Verify GoogleAdsClient.load_from_storage was called (implicitly by @patch)
+        mock_load_from_storage.assert_called_once() # Path is checked by the patch decorator
+
+        # Verify AdGroupAssetService was fetched
+        mock_ads_client.get_service.assert_called_once_with("AdGroupAssetService")
+
+        # Verify path helper methods were called with correct parameters
+        mock_ad_group_asset_service.asset_path.assert_called_once_with(customer_id, asset_id)
+        mock_ad_group_asset_service.ad_group_path.assert_called_once_with(customer_id, ad_group_id)
+        
+        # Verify client.get_type("AdGroupAssetOperation") was called
+        mock_ads_client.get_type.assert_called_once_with("AdGroupAssetOperation")
+
+        # Verify the mutate_ad_group_assets call
+        # The script constructs an operation and passes it.
+        # We need to check the arguments of this call.
+        
+        # Expected operation built by the script:
+        # ad_group_asset_operation = client.get_type("AdGroupAssetOperation") (this is our mock_operation)
+        # ad_group_asset_set = ad_group_asset_operation.create
+        # ad_group_asset_set.asset = expected_asset_resource_name
+        # ad_group_asset_set.field_type = client.enums.AssetFieldTypeEnum.AD_IMAGE
+        # ad_group_asset_set.ad_group = expected_ad_group_resource_name
+        
+        # Check that the 'create' attribute of our mock_operation was configured as expected
+        self.assertEqual(mock_operation.create.asset, expected_asset_resource_name)
+        self.assertEqual(mock_operation.create.ad_group, expected_ad_group_resource_name)
+        self.assertEqual(mock_operation.create.field_type, mock_ads_client.enums.AssetFieldTypeEnum.AD_IMAGE)
+
+        # Now, check if mutate_ad_group_assets was called with this 'mock_operation'
+        mock_ad_group_asset_service.mutate_ad_group_assets.assert_called_once_with(
+            customer_id=customer_id,
+            operations=[mock_operation] # The script passes the operation it built
+        )
+
+        # Verify the printed output
+        expected_print_message = (
+            f"Created ad group asset with resource name: '{mock_created_resource_name}'"
+        )
+        # The script loops through results, so check if any call matches.
+        mock_print.assert_any_call(expected_print_message)
+
+if __name__ == "__main__":
+    unittest.main()

examples/misc/tests/__init__.py

@@ -0,0 +1,133 @@
+import unittest
+from unittest.mock import patch, MagicMock, call
+import csv
+import os
+import sys
+from datetime import datetime
+
+# Adjust sys.path to allow import of the script under test
+_PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", ".."))
+if _PROJECT_ROOT not in sys.path:
+    sys.path.insert(0, _PROJECT_ROOT)
+
+from examples.misc import campaign_report_to_csv
+
+# Define a fixed output directory for tests to manage created files
+# Assuming os.path.join behavior: os.path.join("anything", "/absolute/path") returns "/absolute/path"
+# If _TEST_CSV_FILEPATH is absolute, the script should use it directly.
+_TEST_OUTPUT_DIR = os.path.join(os.path.dirname(__file__), "test_output")
+_TEST_CSV_FILENAME = "test_campaign_report.csv"
+_TEST_CSV_FILEPATH = os.path.abspath(os.path.join(_TEST_OUTPUT_DIR, _TEST_CSV_FILENAME))
+
+
+class TestCampaignReportToCsv(unittest.TestCase):
+
+    def setUp(self):
+        if not os.path.exists(_TEST_OUTPUT_DIR):
+            os.makedirs(_TEST_OUTPUT_DIR)
+        if os.path.exists(_TEST_CSV_FILEPATH):
+            os.remove(_TEST_CSV_FILEPATH)
+
+    def tearDown(self):
+        if os.path.exists(_TEST_CSV_FILEPATH):
+            os.remove(_TEST_CSV_FILEPATH)
+        if os.path.exists(_TEST_OUTPUT_DIR) and not os.listdir(_TEST_OUTPUT_DIR):
+            os.rmdir(_TEST_OUTPUT_DIR)
+
+    def _create_mock_google_ads_row(self, descriptive_name, date, campaign_name, impressions, clicks, cost_micros):
+        """Helper to create a MagicMock object that mimics a GoogleAdsRow based on script's output."""
+        row = MagicMock()
+        row.customer.descriptive_name = descriptive_name
+        row.segments.date = date
+        row.campaign.name = campaign_name
+        row.metrics.impressions = impressions
+        row.metrics.clicks = clicks
+        row.metrics.cost_micros = cost_micros
+        return row
+
+    @patch('examples.misc.campaign_report_to_csv.GoogleAdsClient.load_from_storage')
+    def test_report_generation_with_headers(self, mock_load_from_storage):
+        mock_ads_client = MagicMock()
+        mock_load_from_storage.return_value = mock_ads_client
+        mock_ga_service = mock_ads_client.get_service.return_value
+
+        # Sample data matching script's output fields
+        mock_rows_data = [
+            ("Test Account 1", "2023-10-26", "Campaign Alpha", 1000, 150, 5000000),
+            ("Test Account 1", "2023-10-26", "Campaign Beta", 2500, 300, 12000000),
+        ]
+        # search_stream returns an iterable of batches, each batch has 'results'
+        mock_search_stream_response = [MagicMock(results=[self._create_mock_google_ads_row(*data) for data in mock_rows_data])]
+        mock_ga_service.search_stream.return_value = mock_search_stream_response
+        
+        customer_id = "1234567890"
+        
+        campaign_report_to_csv.main(
+            mock_ads_client, customer_id, output_file=_TEST_CSV_FILEPATH, write_headers=True
+        )
+
+        self.assertTrue(os.path.exists(_TEST_CSV_FILEPATH))
+
+        # Script's actual headers
+        expected_headers = ["Account", "Date", "Campaign", "Impressions", "Clicks", "Cost"]
+        with open(_TEST_CSV_FILEPATH, mode='r', newline='') as csvfile:
+            reader = csv.reader(csvfile)
+            headers = next(reader)
+            self.assertEqual(headers, expected_headers)
+            
+            content = list(reader)
+            self.assertEqual(len(content), len(mock_rows_data))
+            for i, row_data in enumerate(mock_rows_data):
+                self.assertEqual(content[i][0], str(row_data[0])) # Account
+                self.assertEqual(content[i][1], str(row_data[1])) # Date
+                self.assertEqual(content[i][2], str(row_data[2])) # Campaign
+                self.assertEqual(content[i][3], str(row_data[3])) # Impressions
+                self.assertEqual(content[i][4], str(row_data[4])) # Clicks
+                self.assertEqual(content[i][5], str(row_data[5])) # Cost
+        
+        mock_ga_service.search_stream.assert_called_once()
+        # Verify the search_request object passed to search_stream
+        args, _ = mock_ga_service.search_stream.call_args
+        search_request = args[0]
+        self.assertEqual(search_request.customer_id, customer_id)
+        self.assertIn("FROM campaign", search_request.query)
+
+
+    @patch('examples.misc.campaign_report_to_csv.GoogleAdsClient.load_from_storage')
+    def test_report_generation_without_headers(self, mock_load_from_storage):
+        mock_ads_client = MagicMock()
+        mock_load_from_storage.return_value = mock_ads_client
+        mock_ga_service = mock_ads_client.get_service.return_value
+
+        mock_rows_data = [
+            ("Test Account 2", "2023-10-27", "Campaign Gamma", 500, 50, 2000000),
+        ]
+        mock_search_stream_response = [MagicMock(results=[self._create_mock_google_ads_row(*data) for data in mock_rows_data])]
+        mock_ga_service.search_stream.return_value = mock_search_stream_response
+        customer_id = "9876543210"
+
+        campaign_report_to_csv.main(
+            mock_ads_client, customer_id, output_file=_TEST_CSV_FILEPATH, write_headers=False
+        )
+
+        self.assertTrue(os.path.exists(_TEST_CSV_FILEPATH))
+
+        with open(_TEST_CSV_FILEPATH, mode='r', newline='') as csvfile:
+            reader = csv.reader(csvfile)
+            content = list(reader)
+            self.assertEqual(len(content), len(mock_rows_data))
+            for i, row_data in enumerate(mock_rows_data):
+                self.assertEqual(content[i][0], str(row_data[0])) # Account
+                self.assertEqual(content[i][1], str(row_data[1])) # Date
+                self.assertEqual(content[i][2], str(row_data[2])) # Campaign
+                self.assertEqual(content[i][3], str(row_data[3])) # Impressions
+                self.assertEqual(content[i][4], str(row_data[4])) # Clicks
+                self.assertEqual(content[i][5], str(row_data[5])) # Cost
+        
+        mock_ga_service.search_stream.assert_called_once()
+        args, _ = mock_ga_service.search_stream.call_args
+        search_request = args[0]
+        self.assertEqual(search_request.customer_id, customer_id)
+
+if __name__ == "__main__":
+    unittest.main()

examples/misc/tests/test_add_ad_group_image_asset.py

@@ -0,0 +1,139 @@
+import unittest
+from unittest.mock import patch, MagicMock, ANY
+import sys
+import os
+
+from google.api_core.exceptions import DeadlineExceeded
+from google.api_core import retry as api_core_retry
+
+# Adjust sys.path to allow import of the script under test
+_PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", ".."))
+if _PROJECT_ROOT not in sys.path:
+    sys.path.insert(0, _PROJECT_ROOT)
+
+from examples.misc import set_custom_client_timeouts
+
+# _EXPECTED_PAGE_SIZE is not used in the target script.
+
+class TestSetCustomClientTimeouts(unittest.TestCase):
+
+    def _create_mock_row_with_campaign_id(self, campaign_id="test_campaign_id_123"):
+        row = MagicMock()
+        row.campaign.id = campaign_id
+        return row
+
+    def _create_mock_ads_client(self):
+        mock_ads_client = MagicMock()
+        mock_ga_service = mock_ads_client.get_service.return_value
+        
+        # Mock for search_stream (streaming call)
+        mock_batch = MagicMock()
+        mock_batch.results = [self._create_mock_row_with_campaign_id()] 
+        mock_ga_service.search_stream.return_value = [mock_batch]
+
+        # Mock for search (unary call)
+        mock_ga_service.search.return_value = [self._create_mock_row_with_campaign_id()]
+        
+        return mock_ads_client, mock_ga_service
+
+    # --- Tests for make_server_streaming_call ---
+
+    @patch('set_custom_client_timeouts._CLIENT_TIMEOUT_SECONDS', 60.0) # Patch script's constant
+    @patch('builtins.print')
+    def test_streaming_call_success(self, mock_print):
+        mock_ads_client, mock_ga_service = self._create_mock_ads_client()
+        customer_id = "1234567890"
+        
+        # Call the function without timeout arg, it uses the patched script constant
+        set_custom_client_timeouts.make_server_streaming_call(
+            mock_ads_client, customer_id
+        )
+
+        mock_ads_client.get_service.assert_called_once_with("GoogleAdsService") # Version not specified in script func
+        mock_ga_service.search_stream.assert_called_once_with(request=ANY, timeout=60.0) # Check patched timeout
+        
+        # Verify print output for success
+        mock_print.assert_any_call("The server streaming call completed before the timeout.")
+        mock_print.assert_any_call("Total # of campaign IDs retrieved: 1")
+
+
+    @patch('set_custom_client_timeouts._CLIENT_TIMEOUT_SECONDS', 0.01) # Very small timeout
+    @patch('builtins.print')
+    @patch('sys.exit')
+    def test_streaming_call_deadline_exceeded(self, mock_sys_exit, mock_print):
+        mock_ads_client, mock_ga_service = self._create_mock_ads_client()
+        customer_id = "1234567890"
+
+        mock_ga_service.search_stream.side_effect = DeadlineExceeded("Test DeadlineExceeded Streaming")
+
+        set_custom_client_timeouts.make_server_streaming_call(
+            mock_ads_client, customer_id
+        )
+        
+        mock_ga_service.search_stream.assert_called_once_with(request=ANY, timeout=0.01)
+        mock_print.assert_any_call("The server streaming call did not complete before the timeout.")
+        mock_sys_exit.assert_called_once_with(1)
+
+    # --- Tests for make_unary_call ---
+
+    @patch('set_custom_client_timeouts._CLIENT_TIMEOUT_SECONDS', 120.0) # Patch script's constant for deadline
+    @patch('builtins.print')
+    def test_unary_call_success(self, mock_print):
+        mock_ads_client, mock_ga_service = self._create_mock_ads_client()
+        customer_id = "1234567890"
+        
+        # Patched _CLIENT_TIMEOUT_SECONDS will be used by the script function
+        script_timeout = 120.0
+        expected_initial = script_timeout / 10.0
+        expected_maximum = script_timeout / 5.0
+        expected_deadline = script_timeout
+
+        set_custom_client_timeouts.make_unary_call(
+            mock_ads_client, customer_id
+        )
+
+        mock_ads_client.get_service.assert_called_once_with("GoogleAdsService") # Version not specified in script func
+        
+        args, kwargs = mock_ga_service.search.call_args
+        self.assertEqual(kwargs.get('request').customer_id, customer_id)
+        self.assertIn("SELECT campaign.id FROM campaign", kwargs.get('request').query)
+        
+        actual_retry = kwargs.get('retry')
+        self.assertIsNotNone(actual_retry)
+        self.assertIsInstance(actual_retry, api_core_retry.Retry)
+        self.assertEqual(actual_retry._initial, expected_initial)
+        self.assertEqual(actual_retry._maximum, expected_maximum)
+        self.assertEqual(actual_retry._deadline, expected_deadline)
+        # Assert default multiplier and predicate if they are not set in script
+        self.assertEqual(actual_retry._multiplier, 1.3) # google.api_core.retry.Retry._DEFAULT_MULTIPLIER
+        self.assertTrue(callable(actual_retry._predicate)) # Check if it's a callable, like if_transient_error
+
+
+        mock_print.assert_any_call("The unary call completed before the timeout.")
+        mock_print.assert_any_call("Total # of campaign IDs retrieved: 1")
+
+
+    @patch('set_custom_client_timeouts._CLIENT_TIMEOUT_SECONDS', 0.01) # Very small timeout for deadline
+    @patch('builtins.print')
+    @patch('sys.exit')
+    def test_unary_call_deadline_exceeded(self, mock_sys_exit, mock_print):
+        mock_ads_client, mock_ga_service = self._create_mock_ads_client()
+        customer_id = "1234567890"
+
+        mock_ga_service.search.side_effect = DeadlineExceeded("Test DeadlineExceeded Unary")
+
+        set_custom_client_timeouts.make_unary_call(
+            mock_ads_client, customer_id
+        )
+        
+        args, kwargs = mock_ga_service.search.call_args
+        self.assertIsNotNone(kwargs.get('retry'))
+        self.assertEqual(kwargs.get('retry')._deadline, 0.01)
+
+
+        mock_print.assert_any_call("The unary call did not complete before the timeout.")
+        mock_sys_exit.assert_called_once_with(1)
+
+
+if __name__ == "__main__":
+    unittest.main()