Complete bidstream tests

Author: asloobq

tests/test_bidstream_client.py

@@ -2,7 +2,7 @@
 from unittest.mock import patch
 
 from test_utils import *
-from uid2_client import BidstreamClient, EncryptionError, Uid2Base64UrlCoder, DecryptionStatus
+from uid2_client import BidstreamClient, Uid2Base64UrlCoder, DecryptionStatus, Uid2ClientFactory
 from uid2_client.refresh_response import RefreshResponse
 
 
@@ -15,21 +15,30 @@ class TestBidStreamClient(unittest.TestCase):
     _CONST_BASE_URL = 'base_url'
     _CONST_API_KEY = 'api_key'
 
-    def assert_success(self, decryption_response, token_version, scope):
+    def _assert_success(self, decryption_response, token_version, scope):
+        self.assertTrue(decryption_response.success)
+        self.assertEqual(decryption_response.uid, example_uid)
         self.assertEqual(decryption_response.advertising_token_version, token_version)
         if (token_version == AdvertisingTokenVersion.ADVERTISING_TOKEN_V3
                 or token_version == AdvertisingTokenVersion.ADVERTISING_TOKEN_V4):
             self.assertEqual(decryption_response.identity_type, IdentityType.Email)
         else:
             self.assertEqual(decryption_response.identity_type, None)
-        self.assertEqual(decryption_response.uid, example_uid)
         self.assertEqual(decryption_response.identity_scope, scope)
         self.assertEqual(decryption_response.is_client_side_generated, False)
-        self.assertEqual((now - decryption_response.established).total_seconds(), 0)
+
+    def _assert_fails(self, decryption_response, token_version, scope):
+        self.assertFalse(decryption_response.success)
+        self.assertEqual(decryption_response.status, DecryptionStatus.INVALID_TOKEN_LIFETIME)
+        self.assertEqual(decryption_response.advertising_token_version, token_version)
+        self.assertEqual(decryption_response.identity_scope, scope)
+        if (token_version == AdvertisingTokenVersion.ADVERTISING_TOKEN_V3
+                or token_version == AdvertisingTokenVersion.ADVERTISING_TOKEN_V4):
+            self.assertEqual(decryption_response.identity_type, IdentityType.Email)
 
     def decrypt_and_assert_success(self, token, token_version, scope):
         decrypted = self._client.decrypt_token_into_raw_uid(token, None)
-        self.assert_success(decrypted, token_version, scope)
+        self._assert_success(decrypted, token_version, scope)
 
     def setUp(self):
         self._key_collection = create_key_collection(IdentityScope.UID2)
@@ -47,9 +56,8 @@ def test_smoke_test(self, mock_refresh_bidstream_keys):  # SmokeTest
     def test_phone_uids(self, mock_refresh_bidstream_keys):  # PhoneTest
         for expected_scope, expected_version in test_cases_all_scopes_v3_v4_versions:
             with self.subTest(expected_scope=expected_scope, expected_version=expected_version):
-                mock_refresh_bidstream_keys.return_value = RefreshResponse.make_success(EncryptionKeysCollection([master_key, site_key],
-                                                                                    expected_scope, site_id, 1,
-                                                                                  99999, 86400))
+                mock_refresh_bidstream_keys.return_value = RefreshResponse.make_success(
+                    create_key_collection(expected_scope))
                 self._client.refresh()
                 token = generate_uid_token(expected_scope, expected_version, phone_uid)
                 self.assertEqual(IdentityType.Phone, get_identity_type(token))
@@ -72,19 +80,18 @@ def test_token_lifetime_too_long_for_bidstream(self, mock_refresh_bidstream_keys
                                                                                   max_bidstream_lifetime_seconds=max_bidstream_lifetime))
                 self._client.refresh()
                 result = self._client.decrypt_token_into_raw_uid(token, None)
-                self.assertFalse(result.success)
-                self.assertEqual(result.status, DecryptionStatus.INVALID_TOKEN_LIFETIME)
+                self._assert_fails(result, expected_version, expected_scope)
 
     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
         for expected_scope, expected_version in test_cases_all_scopes_all_versions:
             with self.subTest(expected_scope=expected_scope, expected_version=expected_version):
                 token = generate_uid_token(expected_scope, expected_version, created_at=created_at_future)
-                mock_refresh_bidstream_keys.return_value = RefreshResponse.make_success(EncryptionKeysCollection(
-                    [master_key, site_key], expected_scope, site_id, 1,99999, 86400))
+                mock_refresh_bidstream_keys.return_value = RefreshResponse.make_success(
+                    create_key_collection(expected_scope))
                 self._client.refresh()
                 result = self._client.decrypt_token_into_raw_uid(token, None)
-                self.assertFalse(result.success)
+                self._assert_fails(result, expected_version, expected_scope)
                 self.assertEqual(result.status, DecryptionStatus.INVALID_TOKEN_LIFETIME)
 
     def test_token_generated_in_the_future_within_allowed_clock_skew(self, mock_refresh_bidstream_keys):  # TokenGeneratedInTheFutureWithinAllowedClockSkew
