import datetime import functools from unittest import TestCase from requests_mock import Mocker import postgrestutils TOKEN = "JWT_token" default_session = functools.partial( postgrestutils.Session, base_uri="http://example.com/", token=TOKEN ) DEFAULT_HEADERS = { "Authorization": "Bearer {}".format(TOKEN), "Accept": "application/json", } SUPERHERO_TEST_DATA = [ { "id": 68, "name": "Batman", "gender": "Male", "eye_color": "blue", "race": "Human", "hair_color": "black", "height": 188, "publisher": "DC Comics", "skin_color": None, "alignment": "good", "weight": 95, }, { "id": 212, "name": "Deadpool", "gender": "Male", "eye_color": "brown", "race": "Mutant", "hair_color": "No Hair", "height": 188, "publisher": "Marvel Comics", "skin_color": None, "alignment": "neutral", "weight": 95, }, { "id": 345, "name": "Iron Man", "gender": "Male", "eye_color": "blue", "race": "Human", "hair_color": "Black", "height": 198, "publisher": "Marvel Comics", "skin_color": None, "alignment": "good", "weight": 191, }, { "id": 369, "name": "Joker", "gender": "Male", "eye_color": "green", "race": "Human", "hair_color": "Green", "height": 196, "publisher": "DC Comics", "skin_color": "white", "alignment": "bad", "weight": 86, }, { "id": 423, "name": "Magneto", "gender": "Male", "eye_color": "grey", "race": "Mutant", "hair_color": "White", "height": 188, "publisher": "Marvel Comics", "skin_color": None, "alignment": "bad", "weight": 86, }, ] @Mocker() class TestPgrestClientGet(TestCase): def setUp(self): super().setUp() self.data = SUPERHERO_TEST_DATA[0] def test_single_object_returned(self, mock): mock.register_uri( "GET", "http://example.com/superhero?id=eq.1000000000", request_headers={ **DEFAULT_HEADERS, **{"Accept": "application/vnd.pgrst.object+json"}, }, status_code=200, reason="OK", json=self.data, ) with default_session() as s: params = {"id": "eq.1000000000"} res = s.get("superhero", params=params) self.assertEqual(res, self.data) self.assertTrue(mock.called_once) def test_object_does_not_exist(self, mock): mock.register_uri( "GET", "http://example.com/superhero?id=eq.1337", request_headers={ **DEFAULT_HEADERS, **{"Accept": "application/vnd.pgrst.object+json"}, }, status_code=406, reason="Not Acceptable", text="""{"details":"Results contain 0 rows, application/vnd.pgrst.object+json requires 1 row","message":""" """"JSON object requested, multiple (or no) rows returned"}""", ) with default_session() as s, self.assertRaises( postgrestutils.ObjectDoesNotExist ): params = {"id": "eq.1337"} s.get("superhero", params=params) self.assertTrue(mock.called_once) def test_multiple_objects_returned(self, mock): mock.register_uri( "GET", "http://example.com/superhero", request_headers={ **DEFAULT_HEADERS, **{"Accept": "application/vnd.pgrst.object+json"}, }, status_code=406, reason="Not Acceptable", text="""{"details":"Results contain 5 rows, application/vnd.pgrst.object+json requires 1 row","message":""" """"JSON object requested, multiple (or no) rows returned"}""", ) with default_session() as s, self.assertRaises( postgrestutils.MultipleObjectsReturned ): s.get("superhero") self.assertTrue(mock.called_once) def test_datetime_parser(self, mock): expected = { "id": 1337, "random": datetime.datetime( 2020, 5, 20, 8, 35, 6, 659425, tzinfo=datetime.timezone.utc ), } mock.register_uri( "GET", "http://example.com/random_datetime", request_headers={ **DEFAULT_HEADERS, **{"Accept": "application/vnd.pgrst.object+json"}, }, status_code=200, reason="OK", json={"id": 1337, "random": "2020-05-20T08:35:06.659425+00:00"}, ) with default_session() as s: params = {"id": "eq.1337"} res = s.get("random_datetime", params=params) self.assertEqual(res, expected) self.assertTrue(mock.called_once) def test_without_datetime_parser(self, mock): test_json = {"id": 1337, "random": "2020-05-20T08:35:06.659425+00:00"} mock.register_uri( "GET", "http://example.com/random_datetime", request_headers={ **DEFAULT_HEADERS, **{"Accept": "application/vnd.pgrst.object+json"}, }, status_code=200, reason="OK", json=test_json, ) with default_session() as s: params = {"select": "id,random", "id": "eq.1337"} res = s.get("random_datetime", params=params, parse_dt=False) self.assertEqual(res, test_json) self.assertTrue(mock.called_once) @Mocker() class TestPgrestClientFilterStrategyNone(TestCase): def setUp(self): super().setUp() self.data = SUPERHERO_TEST_DATA def test_fetch_all_first(self, mock): mock.register_uri( "GET", "http://example.com/superhero", request_headers=DEFAULT_HEADERS, status_code=200, reason="OK", json=self.data, ) with default_session() as s: res = s.filter("superhero") self.assertIsInstance( res, postgrestutils.JsonResultSet ) # should return lazy object self.assertFalse(mock.called) # no request should have been made yet self.assertEqual(list(res), self.data) # fetch data self.assertTrue(mock.called_once) # should have been called once # fetched data should be cached self.assertEqual(res._result_cache, self.data) # len of fetched data should be cached self.assertEqual(res._len_cache, len(self.data)) self.assertEqual(list(res), self.data) # should utilize cache self.assertEqual(res[:1], self.data[:1]) # should utilize cache self.assertEqual(res[:0], self.data[:0]) # should return empty list self.assertEqual(res[4:2], self.data[4:2]) # should return empty list self.assertEqual(res[2:], self.data[2:]) # should utilize cache self.assertEqual(res[0], self.data[0]) # should utilize cache self.assertTrue(mock.called_once) # should not have been called again def test_fetch_len_first(self, mock): mock.register_uri( "GET", "http://example.com/superhero", request_headers=DEFAULT_HEADERS, status_code=200, reason="OK", json=self.data, ) with default_session() as s: res = s.filter("superhero") # should return lazy object self.assertIsInstance(res, postgrestutils.JsonResultSet) self.assertFalse(mock.called) # no request should have been made yet self.assertEqual(len(res), len(self.data)) # should fetch len self.assertTrue(mock.called_once) # should have been called once # len of fetched data should be cached self.assertEqual(res._len_cache, len(self.data)) # results should be cached (counting strategy none) self.assertEqual(res._result_cache, self.data) self.assertEqual(res[:1], self.data[:1]) # should utilize cache self.assertEqual(res[:0], self.data[:0]) # should return empty list self.assertEqual(res[4:2], self.data[4:2]) # should return empty list self.assertEqual(res[2:], self.data[2:]) # should utilize cache self.assertEqual(res[0], self.data[0]) # should utilize cache self.assertEqual(list(res), self.data) # should utilize cache self.assertTrue(mock.called_once) # should not have been called again def test_cache_fetching_unbounded_slice(self, mock): mock.register_uri( "GET", "http://example.com/superhero", request_headers=DEFAULT_HEADERS, status_code=200, reason="OK", json=self.data, ) with default_session() as s: res = s.filter("superhero") # should return lazy object self.assertIsInstance(res, postgrestutils.JsonResultSet) self.assertFalse(mock.called) # no request should have been made yet self.assertEqual(res[:], self.data) # fetch data self.assertTrue(mock.called_once) # should have been called once # fetched data should be cached self.assertEqual(res._result_cache, self.data) # len of fetched data should be cached self.assertEqual(res._len_cache, len(self.data)) self.assertEqual(res[:], self.data) # should utilize cache self.assertEqual(res[:0], self.data[:0]) # should return empty list self.assertEqual(res[4:2], self.data[4:2]) # should return empty list self.assertEqual(res[2:], self.data[2:]) # should utilize cache self.assertEqual(res[0], self.data[0]) # should utilize cache self.assertTrue(mock.called_once) # should not have been called again @Mocker() class TestPgrestClientFilterCountingStrategies(TestCase): def setUp(self): super().setUp() self.data = SUPERHERO_TEST_DATA self.counting_strategies = ( postgrestutils.Count.EXACT, postgrestutils.Count.PLANNED, postgrestutils.Count.ESTIMATED, ) def test_fetch_all_first(self, mock): # in order to fetch all mock.register_uri( "GET", "http://example.com/superhero", request_headers=DEFAULT_HEADERS, status_code=200, reason="OK", json=self.data, ) for strategy in self.counting_strategies: mock.reset() with default_session(count=strategy) as s: res = s.filter("superhero") # should return lazy object self.assertIsInstance(res, postgrestutils.JsonResultSet) self.assertFalse(mock.called) # no request should have been made yet self.assertEqual(list(res), self.data) # fetch data self.assertTrue(mock.called_once) # should have been called once # fetched data should be cached self.assertEqual(res._result_cache, self.data) # len of fetched data should be cached self.assertEqual(res._len_cache, len(self.data)) self.assertEqual(list(res), self.data) # should utilize cache self.assertEqual(res[:1], self.data[:1]) # should utilize cache self.assertEqual(res[:0], self.data[:0]) # should return empty list self.assertEqual(res[4:2], self.data[4:2]) # should return empty list self.assertEqual(res[2:], self.data[2:]) # should utilize cache self.assertEqual(res[0], self.data[0]) # should utilize cache self.assertTrue(mock.called_once) # should not have been called again def test_fetch_len_first(self, mock): # in order to fetch all mock.register_uri( "GET", "http://example.com/superhero", request_headers=DEFAULT_HEADERS, status_code=200, reason="OK", json=self.data, ) # in order to fetch first mock.register_uri( "GET", "http://example.com/superhero", request_headers={ **DEFAULT_HEADERS, **{"Range-Unit": "items", "Range": "0-0"}, }, status_code=200, reason="OK", headers={"Content-Range": "0-0/*"}, json=[self.data[0]], ) # in order to fetch range since index 2 mock.register_uri( "GET", "http://example.com/superhero", request_headers={ **DEFAULT_HEADERS, **{"Range-Unit": "items", "Range": "2-"}, }, status_code=200, reason="OK", headers={"Content-Range": "2-4/*"}, json=self.data[2:], ) # in order to fetch length using different strategies for strategy in self.counting_strategies: mock.register_uri( "GET", "http://example.com/superhero", request_headers={ **DEFAULT_HEADERS, **{ "Range-Unit": "items", "Range": "0-0", "Prefer": "count={}".format(strategy.value), }, }, status_code=206, reason="Partial Content", headers={"Content-Range": "0-0/5"}, json=self.data[0], ) mock.reset() with default_session(count=strategy) as s: res = s.filter("superhero") # should return lazy object self.assertIsInstance(res, postgrestutils.JsonResultSet) self.assertFalse(mock.called) # no request should have been made yet self.assertEqual(len(res), len(self.data)) # should fetch len self.assertTrue(mock.called_once) # should have been called once # len of fetched data should be cached self.assertEqual(res._len_cache, len(self.data)) # should fetch first element as range self.assertEqual(res[:1], self.data[:1]) self.assertEqual(res[:0], self.data[:0]) # should return empty list self.assertEqual(res[4:2], self.data[4:2]) # should return empty list # should fetch range starting at index 2 self.assertEqual(res[2:], self.data[2:]) # should fetch first element as range but return dict self.assertEqual(res[0], self.data[0]) self.assertEqual(list(res), self.data) # should fetch all elements # should fetch all elements self.assertEqual(res._result_cache, self.data) # should cache all elements self.assertEqual(res._result_cache, self.data) self.assertTrue(mock.called) # should have been called at least once # should have been called 5 times (fetch len, fetch 2 ranges, # fetch first and fetch all) self.assertEqual(mock.call_count, 5) @Mocker() class TestPgrestClientSessionDefaults(TestCase): def setUp(self): super().setUp() self.data = SUPERHERO_TEST_DATA def test_override_parse_dt_session_option(self, mock): test_json = {"id": 1337, "random": "2020-05-20T08:35:06.659425+00:00"} mock.register_uri( "GET", "http://example.com/random_datetime", request_headers={ **DEFAULT_HEADERS, **{"Accept": "application/vnd.pgrst.object+json"}, }, status_code=200, reason="OK", json=test_json, ) with default_session(parse_dt=False) as s: params = {"select": "id,random", "id": "eq.1337"} res = s.get("random_datetime", params=params) self.assertEqual(res, test_json) self.assertTrue(mock.called_once) mock.reset() res2 = s.get("random_datetime", params=params, parse_dt=True) expected = { "id": 1337, "random": datetime.datetime( 2020, 5, 20, 8, 35, 6, 659425, tzinfo=datetime.timezone.utc ), } self.assertEqual(res2, expected) self.assertTrue(mock.called_once) def test_override_count_session_option(self, mock): # in order to fetch all mock.register_uri( "GET", "http://example.com/superhero", request_headers=DEFAULT_HEADERS, status_code=200, reason="OK", json=self.data, ) # in order to fetch length mock.register_uri( "GET", "http://example.com/superhero", request_headers={ **DEFAULT_HEADERS, **{"Range-Unit": "items", "Range": "0-0", "Prefer": "count=exact"}, }, status_code=206, reason="Partial Content", headers={"Content-Range": "0-0/5"}, json=self.data[0], ) with default_session(count=postgrestutils.Count.EXACT) as s: res = s.filter("superhero") # should return lazy object self.assertIsInstance(res, postgrestutils.JsonResultSet) self.assertFalse(mock.called) # no request should have been made yet self.assertEqual(len(res), len(self.data)) # should fetch len self.assertTrue(mock.called_once) # should have been called once # len of fetched data should be cached self.assertEqual(res._len_cache, len(self.data)) # should not have cached all elements self.assertNotEqual(res._result_cache, self.data) mock.reset() # reset mock # override the count session option in this specific request res2 = s.filter("superhero", count=postgrestutils.Count.NONE) # should return lazy object self.assertIsInstance(res2, postgrestutils.JsonResultSet) self.assertFalse(mock.called) # no request should have been made yet # should fetch all elements to get len self.assertEqual(len(res2), len(self.data)) self.assertTrue(mock.called_once) # should have been called once # len of fetched data should be cached self.assertEqual(res2._len_cache, len(self.data)) # should have cached all elements self.assertEqual(res2._result_cache, self.data) def test_override_schema_session_option(self, mock): # in order to fetch all mock.register_uri( "GET", "http://example.com/superhero", request_headers=DEFAULT_HEADERS, status_code=200, reason="OK", json=self.data, ) # in order to fetch all (other schema) mock.register_uri( "GET", "http://example.com/superhero", request_headers={**DEFAULT_HEADERS, **{"Accept-Profile": "other_schema"}}, status_code=200, reason="OK", json=self.data, ) with default_session(schema="other_schema") as s: res = s.filter("superhero") # should return lazy object self.assertIsInstance(res, postgrestutils.JsonResultSet) self.assertFalse(mock.called) # no request should have been made yet self.assertEqual(list(res), self.data) # should fetch all elements self.assertTrue(mock.called_once) # should have been called once # should have cached all elements self.assertEqual(res._result_cache, self.data) # should have cached the length self.assertEqual(res._len_cache, len(self.data)) mock.reset() res2 = s.filter("superhero", schema=postgrestutils.DEFAULT_SCHEMA) # should return lazy object self.assertIsInstance(res2, postgrestutils.JsonResultSet) self.assertFalse(mock.called) # no request.should have been made yet self.assertEqual(list(res2), self.data) # should fetch all elements self.assertTrue(mock.called_once) # should have been called once # should have cached all elements self.assertEqual(res2._result_cache, self.data) # should have cached the length self.assertEqual(res2._len_cache, len(self.data))