xref: /aosp_15_r20/external/grpc-grpc/src/python/grpcio_tests/tests/unit/_auth_test.py (revision cc02d7e222339f7a4f6ba5f422e6413f4bd931f2) 
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 logging
18import threading
19import unittest
20
21from grpc import _auth
22
23
24class MockGoogleCreds(object):
25    def get_access_token(self):
26        token = collections.namedtuple(
27            "MockAccessTokenInfo", ("access_token", "expires_in")
28        )
29        token.access_token = "token"
30        return token
31
32
33class MockExceptionGoogleCreds(object):
34    def get_access_token(self):
35        raise Exception()
36
37
38class GoogleCallCredentialsTest(unittest.TestCase):
39    def test_google_call_credentials_success(self):
40        callback_event = threading.Event()
41
42        def mock_callback(metadata, error):
43            self.assertEqual(metadata, (("authorization", "Bearer token"),))
44            self.assertIsNone(error)
45            callback_event.set()
46
47        call_creds = _auth.GoogleCallCredentials(MockGoogleCreds())
48        call_creds(None, mock_callback)
49        self.assertTrue(callback_event.wait(1.0))
50
51    def test_google_call_credentials_error(self):
52        callback_event = threading.Event()
53
54        def mock_callback(metadata, error):
55            self.assertIsNotNone(error)
56            callback_event.set()
57
58        call_creds = _auth.GoogleCallCredentials(MockExceptionGoogleCreds())
59        call_creds(None, mock_callback)
60        self.assertTrue(callback_event.wait(1.0))
61
62
63class AccessTokenAuthMetadataPluginTest(unittest.TestCase):
64    def test_google_call_credentials_success(self):
65        callback_event = threading.Event()
66
67        def mock_callback(metadata, error):
68            self.assertEqual(metadata, (("authorization", "Bearer token"),))
69            self.assertIsNone(error)
70            callback_event.set()
71
72        metadata_plugin = _auth.AccessTokenAuthMetadataPlugin("token")
73        metadata_plugin(None, mock_callback)
74        self.assertTrue(callback_event.wait(1.0))
75
76
77if __name__ == "__main__":
78    logging.basicConfig()
79    unittest.main(verbosity=2)
80