@@ -948,6 +948,62 @@ def test_session_auth_prefers_security_token_signer(self):
948948 self .assertGreaterEqual (mock_oci .auth .signers .SecurityTokenSigner .call_count , 1 )
949949 mock_oci .signer .Signer .assert_not_called ()
950950
951+ def test_session_token_refreshed_on_subsequent_requests (self ):
952+ """Verify the refreshing signer picks up a new token written to the token file."""
953+ import tempfile
954+ import os
955+ from cohere .oci_client import map_request_to_oci
956+
957+ mock_oci = MagicMock ()
958+ mock_oci .signer .load_private_key_from_file .return_value = "private-key"
959+
960+ # Write initial token to a real temp file so we can overwrite it later.
961+ with tempfile .NamedTemporaryFile ("w" , suffix = ".token" , delete = False ) as tf :
962+ tf .write ("token-v1" )
963+ token_path = tf .name
964+
965+ try :
966+ with patch ("cohere.oci_client.lazy_oci" , return_value = mock_oci ):
967+ hook = map_request_to_oci (
968+ oci_config = {
969+ "security_token_file" : token_path ,
970+ "key_file" : "/irrelevant.pem" ,
971+ },
972+ oci_region = "us-chicago-1" ,
973+ oci_compartment_id = "ocid1.compartment.oc1..example" ,
974+ )
975+
976+ def _make_request ():
977+ req = MagicMock ()
978+ req .url .path = "/v2/embed"
979+ req .read .return_value = b'{"model":"embed-english-v3.0","texts":["hi"]}'
980+ req .method = "POST"
981+ req .extensions = {}
982+ return req
983+
984+ # First request uses token-v1
985+ hook (_make_request ())
986+ calls_after_first = mock_oci .auth .signers .SecurityTokenSigner .call_count
987+
988+ # Simulate token refresh by overwriting the file
989+ with open (token_path , "w" ) as _f :
990+ _f .write ("token-v2" )
991+
992+ # Second request — should re-read and use token-v2
993+ hook (_make_request ())
994+ self .assertGreater (
995+ mock_oci .auth .signers .SecurityTokenSigner .call_count ,
996+ calls_after_first ,
997+ "SecurityTokenSigner should be re-instantiated after token file update" ,
998+ )
999+ # Verify the latest call used the refreshed token
1000+ all_calls = mock_oci .auth .signers .SecurityTokenSigner .call_args_list
1001+ last_call = all_calls [- 1 ]
1002+ last_token = last_call .kwargs .get ("token" ) or (last_call .args [0 ] if last_call .args else None )
1003+ self .assertEqual (last_token , "token-v2" , "Last signing call must use the refreshed token" )
1004+ finally :
1005+ os .unlink (token_path )
1006+
9511007 def test_embed_response_lowercases_embedding_keys (self ):
9521008 """Test embed response uses lowercase keys expected by the SDK model."""
9531009 from cohere .oci_client import transform_oci_response_to_cohere
0 commit comments