1 from test.support import import_helper, threading_helper
2 syslog = import_helper.import_module("syslog") #skip if not supported
3 from test import support
4 import sys
5 import threading
6 import time
7 import unittest
8 from 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
14 class ESC[4;38;5;81mTest(ESC[4;38;5;149munittestESC[4;38;5;149m.ESC[4;38;5;149mTestCase):
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
146 if __name__ == "__main__":
147 unittest.main()