ManagedIdentityClient will send claims and token_sha256_to_refresh to SF

Author: rayluo

tests/test_mi.py

@@ -1,3 +1,4 @@
+import hashlib
 import json
 import os
 import sys
@@ -80,6 +81,10 @@ def assertCacheStatus(self, app):
         self.assertEqual("managed_identity", at["realm"], "Should have expected realm")
 
     def _test_happy_path(self, app, mocked_http, expires_in, resource="R"):
+        """It tests a normal token request that is expected to be sent on wire,
+        a subsequent same token request that is expected to hit cache,
+        and then a request with claims_challenge.
+        """
         result = app.acquire_token_for_client(resource=resource)
         mocked_http.assert_called()
         call_count = mocked_http.call_count
@@ -128,6 +133,12 @@ def test_happy_path(self):
             text='{"access_token": "AT", "expires_in": "%s", "resource": "R"}' % expires_in,
         )) as mocked_method:
             self._test_happy_path(self.app, mocked_method, expires_in)
+            mocked_method.assert_called_with(
+                # The last call contained claims_challenge but IMDS doesn't support claims
+                'http://169.254.169.254/metadata/identity/oauth2/token',
+                params={'api-version': '2018-02-01', 'resource': 'R'},
+                headers={'Metadata': 'true'},
+            )
 
     def test_vm_error_should_be_returned_as_is(self):
         raw_error = '{"raw": "error format is undefined"}'
@@ -238,6 +249,17 @@ def _test_happy_path(self, app):
         )) as mocked_method:
             super(ServiceFabricTestCase, self)._test_happy_path(
                 app, mocked_method, expires_in)
+            mocked_method.assert_called_with(
+                # The last call contained claims_challenge so it should relay both claims and hash to SF
+                'http://localhost',
+                params={
+                    'api-version': '2019-07-01-preview',
+                    'resource': 'R',
+                    'claims': 'foo',
+                    'token_sha256_to_refresh': hashlib.sha256(b"AT").hexdigest(),
+                },
+                headers={'Secret': 'foo'}
+            )
 
     def test_happy_path(self):
         self._test_happy_path(self.app)