• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2016 gRPC authors.
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#     http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14"""Tests of standard AuthMetadataPlugins."""
15
16import collections
17import threading
18import unittest
19import logging
20
21from grpc import _auth
22
23
24class MockGoogleCreds(object):
25
26    def get_access_token(self):
27        token = collections.namedtuple('MockAccessTokenInfo',
28                                       ('access_token', 'expires_in'))
29        token.access_token = 'token'
30        return token
31
32
33class MockExceptionGoogleCreds(object):
34
35    def get_access_token(self):
36        raise Exception()
37
38
39class GoogleCallCredentialsTest(unittest.TestCase):
40
41    def test_google_call_credentials_success(self):
42        callback_event = threading.Event()
43
44        def mock_callback(metadata, error):
45            self.assertEqual(metadata, (('authorization', 'Bearer token'),))
46            self.assertIsNone(error)
47            callback_event.set()
48
49        call_creds = _auth.GoogleCallCredentials(MockGoogleCreds())
50        call_creds(None, mock_callback)
51        self.assertTrue(callback_event.wait(1.0))
52
53    def test_google_call_credentials_error(self):
54        callback_event = threading.Event()
55
56        def mock_callback(metadata, error):
57            self.assertIsNotNone(error)
58            callback_event.set()
59
60        call_creds = _auth.GoogleCallCredentials(MockExceptionGoogleCreds())
61        call_creds(None, mock_callback)
62        self.assertTrue(callback_event.wait(1.0))
63
64
65class AccessTokenAuthMetadataPluginTest(unittest.TestCase):
66
67    def test_google_call_credentials_success(self):
68        callback_event = threading.Event()
69
70        def mock_callback(metadata, error):
71            self.assertEqual(metadata, (('authorization', 'Bearer token'),))
72            self.assertIsNone(error)
73            callback_event.set()
74
75        metadata_plugin = _auth.AccessTokenAuthMetadataPlugin('token')
76        metadata_plugin(None, mock_callback)
77        self.assertTrue(callback_event.wait(1.0))
78
79
80if __name__ == '__main__':
81    logging.basicConfig()
82    unittest.main(verbosity=2)
83