Newer
Older
from unittest import TestCase
from requests_mock import Mocker
default_session = functools.partial(
postgrestutils.Session, base_uri="http://example.com/", token=TOKEN
)
"Authorization": "Bearer {}".format(TOKEN),
"Accept": "application/json",
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
"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"},
},
params = {"id": "eq.1000000000"}
res = s.get("superhero", params=params)
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"},
},
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"},
},
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
),
"GET",
"http://example.com/random_datetime",
request_headers={
**DEFAULT_HEADERS,
**{"Accept": "application/vnd.pgrst.object+json"},
},
reason="OK",
json={"id": 1337, "random": "2020-05-20T08:35:06.659425+00:00"},
params = {"id": "eq.1337"}
res = s.get("random_datetime", params=params)
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"}
"GET",
"http://example.com/random_datetime",
request_headers={
**DEFAULT_HEADERS,
**{"Accept": "application/vnd.pgrst.object+json"},
},
params = {"select": "id,random", "id": "eq.1337"}
res = s.get("random_datetime", params=params, parse_dt=False)
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",
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",
# 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,
# 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
class TestPgrestClientFilterCountingStrategies(TestCase):
def setUp(self):
super().setUp()
self.data = SUPERHERO_TEST_DATA
self.counting_strategies = (
postgrestutils.Count.EXACT,
postgrestutils.Count.PLANNED,
)
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,
for strategy in self.counting_strategies:
mock.reset()
with default_session(count=strategy) as s:
# 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,
)
# in order to fetch first
mock.register_uri(
"GET",
"http://example.com/superhero",
request_headers={
**DEFAULT_HEADERS,
**{"Range-Unit": "items", "Range": "0-0"},
},
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-"},
},
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",
**{
"Range-Unit": "items",
"Range": "0-0",
"Prefer": "count={}".format(strategy.value),
},
reason="Partial Content",
headers={"Content-Range": "0-0/5"},
json=self.data[0],
)
mock.reset()
with default_session(count=strategy) as s:
# 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"}
"GET",
"http://example.com/random_datetime",
request_headers={
**DEFAULT_HEADERS,
**{"Accept": "application/vnd.pgrst.object+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)
"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,
)
# 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"},
},
reason="Partial Content",
headers={"Content-Range": "0-0/5"},
json=self.data[0],
)
with default_session(count=postgrestutils.Count.EXACT) as s:
# 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,
)
# in order to fetch all (other schema)
mock.register_uri(
"GET",
"http://example.com/superhero",
request_headers={**DEFAULT_HEADERS, **{"Accept-Profile": "other_schema"}},
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))
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))