# Copyright 2019 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # https://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # 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. """Tests the configuration helper module.""" from unittest import mock import os import yaml from pyfakefs.fake_filesystem_unittest import TestCase as FileTestCase from google.ads.googleads import config class ConfigTest(FileTestCase): def setUp(self): self.setUpPyfakefs() self.developer_token = "abc123" self.client_id = "client_id_123456789" self.client_secret = "client_secret_987654321" self.refresh_token = "refresh" self.login_customer_id = "1234567890" self.linked_customer_id = "0987654321" self.path_to_private_key_file = "/test/path/to/config.json" self.json_key_file_path = "/another/test/path/to/config.json" self.delegated_account = "delegated@account.com" self.endpoint = "www.testendpoint.com" self.configuration_file_path = "/usr/test/path/google-ads.yaml" self.impersonated_email = "impersonated@account.com" self.use_proto_plus = False self.use_cloud_org_for_api_access = False # The below fields are defaults that include required keys. # They are merged with other keys in individual tests, and isolated # here so that new required keys don't need to be added to each test. self.default_dict_config = { "developer_token": self.developer_token, "use_proto_plus": self.use_proto_plus, } self.default_env_var_config = { "GOOGLE_ADS_DEVELOPER_TOKEN": self.developer_token, "GOOGLE_ADS_USE_PROTO_PLUS": self.use_proto_plus, } def _create_mock_yaml( self, additional_configs, file_path=os.path.join(os.path.expanduser("~"), "google-ads.yaml"), ): merged = {**self.default_dict_config, **additional_configs} self.fs.create_file( file_path, contents=yaml.safe_dump(merged), ) def test_load_from_yaml_file_logging(self): """Should load logging config from yaml if provided.""" self._create_mock_yaml( { "developer_token": self.developer_token, "logging": { "version": 1, "disable_existing_loggers": False, "formatters": { "default_fmt": { "format": "[%(asctime)s - %(levelname)s]", "datefmt": "%Y-%m-%d %H:%M:%S", } }, "handlers": { "default_handler": { "class": "logging.StreamHandler", "formatter": "default_fmt", } }, "loggers": { "": { "handlers": ["default_handler"], "level": "DEBUG", } }, }, } ) result = config.load_from_yaml_file() self.assertEqual(result["developer_token"], self.developer_token) self.assertIsInstance(result["logging"], dict) def test_load_from_yaml_file_logging_invalid_json(self): """Should raise ValueError if logging config is not valid JSON.""" self._create_mock_yaml({"logging": "not JSON"}) self.assertRaises(ValueError, config.load_from_yaml_file) def test_load_from_yaml_file(self): """Should load config from a yaml file in the root directory.""" self._create_mock_yaml( { "client_id": self.client_id, "client_secret": self.client_secret, "refresh_token": self.refresh_token, } ) result = config.load_from_yaml_file() self.assertEqual(result["developer_token"], self.developer_token) self.assertEqual(result["use_proto_plus"], self.use_proto_plus) self.assertEqual(result["client_id"], self.client_id) self.assertEqual(result["client_secret"], self.client_secret) self.assertEqual(result["refresh_token"], self.refresh_token) def test_load_from_yaml_file_missing_developer_token(self): """Should raise ValueError if developer_token key is missing.""" file_path = os.path.join(os.path.expanduser("~"), "google-ads.yaml") # save a YAML file without a required developer_token key self.fs.create_file( file_path, contents=yaml.safe_dump( { "client_id": self.client_id, "client_secret": self.client_secret, "refresh_token": self.refresh_token, "use_proto_plus": self.use_proto_plus, } ), ) self.assertRaises(ValueError, config.load_from_yaml_file) def test_load_from_yaml_file_missing_use_proto_plus_key(self): """Should raise ValueError if use_proto_plus key is missing.""" file_path = os.path.join(os.path.expanduser("~"), "google-ads.yaml") # save a YAML file without a required use_proto_plus key self.fs.create_file( file_path, contents=yaml.safe_dump( { "client_id": self.client_id, "client_secret": self.client_secret, "refresh_token": self.refresh_token, "developer_token": self.developer_token, } ), ) self.assertRaises(ValueError, config.load_from_yaml_file) def test_load_from_yaml_file_with_path(self): """Should load yaml file from a given path passed in directly.""" custom_path = os.path.expanduser("/test/custom/path") file_path = os.path.join(custom_path, "google-ads.yaml") self._create_mock_yaml( { "client_id": self.client_id, "client_secret": self.client_secret, "refresh_token": self.refresh_token, }, file_path=file_path, ) result = config.load_from_yaml_file(path=file_path) self.assertEqual(result["developer_token"], self.developer_token) self.assertEqual(result["use_proto_plus"], self.use_proto_plus) self.assertEqual(result["client_id"], self.client_id) self.assertEqual(result["client_secret"], self.client_secret) self.assertEqual(result["refresh_token"], self.refresh_token) def test_load_from_yaml_file_from_env_var(self): """Should load yaml file from env var path if defined.""" env_var_path = os.path.expanduser("/test/from/env/var") file_path = os.path.join(env_var_path, "google-ads.yaml") self._create_mock_yaml({}, file_path=file_path) environ = {"GOOGLE_ADS_CONFIGURATION_FILE_PATH": file_path} with mock.patch("os.environ", environ): result = config.load_from_yaml_file() self.assertEqual(result["developer_token"], self.developer_token) self.assertEqual(result["use_proto_plus"], self.use_proto_plus) def test_load_from_yaml_file_with_path_and_env_var(self): """Should load yaml file from passed-in path if both are defined.""" env_var_path = os.path.join( os.path.expanduser("/test/from/env/var"), "google-ads.yaml" ) passed_in_path = os.path.join( os.path.expanduser("/test/given/path"), "google-ads.yaml" ) # Create yaml file in location defined by path passed to method self._create_mock_yaml( {"location": "passed in"}, file_path=passed_in_path ) # Create yaml file in location defined by environment variable self._create_mock_yaml({"location": "env var"}, file_path=env_var_path) environ = {"GOOGLE_ADS_CONFIGURATION_FILE_PATH": env_var_path} with mock.patch("os.environ", environ): # Load config and check that it came from path passed to method result = config.load_from_yaml_file(path=passed_in_path) self.assertEqual(result["location"], "passed in") def test_load_from_yaml_file_login_cid_number(self): """Should load login_customer_id key if value is defined as a number.""" login_cid_num = int(self.login_customer_id) self._create_mock_yaml({"login_customer_id": login_cid_num}) result = config.load_from_yaml_file() self.assertEqual(result["login_customer_id"], self.login_customer_id) def test_load_from_yaml_file_linked_cid(self): """Should load liked_customer_id config from yaml.""" self._create_mock_yaml({"linked_customer_id": self.linked_customer_id}) result = config.load_from_yaml_file() self.assertEqual(result["linked_customer_id"], self.linked_customer_id) def test_load_from_yaml_file_secondary_service_account_keys(self): """Should convert secondary keys to primary keys. This test should be removed once the secondary service account keys are deprecated. """ self._create_mock_yaml( { "path_to_private_key_file": self.json_key_file_path, "delegated_account": self.impersonated_email, } ) result = config.load_from_yaml_file() self.assertEqual(result["json_key_file_path"], self.json_key_file_path) self.assertEqual(result["impersonated_email"], self.impersonated_email) def test_parse_yaml_document_to_dict(self): """Should parse configuration from a yaml string""" yaml_doc = f""" client_id: {self.client_id}\n client_secret: {self.client_secret}\n developer_token: {self.developer_token}\n use_proto_plus: {self.use_proto_plus}\n refresh_token: {self.refresh_token}\n """ result = config.parse_yaml_document_to_dict(yaml_doc) self.assertEqual(result["developer_token"], self.developer_token) self.assertEqual(result["use_proto_plus"], self.use_proto_plus) self.assertEqual(result["client_id"], self.client_id) self.assertEqual(result["client_secret"], self.client_secret) self.assertEqual(result["refresh_token"], self.refresh_token) def test_parse_yaml_document_to_dict_missing_required_key(self): """Should raise ValueError if yaml string is missing a required key.""" # YAML document is missing the required developer_token key yaml_doc = f""" client_id: {self.client_id}\n client_secret: {self.client_secret}\n use_proto_plus: {self.use_proto_plus}\n refresh_token: {self.refresh_token}\n """ self.assertRaises( ValueError, config.parse_yaml_document_to_dict, yaml_doc ) def test_load_from_dict(self): """Can load config from a dict.""" config_data = { **self.default_dict_config, **{ "client_id": self.client_id, "client_secret": self.client_secret, "refresh_token": self.refresh_token, }, } self.assertEqual(config.load_from_dict(config_data), config_data) def test_load_from_dict_logging(self): """Should load logging config from dict.""" config_data = { **self.default_dict_config, **{ "logging": { "version": 1, "disable_existing_loggers": False, "formatters": { "default_fmt": { "format": "[%(asctime)s - %(levelname)s]", "datefmt": "%Y-%m-%d %H:%M:%S", } }, "handlers": { "default_handler": { "class": "logging.StreamHandler", "formatter": "default_fmt", } }, "loggers": { "": { "handlers": ["default_handler"], "level": "DEBUG", } }, }, }, } self.assertEqual(config.load_from_dict(config_data), config_data) def test_load_from_dict_secondary_service_account_keys(self): """Should convert secondary keys to primary keys.""" config_data = { **self.default_dict_config, **{ "path_to_private_key_file": self.json_key_file_path, "delegated_account": self.impersonated_email, }, } result = config.load_from_dict(config_data) self.assertEqual(result["json_key_file_path"], self.json_key_file_path) self.assertEqual(result["impersonated_email"], self.impersonated_email) def test_load_from_dict_error(self): """Should raise ValueError if invalid dict is given.""" config_data = 111 self.assertRaises(ValueError, config.load_from_dict, config_data) @mock.patch.object(config, "_logger", mock.Mock()) @mock.patch("logging.config.dictConfig") def test_load_from_env(self, config_spy): """Should load config from environment variables.""" environ = { **self.default_env_var_config, **{ "GOOGLE_ADS_CLIENT_ID": self.client_id, "GOOGLE_ADS_CLIENT_SECRET": self.client_secret, "GOOGLE_ADS_REFRESH_TOKEN": self.refresh_token, "GOOGLE_ADS_LOGGING": '{"test": true}', "GOOGLE_ADS_ENDPOINT": self.endpoint, "GOOGLE_ADS_LOGIN_CUSTOMER_ID": self.login_customer_id, "GOOGLE_ADS_LINKED_CUSTOMER_ID": self.linked_customer_id, "GOOGLE_ADS_LINKED_CUSTOMER_ID": self.linked_customer_id, "GOOGLE_ADS_JSON_KEY_FILE_PATH": self.json_key_file_path, "GOOGLE_ADS_IMPERSONATED_EMAIL": self.impersonated_email, }, } with mock.patch("os.environ", environ): result = config.load_from_env() self.assertEqual( result, { "developer_token": self.developer_token, "use_proto_plus": self.use_proto_plus, "client_id": self.client_id, "client_secret": self.client_secret, "refresh_token": self.refresh_token, "logging": {"test": True}, "endpoint": self.endpoint, "login_customer_id": self.login_customer_id, "linked_customer_id": self.linked_customer_id, "json_key_file_path": self.json_key_file_path, "impersonated_email": self.impersonated_email, }, ) config_spy.assert_called_once() @mock.patch.object(config, "_logger", mock.Mock()) def test_load_from_env_missing_required_key(self): """Should raise ValueError if missing required env var..""" # environ is missing required developer_token key environ = { "GOOGLE_ADS_CLIENT_ID": self.client_id, "GOOGLE_ADS_CLIENT_SECRET": self.client_secret, "GOOGLE_ADS_REFRESH_TOKEN": self.refresh_token, "GOOGLE_ADS_LOGGING": '{"test": true}', "GOOGLE_ADS_ENDPOINT": self.endpoint, "GOOGLE_ADS_LOGIN_CUSTOMER_ID": self.login_customer_id, "GOOGLE_ADS_JSON_KEY_FILE_PATH": self.json_key_file_path, "GOOGLE_ADS_IMPERSONATED_EMAIL": self.impersonated_email, } with mock.patch("os.environ", environ): self.assertRaises(ValueError, config.load_from_env) def test_load_from_env_config_file_path(self): """Should delegate to load_from_yaml_file method.""" environ = { "GOOGLE_ADS_CONFIGURATION_FILE_PATH": self.configuration_file_path } with mock.patch("os.environ", environ): with mock.patch.object( config, "load_from_yaml_file", # Return basic config to pass validation. return_value=self.default_dict_config, ) as spy: config.load_from_env() spy.assert_called_once() def test_load_from_env_linked_cid(self): """Should load linked CID from environment when specified""" environ = { **self.default_env_var_config, **{ "GOOGLE_ADS_LINKED_CUSTOMER_ID": self.linked_customer_id, }, } with mock.patch("os.environ", environ): results = config.load_from_env() self.assertEqual( results["linked_customer_id"], self.linked_customer_id ) def test_load_from_env_config_file_path_added_vars(self): """Should use config from yaml when config file path env var exists. If a configuration file path is defined via an environment variable then the yaml file at that location will be loaded and any other environment variable configuration will be ignored. """ env_dev_token = "123456" environ = { **self.default_env_var_config, **{ "GOOGLE_ADS_CONFIGURATION_FILE_PATH": self.configuration_file_path, "GOOGLE_ADS_DEVELOPER_TOKEN": env_dev_token, }, } with mock.patch("os.environ", environ), mock.patch.object( # Mock load_from_yaml_file return value so it returns # a default dict config that passes validation config, "load_from_yaml_file", return_value=self.default_dict_config, ) as spy: # Assert that the config values were retrieved from the yaml # file and not from environment variables. result = config.load_from_env() self.assertEqual(result["developer_token"], self.developer_token) self.assertEqual(result["use_proto_plus"], self.use_proto_plus) @mock.patch.object(config, "_logger", mock.Mock()) def test_load_from_env_redundant_file_path(self): """JSON_KEY_FILE_PATH takes precedent if both exist.""" environ = { **self.default_env_var_config, # The two below variables represent the same key, and this test # checks that JSON_KEY_FILE_PATH takes precedent and overwrites # the path_to_private_key_file_path key in the returned dict. **{ "GOOGLE_ADS_PATH_TO_PRIVATE_KEY_FILE": self.path_to_private_key_file, "GOOGLE_ADS_JSON_KEY_FILE_PATH": self.json_key_file_path, }, } with mock.patch("os.environ", environ): result = config.load_from_env() self.assertEqual( result["json_key_file_path"], self.json_key_file_path ) @mock.patch.object(config, "_logger", mock.Mock()) def test_load_from_env_secondary_file_path(self): """JSON_KEY_FILE_PATH is used instead of secondary var name.""" environ = { **self.default_env_var_config, **{ "GOOGLE_ADS_PATH_TO_PRIVATE_KEY_FILE": self.path_to_private_key_file, }, } with mock.patch("os.environ", environ): result = config.load_from_env() self.assertEqual( result["json_key_file_path"], self.path_to_private_key_file ) @mock.patch.object(config, "_logger", mock.Mock()) def test_load_from_env_redundant_delegated_email(self): """IMPERSONATED_EMAIL takes precedent if both exist.""" environ = { **self.default_env_var_config, **{ # The two below variables represent the same key, and this test # checks that IMPERSONATED_EMAIL takes precedent and overwrites # the delegate_account key in the returned dict. "GOOGLE_ADS_DELEGATED_ACCOUNT": self.delegated_account, "GOOGLE_ADS_IMPERSONATED_EMAIL": self.impersonated_email, }, } with mock.patch("os.environ", environ): result = config.load_from_env() self.assertEqual( result["impersonated_email"], self.impersonated_email ) @mock.patch.object(config, "_logger", mock.Mock()) def test_load_from_env_secondary_delegated_email(self): """IMPERSONATED_EMAIL is used instead of secondary var name.""" environ = { **self.default_env_var_config, **{ "GOOGLE_ADS_DELEGATED_ACCOUNT": self.delegated_account, }, } with mock.patch("os.environ", environ): result = config.load_from_env() self.assertEqual( result["impersonated_email"], self.delegated_account ) def test_validate_dict(self): config_data = {"invalid": "config"} self.assertRaises(ValueError, config.validate_dict, config_data) def test_validate_dict(self): config_data = {key: "test" for key in config._REQUIRED_KEYS} try: config.validate_dict(config_data) except ValueError as ex: self.fail("test_validate_dict failed unexpectedly: {}".format(ex)) def test_validate_dict_with_invalid_login_cid(self): config_data = {key: "test" for key in config._REQUIRED_KEYS} config_data["login_customer_id"] = "123-456-5789" self.assertRaises(ValueError, config.validate_dict, config_data) def test_validate_dict_with_invalid_linked_cid(self): config_data = {key: "test" for key in config._REQUIRED_KEYS} config_data["linked_customer_id"] = "123-456-5789" self.assertRaises(ValueError, config.validate_dict, config_data) def test_validate_dict_with_valid_login_cid(self): config_data = {key: "test" for key in config._REQUIRED_KEYS} config_data["login_customer_id"] = "1234567893" try: config.validate_dict(config_data) except ValueError as ex: self.fail( "test_validate_dict_with_login_cid failed unexpectedly: " "{}".format(ex) ) def test_validate_dict_with_valid_linked_cid(self): config_data = {key: "test" for key in config._REQUIRED_KEYS} config_data["linked_customer_id"] = "1234567893" try: config.validate_dict(config_data) except ValueError as ex: self.fail( "test_validate_dict_with_linked_cid failed unexpectedly: " "{}".format(ex) ) def test_validate_login_customer_id_invalid(self): self.assertRaises( ValueError, config.validate_login_customer_id, "123-456-7890" ) def test_validate_login_customer_id_unicode(self): self.assertRaises( ValueError, config.validate_login_customer_id, "1230\u00B2" ) def test_validate_login_customer_id_embedded(self): self.assertRaises( ValueError, config.validate_login_customer_id, "abc1234567890def" ) def test_validate_login_customer_id_negative(self): self.assertRaises( ValueError, config.validate_login_customer_id, "-1234567890" ) def test_validate_login_customer_id_too_short(self): self.assertRaises(ValueError, config.validate_login_customer_id, "123") def test_validate_linked_customer_id_too_short(self): self.assertRaises(ValueError, config.validate_linked_customer_id, "123") def test_get_oauth2_installed_app_keys(self): self.assertEqual( config.get_oauth2_installed_app_keys(), config._OAUTH2_INSTALLED_APP_KEYS, ) def test_get_oauth2_required_service_account_keys(self): self.assertEqual( config.get_oauth2_required_service_account_keys(), config._OAUTH2_REQUIRED_SERVICE_ACCOUNT_KEYS, ) def test_convert_login_customer_id_to_str_with_int(self): config_data = {"login_customer_id": 1234567890} expected = {"login_customer_id": "1234567890"} self.assertEqual( config.convert_login_customer_id_to_str(config_data), expected ) def test_parse_login_customer_id_with_str(self): config_data = {"login_customer_id": "1234567890"} self.assertEqual( config.convert_login_customer_id_to_str(config_data), config_data ) def test_parse_login_customer_id_with_none(self): config_data = {"not_login_customer_id": 1234567890} self.assertEqual( config.convert_login_customer_id_to_str(config_data), config_data ) def test_convert_linked_customer_id_to_str_with_int(self): config_data = {"linked_customer_id": 1234567890} expected = {"linked_customer_id": "1234567890"} self.assertEqual( config.convert_linked_customer_id_to_str(config_data), expected ) def test_parse_linked_customer_id_with_str(self): config_data = {"linked_customer_id": "1234567890"} self.assertEqual( config.convert_linked_customer_id_to_str(config_data), config_data ) def test_parse_linked_customer_id_with_none(self): config_data = {"not_linked_customer_id": 1234567890} self.assertEqual( config.convert_linked_customer_id_to_str(config_data), config_data ) def test_disambiguate_string_bool(self): self.assertEqual(config.disambiguate_string_bool(True), True) def test_disambiguate_string_bool_with_str(self): self.assertEqual(config.disambiguate_string_bool("True"), True) def test_disambiguate_string_bool_raises_value_error(self): self.assertRaises( ValueError, config.disambiguate_string_bool, "invalid" ) def test_disambiguate_string_bool_raises_type_error(self): self.assertRaises(TypeError, config.disambiguate_string_bool, {}) def test_load_from_env_use_cloud_org_for_api_access(self): """Should load api access flag from environment when specified""" environ = { **self.default_env_var_config, **{ "GOOGLE_ADS_USE_CLOUD_ORG_FOR_API_ACCESS": self.use_cloud_org_for_api_access, }, } with mock.patch("os.environ", environ): results = config.load_from_env() self.assertEqual( results["use_cloud_org_for_api_access"], self.use_cloud_org_for_api_access, ) def test_load_from_yaml_file_use_cloud_org_for_api_access(self): """Should load "use_cloud_org_for_api_access" config from a yaml.""" self._create_mock_yaml({"use_cloud_org_for_api_access": True}) result = config.load_from_yaml_file() self.assertEqual(result["use_cloud_org_for_api_access"], True) def test_load_from_yaml_file_use_cloud_org_for_api_access_not_set(self): """Should set "use_cloud_org_for_api_access" as None when not set.""" self._create_mock_yaml({}) result = config.load_from_yaml_file() self.assertEqual(result.get("use_cloud_org_for_api_access"), None)