From ef6e47edbdbb560c4634392f11a228623c5f7732 Mon Sep 17 00:00:00 2001 From: nbayati <99771966+nbayati@users.noreply.github.com> Date: Thu, 16 Jul 2026 23:38:25 -0700 Subject: [PATCH 1/4] refactor: exit early when agent cert config is outside well-known directory To prevent unnecessary startup latency, this updates the agent certificate retrieval to only poll when the config points to the well-known workload directory (where Cloud Run dynamically mounts files). For all other paths, it exits early. - Extracts config parsing into a dedicated helper function - Removes the hardcoded fallback to the well-known path when no config is provided - Adds test coverage for new logic --- .../google/auth/_agent_identity_utils.py | 144 ++++++------ .../tests/test_agent_identity_utils.py | 208 +++++++++--------- 2 files changed, 182 insertions(+), 170 deletions(-) diff --git a/packages/google-auth/google/auth/_agent_identity_utils.py b/packages/google-auth/google/auth/_agent_identity_utils.py index f2545f28238e..5978de2bba60 100644 --- a/packages/google-auth/google/auth/_agent_identity_utils.py +++ b/packages/google-auth/google/auth/_agent_identity_utils.py @@ -67,93 +67,80 @@ def _is_certificate_file_ready(path): st = os.stat(path) return stat.S_ISREG(st.st_mode) and st.st_size > 0 except PermissionError: - # Propagate PermissionError to let caller handle it (fail-fast or fallback) + # Propagate PermissionError to let caller handle it (e.g., return early or fallback) raise except OSError: return False def get_agent_identity_certificate_path(): - """Gets the certificate path from the certificate config file. + """Gets the agent certificate path from the certificate config file. The path to the certificate config file is read from the GOOGLE_API_CERTIFICATE_CONFIG environment variable. This function - implements a retry mechanism to handle cases where the environment + can optionally trigger polling to handle cases where the environment variable is set before the files are available on the filesystem. Returns: - str: The path to the leaf certificate file. + Optional[str]: The path to the agent's certificate file, or None if unavailable. Raises: google.auth.exceptions.RefreshError: If the certificate config file or the certificate file cannot be found after retries. """ - import json - cert_config_path = os.environ.get(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG) - # Check if the well-known workload directory is mounted. + if not cert_config_path: + return None + + # We trigger polling only if the config path points to the well-known directory. + # Cloud Run dynamically generates these files in this directory, and both the + # config file and the certificate file may experience a brief startup latency. + # For all other paths, we return early to avoid introducing unnecessary startup + # delays. well_known_dir = os.path.dirname(_WELL_KNOWN_CERT_PATH) - has_well_known_dir = os.path.exists(well_known_dir) + should_poll = cert_config_path.startswith(well_known_dir) - # If we have neither a config path nor a well-known mount directory, exit immediately. - if not cert_config_path and not has_well_known_dir: - return None + return _get_cert_path_with_optional_polling(cert_config_path, should_poll) - # If ECP config path is specified but does not exist, and we are on a workstation, fail-fast immediately. - if ( - cert_config_path - and not has_well_known_dir - and not os.path.exists(cert_config_path) - ): - return None +def _get_cert_path_with_optional_polling(cert_config_path, should_poll): + """Gets the certificate path, optionally polling until it is ready. + + Args: + cert_config_path (str): The path to the certificate configuration file. + should_poll (bool): If True, the function will poll for the file and + certificate to be ready. If False, it will check only once and + return early if they are not immediately available. + + Returns: + str: The path to the certificate file, or None if unavailable. + + Raises: + google.auth.exceptions.RefreshError: If the certificate config file + or the certificate file cannot be found after retries. + """ has_logged_config_warning = False has_logged_cert_warning = False for interval in _POLLING_INTERVALS: try: - # Path A: Config file is explicitly set - if cert_config_path: - with open(cert_config_path, "r") as f: - cert_config = json.load(f) - - cert_configs = ( - cert_config.get("cert_configs") - if isinstance(cert_config, dict) - else None - ) - workload_config = ( - cert_configs.get("workload") - if isinstance(cert_configs, dict) - else None - ) - - if ( - not isinstance(workload_config, dict) - or "cert_path" not in workload_config - ): - return None + cert_path = _parse_cert_path_from_config(cert_config_path) - cert_path = workload_config["cert_path"] - if _is_certificate_file_ready(cert_path): - return cert_path - - # The config was parsed, but the cert file is not ready yet - target_path = cert_path + if cert_path is None: + return None - # Path B: Config is NOT set, fallback to the well-known path - else: - if _is_certificate_file_ready(_WELL_KNOWN_CERT_PATH): - return _WELL_KNOWN_CERT_PATH + if _is_certificate_file_ready(cert_path): + return cert_path - # The well-known cert file is not ready yet - target_path = _WELL_KNOWN_CERT_PATH + # The config was parsed, but the cert file is not ready yet + if not should_poll: + # If polling is disabled, return early. + return None - # Log a warning on the first failed attempt to load the certificate file if not has_logged_cert_warning: warnings.warn( - f"Certificate file not ready at {target_path}. Retrying until startup timeout (up to {_TOTAL_TIMEOUT} seconds total)..." + f"Certificate file not ready at {cert_path}. Retrying until startup timeout (up to {_TOTAL_TIMEOUT} seconds total)..." ) has_logged_cert_warning = True @@ -164,25 +151,24 @@ def get_agent_identity_certificate_path(): ) return None except (IOError, ValueError, KeyError) as e: - if cert_config_path and os.path.exists(cert_config_path): + if os.path.exists(cert_config_path): # If the file exists but has invalid JSON or is unreadable, - # we assume it is in its final format and fail-fast by returning None. + # we assume it is in its final format and return early (returning None). + return None + + if not should_poll: + # If polling is disabled, return early if the file doesn't exist. return None - if not has_logged_config_warning and cert_config_path: + if not has_logged_config_warning: warnings.warn( f"Certificate config file not found or incomplete: {e} (from " f"{environment_vars.GOOGLE_API_CERTIFICATE_CONFIG} environment variable). " f"Retrying until startup timeout (up to {_TOTAL_TIMEOUT} seconds total)..." ) has_logged_config_warning = True - pass - # A sleep is required in two cases: - # 1. The config file is not found (the except block). - # 2. The config file/well-known path is found, but the certificate is not yet available. - # In both cases, we need to poll, so we sleep on every iteration - # that doesn't return a certificate. + # Sleep before the next polling attempt. time.sleep(interval) raise exceptions.RefreshError( @@ -193,6 +179,40 @@ def get_agent_identity_certificate_path(): ) +def _parse_cert_path_from_config(cert_config_path): + """Reads the cert config file and returns the cert_path. + + Args: + cert_config_path (str): The path to the certificate configuration file. + + Returns: + Optional[str]: The path to the certificate file, or None if not found + in the config. + + Raises: + IOError: If the certificate config file cannot be read. + ValueError: If the certificate config file contains invalid JSON. + KeyError: If the certificate config file does not contain the + expected structure. + """ + import json + + with open(cert_config_path, "r") as f: + cert_config = json.load(f) + + cert_configs = ( + cert_config.get("cert_configs") if isinstance(cert_config, dict) else None + ) + workload_config = ( + cert_configs.get("workload") if isinstance(cert_configs, dict) else None + ) + + if not isinstance(workload_config, dict) or "cert_path" not in workload_config: + return None + + return workload_config["cert_path"] + + def get_and_parse_agent_identity_certificate(): """Gets and parses the agent identity certificate if not opted out. diff --git a/packages/google-auth/tests/test_agent_identity_utils.py b/packages/google-auth/tests/test_agent_identity_utils.py index 7394d6914e38..b86c91a218e7 100644 --- a/packages/google-auth/tests/test_agent_identity_utils.py +++ b/packages/google-auth/tests/test_agent_identity_utils.py @@ -65,6 +65,17 @@ def test_parse_certificate(self, mock_load_cert): mock_load_cert.assert_called_once_with(b"cert_bytes") assert result == mock_load_cert.return_value + def test_is_certificate_file_ready_empty_path(self): + result = _agent_identity_utils._is_certificate_file_ready("") + assert result is False + + def test_get_agent_identity_certificate_path_empty_env(self, monkeypatch): + monkeypatch.delenv( + environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, raising=False + ) + result = _agent_identity_utils.get_agent_identity_certificate_path() + assert result is None + @mock.patch("google.auth._agent_identity_utils.os.stat") def test_is_certificate_file_ready_permission_error(self, mock_stat): mock_stat.side_effect = PermissionError("Permission denied") @@ -228,20 +239,18 @@ def test_get_agent_identity_certificate_path_success(self, tmpdir, monkeypatch): def test_get_agent_identity_certificate_path_retry( self, mock_exists, mock_sleep, tmpdir, monkeypatch ): + monkeypatch.setattr( + "google.auth._agent_identity_utils._WELL_KNOWN_CERT_PATH", + str(tmpdir.join("certificates.pem")), + ) config_path = tmpdir.join("config.json") monkeypatch.setenv( environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path) ) - # Simulate workload env (well_known_dir exists) to avoid fail-fast - def exists_side_effect(path): - if path == "/var/run/secrets/workload-spiffe-credentials": - return True - return False - - mock_exists.side_effect = exists_side_effect - # File doesn't exist initially + mock_exists.return_value = False + with pytest.raises(exceptions.RefreshError): _agent_identity_utils.get_agent_identity_certificate_path() @@ -252,18 +261,17 @@ def exists_side_effect(path): def test_get_agent_identity_certificate_path_failure( self, mock_exists, mock_sleep, tmpdir, monkeypatch ): + monkeypatch.setattr( + "google.auth._agent_identity_utils._WELL_KNOWN_CERT_PATH", + str(tmpdir.join("certificates.pem")), + ) config_path = tmpdir.join("non_existent_config.json") monkeypatch.setenv( environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path) ) # Simulate workload env (well_known_dir exists) to avoid fail-fast - def exists_side_effect(path): - if path == "/var/run/secrets/workload-spiffe-credentials": - return True - return False - - mock_exists.side_effect = exists_side_effect + mock_exists.return_value = False with pytest.raises(exceptions.RefreshError) as excinfo: _agent_identity_utils.get_agent_identity_certificate_path() @@ -275,24 +283,73 @@ def exists_side_effect(path): ) assert mock_sleep.call_count == len(_agent_identity_utils._POLLING_INTERVALS) - def test_get_agent_identity_certificate_path_workstation_fail_fast( - self, tmpdir, monkeypatch + + @mock.patch("time.sleep") + @mock.patch("google.auth._agent_identity_utils.os.path.exists") + def test_get_agent_identity_certificate_path_fail_fast_config_missing( + self, mock_exists, mock_sleep, tmpdir, monkeypatch ): - config_path = tmpdir.join("non_existent_config.json") + # Simulate config path outside well-known dir where config file does not exist. + well_known_path = tmpdir.mkdir("well_known").join("certificates.pem") + monkeypatch.setattr( + "google.auth._agent_identity_utils._WELL_KNOWN_CERT_PATH", + str(well_known_path), + ) + + config_path = tmpdir.mkdir("custom").join("custom_config.json") + monkeypatch.setenv( + environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path) + ) + + mock_exists.return_value = False + + result = _agent_identity_utils.get_agent_identity_certificate_path() + + assert result is None + mock_sleep.assert_not_called() + + @mock.patch("time.sleep") + @mock.patch("google.auth._agent_identity_utils.os.path.exists") + def test_get_agent_identity_certificate_path_fail_fast_cert_missing( + self, mock_exists, mock_sleep, tmpdir, monkeypatch + ): + # Simulate config path outside well-known dir where config is valid but cert is missing. + well_known_path = tmpdir.mkdir("well_known_cert").join("certificates.pem") + monkeypatch.setattr( + "google.auth._agent_identity_utils._WELL_KNOWN_CERT_PATH", + str(well_known_path), + ) + + custom_dir = tmpdir.mkdir("custom_cert") + config_path = custom_dir.join("custom_config.json") + cert_path_str = str(custom_dir.join("cert.pem")) + + config_path.write( + json.dumps({"cert_configs": {"workload": {"cert_path": cert_path_str}}}) + ) monkeypatch.setenv( environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path) ) - # On a workstation, well_known_dir does not exist, and config file is missing. - # It should fail-fast and return None immediately. + def exists_side_effect(path): + return path == str(config_path) + + mock_exists.side_effect = exists_side_effect + result = _agent_identity_utils.get_agent_identity_certificate_path() + assert result is None + mock_sleep.assert_not_called() @mock.patch("time.sleep") @mock.patch("google.auth._agent_identity_utils.os.path.exists") def test_get_agent_identity_certificate_path_cert_not_found( self, mock_exists, mock_sleep, tmpdir, monkeypatch ): + monkeypatch.setattr( + "google.auth._agent_identity_utils._WELL_KNOWN_CERT_PATH", + str(tmpdir.join("certificates.pem")), + ) cert_path_str = str(tmpdir.join("cert.pem")) config_path = tmpdir.join("config.json") config_path.write( @@ -380,115 +437,50 @@ def test_get_agent_identity_certificate_path_workload_config_missing_cert_path( @mock.patch("time.sleep") @mock.patch("google.auth._agent_identity_utils.os.path.exists") - @mock.patch("google.auth._agent_identity_utils._is_certificate_file_ready") - def test_get_agent_identity_certificate_path_no_config_but_has_well_known_dir( - self, mock_is_ready, mock_exists, mock_sleep, monkeypatch + def test_get_agent_identity_certificate_path_permission_error_config( + self, mock_exists, mock_sleep, tmpdir, monkeypatch ): - monkeypatch.delenv( - environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, raising=False + config_path = tmpdir.join("config.json") + monkeypatch.setenv( + environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path) ) - - # Simulate that the well-known workload mount directory exists, and the cert is ready + # Mock os.path.exists so ECP workstation fail-fast is not triggered mock_exists.return_value = True - mock_is_ready.return_value = True - - result = _agent_identity_utils.get_agent_identity_certificate_path() - - # Should return the well-known path immediately - assert result == _agent_identity_utils._WELL_KNOWN_CERT_PATH - mock_sleep.assert_not_called() - @mock.patch("time.sleep") - @mock.patch("google.auth._agent_identity_utils.os.path.exists") - def test_get_agent_identity_certificate_path_no_config_no_well_known_dir( - self, mock_exists, mock_sleep, monkeypatch - ): - monkeypatch.delenv( - environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, raising=False - ) - - # Simulate that the well-known mount directory does NOT exist - mock_exists.return_value = False + # Mocking open to raise PermissionError + mock_open = mock.mock_open() + mock_open.side_effect = PermissionError("Permission denied") - result = _agent_identity_utils.get_agent_identity_certificate_path() + with mock.patch("builtins.open", mock_open): + result = _agent_identity_utils.get_agent_identity_certificate_path() - # Should return None immediately without polling assert result is None mock_sleep.assert_not_called() @mock.patch("time.sleep") - @mock.patch("google.auth._agent_identity_utils.os.path.exists") @mock.patch("google.auth._agent_identity_utils._is_certificate_file_ready") - def test_get_agent_identity_certificate_path_no_config_well_known_polling_success( - self, mock_is_ready, mock_exists, mock_sleep, monkeypatch + def test_get_agent_identity_certificate_path_permission_error_cert_file( + self, mock_is_ready, mock_sleep, tmpdir, monkeypatch ): - monkeypatch.delenv( - environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, raising=False + well_known_path = tmpdir.mkdir("well_known").join("certificates.pem") + monkeypatch.setattr( + "google.auth._agent_identity_utils._WELL_KNOWN_CERT_PATH", + str(well_known_path), ) - # Simulate that the directory exists, file appears on 2nd try - mock_exists.return_value = True - mock_is_ready.side_effect = [False, True] - - result = _agent_identity_utils.get_agent_identity_certificate_path() - - assert result == _agent_identity_utils._WELL_KNOWN_CERT_PATH - assert mock_sleep.call_count == 1 - - @mock.patch("time.sleep") - @mock.patch("google.auth._agent_identity_utils.os.path.exists") - @mock.patch("google.auth._agent_identity_utils._is_certificate_file_ready") - def test_get_agent_identity_certificate_path_no_config_well_known_polling_timeout( - self, mock_is_ready, mock_exists, mock_sleep, monkeypatch - ): - monkeypatch.delenv( - environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, raising=False - ) - - # Simulate that the directory exists, but file never appears - mock_exists.return_value = True - mock_is_ready.return_value = False - - with pytest.raises(exceptions.RefreshError): - _agent_identity_utils.get_agent_identity_certificate_path() - - assert mock_sleep.call_count == len(_agent_identity_utils._POLLING_INTERVALS) - - @mock.patch("time.sleep") - @mock.patch("google.auth._agent_identity_utils._is_certificate_file_ready") - @mock.patch("google.auth._agent_identity_utils.os.path.exists") - def test_get_agent_identity_certificate_path_permission_error_well_known( - self, mock_exists, mock_is_ready, mock_sleep, monkeypatch - ): - monkeypatch.delenv( - environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, raising=False + config_path = tmpdir.mkdir("custom").join("custom_config.json") + cert_path_str = str(tmpdir.join("cert.pem")) + config_path.write( + json.dumps({"cert_configs": {"workload": {"cert_path": cert_path_str}}}) ) - mock_exists.return_value = True - mock_is_ready.side_effect = PermissionError("Permission denied") - - # It should fail-fast and return None immediately - result = _agent_identity_utils.get_agent_identity_certificate_path() - assert result is None - mock_sleep.assert_not_called() - - @mock.patch("time.sleep") - @mock.patch("google.auth._agent_identity_utils.os.path.exists") - def test_get_agent_identity_certificate_path_permission_error_config( - self, mock_exists, mock_sleep, tmpdir, monkeypatch - ): - config_path = tmpdir.join("config.json") monkeypatch.setenv( environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path) ) - # Mock os.path.exists so ECP workstation fail-fast is not triggered - mock_exists.return_value = True - # Mocking open to raise PermissionError - mock_open = mock.mock_open() - mock_open.side_effect = PermissionError("Permission denied") + # Mock _is_certificate_file_ready to raise PermissionError + mock_is_ready.side_effect = PermissionError("Permission denied") - with mock.patch("builtins.open", mock_open): - result = _agent_identity_utils.get_agent_identity_certificate_path() + result = _agent_identity_utils.get_agent_identity_certificate_path() assert result is None mock_sleep.assert_not_called() From d43f6de4c562c2cd7124e63fdfeadcfecade4094 Mon Sep 17 00:00:00 2001 From: nbayati <99771966+nbayati@users.noreply.github.com> Date: Thu, 16 Jul 2026 23:44:35 -0700 Subject: [PATCH 2/4] fix lint issue --- packages/google-auth/tests/test_agent_identity_utils.py | 1 - 1 file changed, 1 deletion(-) diff --git a/packages/google-auth/tests/test_agent_identity_utils.py b/packages/google-auth/tests/test_agent_identity_utils.py index b86c91a218e7..6c3f901eb223 100644 --- a/packages/google-auth/tests/test_agent_identity_utils.py +++ b/packages/google-auth/tests/test_agent_identity_utils.py @@ -283,7 +283,6 @@ def test_get_agent_identity_certificate_path_failure( ) assert mock_sleep.call_count == len(_agent_identity_utils._POLLING_INTERVALS) - @mock.patch("time.sleep") @mock.patch("google.auth._agent_identity_utils.os.path.exists") def test_get_agent_identity_certificate_path_fail_fast_config_missing( From fbf145284ec81b7ad33fa6cc72ef8b89f6867b04 Mon Sep 17 00:00:00 2001 From: nbayati <99771966+nbayati@users.noreply.github.com> Date: Fri, 17 Jul 2026 13:04:36 -0700 Subject: [PATCH 3/4] fix: address PR feedback for agent identity utils --- .../google/auth/_agent_identity_utils.py | 12 ++++++++++-- .../tests/test_agent_identity_utils.py | 17 +++++++++++++++++ 2 files changed, 27 insertions(+), 2 deletions(-) diff --git a/packages/google-auth/google/auth/_agent_identity_utils.py b/packages/google-auth/google/auth/_agent_identity_utils.py index 5978de2bba60..4de5d709b4b5 100644 --- a/packages/google-auth/google/auth/_agent_identity_utils.py +++ b/packages/google-auth/google/auth/_agent_identity_utils.py @@ -99,7 +99,15 @@ def get_agent_identity_certificate_path(): # For all other paths, we return early to avoid introducing unnecessary startup # delays. well_known_dir = os.path.dirname(_WELL_KNOWN_CERT_PATH) - should_poll = cert_config_path.startswith(well_known_dir) + try: + abs_cert_path = os.path.abspath(cert_config_path) + abs_well_known_dir = os.path.abspath(well_known_dir) + should_poll = ( + os.path.commonpath([abs_well_known_dir, abs_cert_path]) + == abs_well_known_dir + ) + except ValueError: + should_poll = False return _get_cert_path_with_optional_polling(cert_config_path, should_poll) @@ -197,7 +205,7 @@ def _parse_cert_path_from_config(cert_config_path): """ import json - with open(cert_config_path, "r") as f: + with open(cert_config_path, "r", encoding="utf-8") as f: cert_config = json.load(f) cert_configs = ( diff --git a/packages/google-auth/tests/test_agent_identity_utils.py b/packages/google-auth/tests/test_agent_identity_utils.py index 6c3f901eb223..681dfe40fcfe 100644 --- a/packages/google-auth/tests/test_agent_identity_utils.py +++ b/packages/google-auth/tests/test_agent_identity_utils.py @@ -76,6 +76,23 @@ def test_get_agent_identity_certificate_path_empty_env(self, monkeypatch): result = _agent_identity_utils.get_agent_identity_certificate_path() assert result is None + @mock.patch("google.auth._agent_identity_utils.os.path.commonpath") + @mock.patch( + "google.auth._agent_identity_utils._get_cert_path_with_optional_polling" + ) + def test_get_agent_identity_certificate_path_value_error( + self, mock_get_cert, mock_commonpath, monkeypatch + ): + monkeypatch.setenv( + environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, "/path/to/config.json" + ) + mock_commonpath.side_effect = ValueError("Different drives") + mock_get_cert.return_value = "cert_path" + + result = _agent_identity_utils.get_agent_identity_certificate_path() + assert result == "cert_path" + mock_get_cert.assert_called_once_with("/path/to/config.json", False) + @mock.patch("google.auth._agent_identity_utils.os.stat") def test_is_certificate_file_ready_permission_error(self, mock_stat): mock_stat.side_effect = PermissionError("Permission denied") From 10763adc685972f9731c1e1af3f07475428b8539 Mon Sep 17 00:00:00 2001 From: nbayati <99771966+nbayati@users.noreply.github.com> Date: Fri, 17 Jul 2026 13:41:07 -0700 Subject: [PATCH 4/4] test: add missing unit test --- .../tests/test_agent_identity_utils.py | 58 +++++++++++++++++++ 1 file changed, 58 insertions(+) diff --git a/packages/google-auth/tests/test_agent_identity_utils.py b/packages/google-auth/tests/test_agent_identity_utils.py index 681dfe40fcfe..8cba306fe2d6 100644 --- a/packages/google-auth/tests/test_agent_identity_utils.py +++ b/packages/google-auth/tests/test_agent_identity_utils.py @@ -273,6 +273,64 @@ def test_get_agent_identity_certificate_path_retry( assert mock_sleep.call_count == len(_agent_identity_utils._POLLING_INTERVALS) + @mock.patch("time.sleep") + @mock.patch("google.auth._agent_identity_utils._is_certificate_file_ready") + def test_get_agent_identity_certificate_path_retry_success( + self, mock_is_ready, mock_sleep, tmpdir, monkeypatch + ): + monkeypatch.setattr( + "google.auth._agent_identity_utils._WELL_KNOWN_CERT_PATH", + str(tmpdir.join("certificates.pem")), + ) + cert_path_str = str(tmpdir.join("cert.pem")) + config_path = tmpdir.join("config.json") + config_path.write( + json.dumps({"cert_configs": {"workload": {"cert_path": cert_path_str}}}) + ) + monkeypatch.setenv( + environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path) + ) + + # First attempt: file missing/not ready. Second attempt: succeeds. + mock_is_ready.side_effect = [False, True] + + result = _agent_identity_utils.get_agent_identity_certificate_path() + + assert result == cert_path_str + assert mock_sleep.call_count == 1 + assert mock_is_ready.call_count == 2 + + @mock.patch("time.sleep") + @mock.patch("google.auth._agent_identity_utils._is_certificate_file_ready") + def test_get_agent_identity_certificate_path_config_retry_success( + self, mock_is_ready, mock_sleep, tmpdir, monkeypatch + ): + monkeypatch.setattr( + "google.auth._agent_identity_utils._WELL_KNOWN_CERT_PATH", + str(tmpdir.join("certificates.pem")), + ) + cert_path_str = str(tmpdir.join("cert.pem")) + config_path = tmpdir.join("config.json") + monkeypatch.setenv( + environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, str(config_path) + ) + + mock_is_ready.return_value = True + + def write_config(*args, **kwargs): + config_path.write( + json.dumps({"cert_configs": {"workload": {"cert_path": cert_path_str}}}) + ) + + # First attempt: config file missing. Sleep side effect creates it. + mock_sleep.side_effect = write_config + + result = _agent_identity_utils.get_agent_identity_certificate_path() + + assert result == cert_path_str + assert mock_sleep.call_count == 1 + assert mock_is_ready.call_count == 1 + @mock.patch("time.sleep") @mock.patch("google.auth._agent_identity_utils.os.path.exists") def test_get_agent_identity_certificate_path_failure(