1# Copyright 2018 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"""Test of gRPC Python's interaction with the python logging module"""
15
16import unittest
17import logging
18import grpc
19import subprocess
20import sys
21
22INTERPRETER = sys.executable
23
24
25class LoggingTest(unittest.TestCase):
26
27    def test_logger_not_occupied(self):
28        script = """if True:
29            import logging
30
31            import grpc
32
33            if len(logging.getLogger().handlers) != 0:
34                raise Exception('expected 0 logging handlers')
35
36        """
37        self._verifyScriptSucceeds(script)
38
39    def test_handler_found(self):
40        script = """if True:
41            import logging
42
43            import grpc
44        """
45        out, err = self._verifyScriptSucceeds(script)
46        self.assertEqual(0, len(err), 'unexpected output to stderr')
47
48    def test_can_configure_logger(self):
49        script = """if True:
50            import logging
51            import six
52
53            import grpc
54
55
56            intended_stream = six.StringIO()
57            logging.basicConfig(stream=intended_stream)
58
59            if len(logging.getLogger().handlers) != 1:
60                raise Exception('expected 1 logging handler')
61
62            if logging.getLogger().handlers[0].stream is not intended_stream:
63                raise Exception('wrong handler stream')
64
65        """
66        self._verifyScriptSucceeds(script)
67
68    def test_grpc_logger(self):
69        script = """if True:
70            import logging
71
72            import grpc
73
74            if "grpc" not in logging.Logger.manager.loggerDict:
75                raise Exception('grpc logger not found')
76
77            root_logger = logging.getLogger("grpc")
78            if len(root_logger.handlers) != 1:
79                raise Exception('expected 1 root logger handler')
80            if not isinstance(root_logger.handlers[0], logging.NullHandler):
81                raise Exception('expected logging.NullHandler')
82
83        """
84        self._verifyScriptSucceeds(script)
85
86    def _verifyScriptSucceeds(self, script):
87        process = subprocess.Popen([INTERPRETER, '-c', script],
88                                   stdout=subprocess.PIPE,
89                                   stderr=subprocess.PIPE)
90        out, err = process.communicate()
91        self.assertEqual(
92            0, process.returncode,
93            'process failed with exit code %d (stdout: %s, stderr: %s)' %
94            (process.returncode, out, err))
95        return out, err
96
97
98if __name__ == '__main__':
99    unittest.main(verbosity=2)
100