Skip to content
Snippets Groups Projects
test_postgrestclient.py 20.7 KiB
Newer Older
  • Learn to ignore specific revisions
  • Fynn Becker's avatar
    Fynn Becker committed
    import datetime
    
    import functools
    
    Fynn Becker's avatar
    Fynn Becker committed
    from unittest import TestCase
    
    from requests_mock import Mocker
    
    
    import postgrestutils
    
    Fynn Becker's avatar
    Fynn Becker committed
    
    
    TOKEN = "JWT_token"
    
    default_session = functools.partial(
        postgrestutils.Session, base_uri="http://example.com/", token=TOKEN
    )
    
    Fynn Becker's avatar
    Fynn Becker committed
    DEFAULT_HEADERS = {
    
        "Authorization": "Bearer {}".format(TOKEN),
        "Accept": "application/json",
    
    Fynn Becker's avatar
    Fynn Becker committed
    }
    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,
        },
    
    Fynn Becker's avatar
    Fynn Becker committed
    ]
    
    
    @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"},
                },
    
    Fynn Becker's avatar
    Fynn Becker committed
                status_code=200,
    
                reason="OK",
                json=self.data,
    
    Fynn Becker's avatar
    Fynn Becker committed
            )
    
            with default_session() as s:
    
                params = {"id": "eq.1000000000"}
                res = s.get("superhero", params=params)
    
    Fynn Becker's avatar
    Fynn Becker committed
    
    
            self.assertEqual(res, self.data)
    
    Fynn Becker's avatar
    Fynn Becker committed
            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"},
                },
    
    Fynn Becker's avatar
    Fynn Becker committed
                status_code=406,
    
                reason="Not Acceptable",
    
    Fynn Becker's avatar
    Fynn Becker committed
                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)
    
    Fynn Becker's avatar
    Fynn Becker committed
            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"},
                },
    
    Fynn Becker's avatar
    Fynn Becker committed
                status_code=406,
    
                reason="Not Acceptable",
    
    Fynn Becker's avatar
    Fynn Becker committed
                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")
    
    Fynn Becker's avatar
    Fynn Becker committed
            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
                ),
    
    Fynn Becker's avatar
    Fynn Becker committed
            }
            mock.register_uri(
    
                "GET",
                "http://example.com/random_datetime",
                request_headers={
                    **DEFAULT_HEADERS,
                    **{"Accept": "application/vnd.pgrst.object+json"},
                },
    
    Fynn Becker's avatar
    Fynn Becker committed
                status_code=200,
    
                reason="OK",
                json={"id": 1337, "random": "2020-05-20T08:35:06.659425+00:00"},
    
    Fynn Becker's avatar
    Fynn Becker committed
            )
    
            with default_session() as s:
    
                params = {"id": "eq.1337"}
                res = s.get("random_datetime", params=params)
    
    Fynn Becker's avatar
    Fynn Becker committed
    
    
            self.assertEqual(res, expected)
    
    Fynn Becker's avatar
    Fynn Becker committed
            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"}
    
    Fynn Becker's avatar
    Fynn Becker committed
            mock.register_uri(
    
                "GET",
                "http://example.com/random_datetime",
                request_headers={
                    **DEFAULT_HEADERS,
                    **{"Accept": "application/vnd.pgrst.object+json"},
                },
    
    Fynn Becker's avatar
    Fynn Becker committed
                status_code=200,
    
                reason="OK",
                json=test_json,
    
    Fynn Becker's avatar
    Fynn Becker committed
            )
    
            with default_session() as s:
    
                params = {"select": "id,random", "id": "eq.1337"}
                res = s.get("random_datetime", params=params, parse_dt=False)
    
    Fynn Becker's avatar
    Fynn Becker committed
    
    
            self.assertEqual(res, test_json)
    
    Fynn Becker's avatar
    Fynn Becker committed
            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",
    
    Fynn Becker's avatar
    Fynn Becker committed
                request_headers=DEFAULT_HEADERS,
                status_code=200,
    
                reason="OK",
                json=self.data,
    
    Fynn Becker's avatar
    Fynn Becker committed
            )
    
            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
    
    Fynn Becker's avatar
    Fynn Becker committed
    
        def test_fetch_len_first(self, mock):
            mock.register_uri(
    
                "GET",
                "http://example.com/superhero",
    
    Fynn Becker's avatar
    Fynn Becker committed
                request_headers=DEFAULT_HEADERS,
                status_code=200,
    
                reason="OK",
                json=self.data,
    
    Fynn Becker's avatar
    Fynn Becker committed
            )
    
            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
    
    Fynn Becker's avatar
    Fynn Becker committed
    
    
        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
    
    Fynn Becker's avatar
    Fynn Becker committed
    
    @Mocker()
    
    Fynn Becker's avatar
    Fynn Becker committed
    class TestPgrestClientFilterCountingStrategies(TestCase):
    
    Fynn Becker's avatar
    Fynn Becker committed
        def setUp(self):
            super().setUp()
            self.data = SUPERHERO_TEST_DATA
    
    Fynn Becker's avatar
    Fynn Becker committed
            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,
    
    Fynn Becker's avatar
    Fynn Becker committed
            for strategy in self.counting_strategies:
                mock.reset()
                with default_session(count=strategy) as s:
    
                    res = s.filter("superhero")
    
    Fynn Becker's avatar
    Fynn Becker committed
    
    
                    # should return lazy object
                    self.assertIsInstance(res, postgrestutils.JsonResultSet)
    
    Fynn Becker's avatar
    Fynn Becker committed
                    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))
    
    Fynn Becker's avatar
    Fynn Becker committed
    
                    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:],
    
    Fynn Becker's avatar
    Fynn Becker committed
            # in order to fetch length using different strategies
            for strategy in self.counting_strategies:
                mock.register_uri(
    
                    "GET",
                    "http://example.com/superhero",
    
    Fynn Becker's avatar
    Fynn Becker committed
                    request_headers={
                        **DEFAULT_HEADERS,
    
                        **{
                            "Range-Unit": "items",
                            "Range": "0-0",
                            "Prefer": "count={}".format(strategy.value),
                        },
    
    Fynn Becker's avatar
    Fynn Becker committed
                    },
                    status_code=206,
    
                    reason="Partial Content",
                    headers={"Content-Range": "0-0/5"},
                    json=self.data[0],
    
    Fynn Becker's avatar
    Fynn Becker committed
                )
                mock.reset()
    
                with default_session(count=strategy) as s:
    
                    res = s.filter("superhero")
    
    Fynn Becker's avatar
    Fynn Becker committed
    
    
                    # should return lazy object
                    self.assertIsInstance(res, postgrestutils.JsonResultSet)
    
    Fynn Becker's avatar
    Fynn Becker committed
                    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))
    
    Fynn Becker's avatar
    Fynn Becker committed
    
    
                    # should fetch first element as range
                    self.assertEqual(res[:1], self.data[:1])
    
    Fynn Becker's avatar
    Fynn Becker committed
                    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])
    
    Fynn Becker's avatar
    Fynn Becker committed
                    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)
    
    Fynn Becker's avatar
    Fynn Becker committed
                    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))
    
                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))