• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1from test.support import import_helper, threading_helper
2syslog = import_helper.import_module("syslog") #skip if not supported
3from test import support
4import sys
5import threading
6import time
7import unittest
8from textwrap import dedent
9
10# XXX(nnorwitz): This test sucks.  I don't know of a platform independent way
11# to verify that the messages were really logged.
12# The only purpose of this test is to verify the code doesn't crash or leak.
13
14class Test(unittest.TestCase):
15
16    def tearDown(self):
17        syslog.closelog()
18
19    def test_openlog(self):
20        syslog.openlog('python')
21        # Issue #6697.
22        self.assertRaises(UnicodeEncodeError, syslog.openlog, '\uD800')
23
24    def test_syslog(self):
25        syslog.openlog('python')
26        syslog.syslog('test message from python test_syslog')
27        syslog.syslog(syslog.LOG_ERR, 'test error from python test_syslog')
28
29    def test_syslog_implicit_open(self):
30        syslog.closelog() # Make sure log is closed
31        syslog.syslog('test message from python test_syslog')
32        syslog.syslog(syslog.LOG_ERR, 'test error from python test_syslog')
33
34    def test_closelog(self):
35        syslog.openlog('python')
36        syslog.closelog()
37        syslog.closelog()  # idempotent operation
38
39    def test_setlogmask(self):
40        mask = syslog.LOG_UPTO(syslog.LOG_WARNING)
41        oldmask = syslog.setlogmask(mask)
42        self.assertEqual(syslog.setlogmask(0), mask)
43        self.assertEqual(syslog.setlogmask(oldmask), mask)
44
45    def test_log_mask(self):
46        mask = syslog.LOG_UPTO(syslog.LOG_WARNING)
47        self.assertTrue(mask & syslog.LOG_MASK(syslog.LOG_WARNING))
48        self.assertTrue(mask & syslog.LOG_MASK(syslog.LOG_ERR))
49        self.assertFalse(mask & syslog.LOG_MASK(syslog.LOG_INFO))
50
51    def test_openlog_noargs(self):
52        syslog.openlog()
53        syslog.syslog('test message from python test_syslog')
54
55    @threading_helper.requires_working_threading()
56    def test_syslog_threaded(self):
57        start = threading.Event()
58        stop = False
59        def opener():
60            start.wait(10)
61            i = 1
62            while not stop:
63                syslog.openlog(f'python-test-{i}')  # new string object
64                i += 1
65        def logger():
66            start.wait(10)
67            while not stop:
68                syslog.syslog('test message from python test_syslog')
69
70        orig_si = sys.getswitchinterval()
71        support.setswitchinterval(1e-9)
72        try:
73            threads = [threading.Thread(target=opener)]
74            threads += [threading.Thread(target=logger) for k in range(10)]
75            with threading_helper.start_threads(threads):
76                start.set()
77                time.sleep(0.1)
78                stop = True
79        finally:
80            sys.setswitchinterval(orig_si)
81
82    def test_subinterpreter_syslog(self):
83        # syslog.syslog() is not allowed in subinterpreters, but only if
84        # syslog.openlog() hasn't been called in the main interpreter yet.
85        with self.subTest('before openlog()'):
86            code = dedent('''
87                import syslog
88                caught_error = False
89                try:
90                    syslog.syslog('foo')
91                except RuntimeError:
92                    caught_error = True
93                assert(caught_error)
94            ''')
95            res = support.run_in_subinterp(code)
96            self.assertEqual(res, 0)
97
98        syslog.openlog()
99        try:
100            with self.subTest('after openlog()'):
101                code = dedent('''
102                    import syslog
103                    syslog.syslog('foo')
104                ''')
105                res = support.run_in_subinterp(code)
106                self.assertEqual(res, 0)
107        finally:
108            syslog.closelog()
109
110    def test_subinterpreter_openlog(self):
111        try:
112            code = dedent('''
113                import syslog
114                caught_error = False
115                try:
116                    syslog.openlog()
117                except RuntimeError:
118                    caught_error = True
119
120                assert(caught_error)
121            ''')
122            res = support.run_in_subinterp(code)
123            self.assertEqual(res, 0)
124        finally:
125            syslog.closelog()
126
127    def test_subinterpreter_closelog(self):
128        syslog.openlog('python')
129        try:
130            code = dedent('''
131                import syslog
132                caught_error = False
133                try:
134                    syslog.closelog()
135                except RuntimeError:
136                    caught_error = True
137
138                assert(caught_error)
139            ''')
140            res = support.run_in_subinterp(code)
141            self.assertEqual(res, 0)
142        finally:
143            syslog.closelog()
144
145
146if __name__ == "__main__":
147    unittest.main()
148