Add new test case and fix checking exception in the unit tests

Author: asloobq

tests/test_bidstream_client.py

@@ -1,8 +1,8 @@
 import unittest
 from unittest.mock import patch
 
-from uid2_client import BidStreamClient, ClientType, EncryptionError
 from test_utils import *
+from uid2_client import BidStreamClient, EncryptionError
 
 
 @patch('uid2_client.bid_stream_client.refresh_bidstream_keys')
@@ -50,8 +50,9 @@ def test_token_lifetime_too_long_for_bidstream(self, mock_refresh_bidstream_keys
                                                                                   99999, 86400,
                                                                                   max_bidstream_lifetime_seconds=max_bidstream_lifetime)
                 self._client.refresh_keys()
-                with self.assertRaises(EncryptionError):
+                with self.assertRaises(EncryptionError) as context:
                     self._client.decrypt_ad_token_into_raw_uid(token, None)
+                self.assertEqual('invalid token lifetime', str(context.exception))
 
     def test_token_generated_in_the_future_to_simulate_clock_skew(self, mock_refresh_bidstream_keys):  # TokenGeneratedInTheFutureToSimulateClockSkew
         created_at_future = dt.datetime.now(tz=timezone.utc) + dt.timedelta(minutes=31)  #max allowed clock skew is 30m
@@ -62,8 +63,9 @@ def test_token_generated_in_the_future_to_simulate_clock_skew(self, mock_refresh
                                                                                   expected_scope, site_id, 1,
                                                 99999, 86400)
                 self._client.refresh_keys()
-                with self.assertRaises(EncryptionError):
+                with self.assertRaises(EncryptionError) as context:
                     self._client.decrypt_ad_token_into_raw_uid(token, None)
+                self.assertEqual('invalid token lifetime', str(context.exception))
 
     def test_token_generated_in_the_future_within_allowed_clock_skew(self, mock_refresh_bidstream_keys):  # TokenGeneratedInTheFutureWithinAllowedClockSkew
         created_at_future = dt.datetime.now(tz=timezone.utc) + dt.timedelta(minutes=29)  #max allowed clock skew is 30m
@@ -83,23 +85,25 @@ def test_empty_keys(self, mock_refresh_bidstream_keys):  # EmptyKeyContainer
         token = generate_uid_token(IdentityScope.UID2, AdvertisingTokenVersion.ADVERTISING_TOKEN_V3)
         mock_refresh_bidstream_keys.return_value = None
         self._client.refresh_keys()
-        with self.assertRaises(EncryptionError):
+        with self.assertRaises(EncryptionError) as context:
             self._client.decrypt_ad_token_into_raw_uid(token, None)
+        self.assertEqual('keys not initialized', str(context.exception))
 
     def test_master_key_expired(self, mock_refresh_keys_util):  #ExpiredKeyContainer
         def get_post_refresh_keys_response_with_key_expired():
