Skip to content

Commit 846d65a

Browse files
committed
refactor tests
1 parent b416319 commit 846d65a

File tree

4 files changed

+85
-97
lines changed

4 files changed

+85
-97
lines changed

examples/sample_bidstream_client.py

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,6 @@ def _usage():
3232
print('UID =', decrypt_result.uid)
3333
print('Established =', decrypt_result.established)
3434
print('Site ID =', decrypt_result.site_id)
35-
print('Identity Scope =', decrypt_result.identity_scope)
3635
print('Identity Type =', decrypt_result.identity_type)
3736
print('Advertising Token Version =', decrypt_result.advertising_token_version)
3837
print('Is Client Side Generated =', decrypt_result.is_client_side_generated)

examples/sample_encryption.py

Lines changed: 13 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -12,14 +12,19 @@ def _usage():
1212
sys.exit(1)
1313

1414

15-
if len(sys.argv) <= 5:
16-
_usage()
17-
18-
base_url = sys.argv[1]
19-
auth_key = sys.argv[2]
20-
secret_key = sys.argv[3]
21-
ad_token = sys.argv[4]
22-
str_data = sys.argv[5]
15+
# if len(sys.argv) <= 5:
16+
# _usage()
17+
18+
# base_url = sys.argv[1]
19+
# auth_key = sys.argv[2]
20+
# secret_key = sys.argv[3]
21+
# ad_token = sys.argv[4]
22+
# str_data = sys.argv[5]
23+
base_url = "https://operator-integ.uidapi.com"
24+
auth_key = "UID2-C-I-110-XlrU/d.9MyGWqXP2LbwYijahYYWIkOmtyeauS1Jxjku0="
25+
secret_key = "r5Uw+piQboL3gUsNNtQwlhft3c6zXm35LoVj9GbL6dk="
26+
ad_token = "A4AAAA0l9Yo1fmExhC4dysj34QS0y9G4yAK5j502INSBfM9HED9tfEmEi-yMggOmXcOd8zo99gopfjXXZooYaCQwlvB2StTE1Z0xcFN4wmY7EbKWmg645RrzdJGf5vmGEY4jEIm0HNOT5qU3RIp-uyGRzJ7-dZqgYi5-ThilVMe4WnLjtJnK-xraHnWlD4MkMiN8F5FHbeHBGCaZFzsASzslrw"
27+
str_data = "abcd123"
2328

2429
client = Uid2Client(base_url, auth_key, secret_key)
2530
keys = client.refresh_keys()

tests/test_bidstream_client.py

Lines changed: 29 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@ def encode_keys(keys):
2626
return key_json
2727

2828