@@ -93,13 +100,53 @@ def test_token_generated_in_the_future_within_allowed_clock_skew(self, mock_refr
             with self.subTest(expected_scope=expected_scope, expected_version=expected_version):
                 token = generate_uid_token(expected_scope, expected_version, expires_at=created_at_future)
                 mock_refresh_bidstream_keys.return_value = RefreshResponse.make_success(
-                    EncryptionKeysCollection([master_key, site_key], expected_scope, site_id, 1,
-                                                99999, 86400))
+                    create_key_collection(expected_scope))
                 self._client.refresh()
-                result = self._client.decrypt_token_into_raw_uid(token, None)
-                self.assertIsNotNone(result)
-                self.assertEqual(result.identity_scope, expected_scope)
-                self.assertEqual(result.advertising_token_version, expected_version)
+                self.decrypt_and_assert_success(token, expected_version, expected_scope)
+
+    @patch('uid2_client.client.refresh_sharing_keys')
+    def test_token_generated_in_the_future_legacy_client(self, mock_refresh_keys_util, mock_refresh_bidstream_keys):  # TokenGeneratedInTheFutureLegacyClient
+        created_at_future = dt.datetime.now(tz=timezone.utc) + dt.timedelta(minutes=3)  # max allowed clock skew is 30m
+        legacy_client = Uid2ClientFactory.create(self._CONST_BASE_URL, self._CONST_API_KEY, client_secret)
+
+        for expected_scope, expected_version in test_cases_all_scopes_all_versions:
+            with self.subTest(expected_scope=expected_scope, expected_version=expected_version):
+                mock_refresh_keys_util.return_value = RefreshResponse.make_success(create_key_collection(
+                    expected_scope))
+                legacy_client.refresh_keys()
+                token = generate_uid_token(expected_scope, expected_version, created_at=created_at_future)
+                result = legacy_client.decrypt(token)
+                self._assert_success(result, expected_version, expected_scope)
+
+    @patch('uid2_client.client.refresh_sharing_keys')
+    def test_token_lifetime_too_long_legacy_client(self, mock_refresh_keys_util,
+                                                         mock_refresh_bidstream_keys):  # TokenLifetimeTooLongLegacyClient
+        expires_in_sec = IN_3_DAYS + dt.timedelta(minutes=1)
+        legacy_client = Uid2ClientFactory.create(self._CONST_BASE_URL, self._CONST_API_KEY, client_secret)
+
+        for expected_scope, expected_version in test_cases_all_scopes_all_versions:
+            with self.subTest(expected_scope=expected_scope, expected_version=expected_version):
+                mock_refresh_keys_util.return_value = RefreshResponse.make_success(create_key_collection(
+                    expected_scope))
+                legacy_client.refresh_keys()
+                token = generate_uid_token(expected_scope, expected_version, expires_at=expires_in_sec)
+                result = legacy_client.decrypt(token)
+                self._assert_success(result, expected_version, expected_scope)  # check skipped for legacy clients
+
+    def test_identity_scope_and_types(self, mock_refresh_bidstream_keys):  # IdentityScopeAndType_TestCases
+        test_cases = [
+            [example_email_raw_uid2_v2, IdentityScope.UID2, IdentityType.Email],
+            [example_phone_raw_uid2_v2, IdentityScope.UID2, IdentityType.Phone],
+            [example_email_raw_uid2_v2, IdentityScope.EUID, IdentityType.Email],
+            [example_phone_raw_uid2_v2, IdentityScope.EUID, IdentityType.Phone]
+        ]
+        for uid, identity_scope, identity_type in test_cases:
+            with self.subTest(identity_scope=identity_scope, identity_type=identity_type):
+                token = generate_uid_token(identity_scope, AdvertisingTokenVersion.ADVERTISING_TOKEN_V4)
+                mock_refresh_bidstream_keys.return_value = RefreshResponse.make_success(
+                    create_key_collection(identity_scope))
+                self._client.refresh()
+                self.decrypt_and_assert_success(token, AdvertisingTokenVersion.ADVERTISING_TOKEN_V4, identity_scope)
 
     def test_empty_keys(self, mock_refresh_bidstream_keys):  # EmptyKeyContainer
         token = generate_uid_token(IdentityScope.UID2, AdvertisingTokenVersion.ADVERTISING_TOKEN_V3)

tests/test_utils.py

@@ -35,6 +35,8 @@
 client_secret = "ioG3wKxAokmp+rERx6A4kM/13qhyolUXIu14WN16Spo="
 client_secret_bytes = base64.b64decode(client_secret)
 example_uid = "ywsvDNINiZOVSsfkHpLpSJzXzhr6Jx9Z/4Q0+lsEUvM="
+example_email_raw_uid2_v2 = example_uid
+example_phone_raw_uid2_v2 = "BFOsW2SkK0egqbfyiALtpti5G/cG+PcEvjkoHl56rEV8"
 phone_uid = "BEOGxroPLdcY7LrSiwjY52+X05V0ryELpJmoWAyXiwbZ"
 
 test_cases_all_scopes_all_versions = [