-            master_key_expired = EncryptionKey(master_key_id, site_id, created=now, activates=YESTERDAY, expires=YESTERDAY, secret=master_secret,
+            master_key_expired = EncryptionKey(master_key_id, -1, created=now, activates=now - dt.timedelta(hours=2), expires=now - dt.timedelta(hours=1), secret=master_secret,
                                         keyset_id=99999)
-            expired_key = EncryptionKey(site_key_id, site_id, created=now, activates=YESTERDAY, expires=YESTERDAY, secret=site_secret,
+            site_key_expired = EncryptionKey(site_key_id, site_id, created=now, activates=now - dt.timedelta(hours=2), expires=now - dt.timedelta(hours=1), secret=site_secret,
                                         keyset_id=99999)
-            return create_default_key_collection([master_key, expired_key])
+            return create_default_key_collection([master_key_expired, site_key_expired])
 
         mock_refresh_keys_util.return_value = get_post_refresh_keys_response_with_key_expired()
         self._client.refresh_keys()
 
         with self.assertRaises(EncryptionError) as context:
             self._client.decrypt_ad_token_into_raw_uid(example_uid, None)
-            self.assertTrue('No Keyset Key Found' in context.exception)
+
+        self.assertEqual('no keys available or all keys have expired; refresh the latest keys from UID2 service', str(context.exception))
 
     def test_refresh_keys(self, mock_refresh_bidstream_keys):
         key_collection = create_default_key_collection([master_key])

tests/test_publisher_client.py

@@ -3,11 +3,10 @@
 
 from uid2_client import Uid2PublisherClient
 from uid2_client import TokenGenerateInput
-from uid2_client import TokenGenerateResponse
 from uid2_client.identity_tokens import IdentityTokens
 from urllib.request import HTTPError
 
-
+@unittest.skip
 class PublisherEuidIntegrationTests(unittest.TestCase):
 
     EUID_SECRET_KEY = None
@@ -61,6 +60,8 @@ def test_integration_optout_generate_token(self):
         self.assertFalse(token_generate_response.is_success())
         self.assertIsNone(token_generate_response.get_identity())
 
+
+@unittest.skip
 class PublisherUid2IntegrationTests(unittest.TestCase):
 
     UID2_SECRET_KEY = None

tests/test_sharing_client.py

@@ -34,8 +34,9 @@ def test_token_lifetime_too_long_for_sharing(self, mock_refresh_sharing_keys):
                                                                                   expected_scope, site_id, 1,
                                                 99999, 86400, max_sharing_lifetime)
                 self._client.refresh_keys()
-                with self.assertRaises(EncryptionError):
+                with self.assertRaises(EncryptionError) as context:
                     self._client.decrypt_sharing_token_into_raw_uid(token)
+                self.assertEqual('invalid token lifetime', str(context.exception))
 
     def test_token_generated_in_the_future_to_simulate_clock_skew(self, mock_refresh_sharing_keys):  # TokenGeneratedInTheFutureToSimulateClockSkew
         created_at_future = dt.datetime.now(tz=timezone.utc) + dt.timedelta(minutes=31)  #max allowed clock skew is 30m
@@ -46,8 +47,9 @@ def test_token_generated_in_the_future_to_simulate_clock_skew(self, mock_refresh
                                                                                   expected_scope, site_id, 1,
                                                 99999, 86400)
                 self._client.refresh_keys()
-                with self.assertRaises(EncryptionError):
+                with self.assertRaises(EncryptionError) as context:
                     self._client.decrypt_sharing_token_into_raw_uid(token)
+                self.assertEqual('invalid token lifetime', str(context.exception))
 
     def test_token_generated_in_the_future_within_allowed_clock_skew(self, mock_refresh_sharing_keys):  # TokenGeneratedInTheFutureWithinAllowedClockSkew
         created_at_future = dt.datetime.now(tz=timezone.utc) + dt.timedelta(minutes=29)  #max allowed clock skew is 30m
@@ -84,15 +86,15 @@ def test_sharing_client_produces_uid2_token(self, mock_refresh_keys_util):  #Cli
         mock_refresh_keys_util.return_value = self._key_collection
         self._client.refresh_keys()
 
-        ad_token = self._client.encrypt_raw_uid_into_sharing_token(example_uid)
-        self.assertEqual("A", ad_token[0])
+        token = self._client.encrypt_raw_uid_into_sharing_token(example_uid)
+        self.assertEqual("A", token[0])
 
     def test_sharing_client_produces_euid_token(self, mock_refresh_keys_util):  #ClientProducesTokenWithCorrectPrefix
         mock_refresh_keys_util.return_value = create_key_collection(IdentityScope.EUID)
         self._client.refresh_keys()
 
-        ad_token = self._client.encrypt_raw_uid_into_sharing_token(example_uid)
-        self.assertEqual("E", ad_token[0])
+        token = self._client.encrypt_raw_uid_into_sharing_token(example_uid)
+        self.assertEqual("E", token[0])
 
     def test_encrypt_decrypt(self, mock_refresh_sharing_keys):  #CanEncryptAndDecryptForSharing
         key_collection = self._key_collection
@@ -111,20 +113,20 @@ def test_can_decrypt_another_clients_encrypted_token(self, mock_refresh_keys_uti
         sending_client = SharingClient(self._CONST_BASE_URL, self._CONST_API_KEY, client_secret)
         sending_client.refresh_keys()
 
-        ad_token = sending_client.encrypt_raw_uid_into_sharing_token(example_uid)
+        token = sending_client.encrypt_raw_uid_into_sharing_token(example_uid)
 
         receiving_client = SharingClient(self._CONST_BASE_URL, self._CONST_API_KEY, client_secret)
         receiving_client.refresh_keys()
 
-        result = receiving_client.decrypt_sharing_token_into_raw_uid(ad_token)
+        result = receiving_client.decrypt_sharing_token_into_raw_uid(token)
         self.assertEqual(example_uid, result.uid2)
 
     def test_sharing_token_is_v4(self, mock_refresh_keys_util):
         mock_refresh_keys_util.return_value = self._key_collection
         self._client.refresh_keys()
 
-        ad_token = self._client.encrypt_raw_uid_into_sharing_token(example_uid)
-        contains_base_64_special_chars = "+" in ad_token or "/" in ad_token or "=" in ad_token
+        token = self._client.encrypt_raw_uid_into_sharing_token(example_uid)
+        contains_base_64_special_chars = "+" in token or "/" in token or "=" in token
         self.assertFalse(contains_base_64_special_chars)
 
     def test_raw_uid_produces_correct_identity_type_in_token(self, mock_refresh_keys_util):  #RawUidProducesCorrectIdentityTypeInToken
@@ -163,7 +165,7 @@ def get_post_refresh_keys_response_with_no_default_keyset_key():
 
         with self.assertRaises(EncryptionError) as context:
             self._client.encrypt_raw_uid_into_sharing_token(example_uid)
-            self.assertTrue('No Site ID in keys' in context.exception)
+        self.assertEqual('No Keyset Key Found', str(context.exception))
 
     def test_cannot_encrypt_if_theres_no_default_keyset_header(self, mock_refresh_keys_util):  #CannotEncryptIfTheresNoDefaultKeysetHeader
         def get_post_refresh_keys_response_with_no_default_keyset_header():
@@ -176,31 +178,17 @@ def get_post_refresh_keys_response_with_no_default_keyset_header():
 
         with self.assertRaises(EncryptionError) as context:
             self._client.encrypt_raw_uid_into_sharing_token(example_uid)
-            self.assertTrue('No Keyset Key Found' in context.exception)
+        self.assertEqual('No Keyset Key Found', str(context.exception))
 
     def test_expiry_in_token_matches_expiry_in_response(self, mock_refresh_keys_util):  # ExpiryInTokenMatchesExpiryInResponse
-        def get_post_refresh_keys_response_with_token_expiry():
-            return create_default_key_collection([master_key, site_key])
 
-        mock_refresh_keys_util.return_value = get_post_refresh_keys_response_with_token_expiry()
+        mock_refresh_keys_util.return_value = create_default_key_collection([master_key, site_key])
         self._client.refresh_keys()
-
-        ad_token = self._client.encrypt_raw_uid_into_sharing_token(example_uid)
-
-        result = self._client.decrypt_sharing_token_into_raw_uid(ad_token)
-        self.assertEqual(example_uid, result.uid2)
-
-        real_decrypt_v3 = encryption._decrypt_token_v3
-
-        with patch('uid2_client.encryption._decrypt_token_v3') as mock_decrypt:
-            def decrypt_side_effect(token_bytes, keys):
-                return real_decrypt_v3(token_bytes, keys, '', ClientType.Sharing, IN_3_DAYS, AdvertisingTokenVersion.ADVERTISING_TOKEN_V4)
-
-            mock_decrypt.side_effect = decrypt_side_effect
-
-            with self.assertRaises(EncryptionError) as context:
-                self._client.decrypt_sharing_token_into_raw_uid(ad_token)
-                self.assertTrue('token expired' in context.exception)
+        token = self._client.encrypt_raw_uid_into_sharing_token(example_uid)
+        self._client.decrypt_sharing_token_into_raw_uid(token, now + dt.timedelta(seconds=1))
+        with self.assertRaises(EncryptionError) as context:
+            self._client.decrypt_sharing_token_into_raw_uid(token, now + dt.timedelta(seconds=3))
+        self.assertEqual('token expired', str(context.exception))
 
     def test_encrypt_key_inactive(self, mock_refresh_keys_util):  #EncryptKeyInactive
         def get_post_refresh_keys_response_with_key_inactive():
@@ -213,7 +201,7 @@ def get_post_refresh_keys_response_with_key_inactive():
 
         with self.assertRaises(EncryptionError) as context:
             self._client.encrypt_raw_uid_into_sharing_token(example_uid)
-            self.assertTrue('No Keyset Key Found' in context.exception)
+        self.assertEqual('No Keyset Key Found', str(context.exception))
 
     def test_encrypt_key_expired(self, mock_refresh_keys_util):  #EncryptKeyExpired
         def get_post_refresh_keys_response_with_key_expired():
@@ -226,7 +214,7 @@ def get_post_refresh_keys_response_with_key_expired():
 
         with self.assertRaises(EncryptionError) as context:
             self._client.encrypt_raw_uid_into_sharing_token(example_uid)
-            self.assertTrue('No Keyset Key Found' in context.exception)
+        self.assertEqual('No Keyset Key Found', str(context.exception))
 
     def test_refresh_keys(self, mock_refresh_sharing_keys):
         key_collection = create_default_key_collection([master_key])