Skip to content
Snippets Groups Projects
Select Git revision
  • a2215124bb8cc8744f1a81dc9dcb776184161195
  • master default protected
  • pymilter-1.0.4
  • pymilter-1.0.3
  • pymilter-1.0.2
  • pymilter-1.0.1
  • pymilter-1_0
  • milter-0_8_18
  • pymilter-0_9_8
  • pymilter-0_9_7
  • pymilter-0_9_6
  • pymilter-0_9_5
  • pymilter-0_9_4
  • pymilter-0_9_2
  • pymilter-0_9_1
  • pymilter-0_9_0
  • pymilter-0_8_12
  • pymilter-0_8_11
  • pymilter-0_8_10
  • pymilter-0_8_9
  • milter-0_8_8
  • milter-0_8_7
22 results

testmime.py

Blame
  • test_postgrestclient.py 12.92 KiB
    import datetime
    from unittest import TestCase
    
    from requests_mock import Mocker
    
    from postgrestutils.client import (
        Count, MultipleObjectsReturned, ObjectDoesNotExist,
    )
    from postgrestutils.client.postgrestclient import LazyPostgrestJsonResult
    
    TOKEN = 'JWT_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()
            from postgrestutils.client import pgrest_client
            pgrest_client.configure(TOKEN, base_uri='http://example.com/')
    
            self.pgrest_client = pgrest_client
            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
            )
            params = {'id': 'eq.1000000000'}
            res = self.pgrest_client.get('superhero', params=params)
    
            self.assertDictEqual(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"}"""
            )
            params = {'id': 'eq.1337'}
    
            with self.assertRaises(ObjectDoesNotExist):
                self.pgrest_client.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 self.assertRaises(MultipleObjectsReturned):
                self.pgrest_client.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"}
            )
            params = {'id': 'eq.1337'}
            res = self.pgrest_client.get('random_datetime', params=params)
    
            self.assertDictEqual(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
            )
            params = {'select': 'id,random', 'id': 'eq.1337'}
            res = self.pgrest_client.get('random_datetime', params=params, parse_dt=False)
    
            self.assertDictEqual(res, test_json)
            self.assertTrue(mock.called_once)
    
    
    @Mocker()
    class TestPgrestClientFilterStrategyNone(TestCase):
        def setUp(self):
            super().setUp()
            from postgrestutils.client import pgrest_client
            pgrest_client.configure(TOKEN, base_uri='http://example.com/')
    
            self.pgrest_client = pgrest_client
            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
            )
            res = self.pgrest_client.filter('superhero')
    
            self.assertIsInstance(res, LazyPostgrestJsonResult)  # should return lazy object
            self.assertFalse(mock.called)  # no request should have been made yet
    
            self.assertListEqual(list(res), self.data)  # fetch data
            self.assertTrue(mock.called_once)  # should have been called once
            self.assertListEqual(res._result_cache, self.data)  # fetched data should be cached
            self.assertEqual(res._len_cache, len(self.data))  # len of fetched data should be cached
            self.assertListEqual(list(res), self.data)  # should utilize cache
            self.assertListEqual(res[2:], self.data[2:])  # should utilize cache
            self.assertDictEqual(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
            )
            res = self.pgrest_client.filter('superhero')
    
            self.assertIsInstance(res, LazyPostgrestJsonResult)  # should return lazy object
            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
            self.assertEqual(res._len_cache, len(self.data))  # len of fetched data should be cached
            self.assertListEqual(res._result_cache, self.data)  # results should be cached (counting strategy none)
            self.assertListEqual(res[2:], self.data[2:])  # should utilize cache
            self.assertDictEqual(res[0], self.data[0])  # should utilize cache
            self.assertListEqual(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
            )
            res = self.pgrest_client.filter('superhero')
    
            self.assertIsInstance(res, LazyPostgrestJsonResult)  # should return lazy object
            self.assertFalse(mock.called)  # no request should have been made yet
    
            self.assertListEqual(res[:], self.data)  # fetch data
            self.assertTrue(mock.called_once)  # should have been called once
            self.assertListEqual(res._result_cache, self.data)  # fetched data should be cached
            self.assertEqual(res._len_cache, len(self.data))  # len of fetched data should be cached
            self.assertListEqual(res[:], self.data)  # should utilize cache
            self.assertListEqual(res[2:], self.data[2:])  # should utilize cache
            self.assertDictEqual(res[0], self.data[0])  # should utilize cache
            self.assertTrue(mock.called_once)  # should not have been called again
    
    
    @Mocker()
    class TestPgrestClientFilterStrategyExact(TestCase):
        def setUp(self):
            super().setUp()
            from postgrestutils.client import pgrest_client
            pgrest_client.configure(TOKEN, base_uri='http://example.com/')
    
            self.pgrest_client = pgrest_client
            self.data = SUPERHERO_TEST_DATA
    
        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
            )
            res = self.pgrest_client.filter('superhero', count=Count.EXACT)
    
            self.assertIsInstance(res, LazyPostgrestJsonResult)  # should return lazy object
            self.assertFalse(mock.called)  # no request should have been made yet
    
            self.assertListEqual(list(res), self.data)  # fetch data
            self.assertTrue(mock.called_once)  # should have been called once
            self.assertListEqual(res._result_cache, self.data)  # fetched data should be cached
            self.assertEqual(res._len_cache, len(self.data))  # len of fetched data should also be cached
            self.assertListEqual(list(res), self.data)  # should utilize cache
            self.assertListEqual(res[2:], self.data[2:])  # should utilize cache
            self.assertDictEqual(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
            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]
            )
            res = self.pgrest_client.filter('superhero', count=Count.EXACT)
    
            self.assertIsInstance(res, LazyPostgrestJsonResult)  # should return lazy object
            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
            self.assertEqual(res._len_cache, len(self.data))  # len of fetched data should be cached
            self.assertListEqual(res[2:], self.data[2:])  # should fetch range starting at index 2
            self.assertDictEqual(res[0], self.data[0])  # should fetch first element as range
            self.assertListEqual(list(res), self.data)  # should fetch all elements
            self.assertListEqual(res._result_cache, self.data)  # should cache all elements
            self.assertTrue(mock.called)  # should have been called at least once
            self.assertEqual(mock.call_count, 4)  # should have only been called 4 times (fetch len, range, first and all)