summaryrefslogtreecommitdiff
path: root/tests/oauth2/rfc6749/endpoints/test_base_endpoint.py
blob: 4f78d9b3bf54725ad1ba5f248da3aaf4dc825175 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
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
# -*- coding: utf-8 -*-
from __future__ import absolute_import, unicode_literals

from oauthlib.oauth2 import (FatalClientError, OAuth2Error, RequestValidator,
                             Server)
from oauthlib.oauth2.rfc6749 import (BaseEndpoint,
                                     catch_errors_and_unavailability)

from ....unittest import TestCase


class BaseEndpointTest(TestCase):

    def test_default_config(self):
        endpoint = BaseEndpoint()
        self.assertFalse(endpoint.catch_errors)
        self.assertTrue(endpoint.available)
        endpoint.catch_errors = True
        self.assertTrue(endpoint.catch_errors)
        endpoint.available = False
        self.assertFalse(endpoint.available)

    def test_error_catching(self):
        validator = RequestValidator()
        server = Server(validator)
        server.catch_errors = True
        h, b, s = server.create_token_response(
            'https://example.com?grant_type=authorization_code&code=abc'
        )
        self.assertIn("server_error", b)
        self.assertEqual(s, 500)

    def test_unavailability(self):
        validator = RequestValidator()
        server = Server(validator)
        server.available = False
        h, b, s = server.create_authorization_response('https://example.com')
        self.assertIn("temporarily_unavailable", b)
        self.assertEqual(s, 503)

    def test_wrapper(self):

        class TestServer(Server):

            @catch_errors_and_unavailability
            def throw_error(self, uri):
                raise ValueError()

            @catch_errors_and_unavailability
            def throw_oauth_error(self, uri):
                raise OAuth2Error()

            @catch_errors_and_unavailability
            def throw_fatal_oauth_error(self, uri):
                raise FatalClientError()

        validator = RequestValidator()
        server = TestServer(validator)

        server.catch_errors = True
        h, b, s = server.throw_error('a')
        self.assertIn("server_error", b)
        self.assertEqual(s, 500)

        server.available = False
        h, b, s = server.throw_error('a')
        self.assertIn("temporarily_unavailable", b)
        self.assertEqual(s, 503)

        server.available = True
        self.assertRaises(OAuth2Error, server.throw_oauth_error, 'a')
        self.assertRaises(FatalClientError, server.throw_fatal_oauth_error, 'a')
        server.catch_errors = False
        self.assertRaises(OAuth2Error, server.throw_oauth_error, 'a')
        self.assertRaises(FatalClientError, server.throw_fatal_oauth_error, 'a')