async def test_oauth_flow_completion_with_caching(
      self, manager, mock_tool_context
  ):
    """Test successful OAuth flow completion with proper credential caching.

    This tests the happy path where a user completes the OAuth flow
    and the system successfully creates and caches new credentials
    in both the manager and the tool context state.
    """
    # Mock OAuth response indicating completed flow
    mock_auth_response = Mock()
    mock_auth_response.oauth2.access_token = "new_access_token"
    mock_auth_response.oauth2.refresh_token = "new_refresh_token"
    mock_tool_context.get_auth_response.return_value = mock_auth_response

    # Create a mock credentials instance that will represent our created credentials
    mock_creds = Mock(spec=Credentials)
    # Make the JSON match what a real Credentials object would produce
    mock_creds_json = (
        '{"token": "new_access_token", "refresh_token": "new_refresh_token",'
        ' "token_uri": "https://oauth2.googleapis.com/token", "client_id":'
        ' "test_client_id", "client_secret": "test_client_secret", "scopes":'
        ' ["https://www.googleapis.com/auth/calendar"], "universe_domain":'
        ' "googleapis.com", "account": ""}'
    )
    mock_creds.to_json.return_value = mock_creds_json

    # Use the full module path as it appears in the project structure
    with patch(
        "google.adk.tools.bigquery.bigquery_credentials.Credentials",
        return_value=mock_creds,
    ) as mock_credentials_class:
      result = await manager.get_valid_credentials(mock_tool_context)

      # Verify new credentials were created
      assert result == mock_creds
      # Verify credentials are created with correct parameters
      mock_credentials_class.assert_called_once()
      call_kwargs = mock_credentials_class.call_args[1]
      assert call_kwargs["token"] == "new_access_token"
      assert call_kwargs["refresh_token"] == "new_refresh_token"

      # Verify credentials are not cached in manager
      assert manager.credentials_config.credentials is None
      # Verify credentials are also cached in tool context state
      assert (
          mock_tool_context.state[BIGQUERY_TOKEN_CACHE_KEY] == mock_creds_json
      )
Test successful OAuth flow completion with proper credential caching.

    This tests the happy path where a user completes the OAuth flow
    and the system successfully creates and caches new credentials
    in both the manager and the tool context state.
    
tests/unittests/tools/bigquery/test_bigquery_credentials_manager.py