29-
def create_bidstream_response_json(keys, identity_scope):
29+
def key_bidstream_response_json(keys, identity_scope=IdentityScope.UID2):
3030
encoded_keys = encode_keys(keys)
3131
json_obj = {
3232
"body": {
@@ -52,15 +52,15 @@ def create_bidstream_response_json(keys, identity_scope):
5252
return json.dumps(json_obj)
5353

5454

55-
def create_bidstream_response_json_default_keys(identity_scope):
56-
return create_bidstream_response_json([master_key, site_key], identity_scope)
55+
def key_bidstream_response_json_default_keys(identity_scope=IdentityScope.UID2):
56+
return key_bidstream_response_json([master_key, site_key], identity_scope)
5757

5858

5959
class TestBidStreamClient(unittest.TestCase):
6060
_CONST_BASE_URL = 'base_url'
6161
_CONST_API_KEY = 'api_key'
6262

63-
def _assert_success(self, decryption_response, token_version, scope):
63+
def assert_success(self, decryption_response, token_version, scope):
6464
self.assertTrue(decryption_response.success)
6565
self.assertEqual(decryption_response.uid, example_uid)
6666
self.assertEqual(decryption_response.advertising_token_version, token_version)
@@ -72,7 +72,7 @@ def _assert_success(self, decryption_response, token_version, scope):
7272
self.assertEqual(decryption_response.identity_scope, scope)
7373
self.assertEqual(decryption_response.is_client_side_generated, False)
7474

75-
def _assert_fails(self, decryption_response, token_version, scope):
75+
def assert_fails(self, decryption_response, token_version, scope):
7676
self.assertFalse(decryption_response.success)
7777
self.assertEqual(decryption_response.status, DecryptionStatus.INVALID_TOKEN_LIFETIME)
7878
self.assertEqual(decryption_response.advertising_token_version, token_version)
@@ -81,9 +81,9 @@ def _assert_fails(self, decryption_response, token_version, scope):
8181
or token_version == AdvertisingTokenVersion.ADVERTISING_TOKEN_V4):
8282
self.assertEqual(decryption_response.identity_type, IdentityType.Email)
8383

84-
def decrypt_and_assert_success(self, token, token_version, scope):
84+
def _decrypt_and_assert_success(self, token, token_version, scope):
8585
decrypted = self._client.decrypt_token_into_raw_uid(token, None)
86-
self._assert_success(decrypted, token_version, scope)
86+
self.assert_success(decrypted, token_version, scope)
8787

8888
def setUp(self):
8989
self._client = BidstreamClient(self._CONST_BASE_URL, self._CONST_API_KEY, client_secret)
@@ -92,15 +92,15 @@ def test_smoke_test(self): # SmokeTest
9292
for expected_scope, expected_version in test_cases_all_scopes_all_versions:
9393
with self.subTest(expected_scope=expected_scope, expected_version=expected_version):
9494
token = generate_uid_token(expected_scope, expected_version)
95-
refresh_response = self._client._refresh_json(create_bidstream_response_json_default_keys(
95+
refresh_response = self._client._refresh_json(key_bidstream_response_json_default_keys(
9696
expected_scope))
9797
self.assertTrue(refresh_response.success)
98-
self.decrypt_and_assert_success(token, expected_version, expected_scope)
98+
self._decrypt_and_assert_success(token, expected_version, expected_scope)
9999

100100
def test_phone_uids(self): # PhoneTest
101101
for expected_scope, expected_version in test_cases_all_scopes_v3_v4_versions:
102102
with self.subTest(expected_scope=expected_scope, expected_version=expected_version):
103-
refresh_response = self._client._refresh_json(create_bidstream_response_json_default_keys(
103+
refresh_response = self._client._refresh_json(key_bidstream_response_json_default_keys(
104104
expected_scope))
105105
self.assertTrue(refresh_response.success)
106106
token = generate_uid_token(expected_scope, expected_version, phone_uid)
@@ -117,33 +117,33 @@ def test_token_lifetime_too_long_for_bidstream(self): # TokenLifetimeTooLongFor
117117
for expected_scope, expected_version in test_cases_all_scopes_all_versions:
118118
with self.subTest(expected_scope=expected_scope, expected_version=expected_version):
119119
token = generate_uid_token(expected_scope, expected_version, expires_at=expires_in_sec)
120-
refresh_response = self._client._refresh_json(create_bidstream_response_json_default_keys(
120+
refresh_response = self._client._refresh_json(key_bidstream_response_json_default_keys(
121121
expected_scope))
122122
self.assertTrue(refresh_response.success)
123123
result = self._client.decrypt_token_into_raw_uid(token, None)
124-
self._assert_fails(result, expected_version, expected_scope)
124+
self.assert_fails(result, expected_version, expected_scope)
125125

126126
def test_token_generated_in_the_future_to_simulate_clock_skew(self): # TokenGeneratedInTheFutureToSimulateClockSkew
127127
created_at_future = dt.datetime.now(tz=timezone.utc) + dt.timedelta(minutes=31) #max allowed clock skew is 30m
128128
for expected_scope, expected_version in test_cases_all_scopes_all_versions:
129129
with self.subTest(expected_scope=expected_scope, expected_version=expected_version):
130130
token = generate_uid_token(expected_scope, expected_version, created_at=created_at_future)
131-
refresh_response = self._client._refresh_json(create_bidstream_response_json_default_keys(
131+
refresh_response = self._client._refresh_json(key_bidstream_response_json_default_keys(
132132
expected_scope))
133133
self.assertTrue(refresh_response.success)
134134
result = self._client.decrypt_token_into_raw_uid(token, None)
135-
self._assert_fails(result, expected_version, expected_scope)
135+
self.assert_fails(result, expected_version, expected_scope)
136136
self.assertEqual(result.status, DecryptionStatus.INVALID_TOKEN_LIFETIME)
137137

138138
def test_token_generated_in_the_future_within_allowed_clock_skew(self): # TokenGeneratedInTheFutureWithinAllowedClockSkew
139139
created_at_future = dt.datetime.now(tz=timezone.utc) + dt.timedelta(minutes=29) #max allowed clock skew is 30m
140140
for expected_scope, expected_version in test_cases_all_scopes_all_versions:
141141
with self.subTest(expected_scope=expected_scope, expected_version=expected_version):
142142
token = generate_uid_token(expected_scope, expected_version, expires_at=created_at_future)
143-
refresh_response = self._client._refresh_json(create_bidstream_response_json_default_keys(
143+
refresh_response = self._client._refresh_json(key_bidstream_response_json_default_keys(
144144
expected_scope))
145145
self.assertTrue(refresh_response.success)
146-
self.decrypt_and_assert_success(token, expected_version, expected_scope)
146+
self._decrypt_and_assert_success(token, expected_version, expected_scope)
147147

148148
def test_legacy_response_from_old_operator(self):
149149
test_cases = [AdvertisingTokenVersion.ADVERTISING_TOKEN_V2,
@@ -154,31 +154,31 @@ def test_legacy_response_from_old_operator(self):
154154
for token_version in test_cases:
155155
with self.subTest(token_version=token_version):
156156
token = generate_uid_token(IdentityScope.UID2, token_version)
157-
self.decrypt_and_assert_success(token, token_version, IdentityScope.UID2)
157+
self._decrypt_and_assert_success(token, token_version, IdentityScope.UID2)
158158

159159
def test_token_generated_in_the_future_legacy_client(self): # TokenGeneratedInTheFutureLegacyClient
160160
created_at_future = dt.datetime.now(tz=timezone.utc) + dt.timedelta(minutes=3) # max allowed clock skew is 30m
161161
legacy_client = Uid2ClientFactory.create(self._CONST_BASE_URL, self._CONST_API_KEY, client_secret)
162162

163163
for expected_scope, expected_version in test_cases_all_scopes_all_versions:
164164
with self.subTest(expected_scope=expected_scope, expected_version=expected_version):
165-
legacy_client.refresh_json(create_bidstream_response_json_default_keys(
165+
legacy_client.refresh_json(key_bidstream_response_json_default_keys(
166166
expected_scope))
167167
token = generate_uid_token(expected_scope, expected_version, created_at=created_at_future)
168168
result = legacy_client.decrypt(token)
169-
self._assert_success(result, expected_version, expected_scope)
169+
self.assert_success(result, expected_version, expected_scope)
170170

171171
def test_token_lifetime_too_long_legacy_client(self): # TokenLifetimeTooLongLegacyClient
172172
expires_in_sec = IN_3_DAYS + dt.timedelta(minutes=1)
173173
legacy_client = Uid2ClientFactory.create(self._CONST_BASE_URL, self._CONST_API_KEY, client_secret)
174174

175175
for expected_scope, expected_version in test_cases_all_scopes_all_versions:
176176
with self.subTest(expected_scope=expected_scope, expected_version=expected_version):
177-
legacy_client.refresh_json(create_bidstream_response_json_default_keys(
177+
legacy_client.refresh_json(key_bidstream_response_json_default_keys(
178178
expected_scope))
179179
token = generate_uid_token(expected_scope, expected_version, expires_at=expires_in_sec)
180180
result = legacy_client.decrypt(token)
181-
self._assert_success(result, expected_version, expected_scope) # check skipped for legacy clients
181+
self.assert_success(result, expected_version, expected_scope) # check skipped for legacy clients
182182

183183
def test_identity_scope_and_types(self): # IdentityScopeAndType_TestCases
184184
test_cases = [
@@ -190,10 +190,10 @@ def test_identity_scope_and_types(self): # IdentityScopeAndType_TestCases
190190
for uid, identity_scope, identity_type in test_cases:
191191
with self.subTest(identity_scope=identity_scope, identity_type=identity_type):
192192
token = generate_uid_token(identity_scope, AdvertisingTokenVersion.ADVERTISING_TOKEN_V4)
193-
refresh_response = self._client._refresh_json(create_bidstream_response_json_default_keys(
193+
refresh_response = self._client._refresh_json(key_bidstream_response_json_default_keys(
194194
identity_scope))
195195
self.assertTrue(refresh_response.success)
196-
self.decrypt_and_assert_success(token, AdvertisingTokenVersion.ADVERTISING_TOKEN_V4, identity_scope)
196+
self._decrypt_and_assert_success(token, AdvertisingTokenVersion.ADVERTISING_TOKEN_V4, identity_scope)
197197

198198
def test_empty_keys(self): # EmptyKeyContainer
199199
token = generate_uid_token(IdentityScope.UID2, AdvertisingTokenVersion.ADVERTISING_TOKEN_V3)
@@ -207,8 +207,8 @@ def test_master_key_expired(self): #ExpiredKeyContainer
207207
site_key_expired = EncryptionKey(site_key_id, site_id, created=now, activates=now - dt.timedelta(hours=2),
208208
expires=now - dt.timedelta(hours=1), secret=site_secret, keyset_id=99999)
209209

210-
refresh_response = self._client._refresh_json(create_bidstream_response_json(
211-
[master_key_expired, site_key_expired], IdentityScope.UID2))
210+
refresh_response = self._client._refresh_json(key_bidstream_response_json(
211+
[master_key_expired, site_key_expired]))
212212
self.assertTrue(refresh_response.success)
213213

214214
result = self._client.decrypt_token_into_raw_uid(example_uid, None)
@@ -219,8 +219,8 @@ def test_not_authorized_for_master_key(self): #NotAuthorizedForMasterKey
219219

220220
another_master_key = EncryptionKey(master_key_id + site_key_id + 1, -1, created=now, activates=now, expires=now + dt.timedelta(hours=1), secret=master_secret)
221221
another_site_key = EncryptionKey(master_key_id + site_key_id + 2, site_id, created=now, activates=now, expires=now + dt.timedelta(hours=1), secret=site_secret)
222-
refresh_response = self._client._refresh_json(create_bidstream_response_json(
223-
[another_master_key, another_site_key], IdentityScope.UID2))
222+
refresh_response = self._client._refresh_json(key_bidstream_response_json(
223+
[another_master_key, another_site_key]))
224224
self.assertTrue(refresh_response.success)
225225
token = generate_uid_token(IdentityScope.UID2, AdvertisingTokenVersion.ADVERTISING_TOKEN_V4)
226226

@@ -229,8 +229,7 @@ def test_not_authorized_for_master_key(self): #NotAuthorizedForMasterKey
229229
self.assertEqual(result.status, DecryptionStatus.NOT_AUTHORIZED_FOR_MASTER_KEY)
230230

231231
def test_invalid_payload(self): #InvalidPayload
232-
refresh_response = self._client._refresh_json(create_bidstream_response_json_default_keys(
233-
IdentityScope.UID2))
232+
refresh_response = self._client._refresh_json(key_bidstream_response_json_default_keys())
234233
self.assertTrue(refresh_response.success)
235234
token = generate_uid_token(IdentityScope.UID2, AdvertisingTokenVersion.ADVERTISING_TOKEN_V4)
236235
payload = Uid2Base64UrlCoder.decode(token)
@@ -241,8 +240,7 @@ def test_invalid_payload(self): #InvalidPayload
241240
self.assertEqual(result.status, DecryptionStatus.INVALID_PAYLOAD)
242241

243242
def test_token_expiry_custom_decryption_time(self): #TokenExpiryAndCustomNow
244-
refresh_response = self._client._refresh_json(create_bidstream_response_json_default_keys(
245-
IdentityScope.UID2))
243+
refresh_response = self._client._refresh_json(key_bidstream_response_json_default_keys())
246244
self.assertTrue(refresh_response.success)
247245

248246
expires_at = now - dt.timedelta(days=60)

0 commit comments

Comments
 (0)