Add unit tests

Author: asloobq

tests/test_bidstream_client.py

@@ -88,10 +88,13 @@ def _decrypt_and_assert_success(self, token, token_version, scope):
     def setUp(self):
         self._client = BidstreamClient(self._CONST_BASE_URL, self._CONST_API_KEY, client_secret)
 
-    def test_smoke_test(self):  # SmokeTest
+    def test_smoke_test_for_bidstream(self):  # SmokeTestForBidstream
         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)
+                token = generate_uid_token(expected_scope, expected_version,
+                                           identity_established_at=now - dt.timedelta(days=120),
+                                           generated_at=YESTERDAY,
+                                           expires_at=IN_2_DAYS)
                 refresh_response = self._client._refresh_json(key_bidstream_response_json_default_keys(
                     expected_scope))
                 self.assertTrue(refresh_response.success)
@@ -112,11 +115,29 @@ def test_phone_uids(self):  # PhoneTest
                 self.assertEqual(result.identity_scope, expected_scope)
                 self.assertEqual(result.advertising_token_version, expected_version)
 
-    def test_token_lifetime_too_long_for_bidstream(self):  # TokenLifetimeTooLongForBidstream
-        expires_in_sec = IN_3_DAYS + dt.timedelta(minutes=1)
+    def test_token_lifetime_too_long_for_bidstream_but_remaining_lifetime_allowed(self):  # TokenLifetimeTooLongForBidstreamButRemainingLifetimeAllowed
+        generated = YESTERDAY
+        expires_in_sec = generated + dt.timedelta(days=3) + dt.timedelta(minutes=1)
         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, expires_at=expires_in_sec)
+                token = generate_uid_token(expected_scope, expected_version, generated_at=generated,
+                                           expires_at=expires_in_sec)
+                refresh_response = self._client._refresh_json(key_bidstream_response_json_default_keys(
+                    expected_scope))
+                self.assertTrue(refresh_response.success)
+                result = self._client.decrypt_token_into_raw_uid(token, None)
+                if expected_version == AdvertisingTokenVersion.ADVERTISING_TOKEN_V2:
+                    self.assert_success(result, expected_version, expected_scope)
+                else:
+                    self.assert_fails(result, expected_version, expected_scope)
+
+    def test_token_remaining_lifetime_too_long_for_bidstream(self):  # TokenRemainingLifetimeTooLongForBidstream
+        generated = now
+        expires_in_sec = generated + dt.timedelta(days=3) + dt.timedelta(minutes=1)
+        for expected_scope, expected_version in test_cases_all_scopes_v3_v4_versions:
+            with self.subTest(expected_scope=expected_scope, expected_version=expected_version):
+                token = generate_uid_token(expected_scope, expected_version, generated_at=generated,
+                                           expires_at=expires_in_sec)
                 refresh_response = self._client._refresh_json(key_bidstream_response_json_default_keys(
                     expected_scope))
                 self.assertTrue(refresh_response.success)

tests/test_sharing_client.py

@@ -77,17 +77,35 @@ def decrypt_and_assert_success(self, token, token_version, scope):
         decrypted = self._client.decrypt_token_into_raw_uid(token)
         self._test_bidstream_client.assert_success(decrypted, token_version, scope)
 
-    def test_smoke_test(self):  # SmokeTest
+    def test_smoke_test_for_sharing(self):  # SmokeTestForSharing
         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)
+                token = generate_uid_token(expected_scope, expected_version,
+                                           identity_established_at=now - dt.timedelta(days=120),
+                                           generated_at=YESTERDAY, expires_at=now + dt.timedelta(days=29))
                 refresh_response = self._client._refresh_json(key_sharing_response_json_default_keys(
                     expected_scope))
                 self.assertTrue(refresh_response.success)
                 self.decrypt_and_assert_success(token, expected_version, expected_scope)
 
+    def test_token_lifetime_too_long_for_sharing_but_remaining_lifetime_allowed(self):  # TokenLifetimeTooLongForSharingButRemainingLifetimeAllowed
+        generated = YESTERDAY
+        expires_in_sec = generated + dt.timedelta(days=31)
+        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, generated_at=generated,
+                                           expires_at=expires_in_sec)
+                refresh_response = self._client._refresh_json(key_sharing_response_json_default_keys(
+                    expected_scope))
+                self.assertTrue(refresh_response.success)
+                result = self._client.decrypt_token_into_raw_uid(token)
+                if expected_version == AdvertisingTokenVersion.ADVERTISING_TOKEN_V2:
+                    self._test_bidstream_client.assert_success(result, expected_version, expected_scope)
+                else:
+                    self._test_bidstream_client.assert_fails(result, expected_version, expected_scope)
+
     def test_token_lifetime_too_long_for_sharing(self):  # TokenLifetimeTooLongForSharing
-        expires_in_sec = dt.datetime.now(tz=timezone.utc) + dt.timedelta(days=31)
+        expires_in_sec = now + dt.timedelta(days=30) + dt.timedelta(minutes=1)
         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, expires_at=expires_in_sec)
@@ -277,7 +295,8 @@ def test_expiry_in_token_matches_expiry_in_response(self):  # ExpiryInTokenMatch
         self.assertTrue(result.status)
         self.assertEqual(example_uid, result.uid)
 
-        future_decryption_result = self._client._decrypt_token_into_raw_uid(encryption_data_response.encrypted_data, now + dt.timedelta(seconds=3))
+        future_decryption_result = self._client._decrypt_token_into_raw_uid(encryption_data_response.encrypted_data,
+                                                                            now + dt.timedelta(seconds=3))
         self.assertFalse(future_decryption_result.success)
         self.assertEqual(DecryptionStatus.EXPIRED_TOKEN, future_decryption_result.status)
         self.assertEqual(now + dt.timedelta(seconds=2), future_decryption_result.expiry)