Add test cases for TokenCache and SerializableTokenCache

Author: rayluo

msal/token_cache.py

@@ -50,7 +50,7 @@ def find(self, credential_type, target=None, query=None):
 		    if target else True)
                 ]
 
-    def add(self, event):
+    def add(self, event, now=None):
         # type: (dict) -> None
         # event typically contains: client_id, scope, token_endpoint,
         # resposne, params, data, grant_type
@@ -86,7 +86,7 @@ def add(self, event):
                     realm or "",
 		    target,
                     ]).lower()
-                now = time.time()
+                now = time.time() if now is None else now
                 self._cache.setdefault(self.CredentialType.ACCESS_TOKEN, {})[key] = {
                     "credential_type": self.CredentialType.ACCESS_TOKEN,
                     "secret": access_token,
@@ -202,8 +202,8 @@ class SerializableTokenCache(TokenCache):
         Indicates whether the cache state has changed since last
         :func:`~serialize` or :func:`~deserialize` call.
     """
-    def add(self, event):
-        super(SerializableTokenCache, self).add(event)
+    def add(self, event, **kwargs):
+        super(SerializableTokenCache, self).add(event, **kwargs)
         self.has_state_changed = True
 
     def remove_rt(self, rt_item):

tests/test_token_cache.py

@@ -0,0 +1,119 @@
+import logging
+import base64
+import json
+
+from msal.token_cache import *
+from tests import unittest
+
+
+logger = logging.getLogger(__name__)
+logging.basicConfig(level=logging.DEBUG)
+
+
+class TokenCacheTestCase(unittest.TestCase):
+
+    def setUp(self):
+        self.cache = TokenCache()
+
+    def testAdd(self):
+        client_info = base64.b64encode(b'''
+            {"uid": "uid", "utid": "utid"}
+            ''').decode('utf-8')
+        id_token = "header.%s.signature" % base64.b64encode(b'''{
+            "sub": "subject",
+            "oid": "object1234",
+            "preferred_username": "John Doe"
+            }''').decode('utf-8')
+        self.cache.add({
+            "client_id": "my_client_id",
+            "scope": ["s2", "s1", "s3"],  # Not in particular order
+            "token_endpoint": "https://login.example.com/contoso/v2/token",
+            "response": {
+		"access_token": "an access token",
+		"token_type": "some type",
+		"expires_in": 3600,
+		"refresh_token": "a refresh token",
+                "client_info": client_info,
+                "id_token": id_token,
+		},
+            }, now=1000)
+        self.assertEqual(
+            {
+                'cached_at': 1000,
+                'client_id': 'my_client_id',
+                'credential_type': 'AccessToken',
+                'environment': 'login.example.com',
+                'expires_on': 4600,
+                'extended_expires_on': 1000,
+                'home_account_id': "uid.utid",
+                'realm': 'contoso',
+                'secret': 'an access token',
+                'target': 's2 s1 s3',
+            },
+            self.cache._cache["AccessToken"].get(
+                'uid.utid-login.example.com-accesstoken-my_client_id-contoso-s2 s1 s3')
+            )
+        self.assertEqual(
+            {
+                'client_id': 'my_client_id',
+                'credential_type': 'RefreshToken',
+                'environment': 'login.example.com',
+                'home_account_id': "uid.utid",
+                'secret': 'a refresh token',
+                'target': 's2 s1 s3',
+            },
+            self.cache._cache["RefreshToken"].get(
+                'uid.utid-login.example.com-refreshtoken-my_client_id--s2 s1 s3')
+            )
+        self.assertEqual(
+            {
+                'home_account_id': "uid.utid",
+                'environment': 'login.example.com',
+                'realm': 'contoso',
+                'local_account_id': "object1234",
+                'username': "John Doe",
+                'authority_type': "AAD",
+            },
+            self.cache._cache["Account"].get('uid.utid-login.example.com-contoso')
+            )
+        self.assertEqual(
+            {
+                'credential_type': 'IdToken',
+                'secret': id_token,
+                'home_account_id': "uid.utid",
+                'environment': 'login.example.com',
+                'realm': 'contoso',
+                'client_id': 'my_client_id',
+            },
+            self.cache._cache["IdToken"].get(
+                'uid.utid-login.example.com-idtoken-my_client_id-contoso')
+            )
+
+
+class SerializableTokenCacheTestCase(TokenCacheTestCase):
+    # Run all inherited test methods, and have extra check in tearDown()
+
+    def setUp(self):
+        self.cache = SerializableTokenCache()
+        self.cache.deserialize("""
+            {
+                "AccessToken": {
+                    "an-entry": {
+                        "foo": "bar"
+                        }
+                    },
+                "customized": "whatever"
+            }
+            """)
+
+    def tearDown(self):
+        state = self.cache.serialize()
+        logger.debug("serialize() = %s", state)
+        # Now assert all extended content are kept intact
+        output = json.loads(state)
+        self.assertEqual(output.get("customized"), "whatever",
+            "Undefined cache keys and their values should be intact")
+        self.assertEqual(
+            output.get("AccessToken", {}).get("an-entry"), {"foo": "bar"},
+            "Undefined token keys and their values should be intact")
+