import sys import time import json import random import requests from unittest import TestCase from unittest.mock import Mock, call, patch from pandora.errors import InvalidAuthToken, PandoraException from tests.test_pandora.test_clientbuilder import TestSettingsDictBuilder import pandora.transport as t class SysCallError(Exception): pass class TestTransport(TestCase): def test_test_url_should_return_true_if_request_okay(self): transport = t.APITransport(Mock()) transport._http = Mock() transport._http.head.return_value = Mock( status_code=requests.codes.not_found) self.assertFalse(transport.test_url("foo")) transport._http.head.return_value = Mock(status_code=requests.codes.OK) self.assertTrue(transport.test_url("foo")) def test_call_should_retry_max_times_on_sys_call_error(self): with self.assertRaises(SysCallError): client = TestSettingsDictBuilder._build_minimal() time.sleep = Mock() client.transport._make_http_request = Mock( side_effect=SysCallError("error_mock")) client.transport._start_request = Mock() client("method") client.transport._start_request.assert_has_calls([call("method")]) assert client.transport._start_request.call_count == 3 def test_call_should_not_retry_for_pandora_exceptions(self): with self.assertRaises(PandoraException): client = TestSettingsDictBuilder._build_minimal() time.sleep = Mock() client.transport._make_http_request = Mock( side_effect=PandoraException("error_mock")) client.transport._start_request = Mock() client("method") client.transport._start_request.assert_has_calls([call("method")]) assert client.transport._start_request.call_count == 1 def test_call_should_retry_if_auth_token_expired(self): with self.assertRaises(InvalidAuthToken): client = TestSettingsDictBuilder._build_minimal() time.sleep = Mock() client.transport._make_http_request = Mock( side_effect=InvalidAuthToken("error_mock")) client.transport._start_request = Mock() client._authenticate = Mock() client("method") client.transport._start_request.assert_has_calls([call("method")]) assert client.transport._start_request.call_count == 2 assert client._authenticate.call_count == 1 def test_complete_request(self): transport = t.APITransport(Mock()) transport._http = Mock() http_result = Mock() http_result.content = b'{"stat":"ok","result":"bar"}' transport._http.post.return_value = http_result self.assertEqual( "bar", transport(t.APITransport.NO_ENCRYPT[0], foo="bar")) class TestTransportSetters(TestCase): def setUp(self): self.cryptor = Mock() self.transport = t.APITransport(self.cryptor) def test_set_partner(self): self.cryptor.decrypt_sync_time.return_value = 456 self.transport.set_partner({ "syncTime": "123", "partnerAuthToken": "partner_auth_token", "partnerId": "partner_id", }) self.cryptor.decrypt_sync_time.assert_called_with("123") self.assertEqual("partner_auth_token", self.transport.auth_token) self.assertEqual("partner_id", self.transport.partner_id) self.assertEqual( "partner_auth_token", self.transport.partner_auth_token) self.transport.start_time = 10 with patch.object(time, "time", return_value=30): self.assertEqual(476, self.transport.sync_time) def test_set_user(self): self.transport.set_user({ "userId": "user", "userAuthToken": "auth", }) self.assertEqual("user", self.transport.user_id) self.assertEqual("auth", self.transport.user_auth_token) self.assertEqual("auth", self.transport.auth_token) def test_getting_auth_token_no_login(self): self.assertIsNone(self.transport.auth_token) self.assertIsNone(self.transport.sync_time) class TestDelayExponential(TestCase): def test_fixed_delay(self): self.assertEqual(8, t.delay_exponential(2, 2, 3)) def test_random_delay(self): with patch.object(random, "random", return_value=10): self.assertEqual(20, t.delay_exponential("rand", 2, 2)) def test_fails_with_base_zero_or_below(self): with self.assertRaises(ValueError): t.delay_exponential(0, 1, 1) with self.assertRaises(ValueError): t.delay_exponential(-1, 1, 1) class TestRetries(TestCase): def test_no_retries_returns_none(self): @t.retries(0) def foo(): return True self.assertIsNone(foo()) class TestParseResponse(TestCase): VALID_MSG_NO_BODY_JSON = b'{"stat":"ok"}' VALID_MSG_JSON = b'{"stat":"ok", "result":{"foo":"bar"}}' ERROR_MSG_JSON = b'{"stat":"err", "code":1001, "message":"Details"}' def setUp(self): self.transport = t.APITransport(Mock()) def test_with_valid_response(self): res = self.transport._parse_response(self.VALID_MSG_JSON) self.assertEqual({"foo": "bar"}, res) def test_with_valid_response_no_body(self): res = self.transport._parse_response(self.VALID_MSG_NO_BODY_JSON) self.assertIsNone(res) def test_with_error_response(self): with self.assertRaises(InvalidAuthToken) as ex: self.transport._parse_response(self.ERROR_MSG_JSON) self.assertEqual(1001, ex.exception.code) self.assertEqual("Details", ex.exception.extended_message) class TestTransportRequestPrep(TestCase): def setUp(self): self.cryptor = Mock() self.transport = t.APITransport(self.cryptor) def test_start_request(self): self.transport.start_time = 10 self.transport._start_request("method_name") self.assertEqual(10, self.transport.start_time) def test_start_request_with_reset(self): self.transport.reset = Mock() self.transport._start_request(self.transport.REQUIRE_RESET[0]) self.transport.reset.assert_called_with() def test_start_request_without_time(self): with patch.object(time, "time", return_value=10.0): self.transport._start_request("method_name") self.assertEqual(10, self.transport.start_time) def test_make_http_request(self): # url, data, params http = Mock() retval = Mock() retval.content = "foo" http.post.return_value = retval self.transport._http = http res = self.transport._make_http_request( "/url", b"data", {"a": None, "b": "c"}) http.post.assert_called_with("/url", data=b"data", params={"b": "c"}) retval.raise_for_status.assert_called_with() self.assertEqual("foo", res) def test_build_data_not_logged_in(self): self.cryptor.encrypt = lambda x: x self.transport.partner_auth_token = "pat" self.transport.server_sync_time = 123 self.transport.start_time = 23 with patch.object(time, "time", return_value=20): val = self.transport._build_data("foo", {"a": "b", "c": None}) val = json.loads(val) self.assertEqual("b", val["a"]) self.assertEqual("pat", val["partnerAuthToken"]) self.assertEqual(120, val["syncTime"]) def test_build_data_no_encrypt(self): self.transport.user_auth_token = "uat" self.transport.partner_auth_token = "pat" self.transport.server_sync_time = 123 self.transport.start_time = 23 with patch.object(time, "time", return_value=20): val = self.transport._build_data( t.APITransport.NO_ENCRYPT[0], {"a": "b", "c": None}) val = json.loads(val) self.assertEqual("b", val["a"]) self.assertEqual("uat", val["userAuthToken"]) self.assertEqual(120, val["syncTime"]) # All Cryptor implementations must pass these test cases unmodified class CommonCryptorTestCases: def test_decrypt_invalid_padding(self): with self.assertRaises(ValueError): data = b"12345678\x00" self.assertEqual(b"12345678\x00", self.cryptor.decrypt(data)) def test_decrypt_strip_padding(self): data = b"123456\x02\x02" self.assertEqual(b"123456", self.cryptor.decrypt(data)) def test_decrypt_preserve_padding(self): data = b"123456\x02\x02" self.assertEqual(b"123456\x02\x02", self.cryptor.decrypt(data, False)) def test_encrypt(self): data = "123456" self.assertEqual(b"123456\x02\x02", self.cryptor.encrypt(data)) class TestPurePythonBlowfishCryptor(TestCase, CommonCryptorTestCases): def setUp(self): # Ugh... blowfish can't even be *imported* in python2 if not t.blowfish: t.blowfish = Mock() self.cipher = Mock() self.cipher.decrypt_ecb = lambda x: [x] self.cipher.encrypt_ecb = lambda x: [x] self.cryptor = t.PurePythonBlowfish("keys") self.cryptor.cipher = self.cipher class TestEncryptor(TestCase): ENCODED_JSON = "7b22666f6f223a22626172227d" UNENCODED_JSON = b'{"foo":"bar"}' EXPECTED_TIME = 4111 ENCODED_TIME = "31353037343131313539" class NoopCrypto: def __init__(self, key): pass def decrypt(self, data, strip_padding=True): return data.decode("ascii") def encrypt(self, data): return data def setUp(self): self.cryptor = t.Encryptor("in", "out", self.NoopCrypto) def test_decrypt(self): self.assertEqual( {"foo": "bar"}, self.cryptor.decrypt(self.ENCODED_JSON)) def test_encrypt(self): self.assertEqual( self.ENCODED_JSON.encode("ascii"), self.cryptor.encrypt(self.UNENCODED_JSON)) def test_decrypt_sync_time(self): self.assertEqual( self.EXPECTED_TIME, self.cryptor.decrypt_sync_time(self.ENCODED_TIME))